feat: add spy methods for LazyFlowSubject

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-beta06"
+implementation "com.elveum:container:2.1.0-beta07"
 ```
 
 ## Core Concepts

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

@@ -50,7 +50,7 @@ public interface LazyCache<Arg, T> {
     /**
      * Observe all active args being listened via StateFlow returned by [listen] call.
      */
-    public fun listenActiveCollectorArgs(): StateFlow<Set<Arg>>
+    public fun spyOnArgs(): StateFlow<Set<Arg>>
 
     /**
      * Get a flow for listening values. The load function is started automatically

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

@@ -62,7 +62,7 @@ internal class LazyCacheImpl<Arg, T>(
         return getSubject(arg)?.activeCollectorsCount ?: 0
     }
 
-    override fun listenActiveCollectorArgs(): StateFlow<Set<Arg>> {
+    override fun spyOnArgs(): StateFlow<Set<Arg>> {
         return cacheSlotsFlow.stateMap { it.keys }
     }
 

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

@@ -42,6 +42,7 @@ import kotlinx.coroutines.flow.StateFlow
  * value is passed to a new subscriber and the load doesn't start from scratch.
  * Timeout is specified either by constructor or by [create] method.
  */
+@Suppress("ComplexInterface")
 public interface LazyFlowSubject<T> {
 
     /**
@@ -85,6 +86,15 @@ public interface LazyFlowSubject<T> {
         configuration: ContainerConfiguration = ContainerConfiguration(),
     ): StateFlow<Container<T>>
 
+    /**
+     * Spy on the subject's state. Subscribing to the flow returned by this
+     * call does not trigger data loading. If there is no real listeners
+     * subscribed via [listen] call, the returned flow emits Pending state.
+     */
+    public fun spy(
+        configuration: ContainerConfiguration = ContainerConfiguration(),
+    ): StateFlow<Container<T>>
+
     /**
      * Start a new load which will replace existing value in the flow
      * returned by [listen].
@@ -149,9 +159,16 @@ public interface LazyFlowSubject<T> {
      * a flow returned by [listen] call. The [block] is automatically
      * cancelled when the last observer unsubscribes from the flow.
      *
+     * @param spyMode Whether the `listen` call within the `whenActive { ... }` block
+     *     acts as `spy` call (enabled by default).
+     * @param block A suspending block of code with a CoroutineScope executed
+     *     only when the subject has at least 1 active subscriber. The scope
+     *     is cancelled when the last subscriber is unsubscribed (after cache timeout)
+     *
      * @return this LazyFlowSubject instance.
      */
     public fun whenActive(
+        spyMode: Boolean = true,
         block: suspend ScopedLazyFlowSubject<T>.() -> Unit,
     ): LazyFlowSubject<T>
 

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

@@ -14,6 +14,7 @@ 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.stateMap
 import com.elveum.container.subject.lazy.LoadTask
 import com.elveum.container.subject.lazy.LoadTaskManager
 import com.elveum.container.subject.lazy.ScopedLazyFlowSubjectImpl
@@ -51,7 +52,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 whenActiveRecords = mutableListOf<WhenActiveRecord>()
 
     private val reloadFunctionRef: ReloadFunction = ::reloadAsync
 
@@ -117,12 +118,22 @@ internal class LazyFlowSubjectImpl<T>(
     }
 
     override fun whenActive(
+        spyMode: Boolean,
         block: suspend ScopedLazyFlowSubject<T>.() -> Unit,
     ): LazyFlowSubject<T> = synchronized(loadTaskManager) {
-        whenActiveBlocks.add(block)
+        val newRecord = WhenActiveRecord(spyMode, block)
+        whenActiveRecords.add(newRecord)
         this
     }
 
+    override fun spy(
+        configuration: ContainerConfiguration,
+    ): StateFlow<Container<T>> {
+        return loadTaskManager
+            .listen()
+            .stateMap { it.applyConfiguration(configuration) }
+    }
+
     private fun doNewLoad(
         config: LoadConfig,
         valueLoader: ValueLoader<T>,
@@ -160,14 +171,15 @@ internal class LazyFlowSubjectImpl<T>(
                     scope = scope,
                     flowDependencyStore = flowDependencyStore,
                 )
-                whenActiveBlocks.forEach { block ->
+                whenActiveRecords.forEach { record ->
                     scope.launch {
                         supervisorScope {
                             val scopedSubject = ScopedLazyFlowSubjectImpl(
+                                spyMode = record.spyMode,
                                 coroutineScope = this,
                                 subject = this@LazyFlowSubjectImpl,
                             )
-                            block(scopedSubject)
+                            record.block(scopedSubject)
                         }
                     }
                 }
@@ -221,6 +233,11 @@ internal class LazyFlowSubjectImpl<T>(
         }
     }
 
+    private inner class WhenActiveRecord(
+        val spyMode: Boolean,
+        val block: suspend ScopedLazyFlowSubject<T>.() -> Unit,
+    )
+
     interface LoadTaskFactory {
 
         fun <T> create(

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

@@ -1,5 +1,17 @@
 package com.elveum.container.subject
 
+import com.elveum.container.Container
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.flow.StateFlow
 
-public interface ScopedLazyFlowSubject<T> : LazyFlowSubject<T>, CoroutineScope
+/**
+ * Scoped LazyFlowSubject accessible within [LazyFlowSubject.whenActive]
+ * call.
+ */
+public interface ScopedLazyFlowSubject<T> : LazyFlowSubject<T>, CoroutineScope {
+
+    override fun listen(
+        configuration: ContainerConfiguration,
+    ): StateFlow<Container<T>>
+
+}

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

@@ -1,12 +1,26 @@
 package com.elveum.container.subject.lazy
 
+import com.elveum.container.Container
+import com.elveum.container.subject.ContainerConfiguration
 import com.elveum.container.subject.LazyFlowSubject
 import com.elveum.container.subject.ScopedLazyFlowSubject
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.flow.StateFlow
 
 internal class ScopedLazyFlowSubjectImpl<T>(
+    val spyMode: Boolean,
     val coroutineScope: CoroutineScope,
     val subject: LazyFlowSubject<T>,
 ) : ScopedLazyFlowSubject<T>,
     CoroutineScope by coroutineScope,
-    LazyFlowSubject<T> by subject
+    LazyFlowSubject<T> by subject {
+
+    override fun listen(configuration: ContainerConfiguration): StateFlow<Container<T>> {
+        return if (spyMode) {
+            subject.spy(configuration)
+        } else {
+            subject.listen(configuration)
+        }
+    }
+
+}

container/src/test/java/com/elveum/container/subject/LazyFlowSubjectImplIntegrationTest.kt

@@ -941,6 +941,65 @@ class LazyFlowSubjectImplIntegrationTest {
         assertEquals(2, execCount)
     }
 
+    @Test
+    fun `spy without listeners emits Pending`() = runFlowTest {
+        val subject = createLazyFlowSubject {
+            delay(10)
+            emit("1")
+        }
+
+        val collector = subject.spy().startCollecting()
+        advanceTimeBy(11)
+
+        assertEquals(
+            listOf(pendingContainer()),
+            collector.collectedItems
+        )
+    }
+
+    @Test
+    fun `spy with listeners emits output value`() = runFlowTest {
+        val subject = createLazyFlowSubject {
+            delay(10)
+            emit("1")
+            delay(cacheTimeout + 1)
+            emit("2")
+        }
+
+        val spyCollector = subject.spy().startCollecting()
+        advanceTimeBy(5)
+        val realCollector = subject.listen().startCollecting()
+
+        // nothing is emitted while loading:
+        advanceTimeBy(10)
+        assertEquals(
+            listOf(pendingContainer()),
+            spyCollector.collectedItems
+        )
+
+        // success item is emitted when real collector is attached:
+        advanceTimeBy(1)
+        assertEquals(2, spyCollector.count)
+        assertEquals(successContainer("1"), spyCollector.lastItem.raw())
+
+        realCollector.cancel()
+
+        // spy collector is not detached before cache timeout expires:
+        advanceTimeBy(cacheTimeout)
+        assertEquals(2, spyCollector.count)
+
+        // spy collector is detached after cache timeout expires:
+        advanceTimeBy(1)
+        assertEquals(
+            listOf(
+                pendingContainer(),
+                successContainer("1"),
+                pendingContainer(),
+            ),
+            spyCollector.collectedItems.raw()
+        )
+    }
+
     private fun FlowTestScope.createLazyFlowSubject(
         loader: ValueLoader<String>? = null,
     ): LazyFlowSubjectImpl<String> = createLazyFlowSubject(

gradle-public.properties

@@ -6,7 +6,7 @@ android.nonTransitiveRClass=true
 
 # Maven Central publishing coordinates
 GROUP=com.elveum
-VERSION_NAME=2.1.0-beta06
+VERSION_NAME=2.1.0-beta07
 
 # POM metadata
 POM_DESCRIPTION=A library for simplifying state management and data loading in Android applications