feat: add whenActive optional call to LazyFlowSubject and LazyCache

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:2.1.0-beta05"
+implementation "com.elveum:container:2.1.0-beta06"
 ```
 
 ## Core Concepts
@@ -231,9 +231,9 @@ per-item updates), see [Pagination](docs/paging.md).
 
 ## Detailed Documentation
 
-| Topic | Description |
-|-------|-------------|
-| [Container Type](docs/container-type.md) | States, value extraction, transformations, flow extensions, combining flows |
-| [Reducer Pattern](docs/reducer-pattern.md) | `Reducer`, `ContainerReducer`, combining flows, `ReducerOwner` |
-| [Subjects](docs/subjects.md) | `LazyFlowSubject`, metadata, source types |
-| [Pagination](docs/paging.md) | `PageLoader`, next-page states, pull-to-refresh, flow dependencies, per-item updates |
+| Topic                                      | Description                                                                          |
+|--------------------------------------------|--------------------------------------------------------------------------------------|
+| [Container Type](docs/container-type.md)   | States, value extraction, transformations, flow extensions, combining flows          |
+| [Reducer Pattern](docs/reducer-pattern.md) | `Reducer`, `ContainerReducer`, combining flows, `ReducerOwner`                       |
+| [Subjects](docs/subjects.md)               | `LazyFlowSubject`, metadata, source types                                            |
+| [Pagination](docs/paging.md)               | `PageLoader`, next-page states, pull-to-refresh, flow dependencies, per-item updates |

container/src/main/java/com/elveum/container/cache/LazyCache.kt

@@ -47,6 +47,11 @@ public interface LazyCache<Arg, T> {
         return getActiveCollectorsCount(arg) > 0
     }
 
+    /**
+     * Observe all active args being listened via StateFlow returned by [listen] call.
+     */
+    public fun listenActiveCollectorArgs(): StateFlow<Set<Arg>>
+
     /**
      * Get a flow for listening values. The load function is started automatically
      * when at least 1 collector starts collecting the flow. See [LazyFlowSubject]
@@ -85,6 +90,17 @@ public interface LazyCache<Arg, T> {
      */
     public fun reset()
 
+    /**
+     * Launch the [block] when at least one observer subscribes to
+     * a flow returned by [listen] call. The [block] is automatically
+     * cancelled when the last observer unsubscribes from the flow.
+     *
+     * @return this LazyCache instance.
+     */
+    public fun whenActive(
+        block: suspend ScopedLazyCache<Arg, T>.() -> Unit,
+    ): LazyCache<Arg, T>
+
     public companion object {
 
         /**

container/src/main/java/com/elveum/container/cache/LazyCacheImpl.kt

@@ -6,6 +6,7 @@ import com.elveum.container.Container
 import com.elveum.container.LoadConfig
 import com.elveum.container.factory.CoroutineScopeFactory
 import com.elveum.container.factory.DEFAULT_RELOAD_DEPENDENCIES_PERIOD_MILLIS
+import com.elveum.container.stateMap
 import com.elveum.container.subject.ContainerConfiguration
 import com.elveum.container.subject.LazyFlowSubject
 import com.elveum.container.subject.ValueLoader
@@ -17,9 +18,12 @@ import kotlinx.coroutines.cancel
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.FlowCollector
+import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.emptyFlow
+import kotlinx.coroutines.flow.update
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.supervisorScope
 
 internal class LazyCacheImpl<Arg, T>(
     private val cacheTimeoutMillis: Long,
@@ -32,10 +36,14 @@ internal class LazyCacheImpl<Arg, T>(
     )
 ) : LazyCache<Arg, T> {
 
-    private val cacheSlots = mutableMapOf<Arg, CacheRecord<T>>()
+    private val cacheSlotsFlow = MutableStateFlow<Map<Arg, CacheRecord<T>>>(emptyMap())
+    private val cacheSlots get() = cacheSlotsFlow.value
+
     private val totalCount: Int get() = cacheSlots.values.sumOf { it.count }
     private var scope: CoroutineScope? = null
 
+    private val whenActiveBlocks = mutableListOf<suspend ScopedLazyCache<Arg, T>.() -> Unit>()
+
     override fun listen(
         arg: Arg,
         configuration: ContainerConfiguration,
@@ -54,6 +62,10 @@ internal class LazyCacheImpl<Arg, T>(
         return getSubject(arg)?.activeCollectorsCount ?: 0
     }
 
+    override fun listenActiveCollectorArgs(): StateFlow<Set<Arg>> {
+        return cacheSlotsFlow.stateMap { it.keys }
+    }
+
     override fun reload(arg: Arg, config: LoadConfig): Flow<T> {
         return getSubject(arg)?.reload(config) ?: emptyFlow()
     }
@@ -64,28 +76,60 @@ internal class LazyCacheImpl<Arg, T>(
 
     override fun reset() = synchronized(this) {
         val iterator = cacheSlots.iterator()
+        val argsToRemove = mutableSetOf<Arg>()
         while (iterator.hasNext()) {
             val entry = iterator.next()
             if (entry.value.count == 0) {
-                iterator.remove()
+                argsToRemove += entry.key
             }
         }
+        if (argsToRemove.isNotEmpty()) {
+            cacheSlotsFlow.update { oldMap -> oldMap - argsToRemove }
+        }
         if (totalCount == 0) {
             scope?.cancel()
             scope = null
         }
     }
 
+    override fun whenActive(
+        block: suspend ScopedLazyCache<Arg, T>.() -> Unit,
+    ): LazyCache<Arg, T> = synchronized(this) {
+        whenActiveBlocks.add(block)
+        this
+    }
+
     private fun registerRecord(arg: Arg): CacheRecord<T> = synchronized(this) {
-        val record = cacheSlots.getOrPut(arg) {
+        val existingRecord = cacheSlots[arg]
+        val record = if (existingRecord == null) {
             val subject = subjectFactory.create(
                 valueLoader = valueLoaderFactory.create(arg)
             )
-            CacheRecord(subject)
+            CacheRecord(subject).also {
+                cacheSlotsFlow.update { oldMap ->
+                    oldMap + (arg to it)
+                }
+            }
+        } else {
+            existingRecord
         }
         record.count++
         if (totalCount == 1 && scope == null) {
-            scope = coroutineScopeFactory.createScope()
+            scope = coroutineScopeFactory
+                .createScope()
+                .also { scope ->
+                    whenActiveBlocks.forEach { block ->
+                        scope.launch {
+                            supervisorScope {
+                                val scopedCache = ScopedLazyCacheImpl(
+                                    lazyCache = this@LazyCacheImpl,
+                                    coroutineScope = this,
+                                )
+                                block.invoke(scopedCache)
+                            }
+                        }
+                    }
+                }
         }
         return record
     }
@@ -104,7 +148,7 @@ internal class LazyCacheImpl<Arg, T>(
             synchronized(this@LazyCacheImpl) {
                 val record = cacheSlots[arg]
                 if (record?.count == 0) {
-                    cacheSlots.remove(arg)
+                    cacheSlotsFlow.update { oldMap -> oldMap - arg }
                 }
                 if (totalCount == 0) {
                     scope?.cancel()

container/src/main/java/com/elveum/container/cache/ScopedLazyCache.kt

@@ -0,0 +1,5 @@
+package com.elveum.container.cache
+
+import kotlinx.coroutines.CoroutineScope
+
+public interface ScopedLazyCache<Arg, T> : LazyCache<Arg, T>, CoroutineScope

container/src/main/java/com/elveum/container/cache/ScopedLazyCacheImpl.kt

@@ -0,0 +1,10 @@
+package com.elveum.container.cache
+
+import kotlinx.coroutines.CoroutineScope
+
+internal class ScopedLazyCacheImpl<Arg, T>(
+    private val lazyCache: LazyCache<Arg, T>,
+    private val coroutineScope: CoroutineScope,
+) : ScopedLazyCache<Arg, T>,
+    LazyCache<Arg, T> by lazyCache,
+    CoroutineScope by coroutineScope

container/src/main/java/com/elveum/container/subject/LazyFlowSubject.kt

@@ -144,6 +144,17 @@ public interface LazyFlowSubject<T> {
         metadata: ContainerMetadata = EmptyMetadata,
     ): Flow<T>
 
+    /**
+     * Launch the [block] when at least one observer subscribes to
+     * a flow returned by [listen] call. The [block] is automatically
+     * cancelled when the last observer unsubscribes from the flow.
+     *
+     * @return this LazyFlowSubject instance.
+     */
+    public fun whenActive(
+        block: suspend ScopedLazyFlowSubject<T>.() -> Unit,
+    ): LazyFlowSubject<T>
+
     public companion object {
 
         public fun <T> create(

container/src/main/java/com/elveum/container/subject/LazyFlowSubjectImpl.kt

@@ -6,16 +6,17 @@ import com.elveum.container.BackgroundLoadState
 import com.elveum.container.Container
 import com.elveum.container.ContainerMetadata
 import com.elveum.container.EmptyReloadFunction
+import com.elveum.container.IsReloadDependenciesMetadata
+import com.elveum.container.LoadConfig
 import com.elveum.container.LoadTrigger
 import com.elveum.container.LoadTriggerMetadata
-import com.elveum.container.LoadConfig
-import com.elveum.container.IsReloadDependenciesMetadata
 import com.elveum.container.ReloadFunction
 import com.elveum.container.factory.CoroutineScopeFactory
 import com.elveum.container.factory.DEFAULT_RELOAD_DEPENDENCIES_PERIOD_MILLIS
 import com.elveum.container.internalDistinctUntilChanged
 import com.elveum.container.subject.lazy.LoadTask
 import com.elveum.container.subject.lazy.LoadTaskManager
+import com.elveum.container.subject.lazy.ScopedLazyFlowSubjectImpl
 import com.elveum.container.update
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.ExperimentalForInheritanceCoroutinesApi
@@ -30,6 +31,7 @@ import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.emptyFlow
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.supervisorScope
 
 internal class LazyFlowSubjectImpl<T>(
     private val coroutineScopeFactory: CoroutineScopeFactory,
@@ -49,6 +51,7 @@ internal class LazyFlowSubjectImpl<T>(
     private val collectorsCountFlow = MutableStateFlow(0)
     private var scope: CoroutineScope? = null
     private var cancellationJob: Job? = null
+    private val whenActiveBlocks = mutableListOf<suspend ScopedLazyFlowSubject<T>.() -> Unit>()
 
     private val reloadFunctionRef: ReloadFunction = ::reloadAsync
 
@@ -113,6 +116,13 @@ internal class LazyFlowSubjectImpl<T>(
         } ?: emptyFlow()
     }
 
+    override fun whenActive(
+        block: suspend ScopedLazyFlowSubject<T>.() -> Unit,
+    ): LazyFlowSubject<T> = synchronized(loadTaskManager) {
+        whenActiveBlocks.add(block)
+        this
+    }
+
     private fun doNewLoad(
         config: LoadConfig,
         valueLoader: ValueLoader<T>,
@@ -150,6 +160,17 @@ internal class LazyFlowSubjectImpl<T>(
                     scope = scope,
                     flowDependencyStore = flowDependencyStore,
                 )
+                whenActiveBlocks.forEach { block ->
+                    scope.launch {
+                        supervisorScope {
+                            val scopedSubject = ScopedLazyFlowSubjectImpl(
+                                coroutineScope = this,
+                                subject = this@LazyFlowSubjectImpl,
+                            )
+                            block(scopedSubject)
+                        }
+                    }
+                }
             }
     }
 

container/src/main/java/com/elveum/container/subject/ScopedLazyFlowSubject.kt

@@ -0,0 +1,5 @@
+package com.elveum.container.subject
+
+import kotlinx.coroutines.CoroutineScope
+
+public interface ScopedLazyFlowSubject<T> : LazyFlowSubject<T>, CoroutineScope

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

@@ -0,0 +1,12 @@
+package com.elveum.container.subject.lazy
+
+import com.elveum.container.subject.LazyFlowSubject
+import com.elveum.container.subject.ScopedLazyFlowSubject
+import kotlinx.coroutines.CoroutineScope
+
+internal class ScopedLazyFlowSubjectImpl<T>(
+    val coroutineScope: CoroutineScope,
+    val subject: LazyFlowSubject<T>,
+) : ScopedLazyFlowSubject<T>,
+    CoroutineScope by coroutineScope,
+    LazyFlowSubject<T> by subject

container/src/test/java/com/elveum/container/cache/LazyCacheIntegrationTest.kt

@@ -14,10 +14,15 @@ import io.mockk.coEvery
 import io.mockk.coVerify
 import io.mockk.mockk
 import io.mockk.verify
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.awaitCancellation
 import kotlinx.coroutines.delay
+import kotlinx.coroutines.launch
 import kotlinx.coroutines.test.TestScope
 import org.junit.Assert.assertEquals
 import org.junit.Assert.assertFalse
+import org.junit.Assert.assertNotNull
+import org.junit.Assert.assertNull
 import org.junit.Assert.assertSame
 import org.junit.Assert.assertTrue
 import org.junit.Before
@@ -479,6 +484,75 @@ class LazyCacheIntegrationTest {
         assertEquals(Container.Pending, lazyCache.get(key))
     }
 
+    @Test
+    fun whenActive_executesBlockOnLaunchAndCancelsAfterTimeout() = scope.runFlowTest {
+        var job: Job? = null
+        var rootJobRunning = false
+        lazyCache
+            .whenActive {
+                rootJobRunning = true
+                try {
+                    job = launch { awaitCancellation() }
+                    awaitCancellation()
+                } finally {
+                    rootJobRunning = false
+                }
+            }
+
+        // get flow does not execute whenActive block:
+        val flow = lazyCache.listen("1")
+        runCurrent()
+        assertNull(job)
+        assertFalse(rootJobRunning)
+
+        // start collecting -> executes whenActive block:
+        val collector = flow.startCollecting()
+        runCurrent()
+        assertNotNull(job)
+        assertTrue(job!!.isActive)
+        assertTrue(rootJobRunning)
+
+        // cancel collecting -> does not stop whenActive block before timeout:
+        collector.cancel()
+        advanceTimeBy(timeoutMillis - 1)
+        assertFalse(job.isCancelled)
+        assertTrue(rootJobRunning)
+
+        // after timeout - the block is cancelled:
+        advanceTimeBy(2)
+        assertTrue(job.isCancelled)
+        assertFalse(rootJobRunning)
+    }
+
+    @Test
+    fun whenActive_executedOncePerSession() = scope.runFlowTest {
+        var execCount = 0
+        lazyCache.whenActive { execCount++ }
+
+        // run 2 collectors:
+        val collector1 = lazyCache.listen("1").startCollecting()
+        runCurrent()
+        val collector2 = lazyCache.listen("2").startCollecting()
+        runCurrent()
+        // block executed only once:
+        assertEquals(1, execCount)
+
+        // cancel 1 collector, add third collector -> still 1 execution
+        collector1.cancel()
+        advanceTimeBy(timeoutMillis + 1)
+        val collector3 = lazyCache.listen("3").startCollecting()
+        runCurrent()
+        assertEquals(1, execCount)
+
+        // cancel all collectors, then add fresh collector -> 2nd execution
+        collector2.cancel()
+        collector3.cancel()
+        advanceTimeBy(timeoutMillis + 1)
+        lazyCache.listen("4").startCollecting()
+        runCurrent()
+        assertEquals(2, execCount)
+    }
+
     private fun mockLoaderReturningDifferentResults() {
         var count = 0
         coEvery { loader.invoke(any(), any()) } coAnswers {