Take into account @defer cases (multiple emissions in Flows) in norma…

…lized cache interceptors

apollo-api/src/commonMain/kotlin/com/apollographql/apollo3/api/AdapterContext.kt

@@ -19,7 +19,11 @@ class AdapterContext private constructor(
   }
 
   fun hasDeferredFragment(path: List<Any>, label: String?): Boolean {
-    return mergedDeferredFragmentIds?.contains(DeferredFragmentIdentifier(path, label)) == true
+    if (mergedDeferredFragmentIds == null) {
+      // By default, parse all deferred fragments - this is the case when parsing from the normalized cache.
+      return true
+    }
+    return mergedDeferredFragmentIds.contains(DeferredFragmentIdentifier(path, label))
   }
 
   class Builder {

apollo-normalized-cache/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/FetchPolicyInterceptors.kt

@@ -13,7 +13,9 @@ import com.apollographql.apollo3.interceptor.ApolloInterceptor
 import com.apollographql.apollo3.interceptor.ApolloInterceptorChain
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.catch
+import kotlinx.coroutines.flow.emitAll
 import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.singleOrNull
 import kotlin.jvm.JvmName
 
@@ -72,26 +74,25 @@ val CacheFirstInterceptor = object : ApolloInterceptor {
         } else {
           throw it
         }
-      }.singleOrNull()
+      }.map { response ->
+        response.newBuilder()
+            .cacheInfo(
+                response.cacheInfo!!
+                    .newBuilder()
+                    .cacheMissException(cacheException as? CacheMissException)
+                    .build()
+            )
+            .build()
+      }
 
-      if (networkResponse != null) {
-        emit(
-            networkResponse.newBuilder()
-                .cacheInfo(
-                    networkResponse.cacheInfo!!
-                        .newBuilder()
-                        .cacheMissException(cacheException as? CacheMissException)
-                        .build()
-                )
-                .build()
+      emitAll(networkResponse)
+
+      if (networkException != null) {
+        throw ApolloCompositeException(
+            first = cacheException,
+            second = networkException
         )
-        return@flow
       }
-
-      throw ApolloCompositeException(
-          first = cacheException,
-          second = networkException
-      )
     }
   }
 }
@@ -110,10 +111,10 @@ val NetworkFirstInterceptor = object : ApolloInterceptor {
         } else {
           throw it
         }
-      }.singleOrNull()
+      }
 
-      if (networkResponse != null) {
-        emit(networkResponse)
+      emitAll(networkResponse)
+      if (networkException == null) {
         return@flow
       }
 
@@ -186,28 +187,29 @@ val CacheAndNetworkInterceptor = object : ApolloInterceptor {
             } else {
               throw it
             }
-          }.singleOrNull()
-
-      if (networkResponse != null) {
-        emit(
-            networkResponse.newBuilder()
+          }
+          .map { response ->
+            response.newBuilder()
                 .cacheInfo(
-                    networkResponse.cacheInfo!!
+                    response.cacheInfo!!
                         .newBuilder()
                         .cacheMissException(cacheException as? CacheMissException)
                         .build()
                 )
                 .build()
-        )
-        return@flow
-      }
-      if (cacheException != null) {
-        throw ApolloCompositeException(
-            first = cacheException,
-            second = networkException
-        )
+          }
+
+      emitAll(networkResponse)
+
+      if (networkException != null) {
+        if (cacheException != null) {
+          throw ApolloCompositeException(
+              first = cacheException,
+              second = networkException
+          )
+        }
+        throw networkException!!
       }
-      throw networkException!!
     }
   }
 }

apollo-normalized-cache/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/internal/ApolloCacheInterceptor.kt

@@ -24,10 +24,13 @@ import com.apollographql.apollo3.interceptor.ApolloInterceptor
 import com.apollographql.apollo3.interceptor.ApolloInterceptorChain
 import com.apollographql.apollo3.mpp.currentTimeMillis
 import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.catch
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.flow.emitAll
 import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.flow.flowOn
