Repository: apollographql/apollo-kotlin
Pull request #4385 (2022-09-14 10:33:17)

Ignore empty defer payloads
When receiving a deferred payload with no `incremental` field, ignore it to avoid re-emitting the previously merged data.

Related to https://github.com/apollographql/router/issues/1687

netlify[bot]: ### <span aria-hidden="true">👷</span> Deploy Preview for *apollo-android-docs* processing.


|  Name | Link |
|---------------------------------|------------------------|
|<span aria-hidden="true">🔨</span> Latest commit | 1eb6f91454d6fd5ebceb81431130ffda944104c2 |
|<span aria-hidden="true">🔍</span> Latest deploy log | https://app.netlify.com/sites/apollo-android-docs/deploys/6321adef332c0e00092ac53a |

Patch:
diff --git a/apollo-runtime/src/commonMain/kotlin/com/apollographql/apollo3/internal/DeferredJsonMerger.kt b/apollo-runtime/src/commonMain/kotlin/com/apollographql/apollo3/internal/DeferredJsonMerger.kt
index 83acb0f79e..51aa867a80 100644
--- a/apollo-runtime/src/commonMain/kotlin/com/apollographql/apollo3/internal/DeferredJsonMerger.kt
+++ b/apollo-runtime/src/commonMain/kotlin/com/apollographql/apollo3/internal/DeferredJsonMerger.kt
@@ -34,6 +34,14 @@ class DeferredJsonMerger {
   var hasNext: Boolean = true
     private set
 
+  /**
+   * A payload can sometimes have no `incremental` field, e.g. when the server couldn't predict if there were more data after the last
+   * emitted payload. This field allows to test for this in order to ignore such payloads.
+   * See https://github.com/apollographql/router/issues/1687.
+   */
+  var isEmptyPayload: Boolean = false
+    private set
+
   fun merge(payload: BufferedSource): JsonMap {
     val payloadMap = jsonToMap(payload)
     return merge(payloadMap)
@@ -48,7 +56,10 @@ class DeferredJsonMerger {
     }
 
     val incrementalList = payload["incremental"] as? List<JsonMap>
-    if (incrementalList != null) {
+    if (incrementalList == null) {
+      isEmptyPayload = true
+    } else {
+      isEmptyPayload = false
       val mergedErrors = mutableListOf<JsonMap>()
       val mergedExtensions = mutableListOf<JsonMap>()
       for (incrementalItem in incrementalList) {
@@ -131,6 +142,7 @@ class DeferredJsonMerger {
     _merged.clear()
     _mergedFragmentIds.clear()
     hasNext = true
+    isEmptyPayload = false
   }
 }
 
diff --git a/apollo-runtime/src/commonMain/kotlin/com/apollographql/apollo3/network/http/HttpNetworkTransport.kt b/apollo-runtime/src/commonMain/kotlin/com/apollographql/apollo3/network/http/HttpNetworkTransport.kt
index 37b60b68f9..bc7107e1aa 100644
--- a/apollo-runtime/src/commonMain/kotlin/com/apollographql/apollo3/network/http/HttpNetworkTransport.kt
+++ b/apollo-runtime/src/commonMain/kotlin/com/apollographql/apollo3/network/http/HttpNetworkTransport.kt
@@ -23,6 +23,7 @@ import com.apollographql.apollo3.network.NetworkTransport
 import com.benasher44.uuid.Uuid
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.emitAll
+import kotlinx.coroutines.flow.filterNot
 import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.flow.map
 
@@ -110,19 +111,21 @@ private constructor(
       httpResponse: HttpResponse,
   ): Flow<ApolloResponse<D>> {
     val jsonMerger = DeferredJsonMerger()
-    return multipartBodyFlow(httpResponse).map { part ->
-      try {
-        val merged = jsonMerger.merge(part)
-        val deferredFragmentIds = jsonMerger.mergedFragmentIds
-        val isLast = !jsonMerger.hasNext
-        operation.parseJsonResponse(
-            jsonReader = merged.jsonReader(),
-            customScalarAdapters = customScalarAdapters.withDeferredFragmentIds(deferredFragmentIds)
-        ).newBuilder().isLast(isLast).build()
-      } catch (e: Exception) {
-        throw wrapThrowableIfNeeded(e)
-      }
-    }
+    return multipartBodyFlow(httpResponse)
+        .map { part ->
+          try {
+            val merged = jsonMerger.merge(part)
+            val deferredFragmentIds = jsonMerger.mergedFragmentIds
+            val isLast = !jsonMerger.hasNext
+            operation.parseJsonResponse(
+                jsonReader = merged.jsonReader(),
+                customScalarAdapters = customScalarAdapters.withDeferredFragmentIds(deferredFragmentIds)
+            ).newBuilder().isLast(isLast).build()
+          } catch (e: Exception) {
+            throw wrapThrowableIfNeeded(e)
+          }
+        }
+        .filterNot { jsonMerger.isEmptyPayload }
   }
 
   private fun <D : Operation.Data> ApolloResponse<D>.withHttpInfo(
diff --git a/apollo-runtime/src/commonMain/kotlin/com/apollographql/apollo3/network/ws/WebSocketNetworkTransport.kt b/apollo-runtime/src/commonMain/kotlin/com/apollographql/apollo3/network/ws/WebSocketNetworkTransport.kt
index 49b66a4d5d..683d8b2105 100644
--- a/apollo-runtime/src/commonMain/kotlin/com/apollographql/apollo3/network/ws/WebSocketNetworkTransport.kt
+++ b/apollo-runtime/src/commonMain/kotlin/com/apollographql/apollo3/network/ws/WebSocketNetworkTransport.kt
@@ -40,6 +40,7 @@ import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.asSharedFlow
 import kotlinx.coroutines.flow.filter
+import kotlinx.coroutines.flow.filterNot
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.onCompletion
 import kotlinx.coroutines.flow.onSubscription
@@ -303,12 +304,15 @@ private constructor(
           }
           apolloResponse
         }
+
         is OperationError -> throw SubscriptionOperationException(request.operation.name(), response.payload)
         is NetworkError -> throw ApolloNetworkException("Network error while executing ${request.operation.name()}", response.cause)
 
         // Cannot happen as these events are filtered out upstream
         is OperationComplete, is GeneralError -> error("Unexpected event $response")
       }
+    }.filterNot {
+      deferredJsonMerger.isEmptyPayload
     }.onCompletion {
       messages.send(StopOperation(request))
     }


Test patch:
diff --git a/apollo-runtime/src/commonTest/kotlin/com/apollographql/apollo3/internal/DeferJsonMergerTest.kt b/apollo-runtime/src/commonTest/kotlin/com/apollographql/apollo3/internal/DeferredJsonMergerTest.kt
similarity index 86%
rename from apollo-runtime/src/commonTest/kotlin/com/apollographql/apollo3/internal/DeferJsonMergerTest.kt
rename to apollo-runtime/src/commonTest/kotlin/com/apollographql/apollo3/internal/DeferredJsonMergerTest.kt
index 5b06a81f20..f858e6daaf 100644
--- a/apollo-runtime/src/commonTest/kotlin/com/apollographql/apollo3/internal/DeferJsonMergerTest.kt
+++ b/apollo-runtime/src/commonTest/kotlin/com/apollographql/apollo3/internal/DeferredJsonMergerTest.kt
@@ -9,13 +9,15 @@ import com.apollographql.apollo3.api.json.readAny
 import okio.Buffer
 import kotlin.test.Test
 import kotlin.test.assertEquals
+import kotlin.test.assertFalse
+import kotlin.test.assertTrue
 
-class DeferJsonMergerTest {
-  @Test
-  fun mergeJsonSingleIncrementalItem() {
-      val deferredJsonMerger = DeferredJsonMerger()
+class DeferredJsonMergerTest {
+    @Test
+    fun mergeJsonSingleIncrementalItem() {
+        val deferredJsonMerger = DeferredJsonMerger()
 
-      val payload1 = """
+        val payload1 = """
       {
         "data": {
           "computers": [
@@ -352,14 +354,14 @@ class DeferJsonMergerTest {
         ]
       }
     """
-      deferredJsonMerger.merge(payload5.buffer())
-      assertEquals(jsonToMap(mergedPayloads_1_2_3_4_5), deferredJsonMerger.merged)
-      assertEquals(setOf(
-          DeferredFragmentIdentifier(path = listOf("computers", 0), label = "query:Query1:0"),
-          DeferredFragmentIdentifier(path = listOf("computers", 1), label = "query:Query1:0"),
-          DeferredFragmentIdentifier(path = listOf("computers", 1, "screen"), label = "fragment:ComputerFields:0"),
-      ), deferredJsonMerger.mergedFragmentIds)
-  }
+        deferredJsonMerger.merge(payload5.buffer())
+        assertEquals(jsonToMap(mergedPayloads_1_2_3_4_5), deferredJsonMerger.merged)
+        assertEquals(setOf(
+            DeferredFragmentIdentifier(path = listOf("computers", 0), label = "query:Query1:0"),
+            DeferredFragmentIdentifier(path = listOf("computers", 1), label = "query:Query1:0"),
+            DeferredFragmentIdentifier(path = listOf("computers", 1, "screen"), label = "fragment:ComputerFields:0"),
+        ), deferredJsonMerger.mergedFragmentIds)
+    }
 
     @Test
     fun mergeJsonMultipleIncrementalItems() {
@@ -624,6 +626,81 @@ class DeferJsonMergerTest {
             DeferredFragmentIdentifier(path = listOf("computers", 1, "screen"), label = "fragment:ComputerFields:0"),
         ), deferredJsonMerger.mergedFragmentIds)
     }
+
+    @Test
+    fun emptyPayloads() {
+        val deferredJsonMerger = DeferredJsonMerger()
+
+        val payload1 = """
+      {
+        "data": {
+          "computers": [
+            {
+              "id": "Computer1",
+              "screen": {
+                "isTouch": true
+              }
+            },
+            {
+              "id": "Computer2",
+              "screen": {
+                "isTouch": false
+              }
+            }
+          ]
+        },
+        "hasNext": true
+      }
+    """
+        deferredJsonMerger.merge(payload1.buffer())
+        assertFalse(deferredJsonMerger.isEmptyPayload)
+
+        val payload2 = """
+      {
+        "hasNext": true
+      }
+    """
+        deferredJsonMerger.merge(payload2.buffer())
+        assertTrue(deferredJsonMerger.isEmptyPayload)
+
+        val payload3 = """
+      {
+        "incremental": [
+          {
+            "data": {
+              "cpu": "386",
+              "year": 1993,
+              "screen": {
+                "resolution": "640x480"
+              }
+            },
+            "path": [
+              "computers",
+              0
+            ],
+            "label": "query:Query1:0",
+            "extensions": {
+              "duration": {
+                "amount": 100,
+                "unit": "ms"
+              }
+            }
+          }
+        ],
+        "hasNext": true
+      }
+    """
+        deferredJsonMerger.merge(payload3.buffer())
+        assertFalse(deferredJsonMerger.isEmptyPayload)
+
+        val payload4 = """
+      {
+        "hasNext": false
+      }
+    """
+        deferredJsonMerger.merge(payload4.buffer())
+        assertTrue(deferredJsonMerger.isEmptyPayload)
+    }
 }
 
 private fun String.buffer() = Buffer().writeUtf8(this)
diff --git a/tests/defer/README.md b/tests/defer/README.md
index f592d552d8..ba3ebf7c4d 100644
--- a/tests/defer/README.md
+++ b/tests/defer/README.md
@@ -1,5 +1,4 @@
 # Tests related to `@defer`
 
-At the moment the tests use `StreamingNSURLSessionHttpEngine` on Apple targets, which requires
-the [new Memory Manager](https://github.com/JetBrains/kotlin/blob/master/kotlin-native/NEW_MM.md). It is enabled on this
-project in `gradle.propertis`.
+Note: on Apple targets, the client is configured to use `StreamingNSURLSessionHttpEngine` which supports chunked
+encoding.  
diff --git a/tests/defer/src/commonMain/graphql/base/operation.graphql b/tests/defer/src/commonMain/graphql/base/operation.graphql
index a8f3dfb722..c7a2489148 100644
--- a/tests/defer/src/commonMain/graphql/base/operation.graphql
+++ b/tests/defer/src/commonMain/graphql/base/operation.graphql
@@ -62,3 +62,12 @@ subscription WithFragmentSpreadsSubscription {
 fragment CounterFields on Counter{
   valueTimesTwo
 }
+
+query SimpleDeferQuery {
+  computers {
+    id
+    ... on Computer @defer {
+      cpu
+    }
+  }
+}
diff --git a/tests/defer/src/commonTest/kotlin/test/DeferTest.kt b/tests/defer/src/commonTest/kotlin/test/DeferTest.kt
index 069e01ff62..ba9b353ec6 100644
--- a/tests/defer/src/commonTest/kotlin/test/DeferTest.kt
+++ b/tests/defer/src/commonTest/kotlin/test/DeferTest.kt
@@ -11,6 +11,7 @@ import com.apollographql.apollo3.mpp.currentTimeMillis
 import com.apollographql.apollo3.mpp.platform
 import com.apollographql.apollo3.testing.internal.runTest
 import com.benasher44.uuid.uuid4
+import defer.SimpleDeferQuery
 import defer.WithFragmentSpreadsQuery
 import defer.WithInlineFragmentsQuery
 import defer.fragment.ComputerFields
@@ -285,4 +286,34 @@ class DeferTest {
     }
     job.cancel()
   }
+
+  @Test
+  fun emptyPayloadsAreIgnored() = runTest(before = { setUp() }, after = { tearDown() }) {
+    val jsonWithEmptyPayload = listOf(
+        """{"data":{"computers":[{"__typename":"Computer","id":"computer1"}]},"hasNext":true}""",
+        """{"incremental": [{"data":{"cpu":"386"},"path":["computers",0]}],"hasNext":true}""",
+        """{"hasNext":false}""",
+    )
+    val jsonWithoutEmptyPayload = listOf(
+        """{"data":{"computers":[{"__typename":"Computer","id":"computer1"}]},"hasNext":true}""",
+        """{"incremental": [{"data":{"cpu":"386"},"path":["computers",0]}],"hasNext":false}""",
+    )
+
+    val expectedDataList = listOf(
+        SimpleDeferQuery.Data(
+            listOf(SimpleDeferQuery.Computer("Computer", "computer1", null))
+        ),
+        SimpleDeferQuery.Data(
+            listOf(SimpleDeferQuery.Computer("Computer", "computer1", SimpleDeferQuery.OnComputer("386")))
+        ),
+    )
+
+    mockServer.enqueueMultipart(jsonWithEmptyPayload)
+    var actualDataList = apolloClient.query(SimpleDeferQuery()).toFlow().toList().map { it.dataAssertNoErrors }
+    assertEquals(expectedDataList, actualDataList)
+
+    mockServer.enqueueMultipart(jsonWithoutEmptyPayload)
+    actualDataList = apolloClient.query(SimpleDeferQuery()).toFlow().toList().map { it.dataAssertNoErrors }
+    assertEquals(expectedDataList, actualDataList)
+  }
 }