fix: Add background loading for paged loader - while the first page is being loaded

Additional on-fly improvements:
- invalidateAllAsync helper extension for all keyed stores
- new 'plus' operator for combining StoreResult and any metadata values
- throttleLatest flow operator for smoother UI rendering
- minor core store refactoring

Author: Roman Andrushchenko

README.md

@@ -34,7 +34,7 @@ The full documentation is available [here](https://docs.uandcode.com/container).
 Add the following line to your `build.gradle` file:
 
 ```
-implementation "com.elveum:container:3.3.0"
+implementation "com.elveum:container:3.3.1"
 ```
 
 ## Core Concepts

container/src/main/java/com/elveum/container/FlowExtensions.kt

@@ -0,0 +1,50 @@
+package com.elveum.container
+
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.channels.onClosed
+import kotlinx.coroutines.channels.onSuccess
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.channelFlow
+import kotlinx.coroutines.launch
+
+/**
+ * Throttles emissions so that consecutive downstream values are never closer
+ * than [millis] apart, while keeping isolated updates responsive.
+ *
+ * Behavior:
+ * - The first value (and any value that arrives after at least [millis] of
+ *   silence since the previous emission) is emitted **immediately**.
+ * - While values arrive more frequently than [millis], they are coalesced:
+ *   only the **latest** value is emitted, once per [millis] window.
+ *
+ * This gives two benefits at once:
+ * 1. bursts of frequent updates are optimized down to one emission per window;
+ * 2. rare, standalone updates are delivered without delay.
+ *
+ * Note: for a finite source, completion may be postponed by up to [millis]
+ * after the last value; this does not affect which values are emitted.
+ *
+ * @param T the type of values emitted by the flow.
+ * @param millis minimum period, in milliseconds, between two downstream emissions.
+ */
+public fun <T> Flow<T>.throttleLatest(
+    millis: Long,
+): Flow<T> {
+    val originFlow = this
+    return channelFlow {
+        val channel = Channel<T>(capacity = Channel.CONFLATED)
+        launch {
+            originFlow.collect { channel.send(it) }
+            channel.close()
+        }
+        while (true) {
+            channel.receiveCatching()
+                .onClosed { break }
+                .onSuccess {
+                    send(it)
+                    delay(millis)
+                }
+        }
+    }
+}

container/src/main/java/com/elveum/container/StatefulEmitter.kt

@@ -2,6 +2,8 @@ package com.elveum.container
 
 public interface StatefulEmitter<T> : Emitter<T> {
 
+    public val loadConfig: LoadConfig
+
     public val hasEmittedValues: Boolean
 
     public suspend fun emitPendingState()

container/src/main/java/com/elveum/container/subject/lazy/StatefulEmitterImpl.kt

@@ -18,7 +18,7 @@ import java.util.concurrent.atomic.AtomicBoolean
 internal class StatefulEmitterImpl<T>(
     private val emitter: FlowEmitter<T>,
     private val executeParams: ExecuteParams<T>,
-    private val loadConfig: LoadConfig,
+    override val loadConfig: LoadConfig,
     private val flowCollector: FlowCollector<Container<T>>,
     private val flowSubject: FlowSubject<T>?,
 ) : StatefulEmitter<T>, Emitter<T> by emitter {

container/src/main/java/com/elveum/container/subject/paging/PageEmitter.kt

@@ -3,6 +3,7 @@ package com.elveum.container.subject.paging
 import com.elveum.container.ContainerMetadata
 import com.elveum.container.EmptyMetadata
 import com.elveum.container.FlowComposer
+import com.elveum.container.LoadConfig
 
 /**
  * An instance of this emitter is available within the page loader function
@@ -16,6 +17,13 @@ public interface PageEmitter<Key, T> : FlowComposer {
      */
     public val metadata: ContainerMetadata
 
+    /**
+     * The requested load configuration (e.g. silent loading or not).
+     *
+     * @see LoadConfig
+     */
+    public val loadConfig: LoadConfig
+
     /**
      * Emit data loaded for the current page key.
      *

container/src/main/java/com/elveum/container/subject/paging/internal/PageContext.kt

@@ -1,14 +1,19 @@
 package com.elveum.container.subject.paging.internal
 
+import com.elveum.container.BackgroundLoadMetadata
+import com.elveum.container.BackgroundLoadState
 import com.elveum.container.ContainerMetadata
+import com.elveum.container.LoadConfig
 import com.elveum.container.errorContainer
 import com.elveum.container.pendingContainer
 import com.elveum.container.successContainer
+import com.elveum.container.update
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.StateFlow
 
 internal class PageContext<Key, T>(
     private val state: PageRecordsState<Key, T>,
+    private val loadConfig: LoadConfig,
     val isRetry: Boolean,
     private val onScheduleNextKey: suspend (Int, Key) -> Unit,
     initialRecord: ImmutablePageRecord<Key, T>,
@@ -34,6 +39,15 @@ internal class PageContext<Key, T>(
 
     suspend fun onLoadCompleted() {
         state.markAsCompleted(pageIndex, pageKey)
+        if (loadConfig.isSilentLoadingEnabled) {
+            state.updateRecord(
+                pageIndex = pageIndex,
+                pageKey = pageKey,
+                container = { old ->
+                    old.update { backgroundLoadState = BackgroundLoadState.Idle }
+                }
+            )
+        }
     }
 
     suspend fun onLoadFailed(e: Exception) {
@@ -45,10 +59,22 @@ internal class PageContext<Key, T>(
     }
 
     suspend fun onPageDataLoaded(items: List<T>, metadata: ContainerMetadata) {
+        val container = if (loadConfig.isSilentLoadingEnabled) {
+            val bgLoadState = if (pageIndex == 0) {
+                // report background loading only for the first page
+                BackgroundLoadState.Loading
+            } else {
+                BackgroundLoadState.Idle
+            }
+            val bgLoadMetadata = BackgroundLoadMetadata(bgLoadState)
+            successContainer(items, metadata + bgLoadMetadata)
+        } else {
+            successContainer(items, metadata)
+        }
         state.updateRecord(
             pageIndex = pageIndex,
             pageKey = pageKey,
-            container = { successContainer(items, metadata) },
+            container = { container },
         )
     }
 

container/src/main/java/com/elveum/container/subject/paging/internal/PageEmitterImpl.kt

@@ -2,6 +2,7 @@ package com.elveum.container.subject.paging.internal
 
 import com.elveum.container.Container
 import com.elveum.container.ContainerMetadata
+import com.elveum.container.LoadConfig
 import com.elveum.container.StatefulEmitter
 import com.elveum.container.subject.paging.PageEmitter
 import kotlinx.coroutines.flow.Flow
@@ -21,6 +22,7 @@ internal class PageEmitterImpl<Key, T>(
         private set
 
     override val metadata: ContainerMetadata get() = originEmitter.metadata
+    override val loadConfig: LoadConfig get() = originEmitter.loadConfig
 
     override suspend fun emitPage(list: List<T>, metadata: ContainerMetadata) {
         isPageEmitted = true

container/src/main/java/com/elveum/container/subject/paging/internal/PageLoadSession.kt

@@ -39,6 +39,7 @@ internal class PageLoadSession<Key, T>(
         val record = state.prepareRecord(pageIndex, pageKey)
         val context = PageContext(
             state = state,
+            loadConfig = originEmitter.loadConfig,
             initialRecord = record,
             isRetry = isRetry,
             onScheduleNextKey = { nextIndex, nextKey ->

container/src/main/java/com/elveum/container/subject/paging/internal/PageRecordsState.kt

@@ -1,11 +1,11 @@
 package com.elveum.container.subject.paging.internal
 
 import com.elveum.container.BackgroundLoadMetadata
-import com.elveum.container.BackgroundLoadState
 import com.elveum.container.Container
 import com.elveum.container.ContainerMetadata
 import com.elveum.container.EmptyMetadata
 import com.elveum.container.StatefulEmitter
+import com.elveum.container.get
 import com.elveum.container.getOrNull
 import com.elveum.container.isCompleted
 import com.elveum.container.isError
@@ -163,10 +163,12 @@ internal class PageRecordsState<Key, T>(
             onNextPageStateChanged(pageState)
             val outputList = outputListSnapshot()
 
-            val finalMetadata = BackgroundLoadMetadata(BackgroundLoadState.Idle) + if (config.emitMetadata) {
+            val bgLoadMetadata = containers.firstOrNull()?.metadata?.get<BackgroundLoadMetadata>()
+            val finalMetadata = if (config.emitMetadata) {
                 OnItemRenderedCallbackMetadata(onItemRenderedRef) +
                         NextPageStateMetadata(pageState) +
-                        outputMergedMetadata(containers)
+                        outputMergedMetadata(containers) +
+                        bgLoadMetadata
             } else {
                 EmptyMetadata
             }

container/src/main/java/com/elveum/container/subject/paging/internal/ScopedPageLoader.kt

@@ -6,6 +6,7 @@ import com.elveum.container.StatefulEmitter
 import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.currentCoroutineContext
 import kotlinx.coroutines.ensureActive
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.combine
@@ -27,29 +28,22 @@ internal class ScopedPageLoader<Key, T>(
     fun loadPage(context: PageContext<Key, T>) {
         coroutineScope.launch {
             if (!context.isRetry) awaitFetchDistanceSatisfied(context)
+            val emitter = PageEmitterImpl(
+                context = context,
+                originEmitter = originEmitter,
+            )
             try {
                 context.onLoadStarted(context.isRetry)
-                val emitter = PageEmitterImpl(
-                    context = context,
-                    originEmitter = originEmitter,
-                )
                 config.block(emitter, context.pageKey)
                 if (!emitter.isPageEmitted) {
                     sessionCompleteDeferred.completeExceptionally(
                         IllegalStateException("emitPage() must be called at least once.")
                     )
                     return@launch
                 }
-                context.onLoadCompleted()
-                if (context.isAllPagesCompleted()) {
-                    sessionCompleteDeferred.complete(Unit)
-                }
+                completeLoad(context)
             } catch (e: Exception) {
-                ensureActive()
-                if (context.pageIndex == 0) {
-                    sessionCompleteDeferred.completeExceptionally(e)
-                }
-                context.onLoadFailed(e)
+                handleLoadError(context, emitter.isPageEmitted, e)
             }
         }
     }
@@ -64,6 +58,31 @@ internal class ScopedPageLoader<Key, T>(
         }
     }
 
+    private suspend fun handleLoadError(
+        context: PageContext<Key, T>,
+        isPageEmitted: Boolean,
+        e: Exception,
+    ) {
+        currentCoroutineContext().ensureActive()
+        val isSilentErrors = originEmitter.loadConfig.isSilentErrorsEnabled
+        if (isSilentErrors && isPageEmitted) {
+            // page data already exists, errors can be suppressed by silent flag
+            completeLoad(context)
+        } else {
+            if (context.pageIndex == 0) {
+                sessionCompleteDeferred.completeExceptionally(e)
+            }
+            context.onLoadFailed(e)
+        }
+    }
+
+    private suspend fun completeLoad(context: PageContext<Key, T>) {
+        context.onLoadCompleted()
+        if (context.isAllPagesCompleted()) {
+            sessionCompleteDeferred.complete(Unit)
+        }
+    }
+
     private suspend fun awaitFetchDistanceSatisfied(context: PageContext<Key, T>) {
         if (context.pageIndex == 0) return // immediate load of the initial page
 
@@ -89,7 +108,8 @@ internal class ScopedPageLoader<Key, T>(
         ) { indexOfLastItemOfPage, lastRenderedIndex -> indexOfLastItemOfPage to lastRenderedIndex }
 
         finalFlow.first { (indexOfLastItemOfPage, lastRenderedIndex) ->
-            lastRenderedIndex > indexOfLastItemOfPage - config.finalFetchDistance
+            val result = lastRenderedIndex > indexOfLastItemOfPage - config.finalFetchDistance
+            result
         }
     }