+import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.onEach
-import kotlinx.coroutines.flow.single
 import kotlinx.coroutines.launch
 
 internal class ApolloCacheInterceptor(
@@ -134,26 +137,38 @@ internal class ApolloCacheInterceptor(
       /**
        * This doesn't use [readFromNetwork] so that we can publish all keys all at once after the keys have been rolled back
        */
-      var response: ApolloResponse<D>? = null
-      var exception: ApolloException? = null
-      try {
-        response = chain.proceed(request).single()
-      } catch (e: ApolloException) {
-        exception = e
-      }
+      var networkException: ApolloException? = null
+      val networkResponse: Flow<ApolloResponse<D>> = chain.proceed(request)
+          .catch {
+            if (it is ApolloException) {
+              networkException = it
+            } else {
+              throw it
+            }
+          }
 
-      val optimisticKeys = if (optimisticData != null) {
-        store.rollbackOptimisticUpdates(request.requestUuid, publish = false)
-      } else {
-        emptySet()
-      }
+      var optimisticKeys: Set<String>? = null
+
+      networkResponse.collect { response ->
+        if (optimisticKeys == null) optimisticKeys = if (optimisticData != null) {
+          store.rollbackOptimisticUpdates(request.requestUuid, publish = false)
+        } else {
+          emptySet()
+        }
 
-      if (response != null) {
-        maybeWriteToCache(request, response, customScalarAdapters, optimisticKeys)
+        maybeWriteToCache(request, response, customScalarAdapters, optimisticKeys!!)
         emit(response)
-      } else {
-        store.publish(optimisticKeys)
-        throw exception!!
+      }
+
+      if (networkException != null) {
+        if (optimisticKeys == null) optimisticKeys = if (optimisticData != null) {
+          store.rollbackOptimisticUpdates(request.requestUuid, publish = false)
+        } else {
+          emptySet()
+        }
+
+        store.publish(optimisticKeys!!)
+        throw networkException!!
       }
     }
   }
@@ -163,13 +178,11 @@ internal class ApolloCacheInterceptor(
     val fetchFromCache = request.fetchFromCache
 
     return flow {
-      emit(
-          if (fetchFromCache) {
-            readFromCache(request, customScalarAdapters)
-          } else {
-            readFromNetwork(request, chain, customScalarAdapters)
-          }
-      )
+      if (fetchFromCache) {
+        emit(readFromCache(request, customScalarAdapters))
+      } else {
+        emitAll(readFromNetwork(request, chain, customScalarAdapters))
+      }
     }
   }
 
@@ -229,18 +242,18 @@ internal class ApolloCacheInterceptor(
       request: ApolloRequest<D>,
       chain: ApolloInterceptorChain,
       customScalarAdapters: CustomScalarAdapters,
-  ): ApolloResponse<D> {
+  ): Flow<ApolloResponse<D>> {
     val startMillis = currentTimeMillis()
-    val networkResponse = chain.proceed(request).onEach {
+    return chain.proceed(request).onEach {
       maybeWriteToCache(request, it, customScalarAdapters)
-    }.single()
-
-    return networkResponse.newBuilder()
-        .cacheInfo(
-            CacheInfo.Builder()
-                .networkStartMillis(startMillis)
-                .networkEndMillis(currentTimeMillis())
-                .build()
-        ).build()
+    }.map { networkResponse ->
+      networkResponse.newBuilder()
+          .cacheInfo(
+              CacheInfo.Builder()
+                  .networkStartMillis(startMillis)
+                  .networkEndMillis(currentTimeMillis())
+                  .build()
+          ).build()
+    }
   }
 }

tests/defer/build.gradle.kts

@@ -17,6 +17,7 @@ kotlin {
     val commonMain by getting {
       dependencies {
         implementation("com.apollographql.apollo3:apollo-runtime")
+        implementation("com.apollographql.apollo3:apollo-normalized-cache")
       }
     }
 

tests/defer/src/commonMain/graphql/operation.graphql

@@ -1,9 +1,10 @@
-query WithFragmentSpreads {
+query WithFragmentSpreadsQuery {
   computers {
     id
     ...ComputerFields @defer
   }
 }
+
 fragment ComputerFields on Computer {
   cpu
   year
@@ -12,13 +13,13 @@ fragment ComputerFields on Computer {
     ...ScreenFields @defer(label: "a")
   }
 }
+
 fragment ScreenFields on Screen {
   isColor
 }
 
 
-
-query WithInlineFragments {
+query WithInlineFragmentsQuery {
   computers {
     id
     ... on Computer @defer {
@@ -33,3 +34,11 @@ query WithInlineFragments {
     }
   }
 }
+
+
+mutation WithFragmentSpreadsMutation {
+  computers {
+    id
+    ...ComputerFields @defer(label: "c")
+  }
+}

tests/defer/src/commonMain/graphql/schema.graphqls

@@ -1,12 +1,18 @@
 type Query {
   computers: [Computer!]!
 }
+
+type Mutation {
+  computers: [Computer!]!
+}
+
 type Computer {
   id: ID!
   cpu: String!
   year: Int!
   screen: Screen!
 }
+
 type Screen {
   resolution: String!
   isColor: Boolean!