diff --git a/README.md b/README.md index a8a8178..314315c 100644 --- a/README.md +++ b/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 diff --git a/container/src/main/java/com/elveum/container/FlowExtensions.kt b/container/src/main/java/com/elveum/container/FlowExtensions.kt new file mode 100644 index 0000000..badf23e --- /dev/null +++ b/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 Flow.throttleLatest( + millis: Long, +): Flow { + val originFlow = this + return channelFlow { + val channel = Channel(capacity = Channel.CONFLATED) + launch { + originFlow.collect { channel.send(it) } + channel.close() + } + while (true) { + channel.receiveCatching() + .onClosed { break } + .onSuccess { + send(it) + delay(millis) + } + } + } +} diff --git a/container/src/main/java/com/elveum/container/StatefulEmitter.kt b/container/src/main/java/com/elveum/container/StatefulEmitter.kt index 98792ce..31adfc6 100644 --- a/container/src/main/java/com/elveum/container/StatefulEmitter.kt +++ b/container/src/main/java/com/elveum/container/StatefulEmitter.kt @@ -2,6 +2,8 @@ package com.elveum.container public interface StatefulEmitter : Emitter { + public val loadConfig: LoadConfig + public val hasEmittedValues: Boolean public suspend fun emitPendingState() diff --git a/container/src/main/java/com/elveum/container/subject/lazy/StatefulEmitterImpl.kt b/container/src/main/java/com/elveum/container/subject/lazy/StatefulEmitterImpl.kt index 16d5e5b..9615e52 100644 --- a/container/src/main/java/com/elveum/container/subject/lazy/StatefulEmitterImpl.kt +++ b/container/src/main/java/com/elveum/container/subject/lazy/StatefulEmitterImpl.kt @@ -18,7 +18,7 @@ import java.util.concurrent.atomic.AtomicBoolean internal class StatefulEmitterImpl( private val emitter: FlowEmitter, private val executeParams: ExecuteParams, - private val loadConfig: LoadConfig, + override val loadConfig: LoadConfig, private val flowCollector: FlowCollector>, private val flowSubject: FlowSubject?, ) : StatefulEmitter, Emitter by emitter { diff --git a/container/src/main/java/com/elveum/container/subject/paging/PageEmitter.kt b/container/src/main/java/com/elveum/container/subject/paging/PageEmitter.kt index 85d183b..fc878c6 100644 --- a/container/src/main/java/com/elveum/container/subject/paging/PageEmitter.kt +++ b/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 : 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. * diff --git a/container/src/main/java/com/elveum/container/subject/paging/internal/PageContext.kt b/container/src/main/java/com/elveum/container/subject/paging/internal/PageContext.kt index 3c6e18c..115e372 100644 --- a/container/src/main/java/com/elveum/container/subject/paging/internal/PageContext.kt +++ b/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( private val state: PageRecordsState, + private val loadConfig: LoadConfig, val isRetry: Boolean, private val onScheduleNextKey: suspend (Int, Key) -> Unit, initialRecord: ImmutablePageRecord, @@ -34,6 +39,15 @@ internal class PageContext( 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( } suspend fun onPageDataLoaded(items: List, 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 }, ) } diff --git a/container/src/main/java/com/elveum/container/subject/paging/internal/PageEmitterImpl.kt b/container/src/main/java/com/elveum/container/subject/paging/internal/PageEmitterImpl.kt index e8388f9..d362d21 100644 --- a/container/src/main/java/com/elveum/container/subject/paging/internal/PageEmitterImpl.kt +++ b/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( private set override val metadata: ContainerMetadata get() = originEmitter.metadata + override val loadConfig: LoadConfig get() = originEmitter.loadConfig override suspend fun emitPage(list: List, metadata: ContainerMetadata) { isPageEmitted = true diff --git a/container/src/main/java/com/elveum/container/subject/paging/internal/PageLoadSession.kt b/container/src/main/java/com/elveum/container/subject/paging/internal/PageLoadSession.kt index 835ace1..4cfe353 100644 --- a/container/src/main/java/com/elveum/container/subject/paging/internal/PageLoadSession.kt +++ b/container/src/main/java/com/elveum/container/subject/paging/internal/PageLoadSession.kt @@ -39,6 +39,7 @@ internal class PageLoadSession( val record = state.prepareRecord(pageIndex, pageKey) val context = PageContext( state = state, + loadConfig = originEmitter.loadConfig, initialRecord = record, isRetry = isRetry, onScheduleNextKey = { nextIndex, nextKey -> diff --git a/container/src/main/java/com/elveum/container/subject/paging/internal/PageRecordsState.kt b/container/src/main/java/com/elveum/container/subject/paging/internal/PageRecordsState.kt index 416ac8d..68e68aa 100644 --- a/container/src/main/java/com/elveum/container/subject/paging/internal/PageRecordsState.kt +++ b/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( onNextPageStateChanged(pageState) val outputList = outputListSnapshot() - val finalMetadata = BackgroundLoadMetadata(BackgroundLoadState.Idle) + if (config.emitMetadata) { + val bgLoadMetadata = containers.firstOrNull()?.metadata?.get() + val finalMetadata = if (config.emitMetadata) { OnItemRenderedCallbackMetadata(onItemRenderedRef) + NextPageStateMetadata(pageState) + - outputMergedMetadata(containers) + outputMergedMetadata(containers) + + bgLoadMetadata } else { EmptyMetadata } diff --git a/container/src/main/java/com/elveum/container/subject/paging/internal/ScopedPageLoader.kt b/container/src/main/java/com/elveum/container/subject/paging/internal/ScopedPageLoader.kt index b625ccf..e15fdb1 100644 --- a/container/src/main/java/com/elveum/container/subject/paging/internal/ScopedPageLoader.kt +++ b/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,12 +28,12 @@ internal class ScopedPageLoader( fun loadPage(context: PageContext) { 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( @@ -40,16 +41,9 @@ internal class ScopedPageLoader( ) 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( } } + private suspend fun handleLoadError( + context: PageContext, + 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) { + context.onLoadCompleted() + if (context.isAllPagesCompleted()) { + sessionCompleteDeferred.complete(Unit) + } + } + private suspend fun awaitFetchDistanceSatisfied(context: PageContext) { if (context.pageIndex == 0) return // immediate load of the initial page @@ -89,7 +108,8 @@ internal class ScopedPageLoader( ) { indexOfLastItemOfPage, lastRenderedIndex -> indexOfLastItemOfPage to lastRenderedIndex } finalFlow.first { (indexOfLastItemOfPage, lastRenderedIndex) -> - lastRenderedIndex > indexOfLastItemOfPage - config.finalFetchDistance + val result = lastRenderedIndex > indexOfLastItemOfPage - config.finalFetchDistance + result } } diff --git a/container/src/test/java/com/elveum/container/ThrottleLatestTest.kt b/container/src/test/java/com/elveum/container/ThrottleLatestTest.kt new file mode 100644 index 0000000..3af67f3 --- /dev/null +++ b/container/src/test/java/com/elveum/container/ThrottleLatestTest.kt @@ -0,0 +1,184 @@ +package com.elveum.container + +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.flow.flow +import kotlinx.coroutines.flow.flowOf +import kotlinx.coroutines.delay +import kotlinx.coroutines.test.TestScope +import kotlinx.coroutines.test.runTest +import org.junit.Assert.assertEquals +import org.junit.Assert.assertTrue +import org.junit.Assert.fail +import org.junit.Test + +@OptIn(ExperimentalCoroutinesApi::class) +class ThrottleLatestTest { + + /** + * Collects the throttled flow and records the virtual time at which + * each value is emitted downstream. + */ + private suspend fun TestScope.collectWithTime( + flow: kotlinx.coroutines.flow.Flow, + ): List> { + val result = mutableListOf>() + flow.collect { result.add(testScheduler.currentTime to it) } + return result + } + + @Test + fun firstValue_isEmittedImmediately() = runTest { + val origin = flow { + emit(1) + delay(1000) + } + + val result = collectWithTime(origin.throttleLatest(100)) + + assertEquals(listOf(0L to 1), result) + } + + @Test + fun infrequentValues_areAllEmittedImmediately() = runTest { + // Every value follows >= millis of silence -> each is a leading value. + val origin = flow { + emit(1) + delay(150) + emit(2) + delay(150) + emit(3) + delay(150) + } + + val result = collectWithTime(origin.throttleLatest(100)) + + assertEquals(listOf(0L to 1, 150L to 2, 300L to 3), result) + } + + @Test + fun frequentValues_onlyLatestIsEmittedOncePerPeriod() = runTest { + // millis = 100, values every 30ms: 1@0, 2@30, 3@60, 4@90, 5@120, 6@150, 7@180 + val origin = flow { + emit(1); delay(30) + emit(2); delay(30) + emit(3); delay(30) + emit(4); delay(30) + emit(5); delay(30) + emit(6); delay(30) + emit(7); delay(30) + } + + val result = collectWithTime(origin.throttleLatest(100)) + + // 1 immediately; latest at each period boundary: 4@100, 7@200. + assertEquals(listOf(0L to 1, 100L to 4, 200L to 7), result) + } + + @Test + fun synchronousBurst_emitsFirstThenLatest() = runTest { + // All values arrive at t=0. First (1) immediately, latest (3) one period later. + val origin = flowOf(1, 2, 3) + + val result = collectWithTime(origin.throttleLatest(100)) + + assertEquals(listOf(0L to 1, 100L to 3), result) + } + + @Test + fun valueArrivingSoonAfterEmission_isThrottledToMinPeriod() = runTest { + // 1@0 (leading), 2@50 (frequent) -> latest emitted at period end @100. + // 3@110 arrives only 10ms after the previous EMISSION, so it is a frequent + // update and must wait until the next period boundary @200 (not immediate). + val origin = flow { + emit(1); delay(50) + emit(2); delay(60) + emit(3); delay(500) + } + + val result = collectWithTime(origin.throttleLatest(100)) + + assertEquals(listOf(0L to 1, 100L to 2, 200L to 3), result) + } + + @Test + fun singleValueAfterLongIdle_isEmittedImmediately() = runTest { + // Advantage #2: a lone non-frequent update after a long quiet period + // must be rendered immediately, not delayed. + val origin = flow { + emit(1); delay(50) + emit(2) // throttled with 1 -> emitted at period boundary @100 + delay(1000) + emit(9) // long idle -> leading -> immediate + delay(50) + } + + val result = collectWithTime(origin.throttleLatest(100)) + + assertEquals(listOf(0L to 1, 100L to 2, 1050L to 9), result) + } + + @Test + fun emissionPeriod_isNeverLessThanMillis() = runTest { + // A dense stream must never produce two emissions closer than millis apart. + val origin = flow { + repeat(50) { i -> + emit(i) + delay(10) + } + } + + val result = collectWithTime(origin.throttleLatest(100)) + val times = result.map { it.first } + + for (i in 1 until times.size) { + val gap = times[i] - times[i - 1] + assertTrue("emission $i came only ${gap}ms after previous", gap >= 100) + } + } + + @Test + fun lastValue_isNotLostOnCompletion() = runTest { + // 1@0, 2@50 then completes. 2 must still be delivered. + val origin = flow { + emit(1); delay(50) + emit(2) + } + + val result = collectWithTime(origin.throttleLatest(100)) + + assertEquals(listOf(1, 2), result.map { it.second }) + } + + @Test + fun finiteFlow_completionIsDelayedByOnePeriodAfterLastValue() = runTest { + // Documents the trailing-delay quirk: after the LAST value is emitted, the + // loop still runs delay(millis) before it observes the closed channel, so a + // finite source's terminal is postponed by `millis` even though no further + // value can ever arrive. + val origin = flowOf(1) // completes immediately after emitting 1 + + val emittedAt = mutableListOf() + origin.throttleLatest(100).collect { emittedAt.add(testScheduler.currentTime) } + val completedAt = testScheduler.currentTime + + assertEquals(listOf(0L), emittedAt) // value emitted immediately + assertEquals(100L, completedAt) // ...but collection returns 100ms later + } + + @Test + fun exceptionFromOrigin_isPropagatedDownstream() = runTest { + val boom = IllegalStateException("boom") + val origin = flow { + emit(1) + delay(10) + throw boom + } + + try { + collectWithTime(origin.throttleLatest(100)) + fail("Expected the origin exception to propagate") + } catch (e: IllegalStateException) { + assertEquals("boom", e.message) + } + } +} diff --git a/container/src/test/java/com/elveum/container/subject/paging/internal/PageContextTest.kt b/container/src/test/java/com/elveum/container/subject/paging/internal/PageContextTest.kt index 1cc1cc5..7c66bee 100644 --- a/container/src/test/java/com/elveum/container/subject/paging/internal/PageContextTest.kt +++ b/container/src/test/java/com/elveum/container/subject/paging/internal/PageContextTest.kt @@ -2,6 +2,7 @@ package com.elveum.container.subject.paging.internal import com.elveum.container.Container import com.elveum.container.EmptyMetadata +import com.elveum.container.LoadConfig import com.elveum.container.errorContainer import com.elveum.container.pendingContainer import com.elveum.container.subject.paging.TotalPagedItemsCountMetadata @@ -264,7 +265,7 @@ class PageContextTest { isCompleted: Boolean = false, ): PageContext { val record = ImmutablePageRecord(pageIndex, pageKey, priority, container, isCompleted) - return PageContext(state, isRetry, onScheduleNextKey, record) + return PageContext(state, LoadConfig.Normal, isRetry, onScheduleNextKey, record) } } diff --git a/container/src/test/java/com/elveum/container/subject/paging/internal/PageLoaderIntegrationTest.kt b/container/src/test/java/com/elveum/container/subject/paging/internal/PageLoaderIntegrationTest.kt index 5c94358..ccb1192 100644 --- a/container/src/test/java/com/elveum/container/subject/paging/internal/PageLoaderIntegrationTest.kt +++ b/container/src/test/java/com/elveum/container/subject/paging/internal/PageLoaderIntegrationTest.kt @@ -45,6 +45,7 @@ class PageLoaderIntegrationTest { executeInBackground { pageLoader.statefulInvoke() } coVerifySequence { + emitter.loadConfig emitter.emitPendingState() } } diff --git a/gradle-public.properties b/gradle-public.properties index 1db9ddc..b352136 100644 --- a/gradle-public.properties +++ b/gradle-public.properties @@ -6,7 +6,7 @@ android.nonTransitiveRClass=true # Maven Central publishing coordinates GROUP=com.elveum -VERSION_NAME=3.3.0 +VERSION_NAME=3.3.1 # POM metadata POM_DESCRIPTION=A library for simplifying state management and data loading in Android applications diff --git a/skills/container-store/README.md b/skills/container-store/README.md index f3b1ad8..371ce2d 100644 --- a/skills/container-store/README.md +++ b/skills/container-store/README.md @@ -43,4 +43,4 @@ discovers skills (e.g. `~/.agents/skills/` for Codex), or paste | `references/api.md` | Every public import, `StoreResult` / `LoadRequest` semantics, typical operations | | `references/patterns.md` | Layer-by-layer patterns: data sources, repositories, ViewModels, Compose, DI, scoping | -The skill matches library version `3.3.0`. +The skill matches library version `3.3.1`. diff --git a/skills/container-store/SKILL.md b/skills/container-store/SKILL.md index 8486bb8..2dec518 100644 --- a/skills/container-store/SKILL.md +++ b/skills/container-store/SKILL.md @@ -22,13 +22,13 @@ directly - no decompilation or dependency tree inspection needed. ## Dependency Setup -Maven coordinates: `com.elveum:store:3.3.0` (transitively brings +Maven coordinates: `com.elveum:store:3.3.1` (transitively brings `com.elveum:container`, whose types are part of the public API). ```toml # gradle/libs.versions.toml [versions] -store = "3.3.0" +store = "3.3.1" [libraries] store = { module = "com.elveum:store", version.ref = "store" } ``` diff --git a/store/README.md b/store/README.md index ff7ba61..4b60827 100644 --- a/store/README.md +++ b/store/README.md @@ -38,7 +38,7 @@ the hood. Add the following line to your `build.gradle` file: ``` -implementation "com.elveum:store:3.3.0" +implementation "com.elveum:store:3.3.1" ``` The `store` artifact depends on `com.elveum:container`, so the Container diff --git a/store/src/main/java/com/elveum/store/internal/stores/PageLoaderProvider.kt b/store/src/main/java/com/elveum/store/internal/stores/PageLoaderProvider.kt new file mode 100644 index 0000000..47dbd3c --- /dev/null +++ b/store/src/main/java/com/elveum/store/internal/stores/PageLoaderProvider.kt @@ -0,0 +1,50 @@ +package com.elveum.store.internal.stores + +import com.elveum.container.subject.ValueLoader +import com.elveum.container.subject.paging.pageLoader +import com.elveum.store.internal.builders.paged.SharedPageConfig +import com.elveum.store.internal.stores.common.CoreEmitter +import com.elveum.store.internal.stores.common.CoreLoaderDelegate +import com.elveum.store.internal.stores.common.CorePageFetcher +import com.elveum.store.internal.stores.common.CoreValueLoaderProvider +import com.elveum.store.load.LoadRequestSource +import com.elveum.store.stores.paged.PagedList +import kotlinx.coroutines.flow.flowOf + +internal class PageLoaderProvider constructor( + private val config: SharedPageConfig, + private val fetcher: CorePageFetcher, + private val loader: suspend (Key, Q, PageKey) -> PagedList?, + private val saver: suspend (Key, Q, PageKey, PagedList) -> Unit, +) : CoreValueLoaderProvider, PagedList> { + + override fun provideValueLoader( + key: Key, + querySource: () -> Q, + requestSource: LoadRequestSource, + delegate: CoreLoaderDelegate> + ): ValueLoader> { + return pageLoader( + initialKey = config.initialKey, + itemId = config.itemId, + fetchDistance = config.fetchDistance, + block = { pageKey -> + val query = querySource() + when (fetcher) { + is CorePageFetcher.Default -> { + delegate.processDataLoad( + emitter = CoreEmitter.fromPageEmitter(this), + requestSource = requestSource, + fetcher = { fetcher.fetcher(key, query, pageKey) }, + loader = { loader(key, query, pageKey) }, + saver = { pageList -> saver(key, query, pageKey, pageList) }, + observer = { flowOf(null) }, + ) + } + is CorePageFetcher.Custom -> fetcher.fetcher(this, key, query, pageKey) + } + } + ) + } + +} diff --git a/store/src/main/java/com/elveum/store/internal/stores/PagedKeyedQueryStoreImpl.kt b/store/src/main/java/com/elveum/store/internal/stores/PagedKeyedQueryStoreImpl.kt index f508578..ee33659 100644 --- a/store/src/main/java/com/elveum/store/internal/stores/PagedKeyedQueryStoreImpl.kt +++ b/store/src/main/java/com/elveum/store/internal/stores/PagedKeyedQueryStoreImpl.kt @@ -2,16 +2,10 @@ package com.elveum.store.internal.stores -import com.elveum.container.subject.ValueLoader -import com.elveum.container.subject.paging.pageLoader import com.elveum.store.internal.builders.paged.SharedPageConfig -import com.elveum.store.internal.stores.common.CoreEmitter -import com.elveum.store.internal.stores.common.CoreLoaderDelegate import com.elveum.store.internal.stores.common.CorePageFetcher import com.elveum.store.internal.stores.common.CoreStore -import com.elveum.store.internal.stores.common.CoreValueLoaderProvider import com.elveum.store.load.LoadRequest -import com.elveum.store.load.LoadRequestSource import com.elveum.store.load.StoreResult import com.elveum.store.stores.base.OptimisticUpdateScope import com.elveum.store.stores.keyed.KeyedStore @@ -33,42 +27,20 @@ internal class PagedKeyedQueryStoreImpl Flow)? = null, ) : PagedKeyedQueryStore { + private val pageLoaderProvider = PageLoaderProvider( + config = config, + fetcher = fetcher, + loader = loader, + saver = saver, + ) + private val coreStore = CoreStore, PagedList>( initialQueryProvider = initialQueryProvider, queryDebounceMillis = queryDebounceMillis, config = config, observer = { _, _ -> flowOf(null) }, externalQueryProvider = externalQueryProvider, - valueLoaderProvider = object : CoreValueLoaderProvider, PagedList> { - override fun provideValueLoader( - key: Key, - querySource: () -> Q, - requestSource: LoadRequestSource, - delegate: CoreLoaderDelegate> - ): ValueLoader> { - return pageLoader( - initialKey = config.initialKey, - itemId = config.itemId, - fetchDistance = config.fetchDistance, - block = { pageKey -> - val query = querySource() - when (fetcher) { - is CorePageFetcher.Default -> { - delegate.processDataLoad( - emitter = CoreEmitter.fromPageEmitter(this), - requestSource = requestSource, - fetcher = { fetcher.fetcher(key, query, pageKey) }, - loader = { loader(key, query, pageKey) }, - saver = { pageList -> saver(key, query, pageKey, pageList) }, - observer = { flowOf(null) }, - ) - } - is CorePageFetcher.Custom -> fetcher.fetcher(this, key, query, pageKey) - } - } - ) - } - } + valueLoaderProvider = pageLoaderProvider, ) override val activeKeys: StateFlow> = coreStore.activeKeys diff --git a/store/src/main/java/com/elveum/store/internal/stores/common/CoreStore.kt b/store/src/main/java/com/elveum/store/internal/stores/common/CoreStore.kt index 872448f..5383ab9 100644 --- a/store/src/main/java/com/elveum/store/internal/stores/common/CoreStore.kt +++ b/store/src/main/java/com/elveum/store/internal/stores/common/CoreStore.kt @@ -46,9 +46,6 @@ import kotlinx.coroutines.flow.update import kotlinx.coroutines.launch import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.atomic.AtomicBoolean -import kotlin.collections.firstOrNull -import kotlin.collections.forEach -import kotlin.collections.map @Suppress("TooManyFunctions") internal class CoreStore( @@ -230,6 +227,24 @@ internal class CoreStore( .distinctUntilChanged() } + fun updateWith(key: Key, storeResult: StoreResult) { + getKeyRecords(key).forEach { + cache.updateWith(it, storeResult.toContainer()) + } + } + + fun observeQueryFlow(key: Key): StateFlow { + return currentQueriesFlow.stateMap { it[key] ?: initialQueryProvider(key) } + } + + fun whenActive(block: suspend () -> Unit) { + cache.whenActive { block() } + } + + fun onItemRendered(key: Key, index: Int) { + get(key).onItemRendered(index) + } + private fun updateQuery(key: Key, query: Q) { val queries = currentQueriesFlow.value if (!queries.contains(key)) return @@ -250,11 +265,6 @@ internal class CoreStore( ) } - - fun observeQueryFlow(key: Key): StateFlow { - return currentQueriesFlow.stateMap { it[key] ?: initialQueryProvider(key) } - } - private fun createDelegate(): CoreLoaderDelegate { return DefaultCoreLoaderDelegate() } @@ -290,26 +300,12 @@ internal class CoreStore( ) } - fun updateWith(key: Key, storeResult: StoreResult) { - getKeyRecords(key).forEach { - cache.updateWith(it, storeResult.toContainer()) - } - } - private fun updateIfSuccess(key: Key, updater: (T) -> T) { getKeyRecords(key).forEach { cache.updateIfSuccess(it, updater = updater) } } - fun whenActive(block: suspend () -> Unit) { - cache.whenActive { block() } - } - - fun onItemRendered(key: Key, index: Int) { - get(key).onItemRendered(index) - } - private data class KeyedQuery( val key: Key, val query: Q, diff --git a/store/src/main/java/com/elveum/store/load/StoreResult.kt b/store/src/main/java/com/elveum/store/load/StoreResult.kt index 4d211c4..5cb5b27 100644 --- a/store/src/main/java/com/elveum/store/load/StoreResult.kt +++ b/store/src/main/java/com/elveum/store/load/StoreResult.kt @@ -63,6 +63,20 @@ public sealed class StoreResult { * loads using this property. */ public val backgroundLoadState: BackgroundLoadState get() = metadata.backgroundLoadState + + /** + * Create a new [StoreResult] instance from the existing one, but with + * additional [metadata] values. + */ + public operator fun plus(metadata: ContainerMetadata): StoreResult { + val combinedMetadata = this.metadata + metadata + return when (this) { + is Failed -> Failed(exception, combinedMetadata) + is Loaded -> Loaded(value, combinedMetadata) + Loading -> Loading + } + } + } /** diff --git a/store/src/main/java/com/elveum/store/load/StoreResultExtensions.kt b/store/src/main/java/com/elveum/store/load/StoreResultExtensions.kt index 6215010..c6dfb28 100644 --- a/store/src/main/java/com/elveum/store/load/StoreResultExtensions.kt +++ b/store/src/main/java/com/elveum/store/load/StoreResultExtensions.kt @@ -112,3 +112,14 @@ public fun StoreResult.isBackgroundLoading(): Boolean = * or foreground). */ public fun StoreResult.hasAnyLoading(): Boolean = isForegroundLoading() || isBackgroundLoading() + +/** + * Exclude all metadata from the [StoreResult]. Useful for comparing results and testing. + */ +public fun StoreResult.raw(): StoreResult { + return when (this) { + is StoreResult.Failed -> StoreResult.Failed(exception) + is StoreResult.Loaded -> StoreResult.Loaded(value) + StoreResult.Loading -> StoreResult.Loading + } +} diff --git a/store/src/main/java/com/elveum/store/stores/keyed/KeyedStoreExtensions.kt b/store/src/main/java/com/elveum/store/stores/keyed/KeyedStoreExtensions.kt new file mode 100644 index 0000000..a4f7dff --- /dev/null +++ b/store/src/main/java/com/elveum/store/stores/keyed/KeyedStoreExtensions.kt @@ -0,0 +1,10 @@ +package com.elveum.store.stores.keyed + +/** + * Invalidate all keys in the [KeyedStore]. + */ +public fun KeyedStore.invalidateAllAsync() { + activeKeys.value.forEach { + invalidateAsync(it) + } +}