[🗄️ cache] Make ApolloStore.publish() suspend (#5755)

* Make ApolloStore.publish() suspend * fix KDoc
apollographql · Mar 22, 2024 · b54afc5 · b54afc5
1 parent 3fe0b15
commit b54afc5
Show file tree

Hide file tree

Showing 8 changed files with 177 additions and 123 deletions.
diff --git a/...ncubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/ApolloStore.kt b/...ncubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/ApolloStore.kt
@@ -72,41 +72,39 @@ interface ApolloStore {
   ): D
 
   /**
-   * Write an operation data to the store and optionally publish changes of [Record] which have changed,
-   * that will notify any watcher that depends on these [Record] to re-fetch.
+   * Write an operation data to the store.
    * This is a synchronous operation that might block if the underlying cache is doing IO
    *
    * @param operation     [Operation] response data of which should be written to the store
    * @param operationData [Operation.Data] operation response data to be written to the store
-   * @param publish       whether to publish the changed keys to listeners
    * @return the changed keys
+   *
+   * @see publish
    */
   fun <D : Operation.Data> writeOperation(
       operation: Operation<D>,
       operationData: D,
       customScalarAdapters: CustomScalarAdapters = CustomScalarAdapters.Empty,
       cacheHeaders: CacheHeaders = CacheHeaders.NONE,
-      publish: Boolean = true,
   ): Set<String>
 
   /**
-   * Write a fragment data to the store and optionally publish changes of [Record] which have changed,
-   * that will notify any watcher that depends on these [Record] to re-fetch.
+   * Write a fragment data to the store.
    * This is a synchronous operation that might block if the underlying cache is doing IO
    *
    * @param fragment data to be written to the store
    * @param cacheKey [CacheKey] to be used as root record key
    * @param fragmentData [Fragment.Data] to be written to the store
-   * @param publish whether to publish the changed keys to listeners
    * @return the changed keys
+   *
+   * @see publish
    */
   fun <D : Fragment.Data> writeFragment(
       fragment: Fragment<D>,
       cacheKey: CacheKey,
       fragmentData: D,
       customScalarAdapters: CustomScalarAdapters = CustomScalarAdapters.Empty,
       cacheHeaders: CacheHeaders = CacheHeaders.NONE,
-      publish: Boolean = true,
   ): Set<String>
 
   /**
@@ -117,13 +115,14 @@ interface ApolloStore {
    * @param operationData [Operation.Data] operation response data to be written to the store
    * @param mutationId    mutation unique identifier
    * @return the changed keys
+   *
+   * @see publish
    */
   fun <D : Operation.Data> writeOptimisticUpdates(
       operation: Operation<D>,
       operationData: D,
       mutationId: Uuid,
       customScalarAdapters: CustomScalarAdapters = CustomScalarAdapters.Empty,
-      publish: Boolean = true,
   ): Set<String>
 
   /**
@@ -135,7 +134,6 @@ interface ApolloStore {
    */
   fun rollbackOptimisticUpdates(
       mutationId: Uuid,
-      publish: Boolean = true,
   ): Set<String>
 
   /**
@@ -178,7 +176,7 @@ interface ApolloStore {
   /**
    * @param keys A set of keys of [Record] which have changed.
    */
-  fun publish(keys: Set<String>)
+  suspend fun publish(keys: Set<String>)
 
   /**
    * Direct access to the cache.

diff --git a/...Main/kotlin/com/apollographql/apollo3/cache/normalized/internal/ApolloCacheInterceptor.kt b/...Main/kotlin/com/apollographql/apollo3/cache/normalized/internal/ApolloCacheInterceptor.kt
@@ -37,7 +37,7 @@ import kotlinx.coroutines.launch
 internal class ApolloCacheInterceptor(
     val store: ApolloStore,
 ) : ApolloInterceptor {
-  private fun <D : Operation.Data> maybeAsync(request: ApolloRequest<D>, block: () -> Unit) {
+  private suspend fun <D : Operation.Data> maybeAsync(request: ApolloRequest<D>, block: suspend () -> Unit) {
     if (request.writeToCacheAsynchronously) {
       val scope = request.executionContext[ConcurrencyInfo]!!.coroutineScope
       scope.launch {
@@ -55,7 +55,7 @@ internal class ApolloCacheInterceptor(
   /**
    * @param extraKeys extra keys to publish in case there is optimistic data
    */
-  private fun <D : Operation.Data> maybeWriteToCache(
+  private suspend fun <D : Operation.Data> maybeWriteToCache(
       request: ApolloRequest<D>,
       response: ApolloResponse<D>,
       customScalarAdapters: CustomScalarAdapters,
@@ -77,7 +77,7 @@ internal class ApolloCacheInterceptor(
         if (request.storeReceiveDate) {
           cacheHeaders += nowDateCacheHeaders()
         }
-        store.writeOperation(request.operation, response.data!!, customScalarAdapters, cacheHeaders, publish = false)
+        store.writeOperation(request.operation, response.data!!, customScalarAdapters, cacheHeaders)
       } else {
         emptySet()
       }
@@ -141,8 +141,7 @@ internal class ApolloCacheInterceptor(
             operationData = optimisticData as D,
             mutationId = request.requestUuid,
             customScalarAdapters = customScalarAdapters,
-            publish = true
-        )
+        ).also { store.publish(it) }
       }
 
       /**
@@ -163,7 +162,7 @@ internal class ApolloCacheInterceptor(
         }
         previousResponse = response
         if (optimisticKeys == null) optimisticKeys = if (optimisticData != null) {
-          store.rollbackOptimisticUpdates(request.requestUuid, publish = false)
+          store.rollbackOptimisticUpdates(request.requestUuid)
         } else {
           emptySet()
         }
@@ -174,7 +173,7 @@ internal class ApolloCacheInterceptor(
 
       if (networkException != null) {
         if (optimisticKeys == null) optimisticKeys = if (optimisticData != null) {
-          store.rollbackOptimisticUpdates(request.requestUuid, publish = false)
+          store.rollbackOptimisticUpdates(request.requestUuid)
         } else {
           emptySet()
         }

diff --git a/...mmonMain/kotlin/com/apollographql/apollo3/cache/normalized/internal/DefaultApolloStore.kt b/...mmonMain/kotlin/com/apollographql/apollo3/cache/normalized/internal/DefaultApolloStore.kt
@@ -64,14 +64,12 @@ internal class DefaultApolloStore(
     OptimisticNormalizedCache(normalizedCacheFactory.create())
   }
 
-  override fun publish(keys: Set<String>) {
+  override suspend fun publish(keys: Set<String>) {
     if (keys.isEmpty()) {
       return
     }
 
-    if (!changedKeysEvents.tryEmit(keys)) {
-      println("Apollo: changedKeys event lost, your watchers may be collecting too slowly")
-    }
+    changedKeysEvents.emit(keys)
   }
 
   override fun clearAll(): Boolean {
@@ -153,15 +151,15 @@ internal class DefaultApolloStore(
       operationData: D,
       customScalarAdapters: CustomScalarAdapters,
       cacheHeaders: CacheHeaders,
-      publish: Boolean,
   ): Set<String> {
-    return writeOperationWithRecords(
-        operation = operation,
-        operationData = operationData,
-        cacheHeaders = cacheHeaders,
-        publish = publish,
-        customScalarAdapters = customScalarAdapters
-    ).second
+    val records = operation.normalize(
+        data = operationData,
+        customScalarAdapters = customScalarAdapters,
+        cacheKeyGenerator = cacheKeyGenerator,
+        metadataGenerator = metadataGenerator,
+    ).values.toSet()
+
+    return cache.merge(records, cacheHeaders, recordMerger)
   }
 
   override fun <D : Fragment.Data> writeFragment(
@@ -170,7 +168,6 @@ internal class DefaultApolloStore(
       fragmentData: D,
       customScalarAdapters: CustomScalarAdapters,
       cacheHeaders: CacheHeaders,
-      publish: Boolean,
   ): Set<String> {
     val records = fragment.normalize(
         data = fragmentData,
@@ -180,43 +177,14 @@ internal class DefaultApolloStore(
         rootKey = cacheKey.key
     ).values
 
-    val changedKeys = cache.merge(records, cacheHeaders, recordMerger)
-    if (publish) {
-      publish(changedKeys)
-    }
-
-    return changedKeys
+    return cache.merge(records, cacheHeaders, recordMerger)
   }
 
-  private fun <D : Operation.Data> writeOperationWithRecords(
-      operation: Operation<D>,
-      operationData: D,
-      cacheHeaders: CacheHeaders,
-      publish: Boolean,
-      customScalarAdapters: CustomScalarAdapters,
-  ): Pair<Set<Record>, Set<String>> {
-    val records = operation.normalize(
-        data = operationData,
-        customScalarAdapters = customScalarAdapters,
-        cacheKeyGenerator = cacheKeyGenerator,
-        metadataGenerator = metadataGenerator,
-    ).values.toSet()
-
-    val changedKeys = cache.merge(records, cacheHeaders, recordMerger)
-    if (publish) {
-      publish(changedKeys)
-    }
-
-    return records to changedKeys
-  }
-
-
   override fun <D : Operation.Data> writeOptimisticUpdates(
       operation: Operation<D>,
       operationData: D,
       mutationId: Uuid,
       customScalarAdapters: CustomScalarAdapters,
-      publish: Boolean,
   ): Set<String> {
     val records = operation.normalize(
         data = operationData,
@@ -234,24 +202,13 @@ internal class DefaultApolloStore(
     /**
      * TODO: should we forward the cache headers to the optimistic store?
      */
-    val changedKeys = cache.addOptimisticUpdates(records)
-    if (publish) {
-      publish(changedKeys)
-    }
-
-    return changedKeys
+    return cache.addOptimisticUpdates(records)
   }
 
   override fun rollbackOptimisticUpdates(
       mutationId: Uuid,
-      publish: Boolean,
   ): Set<String> {
-    val changedKeys = cache.removeOptimisticUpdates(mutationId)
-    if (publish) {
-      publish(changedKeys)
-    }
-
-    return changedKeys
+    return cache.removeOptimisticUpdates(mutationId)
   }
 
   fun merge(record: Record, cacheHeaders: CacheHeaders): Set<String> {

diff --git a/libraries/apollo-normalized-cache/api/apollo-normalized-cache.api b/libraries/apollo-normalized-cache/api/apollo-normalized-cache.api
@@ -5,7 +5,7 @@ public abstract interface class com/apollographql/apollo3/cache/normalized/Apoll
 	public abstract fun dump ()Ljava/util/Map;
 	public abstract fun getChangedKeys ()Lkotlinx/coroutines/flow/SharedFlow;
 	public abstract fun normalize (Lcom/apollographql/apollo3/api/Operation;Lcom/apollographql/apollo3/api/Operation$Data;Lcom/apollographql/apollo3/api/CustomScalarAdapters;)Ljava/util/Map;
-	public abstract fun publish (Ljava/util/Set;)V
+	public abstract fun publish (Ljava/util/Set;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public abstract fun readFragment (Lcom/apollographql/apollo3/api/Fragment;Lcom/apollographql/apollo3/cache/normalized/api/CacheKey;Lcom/apollographql/apollo3/api/CustomScalarAdapters;Lcom/apollographql/apollo3/cache/normalized/api/CacheHeaders;)Lcom/apollographql/apollo3/api/Fragment$Data;
 	public static synthetic fun readFragment$default (Lcom/apollographql/apollo3/cache/normalized/ApolloStore;Lcom/apollographql/apollo3/api/Fragment;Lcom/apollographql/apollo3/cache/normalized/api/CacheKey;Lcom/apollographql/apollo3/api/CustomScalarAdapters;Lcom/apollographql/apollo3/cache/normalized/api/CacheHeaders;ILjava/lang/Object;)Lcom/apollographql/apollo3/api/Fragment$Data;
 	public abstract fun readOperation (Lcom/apollographql/apollo3/api/Operation;Lcom/apollographql/apollo3/api/CustomScalarAdapters;Lcom/apollographql/apollo3/cache/normalized/api/CacheHeaders;)Lcom/apollographql/apollo3/api/Operation$Data;
@@ -14,14 +14,21 @@ public abstract interface class com/apollographql/apollo3/cache/normalized/Apoll
 	public abstract fun remove (Ljava/util/List;Z)I
 	public static synthetic fun remove$default (Lcom/apollographql/apollo3/cache/normalized/ApolloStore;Lcom/apollographql/apollo3/cache/normalized/api/CacheKey;ZILjava/lang/Object;)Z
 	public static synthetic fun remove$default (Lcom/apollographql/apollo3/cache/normalized/ApolloStore;Ljava/util/List;ZILjava/lang/Object;)I
-	public abstract fun rollbackOptimisticUpdates (Ljava/util/UUID;Z)Ljava/util/Set;
-	public static synthetic fun rollbackOptimisticUpdates$default (Lcom/apollographql/apollo3/cache/normalized/ApolloStore;Ljava/util/UUID;ZILjava/lang/Object;)Ljava/util/Set;
-	public abstract fun writeFragment (Lcom/apollographql/apollo3/api/Fragment;Lcom/apollographql/apollo3/cache/normalized/api/CacheKey;Lcom/apollographql/apollo3/api/Fragment$Data;Lcom/apollographql/apollo3/api/CustomScalarAdapters;Lcom/apollographql/apollo3/cache/normalized/api/CacheHeaders;Z)Ljava/util/Set;
-	public static synthetic fun writeFragment$default (Lcom/apollographql/apollo3/cache/normalized/ApolloStore;Lcom/apollographql/apollo3/api/Fragment;Lcom/apollographql/apollo3/cache/normalized/api/CacheKey;Lcom/apollographql/apollo3/api/Fragment$Data;Lcom/apollographql/apollo3/api/CustomScalarAdapters;Lcom/apollographql/apollo3/cache/normalized/api/CacheHeaders;ZILjava/lang/Object;)Ljava/util/Set;
-	public abstract fun writeOperation (Lcom/apollographql/apollo3/api/Operation;Lcom/apollographql/apollo3/api/Operation$Data;Lcom/apollographql/apollo3/api/CustomScalarAdapters;Lcom/apollographql/apollo3/cache/normalized/api/CacheHeaders;Z)Ljava/util/Set;
-	public static synthetic fun writeOperation$default (Lcom/apollographql/apollo3/cache/normalized/ApolloStore;Lcom/apollographql/apollo3/api/Operation;Lcom/apollographql/apollo3/api/Operation$Data;Lcom/apollographql/apollo3/api/CustomScalarAdapters;Lcom/apollographql/apollo3/cache/normalized/api/CacheHeaders;ZILjava/lang/Object;)Ljava/util/Set;
-	public abstract fun writeOptimisticUpdates (Lcom/apollographql/apollo3/api/Operation;Lcom/apollographql/apollo3/api/Operation$Data;Ljava/util/UUID;Lcom/apollographql/apollo3/api/CustomScalarAdapters;Z)Ljava/util/Set;
-	public static synthetic fun writeOptimisticUpdates$default (Lcom/apollographql/apollo3/cache/normalized/ApolloStore;Lcom/apollographql/apollo3/api/Operation;Lcom/apollographql/apollo3/api/Operation$Data;Ljava/util/UUID;Lcom/apollographql/apollo3/api/CustomScalarAdapters;ZILjava/lang/Object;)Ljava/util/Set;
+	public abstract fun rollbackOptimisticUpdates (Ljava/util/UUID;)Ljava/util/Set;
+	public abstract fun rollbackOptimisticUpdates (Ljava/util/UUID;ZLkotlin/coroutines/Continuation;)Ljava/lang/Object;
+	public static synthetic fun rollbackOptimisticUpdates$default (Lcom/apollographql/apollo3/cache/normalized/ApolloStore;Ljava/util/UUID;ZLkotlin/coroutines/Continuation;ILjava/lang/Object;)Ljava/lang/Object;
+	public abstract fun writeFragment (Lcom/apollographql/apollo3/api/Fragment;Lcom/apollographql/apollo3/cache/normalized/api/CacheKey;Lcom/apollographql/apollo3/api/Fragment$Data;Lcom/apollographql/apollo3/api/CustomScalarAdapters;Lcom/apollographql/apollo3/cache/normalized/api/CacheHeaders;)Ljava/util/Set;
+	public abstract fun writeFragment (Lcom/apollographql/apollo3/api/Fragment;Lcom/apollographql/apollo3/cache/normalized/api/CacheKey;Lcom/apollographql/apollo3/api/Fragment$Data;Lcom/apollographql/apollo3/api/CustomScalarAdapters;Lcom/apollographql/apollo3/cache/normalized/api/CacheHeaders;ZLkotlin/coroutines/Continuation;)Ljava/lang/Object;
+	public static synthetic fun writeFragment$default (Lcom/apollographql/apollo3/cache/normalized/ApolloStore;Lcom/apollographql/apollo3/api/Fragment;Lcom/apollographql/apollo3/cache/normalized/api/CacheKey;Lcom/apollographql/apollo3/api/Fragment$Data;Lcom/apollographql/apollo3/api/CustomScalarAdapters;Lcom/apollographql/apollo3/cache/normalized/api/CacheHeaders;ILjava/lang/Object;)Ljava/util/Set;
+	public static synthetic fun writeFragment$default (Lcom/apollographql/apollo3/cache/normalized/ApolloStore;Lcom/apollographql/apollo3/api/Fragment;Lcom/apollographql/apollo3/cache/normalized/api/CacheKey;Lcom/apollographql/apollo3/api/Fragment$Data;Lcom/apollographql/apollo3/api/CustomScalarAdapters;Lcom/apollographql/apollo3/cache/normalized/api/CacheHeaders;ZLkotlin/coroutines/Continuation;ILjava/lang/Object;)Ljava/lang/Object;
+	public abstract fun writeOperation (Lcom/apollographql/apollo3/api/Operation;Lcom/apollographql/apollo3/api/Operation$Data;Lcom/apollographql/apollo3/api/CustomScalarAdapters;Lcom/apollographql/apollo3/cache/normalized/api/CacheHeaders;)Ljava/util/Set;
+	public abstract fun writeOperation (Lcom/apollographql/apollo3/api/Operation;Lcom/apollographql/apollo3/api/Operation$Data;Lcom/apollographql/apollo3/api/CustomScalarAdapters;Lcom/apollographql/apollo3/cache/normalized/api/CacheHeaders;ZLkotlin/coroutines/Continuation;)Ljava/lang/Object;
+	public static synthetic fun writeOperation$default (Lcom/apollographql/apollo3/cache/normalized/ApolloStore;Lcom/apollographql/apollo3/api/Operation;Lcom/apollographql/apollo3/api/Operation$Data;Lcom/apollographql/apollo3/api/CustomScalarAdapters;Lcom/apollographql/apollo3/cache/normalized/api/CacheHeaders;ILjava/lang/Object;)Ljava/util/Set;
+	public static synthetic fun writeOperation$default (Lcom/apollographql/apollo3/cache/normalized/ApolloStore;Lcom/apollographql/apollo3/api/Operation;Lcom/apollographql/apollo3/api/Operation$Data;Lcom/apollographql/apollo3/api/CustomScalarAdapters;Lcom/apollographql/apollo3/cache/normalized/api/CacheHeaders;ZLkotlin/coroutines/Continuation;ILjava/lang/Object;)Ljava/lang/Object;
+	public abstract fun writeOptimisticUpdates (Lcom/apollographql/apollo3/api/Operation;Lcom/apollographql/apollo3/api/Operation$Data;Ljava/util/UUID;Lcom/apollographql/apollo3/api/CustomScalarAdapters;)Ljava/util/Set;
+	public abstract fun writeOptimisticUpdates (Lcom/apollographql/apollo3/api/Operation;Lcom/apollographql/apollo3/api/Operation$Data;Ljava/util/UUID;Lcom/apollographql/apollo3/api/CustomScalarAdapters;ZLkotlin/coroutines/Continuation;)Ljava/lang/Object;
+	public static synthetic fun writeOptimisticUpdates$default (Lcom/apollographql/apollo3/cache/normalized/ApolloStore;Lcom/apollographql/apollo3/api/Operation;Lcom/apollographql/apollo3/api/Operation$Data;Ljava/util/UUID;Lcom/apollographql/apollo3/api/CustomScalarAdapters;ILjava/lang/Object;)Ljava/util/Set;
+	public static synthetic fun writeOptimisticUpdates$default (Lcom/apollographql/apollo3/cache/normalized/ApolloStore;Lcom/apollographql/apollo3/api/Operation;Lcom/apollographql/apollo3/api/Operation$Data;Ljava/util/UUID;Lcom/apollographql/apollo3/api/CustomScalarAdapters;ZLkotlin/coroutines/Continuation;ILjava/lang/Object;)Ljava/lang/Object;
 }
 
 public final class com/apollographql/apollo3/cache/normalized/ApolloStoreKt {