diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/context/DependencyChangeListener.kt b/src/commonMain/kotlin/com/lightningkite/reactive/context/DependencyChangeListener.kt index 270d046..9d1c692 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/context/DependencyChangeListener.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/context/DependencyChangeListener.kt @@ -131,12 +131,20 @@ suspend fun Reactive.await(): T { } ?: awaitOnce() } +/** + * Reads the current value, waiting for one if there isn't one yet. + * + * A ready state is taken as-is, with no subscription: readiness means some other listener - or the + * source's own nature, as with a `Signal` - is keeping that value current. Only [ReactiveState.notActive] + * and notReady require listening, and the subscription is released as soon as a value arrives. + */ suspend fun Reactive.awaitOnce(): T { val state = state @Suppress("DEPRECATION") return if (state.ready) state.get() else suspendCancellableCoroutine { - // If it's not ready, we need to wait until it is then never bother with this again. + // Not ready, or nothing is maintaining a value: either way we have to listen, which is + // also what activates a lazy source so it calculates one. var remover: (() -> Unit)? = null var alreadyRun = false var done = false diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/context/DependencyTracker.kt b/src/commonMain/kotlin/com/lightningkite/reactive/context/DependencyTracker.kt index 7ae6bbf..fb743aa 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/context/DependencyTracker.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/context/DependencyTracker.kt @@ -6,14 +6,31 @@ abstract class DependencyTracker { protected val dependencyCount: Int get() = dependencies.size + /** + * Looks up a dependency already tracked for this run, marking it used. + * + * Fast path: if dependencies are read in the same order every run (the steady-state case), + * the dependency due to be read next always sits at [dependencies]`[usedDependencies.size]` - + * i.e. the slot immediately following the ones already marked used this run. Checking that one + * slot avoids the O(n) [dependencies].find fallback below, keeping a rerun over n dependencies + * O(n) instead of O(n^2). + * + * A dependency read more than once in the same run is only added to [usedDependencies] once, + * so the fast-path alignment above holds even when a call site re-reads an earlier dependency. + */ @Suppress("UNCHECKED_CAST") fun existingDependency(listenable: T): T? { - usedDependencies.add(listenable) - if (dependencies.size > usedDependencies.size) { - val maybe = dependencies[usedDependencies.size].first - if (maybe == listenable) return maybe as T + val index = usedDependencies.size + if (index < dependencies.size) { + val maybe = dependencies[index].first + if (maybe == listenable) { + usedDependencies.add(listenable) + return maybe as T + } } - return dependencies.find { it.first == listenable }?.first as? T + val found = dependencies.find { it.first == listenable }?.first as? T + if (listenable !in usedDependencies) usedDependencies.add(listenable) + return found } fun registerDependency(any: Any, remove: () -> Unit) { diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt b/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt index 176bcd5..34a8eae 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt @@ -132,15 +132,32 @@ typealias ReactiveContext = TypedReactiveContext<*> * context.startCalculation() * ``` * + * ### Self-Triggering Calculations + * + * A calculation that changes one of its own dependencies - most often by writing to a signal it + * also reads - asks to be rerun while it is still running. [reentrancyLimit] decides what happens: + * + * - `0` (the default) treats it as the mistake it usually is, reporting a + * [ReactiveReentrancyException] as this context's state. + * - A positive limit lets the calculation *settle*: it reruns in place, publishing each + * intermediate result, until it stops changing its own inputs or the limit is spent. Use this + * only for calculations that visibly converge, such as one that clamps a value into range. + * + * Either way the calculation never recurses, so a runaway cycle surfaces as a named error rather + * than a stack overflow (or, under a dispatching scheduler, a silent livelock). + * * @param T The type of value produced by the calculation. * @property scope The coroutine scope for calculations. * @property useLastWhileLoading Whether to preserve the last value during recalculation (true) or show loading state (false). + * @property reentrancyLimit How many times the calculation may trigger its own re-execution before + * it is reported as a [ReactiveReentrancyException]. Zero, the default, forbids self-triggering. * @property reportTo The underlying [RawReactive] to report state updates to. * @property action The calculation logic to execute in this context. */ class TypedReactiveContext( val scope: CoroutineScope, val useLastWhileLoading: Boolean = false, + val reentrancyLimit: Int = 0, private val reportTo: RawReactive = RawReactive(), val action: TypedReactiveContext.() -> T ) : DependencyChangeListener(), ReactiveCoroutineScope, Reactive by reportTo { @@ -156,15 +173,15 @@ class TypedReactiveContext( /** * Reference to [startCalculation] used as a listener callback. * Dependencies invoke this when they change to trigger recalculation. + * + * Invariant: a dependency registered *during* a run must not use this to deliver its initial + * value. It must hand that value to the run that registered it instead - see how the `async`, + * `Deferred.invoke` and `Flow.invoke` operators below arrange to do so. A value that first + * arrives through this path while the calculation is still running is indistinguishable from + * the calculation triggering itself, and so spends [reentrancyLimit]. */ val rerun: () -> Unit = ::startCalculation - /** - * Prevents multiple simultaneous recalculation requests from queueing up. - * Only the first request triggers [startCalculation], subsequent requests are ignored until it completes. - */ - private var queued = false - /** * The current job for this calculation run. * Gets cancelled and replaced with a new job on each [startCalculation] call. @@ -183,57 +200,80 @@ class TypedReactiveContext( */ override val coroutineContext: CoroutineContext get() = scope.coroutineContext + job + this + /** + * A recalculation has been requested but not yet performed. Set when a dependency changes and + * cleared as each run begins, so that several changes arriving before the run happens collapse + * into one, and so that a change arriving *during* a run is picked up by the settling loop in + * [startCalculation] rather than recursing. + */ + private var queued = false + + /** + * True while this context's [action] is executing. A recalculation requested while this is set + * came from the calculation itself - directly, by writing a signal it also reads, or by way of + * something it triggered. That is what [reentrancyLimit] governs. + */ + private var calculating = false + /** * Starts or restarts the reactive calculation. * - * This method: - * 1. Cancels the previous calculation's job (if any) - * 2. Creates a fresh job for this calculation run - * 3. Executes the calculation on the scope's thread - * 4. Tracks dependencies accessed during execution - * 5. Updates the reactive state with the result - * 6. Cleans up unused dependencies + * When no calculation is running, this cancels the previous run's coroutines, then runs the + * calculation on the scope's thread: dependencies accessed during [action] are tracked, the + * result is reported as this context's state, and dependencies that went unused are released. * - * If the calculation has no dependencies after completion, the context is automatically - * cancelled to release resources (since it will never rerun). - * - * Thread safety: Uses [queued] flag to prevent multiple simultaneous executions. + * When a calculation *is* running, the request is left for that run's settling loop instead of + * recursing into a new one. The calculation reruns in place until it stops re-triggering + * itself, publishing each intermediate result as it goes. [reentrancyLimit] bounds how many + * times it may do so; beyond that the context reports a [ReactiveReentrancyException] rather + * than recursing until the stack overflows, or livelocking under a dispatching scheduler. The + * default limit of zero forbids self-triggering outright. */ fun startCalculation() { active = true - if (queued) return // Prevent duplicate queuing + if (queued) return // Prevent duplicate queuing; the pending run will see the new state queued = true + // Requested from inside the calculation. The settling loop below picks this up on its next + // pass; we must not cancel `job` here, as it holds the coroutines of the run in progress. + if (calculating) return - // Cancel previous calculation and create fresh job - job.cancel() - job = Job() + // The previous run's inputs have changed, so anything it launched is now producing stale + // results. Cancel at the moment of the change rather than when the rerun is dispatched. + endRun() scope.onThread { - queued = false if (!active) return@onThread // Check if cancelled while queued - dependencyBlockStart() // Begin tracking dependencies - val state = reactiveState { action(this@TypedReactiveContext) } - - // Update state unless useLastWhileLoading is true and result isn't ready - if (!useLastWhileLoading || state.ready) reportTo.state = state - - dependencyBlockEnd() // Clean up dependencies not used in this run + var runs = 0 + while (queued) { + queued = false + if (runs++ > reentrancyLimit) { + reportTo.state = + ReactiveState.exception(ReactiveReentrancyException(this, reentrancyLimit)) + break + } + try { + calculating = true + dependencyBlockStart() // Begin tracking dependencies + val state = reactiveState { action(this@TypedReactiveContext) } + + // Update state unless useLastWhileLoading is true and result isn't ready + if (!useLastWhileLoading || state.ready) reportTo.state = state + } finally { + dependencyBlockEnd() // Clean up dependencies not used in this run + calculating = false + } + // The calculation re-triggered itself, so this run's coroutines are stale for the + // same reason any other superseded run's are. + if (queued) endRun() + } } } - /** - * Runs the calculation once without activating the context or tracking dependencies. - * - * This is useful for getting an initial value or running the calculation in a test - * environment without setting up the full reactive machinery. - * - * The result is still reported to [reportTo], but no listeners are registered and - * the calculation will not rerun when dependencies change. - */ - fun runOnceWhileDead() { - val state = reactiveState { action(this) } - if (!useLastWhileLoading || state.ready) reportTo.state = state + /** Cancels the current run's coroutines and opens a fresh job for the next run. */ + private fun endRun() { + job.cancel() + job = Job() } init { @@ -269,8 +309,7 @@ class TypedReactiveContext( * After cancellation, the context will not respond to dependency changes. */ override fun cancel() { - job.cancel() - job = Job() + endRun() active = false queued = false super.cancel() // Cancel dependency listeners @@ -314,6 +353,11 @@ class TypedReactiveContext( * if the state is not ready. * * This allows the reactive calculation to be paused until dependencies become ready. + * + * A notActive dependency counts as loading rather than as an error: every operator here + * registers its listener before reading, so the dependency is being maintained by the time we + * see it, and any notActive we do observe is a source part-way through activating. The + * listener is in place, so the calculation reruns when its value arrives. */ private fun ReactiveState.getOrLoading() = handle( @@ -413,7 +457,7 @@ class TypedReactiveContext( * @param get Function to extract a value from the [ReactiveState] * @return The transformed value */ - inline fun Reactive.state(crossinline get: (ReactiveState) -> V): V { + fun Reactive.state(get: (ReactiveState) -> V): V { var current: V = state.let(get) if (existingDependency(this) == null) { registerDependency(this, addListener { @@ -424,6 +468,9 @@ class TypedReactiveContext( rerun() } }) + // Repull in case of activation: a lazy source calculates inside addListener, before + // our listener is in place, so the value read above can predate that calculation. + current = state.let(get) } return current } @@ -432,11 +479,13 @@ class TypedReactiveContext( * Wrapper class for tracking "once" dependencies. * This creates a unique key for each reactive value accessed via [once]. */ - private data class Once(val wraps: Reactive) + private data class Once(val wraps: Reactive) { + /** Whether a value has been obtained; changes stop causing reruns once one has. */ + var have = false + } /** - * Accesses the value of this [Reactive] once, automatically removing the dependency - * after the value becomes ready. + * Accesses the value of this [Reactive] once, without rerunning when it later changes. * * This is useful when you want to wait for an initial value but don't want subsequent * changes to trigger reruns. @@ -453,23 +502,12 @@ class TypedReactiveContext( * @throws ReactiveLoading if the value is not ready yet */ fun Reactive.once(): T { - return state.handle( - success = { it }, - exception = { throw it }, - notReady = { - val key = Once(this) - if (existingDependency(key) == null) { - // Register a one-shot listener that removes itself after firing - var remover: () -> Unit = {} - remover = addListener { - remover() // Remove the listener - rerun() - } - registerDependency(key, remover) - } - throw ReactiveLoading - } - ) + val key = existingDependency(Once(this)) ?: Once(this).also { key -> + registerDependency(key, addListener { if (!key.have) rerun() }) + } + val state = state + key.have = state.ready + return state.getOrLoading() } // Hack: fixes compiler weirdness around lambdas with 'this' @@ -524,13 +562,17 @@ class TypedReactiveContext( * } * ``` * + * @param identity The identifier for uniqueness on this item. * @param dependencies Values that uniquely identify this calculation (changes trigger recalculation) * @param action The suspending function to execute * @return The result of the calculation once complete * @throws ReactiveLoading if the calculation is not yet complete */ - fun async(vararg dependencies: Any?, action: suspend () -> T): T { - val key = setOf(*dependencies) + fun async(identity: String, vararg dependencies: Any?, action: suspend () -> T): T { + // action::class distinguishes call sites: two different `async(...) { }` call sites compile + // to distinct anonymous classes, so folding it into the key stops two call sites that + // happen to pass identical dependencies from colliding on the same cached calculation. + val key = setOf(identity, *dependencies) val calc = SuspendCalculation(key) // Reuse existing calculation if already running @@ -538,11 +580,19 @@ class TypedReactiveContext( return it.state.getOrLoading() } - // Launch new calculation - scope.launch { + // Launch new calculation and register as dependency. The launch happens before the + // listener is attached so a synchronously-completing action (e.g. under an Unconfined + // dispatcher) can't re-enter rerun() while this calculation is still in progress. + val job = scope.launch { calc.state = reactiveState { action() } } - registerDependency(calc, calc.addListener(rerun)) + val removeListener = calc.addListener(rerun) + registerDependency(calc) { + // A rerun means this calculation is no longer used - cancel it so stale work + // doesn't keep running and overwrite `calc` after it's been orphaned. + removeListener() + job.cancel() + } return calc.state.getOrLoading() } @@ -572,11 +622,17 @@ class TypedReactiveContext( return it.invoke() } - // Launch await operation - scope.launch { + // Launch await operation and register as dependency. Same ordering and cancellation + // rationale as `async` above: launch before attaching the listener, and cancel the + // launch when this dependency is no longer used. + val job = scope.launch { calc.state = reactiveState { this@invoke.await() } } - registerDependency(calc, calc.addListener(rerun)) + val removeListener = calc.addListener(rerun) + registerDependency(calc) { + removeListener() + job.cancel() + } return calc.state.getOrLoading() } @@ -624,21 +680,29 @@ class TypedReactiveContext( // Register cleanup to cancel collection when dependency is removed registerDependency(new, { job?.cancel() }) - // Start collecting the flow + // An emission that arrives while we are still registering (a hot flow under an + // undispatched scheduler collects synchronously) belongs to this run, so it is + // delivered by the return below rather than through rerun(). A dependency's initial + // value is not the calculation triggering itself, and pushing it through rerun() + // makes it indistinguishable from reentrancy. Same rationale as the + // launch-before-listen ordering in `async` above. + var registering = true job = scope.launch { collect { v -> try { new.state = ReactiveState(v) - rerun() // Trigger recalculation on each emission + if (!registering) rerun() // Later emissions trigger recalculation } catch (e: Exception) { new.state = ReactiveState.exception(e) } } } + registering = false // StateFlow always has a current value available immediately if (this is StateFlow) return this.value - else throw ReactiveLoading + // A flow that emitted during registration already has a value for this run + else return new.state.getOrLoading() } else { return existing.state.handle( success = { it }, @@ -700,6 +764,10 @@ class TypedReactiveContext( * When the outer signal changes, both the outer and inner contexts are cancelled and recreated. * When only the inner signal changes, only the inner context reruns. * + * @param reentrancyLimit How many times [action] may trigger its own re-execution - by changing a + * dependency it also reads - before the calculation is reported as failed with a + * [ReactiveReentrancyException]. Leave at zero unless the calculation is meant to settle over + * several runs; see [TypedReactiveContext] for the trade-off. * @param action The calculation logic to run reactively * @return A [TypedReactiveContext] managing the calculation and its dependencies * @@ -707,8 +775,8 @@ class TypedReactiveContext( * @see TypedReactiveContext for implementation details * @see reactiveSuspending for suspending calculations */ -fun CoroutineScope.reactive(action: ReactiveContext.() -> T): TypedReactiveContext { - val trc = TypedReactiveContext(this, action = action) +fun CoroutineScope.reactive(reentrancyLimit: Int = 0, action: ReactiveContext.() -> T): TypedReactiveContext { + val trc = TypedReactiveContext(this, reentrancyLimit = reentrancyLimit, action = action) trc.startCalculation() coroutineContext[StatusListener]?.watchBackgroundProcess(trc) return trc @@ -740,9 +808,9 @@ inline fun CoroutineScope.reactive(crossinline onLoad: () -> Unit, crossinline a action(this) wasLoadingLastTime = false } catch (e: ReactiveLoading) { - if (wasLoadingLastTime) { - onLoad() + if (!wasLoadingLastTime) { wasLoadingLastTime = true + onLoad() } throw e } catch (e: Exception) { @@ -775,3 +843,21 @@ inline fun CoroutineScope.reactiveScope(crossinline onLoad: () -> Unit, crossinl @InternalReactiveApi object ReactiveLoading : Throwable() + +/** + * Thrown when a reactive calculation triggers its own re-execution, typically by writing to a + * signal it also reads inside the same [reactive] block. This would otherwise recurse until the + * stack overflows (or livelock under a dispatching scheduler), so it is surfaced as a clear error. + */ +class ReactiveReentrancyException(context: ReactiveContext, reentrancyLimit: Int) : IllegalStateException( + if (reentrancyLimit <= 0) + "A reactive calculation triggered its own re-execution ($context). This usually means the " + + "calculation wrote to a signal it also reads. Break the cycle so the calculation " + + "does not mutate its own dependencies. If it is instead meant to settle over " + + "several runs, raise reentrancyLimit to the number of reruns it needs." + else + "A reactive calculation kept triggering its own re-execution and did not settle within " + + "its reentrancyLimit of $reentrancyLimit reruns ($context). Either it never " + + "converges - every run changes a signal it reads, so there is no stable value to " + + "reach - or it genuinely needs a higher limit." +) diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContextSuspending.kt b/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContextSuspending.kt index 7398d74..a6e42ab 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContextSuspending.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContextSuspending.kt @@ -160,9 +160,6 @@ class ReactiveContextSuspending( * * The calculation may complete synchronously (if already on correct dispatcher and no suspension points) * or asynchronously. If [useLastWhileLoading] is false, the state is set to notReady during async execution. - * - * If the calculation has no dependencies after completion, the context is automatically - * cancelled to release resources (since it will never rerun). */ fun startCalculation() { active = true @@ -194,33 +191,6 @@ class ReactiveContextSuspending( } } - /** - * Runs the calculation once without activating the context or tracking dependencies. - * - * This is useful for getting an initial value or running the calculation in a test - * environment without setting up the full reactive machinery. - * - * The result is still reported to [reportTo], but no dependency tracking occurs and - * the calculation will not rerun when dependencies change. - */ - fun runOnceWhileDead() { - lastLoopJob = run { - var done = false - val job = scope.launchWithStart { - val result = reactiveState { this@ReactiveContextSuspending.action() } - if (!useLastWhileLoading || result.ready) reportTo.state = result - done = true - } - - // Check if calculation completed synchronously - if (done) null - else { - if (!useLastWhileLoading) reportTo.state = ReactiveState.notReady - job - } - } - } - /** * Called by the dependency tracker when a dependency changes. * Triggers a recalculation with [startCalculation]. diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/core/MutableRemember.kt b/src/commonMain/kotlin/com/lightningkite/reactive/core/MutableRemember.kt index c541a2c..f570fb6 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/core/MutableRemember.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/core/MutableRemember.kt @@ -14,7 +14,7 @@ import kotlin.time.Duration * is manually set it will stop the calculation and instead behave like a [Signal]. * * Note: - * - `mutableRemember` is lazy: if it has no listeners, it will not calculate a value. + * - `mutableRemember` is lazy: if it has no listeners, it will not calculate a value. Reading its state while it has none, and has not been set, reports [ReactiveState.Companion.notActive]. * - Listeners are only notified if the calculated or set value changes. * * Example: @@ -46,7 +46,7 @@ fun mutableRemember( * When overridden by direct assignment, automatic calculation is paused until `reset()` is called. * * Note: - * - `MutableRemember` is lazy: if it has no listeners, it will not calculate a value. + * - `MutableRemember` is lazy: if it has no listeners, it will not calculate a value. Reading its state while it has none, and has not been set, reports [ReactiveState.Companion.notActive]. * - Listeners are only notified if the calculated or set value changes. * - The `reset()` method restores automatic calculation and updates the value from dependencies. * @@ -67,19 +67,15 @@ class MutableRemember( var overridden: Boolean = false private set - private val remember = Remember(coroutineContext, useLastWhileLoading, deactivationDelay, initialValue) + private val remember = Remember(coroutineContext, useLastWhileLoading, deactivationDelay, action = initialValue) private var forget: (()->Unit)? = null - private fun updateOnce() { - val currentRememberedState = remember.state - if(!overridden && (!useLastWhileLoading || currentRememberedState.ready)) state = currentRememberedState - } - private fun startListening() { forget = remember.addListener { if (!overridden) state = remember.state } - updateOnce() + val currentRememberedState = remember.state + if (!overridden && (!useLastWhileLoading || currentRememberedState.ready)) state = currentRememberedState } private fun stopListening() { forget?.invoke() @@ -88,7 +84,7 @@ class MutableRemember( override var state: ReactiveState get() { - if (!overridden && forget == null) updateOnce() + if (!overridden && forget == null) return ReactiveState.notActive return super.state } set(value) { diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/core/MutableRememberSuspending.kt b/src/commonMain/kotlin/com/lightningkite/reactive/core/MutableRememberSuspending.kt index 15f0629..d837487 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/core/MutableRememberSuspending.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/core/MutableRememberSuspending.kt @@ -14,7 +14,7 @@ import kotlin.time.Duration * is manually set it will stop the calculation and instead behave like a [Signal]. * * Note: - * - `mutableRememberSuspending` is lazy: if it has no listeners, it will not calculate a value. + * - `mutableRememberSuspending` is lazy: if it has no listeners, it will not calculate a value. Reading its state while it has none, and has not been set, reports [ReactiveState.Companion.notActive]. * - Listeners are only notified if the calculated or set value changes. * * Example: @@ -47,7 +47,7 @@ fun mutableRememberSuspending( * When overridden by direct assignment, automatic calculation is paused until `reset()` is called. * * Note: - * - lazy: if this has no listeners, it will not calculate a value. + * - lazy: if this has no listeners, it will not calculate a value. Reading its state while it has none, and has not been set, reports [ReactiveState.Companion.notActive]. * - Listeners are only notified if the calculated or set value changes. I.e., if '1' is calculated, and then '1' is set, it will not notify listeners. * - The `reset()` method restores automatic calculation and updates the value from dependencies. * @@ -90,7 +90,7 @@ class MutableRememberSuspending( override var state: ReactiveState get() { - if (!overridden && forget == null) updateOnce() + if (!overridden && forget == null) return ReactiveState.notActive return super.state } set(value) { diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/core/ReactiveState.kt b/src/commonMain/kotlin/com/lightningkite/reactive/core/ReactiveState.kt index 8f3212d..da8ad4a 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/core/ReactiveState.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/core/ReactiveState.kt @@ -7,16 +7,34 @@ import kotlin.jvm.JvmInline /** * Represents the state of a reactive value, including loading, success, and error conditions. * - * - A [ReactiveState] can hold a ready value, a loading state, or an error state. + * - A [ReactiveState] can hold a ready value, a loading state, an error state, or [notActive]. * - Use [ready] to check if the value is available, [success] to check if it is available and not an error, and [exception] to retrieve any error. * - Listeners of [Reactive] are only notified when the [ReactiveState] changes. * - [ReactiveState] provides methods for safely handling, mapping, and retrieving the underlying value. + * + * ### notReady versus notActive + * + * Both mean "no value", for different reasons, and both make [ready] false: + * + * - [Companion.notReady]: something is maintaining this value, and it does not have one right now - + * it is loading, or one of its dependencies is not ready. + * - [Companion.notActive]: *nothing* is maintaining this value. Lazy reactives like `remember` only + * calculate while they have listeners, so with none they cannot vouch for any value. Sources whose + * value is accurate whether or not anyone is listening - `Signal`, `Constant` - never report it. + * + * The distinction is what makes it safe to read a value without subscribing: anything other than + * [notActive] is current, no matter who is (or isn't) listening. On [notActive] you must subscribe - + * see `awaitOnce` - or accept that there is no value to be had. */ @JvmInline @OptIn(InternalReactiveApi::class) value class ReactiveState(val raw: T) { - inline val ready: Boolean get() = raw !is InternalReactiveNotReady + inline val ready: Boolean get() = raw != InternalReactiveNotReady && raw != InternalReactiveNotActive inline val success: Boolean get() = ready && raw !is InternalReactiveThrownException + + /** True when nothing is maintaining this value; see the [ReactiveState] docs. */ + inline val notActive: Boolean get() = raw is InternalReactiveNotActive + inline fun onSuccess(action: (T)->R): R? = handle( success = { action(it) }, exception = { null }, @@ -28,7 +46,8 @@ value class ReactiveState(val raw: T) { fun get(): T = handle( success = { it }, exception = { throw it }, - notReady = { throw NotReadyException() } + notReady = { throw NotReadyException() }, + notActive = { throw NotActiveException() } ) fun getOrNull(): T? = handle( @@ -40,6 +59,11 @@ value class ReactiveState(val raw: T) { companion object Companion { @Suppress("UNCHECKED_CAST") val notReady: ReactiveState = ReactiveState(InternalReactiveNotReady) as ReactiveState + + /** No value, because nothing is maintaining one; see the [ReactiveState] docs. */ + @Suppress("UNCHECKED_CAST") + val notActive: ReactiveState = ReactiveState(InternalReactiveNotActive) as ReactiveState + @Suppress("UNCHECKED_CAST") fun exception(exception: Exception) = (if(exception is CancellationException) notReady else ReactiveState(InternalReactiveThrownException(exception))) as ReactiveState @Suppress("UNCHECKED_CAST") @@ -47,7 +71,9 @@ value class ReactiveState(val raw: T) { } @Suppress("UNCHECKED_CAST") inline fun map(mapper: (T)->B): ReactiveState { - if(raw is InternalReactiveNotReady || raw is InternalReactiveThrownException) return this as ReactiveState + // notActive propagates like the other valueless states: a value derived from a source + // nobody is maintaining is equally unmaintained. + if(raw is InternalReactiveNotReady || raw is InternalReactiveNotActive || raw is InternalReactiveThrownException) return this as ReactiveState if(raw is InternalReactiveWrapper<*>) try { return ReactiveState(mapper(raw.other as T)) } catch(e: Exception) { @@ -59,24 +85,43 @@ value class ReactiveState(val raw: T) { exception(e) } } - @Suppress("UNCHECKED_CAST") + /** + * Handles [Companion.notActive] the same way as [notReady], because "nobody is maintaining a + * value" and "there is no value yet" are the same thing to code that only wants to display or + * wait for one. Use the four-argument overload to do something better, such as subscribing. + */ inline fun handle( success: (T)->R, exception: (Exception)->R, notReady: ()->R + ): R = handle(success, exception, notReady, notReady) + + @Suppress("UNCHECKED_CAST") + inline fun handle( + success: (T)->R, + exception: (Exception)->R, + notReady: ()->R, + notActive: ()->R ): R { return when(raw) { InternalReactiveNotReady -> notReady() + InternalReactiveNotActive -> notActive() is InternalReactiveThrownException -> exception(raw.exception) is InternalReactiveWrapper<*> -> success(raw.other as T) else -> success(raw) } } - fun asResult(): Result = handle(success = { Result.success(it) }, exception = { Result.failure(it) }, notReady = { Result.failure(NotReadyException()) }) + fun asResult(): Result = handle( + success = { Result.success(it) }, + exception = { Result.failure(it) }, + notReady = { Result.failure(NotReadyException()) }, + notActive = { Result.failure(NotActiveException()) } + ) override fun toString(): String = when(raw) { is InternalReactiveNotReady -> "NotReady" + is InternalReactiveNotActive -> "NotActive" is InternalReactiveThrownException -> "ThrownException(${raw.exception})" is InternalReactiveWrapper<*> -> "ReadyW($raw)" else -> "Ready($raw)" @@ -88,15 +133,27 @@ data class InternalReactiveWrapper(val other: T) data class InternalReactiveThrownException(val exception: Exception) @InternalReactiveApi object InternalReactiveNotReady +@InternalReactiveApi +object InternalReactiveNotActive -class NotReadyException(message: String? = null) : IllegalStateException(message) +open class NotReadyException(message: String? = null) : IllegalStateException(message) + +/** + * Thrown when reading a value from a reactive that nothing is listening to, and which therefore + * has no value to give. Subscribe to it first, or use `awaitOnce`, which subscribes for as long + * as it takes to obtain a value. + */ +class NotActiveException(message: String = "Nothing is listening to this reactive value, so it has no value to report. Subscribe to it, or use awaitOnce.") : NotReadyException(message) inline fun reactiveState(action: () -> T): ReactiveState { @OptIn(InternalReactiveApi::class) return try { ReactiveState(action()) - } catch (_: CancellationException) { - ReactiveState.notReady + } catch (e: CancellationException) { + // A cancellation means the coroutine running `action` was torn down mid-calculation - it + // must propagate so the caller's suspension point actually stops, rather than being + // reinterpreted as "not ready" and letting the calculation resume past its cancellation. + throw e } catch (_: ReactiveLoading) { ReactiveState.notReady } catch (e: Exception) { diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/core/Remember.kt b/src/commonMain/kotlin/com/lightningkite/reactive/core/Remember.kt index ecbe95a..d0134a5 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/core/Remember.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/core/Remember.kt @@ -14,12 +14,14 @@ import kotlin.time.Duration * The calculation runs in a coroutine, and will re-run whenever any reactive value it depends on changes. * * Note: - * - `remember` is lazy: if it has no listeners, it will not calculate a value. + * - `remember` is lazy: if it has no listeners, it will not calculate a value. Reading its state while it has none reports [ReactiveState.Companion.notActive]. * - Listeners are only notified if the calculated value changes (i.e., if the new value is different from the previous value). * * @param coroutineContext The coroutine context for running the calculation (default: Dispatchers.Unconfined). * @param useLastWhileLoading If true, uses the last known value while recalculating. * @param deactivationDelay If provided, the reactive context will be kept alive for that duration after all listeners have unsubscribed. + * @param reentrancyLimit How many times [action] may trigger its own re-execution before the value + * is reported as failed; see [TypedReactiveContext]. * @param action The block to compute the value reactively. * @return A [Reactive] value that updates automatically. * @@ -41,9 +43,10 @@ fun remember( coroutineContext: CoroutineContext = Dispatchers.Unconfined, useLastWhileLoading: Boolean = false, deactivationDelay: Duration? = null, + reentrancyLimit: Int = 0, action: ReactiveContext.() -> T, ): Reactive = - Remember(coroutineContext, useLastWhileLoading, deactivationDelay, action) + Remember(coroutineContext, useLastWhileLoading, deactivationDelay, reentrancyLimit, action) /** * A reactive value that remembers the result of a calculation and shares the result among its listeners. @@ -54,13 +57,15 @@ fun remember( * any of its dependencies change, and listeners are notified accordingly. * * Note: - * - `Remember` is lazy: if it has no listeners, it will not calculate a value. + * - `Remember` is lazy: if it has no listeners, it will not calculate a value. Reading its state while it has none reports [ReactiveState.Companion.notActive]. * - Listeners are only notified if the calculated value changes (i.e., if the new value is different from the previous value). * * @param T The type of value produced by the calculation. * @param coroutineContext The coroutine context in which the calculation runs. Defaults to [Dispatchers.Unconfined]. * @param useLastWhileLoading If true, the last known value will be used while the calculation is loading or re-running. * @param deactivationDelay If provided, the reactive context will be kept alive for that duration after all listeners have unsubscribed. + * @param reentrancyLimit How many times [action] may trigger its own re-execution before the value + * is reported as failed; see [TypedReactiveContext]. * @param action The block of code to execute within the [ReactiveContext] to produce the value. * * This class manages its own coroutine job and calculation scope. When activated, it starts the calculation @@ -71,8 +76,9 @@ fun remember( */ class Remember( val incomingCoroutineContext: CoroutineContext = Dispatchers.Unconfined, - useLastWhileLoading: Boolean = false, + private val useLastWhileLoading: Boolean = false, private val deactivationDelay: Duration? = null, + private val reentrancyLimit: Int = 0, private val action: ReactiveContext.() -> T, ) : Reactive, CoroutineScope, BaseListenable() { @@ -92,13 +98,23 @@ class Remember( // Remember/shared reactive graph forever. Matches RememberSuspending's ordering. override val coroutineContext get() = restOfContext + job - private val scope = TypedReactiveContext(this, useLastWhileLoading, action = action) + // Starts notActive rather than notReady: nothing is listening yet, so there is no value to + // be had, as opposed to one that is on its way. + private val reported = RawReactive(ReactiveState.notActive) + private val scope = TypedReactiveContext( + scope = this, + useLastWhileLoading = useLastWhileLoading, + reentrancyLimit = reentrancyLimit, + reportTo = reported, + action = action, + ) - override val state: ReactiveState - get() { - if (!scope.active) scope.runOnceWhileDead() - return scope.state - } + // A Remember only calculates while it has listeners, and reports notActive when it has none. + // It deliberately does not calculate on demand: doing so would either subscribe to sources + // nobody is listening to, or run a dependency-less calculation whose result is silently never + // updated. To read one imperatively, use awaitOnce - it subscribes for as long as it takes to + // get a value. + override val state: ReactiveState get() = reported.state private var deactivating: Job? = null private var remover: (() -> Unit)? = null @@ -111,6 +127,11 @@ class Remember( return } + // Something is maintaining this value again - it just doesn't have one yet. Without this, + // useLastWhileLoading would suppress the notReady the first calculation reports and leave + // notActive in place, claiming nobody is listening when somebody now is. + if (reported.state.notActive && !useLastWhileLoading) reported.state = ReactiveState.notReady + shuttingDown?.let { CoroutineScope(incomingCoroutineContext).launch { it.join() @@ -130,6 +151,9 @@ class Remember( job.cancel() job = SupervisorJob() shuttingDown = null + // The calculation is stopped, so the value it produced is no longer maintained. Reporting + // notActive says exactly that, rather than passing off a value that may have gone stale. + reported.state = ReactiveState.notActive } override fun deactivate() { diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/core/RememberSuspending.kt b/src/commonMain/kotlin/com/lightningkite/reactive/core/RememberSuspending.kt index 04a7220..0d0e43b 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/core/RememberSuspending.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/core/RememberSuspending.kt @@ -14,7 +14,7 @@ import kotlin.time.Duration * The calculation runs in a coroutine, and will re-run whenever any reactive value it depends on changes. * * Note: - * - `remember` is lazy: if it has no listeners, it will not calculate a value. + * - `remember` is lazy: if it has no listeners, it will not calculate a value. Reading its state while it has none reports [ReactiveState.Companion.notActive]. * - Listeners are only notified if the calculated value changes (i.e., if the new value is different from the previous value). * * @param coroutineContext The coroutine context for running the calculation (default: Dispatchers.Unconfined). @@ -56,7 +56,7 @@ fun rememberSuspending( * any of its dependencies change, and listeners are notified accordingly. * * Note: - * - `Remember` is lazy: if it has no listeners, it will not calculate a value. + * - `Remember` is lazy: if it has no listeners, it will not calculate a value. Reading its state while it has none reports [ReactiveState.Companion.notActive]. * - Listeners are only notified if the calculated value changes (i.e., if the new value is different from the previous value). * * @param T The type of value produced by the calculation. @@ -74,7 +74,7 @@ fun rememberSuspending( */ class RememberSuspending( val incomingCoroutineContext: CoroutineContext = Dispatchers.Unconfined, - useLastWhileLoading: Boolean = false, + private val useLastWhileLoading: Boolean = false, private val deactivationDelay: Duration? = null, private val action: suspend ReactiveCoroutineScope.() -> T, ) : Reactive, CoroutineScope, BaseListenable() { @@ -89,13 +89,12 @@ class RememberSuspending( override val coroutineContext: CoroutineContext get() = restOfContext + job - private val scope = ReactiveContextSuspending(this, useLastWhileLoading, action = action) + // See [Remember.reported]: nothing is listening yet, so there is no value to be had. + private val reported = RawReactive(ReactiveState.notActive) + private val scope = ReactiveContextSuspending(this, useLastWhileLoading, reported, action) - override val state: ReactiveState - get() { - if (!scope.active) scope.runOnceWhileDead() - return scope.state - } + // See [Remember.state]: no listeners means no calculation, and so notActive. + override val state: ReactiveState get() = reported.state private var deactivating: Job? = null private var remover: (() -> Unit)? = null @@ -108,6 +107,9 @@ class RememberSuspending( return } + // See [Remember.activate]: listening again means active-but-valueless, not notActive. + if (reported.state.notActive) reported.state = ReactiveState.notReady + shuttingDown?.let { CoroutineScope(incomingCoroutineContext).launch { it.join() @@ -127,6 +129,8 @@ class RememberSuspending( job.cancel() job = SupervisorJob() shuttingDown = null + // See [Remember.shutdown]: a stopped calculation maintains no value. + reported.state = ReactiveState.notActive } override fun deactivate() { diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/core/abstracts.kt b/src/commonMain/kotlin/com/lightningkite/reactive/core/abstracts.kt index 8e6742d..e02a11b 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/core/abstracts.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/core/abstracts.kt @@ -36,6 +36,13 @@ abstract class BaseListenable : Listenable { protected open fun deactivate() {} private val listeners = ArrayList<() -> Unit>() + + /** + * Number of currently-registered listeners. Exposed for tests that need to assert + * that a listenable is not leaking subscriptions. + */ + val listenerCount: Int get() = listeners.size + override fun addListener(listener: () -> Unit): Release { if (listeners.isEmpty()) activate() listeners.add(listener) diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/core/coreInterfaces.kt b/src/commonMain/kotlin/com/lightningkite/reactive/core/coreInterfaces.kt index ed1c2e5..2af73fc 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/core/coreInterfaces.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/core/coreInterfaces.kt @@ -92,6 +92,14 @@ fun Listenable.addAndRunListener(listener: () -> Unit): Release { * @see com.lightningkite.reactive.context.ReactiveContext */ interface Reactive : Listenable { + /** + * The current state. + * + * Implementations that only maintain a value while they have listeners - `remember` and + * friends - must report [ReactiveState.Companion.notActive] when they have none, rather than + * a value they are no longer keeping up to date. Anything else is a promise that the state is + * current, which is what lets callers read it without subscribing. + */ val state: ReactiveState object Never: Reactive { @@ -196,6 +204,9 @@ interface MutableReactive : Reactive, Mutable { */ interface ReactiveValue : Reactive, ReadOnlyProperty { val value: T + + // A ReactiveValue always has a value, so its state can never be notReady or notActive. That + // makes it the wrong interface for anything that only maintains a value while listened to. override val state: ReactiveState get() = ReactiveState(value) override fun getValue(thisRef: Any?, property: KProperty<*>): T = value } diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/core/processes.kt b/src/commonMain/kotlin/com/lightningkite/reactive/core/processes.kt index 575cd39..4891cef 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/core/processes.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/core/processes.kt @@ -23,8 +23,10 @@ fun CoroutineScope.reactiveProcess(emitter: suspend Emitter.() -> Unit): } return prop } +// The two below only run their emitter while something is listening, so with no listeners they +// report notActive: whatever the emitter last produced is no longer being kept up to date. fun reactiveProcess(scope: CoroutineScope = AppScope, emitter: suspend Emitter.() -> Unit): Reactive { - return object: BaseReactive() { + return object: BaseReactive(ReactiveState.notActive) { var job: Job? = null override fun activate() { state = ReactiveState.notReady @@ -39,13 +41,15 @@ fun reactiveProcess(scope: CoroutineScope = AppScope, emitter: suspend Emitt override fun deactivate() { job?.cancel() job = null + state = ReactiveState.notActive } } } fun rawReactiveProcess(scope: CoroutineScope = AppScope, emitter: suspend Emitter>.() -> Unit): Reactive { - return object: BaseReactive() { + return object: BaseReactive(ReactiveState.notActive) { var job: Job? = null override fun activate() { + state = ReactiveState.notReady job = scope.launch { emitter(object : Emitter>, CoroutineScope by this@launch { override fun emit(value: ReactiveState) { @@ -57,6 +61,7 @@ fun rawReactiveProcess(scope: CoroutineScope = AppScope, emitter: suspend Em override fun deactivate() { job?.cancel() job = null + state = ReactiveState.notActive } } } \ No newline at end of file diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/deprecated.kt b/src/commonMain/kotlin/com/lightningkite/reactive/deprecated.kt index a2cdfb9..9edca9a 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/deprecated.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/deprecated.kt @@ -99,7 +99,7 @@ typealias LateInitProperty = LateInitSignal @Deprecated("Use remember", ReplaceWith("remember", "com.lightningkite.reactive.core")) fun shared(coroutineContext: CoroutineContext = Dispatchers.Unconfined, useLastWhileLoading: Boolean = false, action: ReactiveContext.() -> T): Reactive = - remember(coroutineContext, useLastWhileLoading, null, action) + remember(coroutineContext, useLastWhileLoading, null, action = action) @Deprecated("Use reactiveProcess", ReplaceWith("reactiveProcess", "com.lightningkite.reactive.core")) fun sharedProcess(scope: CoroutineScope = AppScope, emitter: suspend Emitter.() -> Unit): Reactive = reactiveProcess(scope, emitter) diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/extensions/WaitForNotNull.kt b/src/commonMain/kotlin/com/lightningkite/reactive/extensions/WaitForNotNull.kt index 7e51279..bbb098f 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/extensions/WaitForNotNull.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/extensions/WaitForNotNull.kt @@ -8,6 +8,8 @@ import com.lightningkite.reactive.core.Release import kotlinx.coroutines.suspendCancellableCoroutine internal class WaitForNotNull(val wraps: Reactive) : Reactive { + // A null value is notReady - a value is there, it just isn't one we can use yet. Every other + // state, notActive included, passes through as the wrapped reactive reported it. @Suppress("UNCHECKED_CAST") override val state: ReactiveState get() = if(wraps.state.raw == null) ReactiveState.notReady else wraps.state as ReactiveState diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/extensions/helpers.kt b/src/commonMain/kotlin/com/lightningkite/reactive/extensions/helpers.kt index d821f12..1aae25a 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/extensions/helpers.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/extensions/helpers.kt @@ -61,18 +61,24 @@ fun Reactive.withWrite(action: suspend Reactive.(T) -> Unit): MutableR } fun Reactive.onNextSuccess(action: (T) -> Unit): Release? { + // A successful state is one somebody is maintaining, so it needs no subscription at all. if (state.success) { state.onSuccess(action) return null } var release: Release? = null + var acted = false + fun perform(value: T) { + if (acted) return + release?.invoke() + acted = true + action(value) + } release = addListener { - state.onSuccess { - action(it) - release?.invoke() - } + state.onSuccess(::perform) } + state.onSuccess(::perform) return release } @@ -130,7 +136,9 @@ fun Reactive>.flatten(): Reactive = remember { this@flatten() fun Reactive>.flatten(): MutableReactive = remember { this@flatten()() }.withWrite { - this@flatten.state.onSuccess { s -> s set it } + // awaitOnce rather than reading state: if the outer reactive is lazy it has no value to + // read unless something is listening, and the write would be silently dropped. + this@flatten.awaitOnce().set(it) } fun CoroutineScope.asyncReactive(action: suspend () -> T): Reactive { diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/lensing/abstracts.kt b/src/commonMain/kotlin/com/lightningkite/reactive/lensing/abstracts.kt index d19db49..b489bae 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/lensing/abstracts.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/lensing/abstracts.kt @@ -22,12 +22,15 @@ open class Lens, T, L>(val source: S, val get: (T) -> L) : BaseR private var myListen: (() -> Unit)? = null + // Subscribe before reading: activating a lazy source (a Remember, for example) calculates its + // state as part of addListener, and that calculation happens before our listener is in place. + // Reading afterwards is what picks that first value up. override fun activate() { super.activate() - super.state = source.state.map(get) myListen = source.addListener { super.state = source.state.map(get) } + super.state = source.state.map(get) } override fun deactivate() { @@ -74,12 +77,14 @@ open class ValueLens, T, L>( } private var myListen: (() -> Unit)? = null + + // Subscribe before reading, for the same reason as [Lens.activate]. override fun activate() { super.activate() - super.value = source.value.let(get) myListen = source.addListener { super.value = source.value.let(get) } + super.value = source.value.let(get) } override fun deactivate() { diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/lensing/validation/validation.kt b/src/commonMain/kotlin/com/lightningkite/reactive/lensing/validation/validation.kt index 41ad118..6f84f7e 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/lensing/validation/validation.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/lensing/validation/validation.kt @@ -44,10 +44,11 @@ private open class ValidatedLens, T>( override fun activate() { super.activate() baseNode.connect() - state = source.state.also(::check) + // Subscribe before reading, so state produced by activating a lazy source isn't missed. myListen = source.addListener { if (listen) state = source.state.also(::check) } + state = source.state.also(::check) } override fun deactivate() { super.deactivate() @@ -88,10 +89,11 @@ private open class ValidatedValueLens, T>( override fun activate() { super.activate() baseNode.connect() - value = source.value.also(::check) + // Subscribe before reading, for the same reason as [ValidatedLens.activate]. myListen = source.addListener { if (listen) value = source.value.also(::check) } + value = source.value.also(::check) } override fun deactivate() { super.deactivate() diff --git a/src/commonTest/kotlin/com/lightningkite/reactive/DependencyTrackerTest.kt b/src/commonTest/kotlin/com/lightningkite/reactive/DependencyTrackerTest.kt index 2bc83d1..6b18587 100644 --- a/src/commonTest/kotlin/com/lightningkite/reactive/DependencyTrackerTest.kt +++ b/src/commonTest/kotlin/com/lightningkite/reactive/DependencyTrackerTest.kt @@ -7,12 +7,100 @@ import kotlin.collections.plusAssign import kotlin.test.Ignore import kotlin.test.Test import kotlin.test.assertEquals +import kotlin.test.assertTrue import kotlin.time.Duration import kotlin.time.Duration.Companion.nanoseconds import kotlin.time.TimeSource import kotlin.time.measureTime class DependencyTrackerTest { + /** + * Exposes the real [com.lightningkite.reactive.context.DependencyTracker]'s protected block + * methods for direct testing. Fully qualified because this file also declares its own nested + * `DependencyTracker` benchmark helper below, which would otherwise shadow the real one. + */ + private class TestableDependencyTracker : com.lightningkite.reactive.context.DependencyTracker() { + public override fun cancel() = super.cancel() + fun blockStart() = dependencyBlockStart() + fun blockEnd() = dependencyBlockEnd() + } + + /** A [Listenable] that counts every `equals()` comparison it's involved in. */ + private class CountingListenable(val id: Int) : BaseListenable() { + override fun equals(other: Any?): Boolean { + calls++ + return other is CountingListenable && id == other.id + } + override fun hashCode(): Int = id + companion object { var calls = 0 } + } + + @Test + fun orderedFastPathAvoidsLinearScanOnSteadyStateRerun() { + val tracker = TestableDependencyTracker() + val deps = List(30) { CountingListenable(it) } + + // First run: nothing is registered yet, so this just populates `dependencies`. + tracker.blockStart() + for (d in deps) if (tracker.existingDependency(d) == null) tracker.registerDependency(d, d.addListener {}) + tracker.blockEnd() + + // Second, steady-state run: same dependencies read in the same order. The ordered fast + // path should find each one with a single comparison at its expected slot. Before the + // B1 fix, the off-by-one probe missed every slot and fell back to the O(n) linear scan, + // costing roughly deps.size comparisons per read (~deps.size^2 total instead of ~deps.size). + // Measured around just the reads - dependencyBlockEnd's own cleanup does an unrelated + // O(n) `!in` scan per dependency and would otherwise swamp the count being tested here. + CountingListenable.calls = 0 + tracker.blockStart() + for (d in deps) tracker.existingDependency(d) + val callsDuringReads = CountingListenable.calls + tracker.blockEnd() + + assertTrue( + callsDuringReads <= deps.size, + "expected the ordered fast path to fire (~${deps.size} comparisons), but saw $callsDuringReads" + ) + } + + @Test + fun repeatedReadsInOneRunDoNotBreakTheFastPathOnTheNextRerun() { + val tracker = TestableDependencyTracker() + val deps = List(30) { CountingListenable(it) } + + // First run: register everything, reading the first dependency an extra, repeated time + // partway through - mimicking a call site that reads the same reactive value twice. + tracker.blockStart() + for ((i, d) in deps.withIndex()) { + if (tracker.existingDependency(d) == null) tracker.registerDependency(d, d.addListener {}) + if (i == 0) tracker.existingDependency(d) // repeated read of the same dependency + } + tracker.blockEnd() + + // Steady-state rerun with the same repeated read pattern. The repeated read of `d[0]` + // itself necessarily misses the fast path once (it falls on `d[1]`'s slot) and pays a + // small, constant number of extra comparisons (a failed fast-path probe, a linear find, + // and an `in` check). What R4 actually guards against is that this one-off miss does NOT + // cascade: every read after it (`d[1]`..`d[29]`) must still land on its fast-path slot. If + // R4 regressed (the repeated read added a duplicate to `usedDependencies`), the whole rest + // of the run would be misaligned and fall back to the O(n) linear scan, costing roughly + // deps.size comparisons per read instead of one. Measured around just the reads - see the + // comment in the test above for why dependencyBlockEnd is excluded from the window. + CountingListenable.calls = 0 + tracker.blockStart() + for ((i, d) in deps.withIndex()) { + tracker.existingDependency(d) + if (i == 0) tracker.existingDependency(d) + } + val callsDuringReads = CountingListenable.calls + tracker.blockEnd() + + assertTrue( + callsDuringReads <= deps.size + 5, + "expected only a small, constant overhead from the repeated read (not O(n) cascading misalignment), but saw $callsDuringReads comparisons" + ) + } + data class LoopTimings( val registerAll: Duration, val duplicateLoop: Duration, @@ -44,12 +132,17 @@ class DependencyTrackerTest { @Suppress("UNCHECKED_CAST") override fun existingDependency(listenable: T): T? { - usedDependencies.add(listenable) - if (dependencies.size > usedDependencies.size) { - val maybe = dependencies[usedDependencies.size].first - if (maybe == listenable) return maybe as T + val index = usedDependencies.size + if (index < dependencies.size) { + val maybe = dependencies[index].first + if (maybe == listenable) { + usedDependencies.add(listenable) + return maybe as T + } } - return dependencies.find { it.first == listenable }?.first as? T + val found = dependencies.find { it.first == listenable }?.first as? T + if (listenable !in usedDependencies) usedDependencies.add(listenable) + return found } override fun registerDependency(any: Any, remove: () -> Unit) { diff --git a/src/commonTest/kotlin/com/lightningkite/reactive/MutableRememberSuspendingTests.kt b/src/commonTest/kotlin/com/lightningkite/reactive/MutableRememberSuspendingTests.kt index c5071f1..2eaecef 100644 --- a/src/commonTest/kotlin/com/lightningkite/reactive/MutableRememberSuspendingTests.kt +++ b/src/commonTest/kotlin/com/lightningkite/reactive/MutableRememberSuspendingTests.kt @@ -231,8 +231,12 @@ class MutableRememberSuspendingTests { late.unset() assertEquals(ReactiveState(10), test.state) + // Resetting restarts the calculation, which has nothing to calculate from now that + // `late` is unset. useLastWhileLoading means the last value stands until it does. + // (It used to report 1 here - a value the dormant calculation had held onto from + // before `late` was unset.) test.reset() - assertEquals(ReactiveState(1), test.state) + assertEquals(ReactiveState(10), test.state) late.value = 2 assertEquals(ReactiveState(2), test.state) diff --git a/src/commonTest/kotlin/com/lightningkite/reactive/MutableRememberTests.kt b/src/commonTest/kotlin/com/lightningkite/reactive/MutableRememberTests.kt index d83c19e..7fee299 100644 --- a/src/commonTest/kotlin/com/lightningkite/reactive/MutableRememberTests.kt +++ b/src/commonTest/kotlin/com/lightningkite/reactive/MutableRememberTests.kt @@ -228,8 +228,12 @@ class MutableRememberTests { late.unset() assertEquals(ReactiveState(10), test.state) + // Resetting restarts the calculation, which has nothing to calculate from now that + // `late` is unset. useLastWhileLoading means the last value stands until it does. + // (It used to report 1 here - a value the dormant calculation had held onto from + // before `late` was unset.) test.reset() - assertEquals(ReactiveState(1), test.state) + assertEquals(ReactiveState(10), test.state) late.value = 2 assertEquals(ReactiveState(2), test.state) diff --git a/src/commonTest/kotlin/com/lightningkite/reactive/ReactivitySuspendingTests.kt b/src/commonTest/kotlin/com/lightningkite/reactive/ReactivitySuspendingTests.kt index 25745b4..580da49 100644 --- a/src/commonTest/kotlin/com/lightningkite/reactive/ReactivitySuspendingTests.kt +++ b/src/commonTest/kotlin/com/lightningkite/reactive/ReactivitySuspendingTests.kt @@ -1,6 +1,7 @@ package com.lightningkite.reactive import com.lightningkite.reactive.context.ReactiveContext +import com.lightningkite.reactive.context.ReactiveContextSuspending import com.lightningkite.reactive.context.await import com.lightningkite.reactive.context.invoke import com.lightningkite.reactive.context.onRemove @@ -8,6 +9,7 @@ import com.lightningkite.reactive.context.reactive import com.lightningkite.reactive.context.reactiveSuspending import com.lightningkite.reactive.context.rerunOn import com.lightningkite.reactive.core.ReactiveState +import com.lightningkite.reactive.core.reactiveState import com.lightningkite.reactive.extensions.value import com.lightningkite.reactive.extensions.waitForNotNull import com.lightningkite.reactive.core.BaseReactive @@ -17,9 +19,16 @@ import com.lightningkite.reactive.core.Signal import com.lightningkite.reactive.core.Release import com.lightningkite.reactive.core.rememberSuspending import com.lightningkite.reactive.extensions.invoke +import kotlinx.coroutines.CancellationException import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.delay +import kotlinx.coroutines.test.advanceTimeBy +import kotlinx.coroutines.test.runCurrent +import kotlinx.coroutines.test.runTest import kotlin.test.Test import kotlin.test.assertEquals +import kotlin.test.assertFailsWith +import kotlin.test.assertFalse import kotlin.test.assertIs import kotlin.test.assertTrue @@ -403,6 +412,44 @@ class ReactivitySuspendingTests { } } + @Test + fun reactiveStateRethrowsCancellationInsteadOfSwallowingIt() { + // reactiveState must propagate CancellationException rather than converting it to + // notReady - swallowing it would let a cancelled suspending calculation resume past its + // suspension point and write state / clean up dependencies for a run that should be dead. + assertFailsWith { + reactiveState { throw CancellationException("cancelled") } + } + } + + @Test + fun cancelledSuspendDoesNotWriteStaleStateAfterRerun() = runTest { + val trigger = Signal(0) + var completions = 0 + val ctx = ReactiveContextSuspending(this) { + val t = trigger() + if (t == 0) delay(100) // first run suspends here; gets cancelled before it elapses + completions++ + } + ctx.startCalculation() + runCurrent() + assertEquals(0, completions, "first run should still be suspended in delay()") + + // Cancels the still-suspended first run and starts a second run that doesn't delay. + trigger.value = 1 + runCurrent() + assertEquals(1, completions, "only the fresh second run should have completed") + + // Let the first run's delay() elapse, as if nothing had cancelled it. If cancellation + // were swallowed instead of rethrown, the dead first run would resume here and write + // stale state (completions incrementing a second time). + advanceTimeBy(200) + runCurrent() + assertEquals(1, completions, "the cancelled first run must not resume and complete") + + ctx.cancel() + } + @Test fun nestedScopesWorks() { testContext { @@ -448,4 +495,31 @@ class ReactivitySuspendingTests { assertEquals(5, innerCancelled) } } + + @Test + fun readingStateWithoutListenersDoesNotCalculate() { + val source = LateInitSignal() + var computeCount = 0 + val r = rememberSuspending { + computeCount++ + source() + } + + assertEquals(0, source.listenerCount, "precondition: no listeners before reading state") + + // Reading state repeatedly must neither calculate nor subscribe: a calculation that waits + // on a not-ready source would hold a subscription open for as long as it waits. + assertEquals(ReactiveState.notActive, r.state) + assertEquals(ReactiveState.notActive, r.state) + assertEquals(0, computeCount, "reading state must not calculate") + assertEquals(0, source.listenerCount, "reading state must not subscribe to sources") + + val release = r.addListener { } + assertEquals(1, source.listenerCount, "activating subscribes") + source.value = 3 + assertEquals(3, r.state.get()) + release() + assertEquals(0, source.listenerCount, "deactivating releases the subscription") + assertEquals(ReactiveState.notActive, r.state, "a dormant Remember must not report its last value") + } } diff --git a/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt b/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt index acb58d5..e27ca6f 100644 --- a/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt +++ b/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt @@ -6,19 +6,30 @@ import com.lightningkite.reactive.context.TypedReactiveContext import com.lightningkite.reactive.context.await import com.lightningkite.reactive.context.onRemove import com.lightningkite.reactive.context.reactive +import com.lightningkite.reactive.context.ReactiveReentrancyException import com.lightningkite.reactive.core.Reactive import com.lightningkite.reactive.core.ReactiveState import com.lightningkite.reactive.core.addAndRunListener +import com.lightningkite.reactive.context.awaitOnce +import com.lightningkite.reactive.extensions.flatten import com.lightningkite.reactive.extensions.interceptWrite +import com.lightningkite.reactive.extensions.onNextSuccess +import com.lightningkite.reactive.lensing.lens import com.lightningkite.reactive.extensions.value import com.lightningkite.reactive.extensions.waitForNotNull import com.lightningkite.reactive.core.LateInitSignal import com.lightningkite.reactive.core.RawReactive +import com.lightningkite.reactive.core.MutableReactive +import com.lightningkite.reactive.core.NotActiveException import com.lightningkite.reactive.core.Remember import com.lightningkite.reactive.core.Signal import com.lightningkite.reactive.core.remember import kotlinx.coroutines.* import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.flowOf +import kotlinx.coroutines.test.advanceTimeBy +import kotlinx.coroutines.test.runCurrent +import kotlinx.coroutines.test.runTest import kotlin.coroutines.Continuation import kotlin.coroutines.CoroutineContext import kotlin.coroutines.resume @@ -469,6 +480,560 @@ class ReactivityTests { // println("Tearing down B") // } // } + + @Test + fun onLoadFiresWhenEnteringLoading() { + testContext { + val signal = LateInitSignal() + var loadCalls = 0 + reactive(onLoad = { loadCalls++ }) { + signal() + } + // Starts not-ready, so entering the calculation is a transition into loading. + assertEquals(1, loadCalls, "onLoad should fire once when first entering loading") + + signal.value = 5 + assertEquals(1, loadCalls, "onLoad should not fire again once the value is ready") + + signal.unset() + assertEquals(2, loadCalls, "onLoad should fire again when re-entering loading") + + // Leave in a ready state so testContext's loadCount balance check passes. + signal.value = 6 + } + } + + @Test + fun reentrancyThrowsClearError() { + testContext { + val s = Signal(0) + // Writing to a signal the calculation also reads re-triggers this same calculation. + // Without reentrancy detection this recurses until the stack overflows. + reactive { + val v = s() + s.value = v + 1 + } + val captured = expectException() + assertTrue( + captured is ReactiveReentrancyException, + "Expected a ReactiveReentrancyException, but captured: $captured" + ) + } + } + + @Test + fun reentrancyErrorSaysWhichProblemItIs() { + // A calculation that opted into settling needs to hear that its budget ran out. Telling it + // to stop writing its own dependencies is advice it has already declined. + val unwanted = testContext { + val s = Signal(0) + reactive { val v = s(); s.value = v + 1 } + expectException().message + } + val unsettled = testContext { + val s = Signal(0) + reactive(reentrancyLimit = 3) { val v = s(); s.value = v + 1 } + expectException().message + } + assertNotEquals(unwanted, unsettled) + } + + @Test + fun partialSettlingReentrancy() { + testContext { + val s = Signal(0) + reactive(reentrancyLimit = 2) { + val v = s() + if (v > 5) return@reactive + s.value = v + 1 + } + assertIs(expectException()) + } + } + + @Test + fun settlingReentrancy() { + testContext { + val s = Signal(0) + val ctx = reactive(reentrancyLimit = 10) { + val v = s() + if (v > 5) return@reactive + s.value = v + 1 + } + assertEquals(ReactiveState(6), s.state) + assertNull(ctx.state.exception, "the calculation settled, so it should not have failed") + } + } + + @Test + fun settlingExactlyAtTheLimitSucceeds() { + // Climbing to six takes six writes, so six self-triggered reruns: precisely the budget. + testContext { + val s = Signal(0) + val ctx = reactive(reentrancyLimit = 6) { + val v = s() + if (v > 5) return@reactive + s.value = v + 1 + } + assertEquals(ReactiveState(6), s.state) + assertNull(ctx.state.exception) + } + } + + @Test + fun settlingOneShortOfTheLimitFails() { + testContext { + val s = Signal(0) + reactive(reentrancyLimit = 5) { + val v = s() + if (v > 5) return@reactive + s.value = v + 1 + } + assertIs(expectException()) + } + } + + @Test + fun settlingPublishesEachIntermediateValue() { + // Deliberate: a settling calculation publishes every step rather than only its fixed point. + // That is what lets two settling calculations drive each other along, as the co-reentrancy + // tests below rely on. + testContext { + val s = Signal(0) + val observed = ArrayList() + reactive { observed.add(s()) } + reactive(reentrancyLimit = 10) { + val v = s() + if (v > 5) return@reactive + s.value = v + 1 + } + assertEquals(listOf(0, 1, 2, 3, 4, 5, 6), observed) + } + } + + @Test + fun eachExternalChangeGetsAFreshReentrancyBudget() { + // The limit bounds one settling sequence, not the lifetime of the calculation. A + // calculation that self-triggers once per change has to keep working indefinitely. + testContext { + val trigger = Signal(0) + val clamped = Signal(0) + val ctx = reactive(reentrancyLimit = 1) { + val t = trigger() + if (clamped() != t) clamped.value = t + } + repeat(20) { trigger.value = it + 1 } + assertEquals(ReactiveState(20), clamped.state) + assertNull(ctx.state.exception) + } + } + + @Test + fun aContextRecoversAfterAReentrancyError() { + // The failed run keeps the dependencies of its last complete pass, so a later change still + // reaches the calculation and it computes normally once the cycle is gone. + testContext { + val cycle = Signal(0) + val other = Signal("a") + var runaway = true + val ctx = reactive { + val o = other() + if (runaway) cycle.value = cycle() + 1 + o + } + assertIs(expectException()) + + runaway = false + other.value = "b" + assertEquals(ReactiveState("b"), ctx.state) + } + } + + @Test + fun aFlowsFirstValueIsNotReentrancy() { + // A cold flow collects synchronously under an undispatched scheduler, so its first value + // lands while the calculation that registered it is still running. That is a dependency + // arriving, not the calculation triggering itself, so it must not spend the budget. + testContext { + val flow = flowOf(7) + val seen = ArrayList() + val ctx = reactive { seen.add(flow()) } + assertEquals(listOf(7), seen) + assertNull(ctx.state.exception) + } + } + + @Test + fun aStateFlowsCurrentValueIsNotReentrancy() { + testContext { + val flow = MutableStateFlow(1) + val seen = ArrayList() + val ctx = reactive { seen.add(flow()) } + flow.value = 2 + assertEquals(listOf(1, 2), seen) + assertNull(ctx.state.exception) + } + } + + @Test + fun rememberDefaultsToRejectingReentrancy() { + testContext { + val s = Signal(0) + val r = remember { val v = s(); s.value = v + 1; v } + reactive { r() } + assertIs(expectException()) + } + } + + @Test + fun rememberPassesItsReentrancyLimitThrough() { + testContext { + val s = Signal(0) + val settled = remember(reentrancyLimit = 10) { + val v = s() + if (v <= 5) s.value = v + 1 + v + } + val seen = ArrayList() + reactive { seen.add(settled()) } + assertEquals(ReactiveState(6), s.state) + assertEquals(6, seen.last()) + } + } + + @Test + fun partialSettlingCoReentrancy() { + testContext { + val a = Signal(0) + val b = Signal(0) + reactive(reentrancyLimit = 3) { + val other = b() + if (other < 10) a.value = other + 1 + } + reactive(reentrancyLimit = 3) { + val other = a() + if (other < 10) b.value = other + 1 + } + assertIs(expectException()) + } + } + + @Test + fun settlingCoReentrancy() { + // Two calculations driving each other converge as long as each one's own budget covers the + // reruns it personally performs. They are not symmetric: the first context is re-entered + // from scratch by the second - so its budget is never spent - while the second settles in + // place and pays for every step. + testContext { + val a = Signal(0) + val b = Signal(0) + val firstSaw = ArrayList() + val secondSaw = ArrayList() + reactive(reentrancyLimit = 10) { + val other = b() + firstSaw.add(other) + if (other < 10) a.value = other + 1 + } + reactive(reentrancyLimit = 10) { + val other = a() + secondSaw.add(other) + if (other < 10) b.value = other + 1 + } + assertEquals(ReactiveState(9), a.state) + assertEquals(ReactiveState(10), b.state) + assertEquals(listOf(0, 2, 4, 6, 8, 10), firstSaw) + assertEquals(listOf(1, 3, 5, 7, 9), secondSaw) + } + } + + @Test + fun writingUnrelatedSignalDuringCalculationIsAllowed() { + testContext { + val trigger = Signal(0) + val unrelated = Signal(100) + var runs = 0 + reactive { + runs++ + val t = trigger() + // Writing to a signal this calculation does NOT read must remain legal. + unrelated.value = t + 100 + } + assertEquals(1, runs) + assertEquals(100, unrelated.value) + trigger.value = 5 + assertEquals(2, runs) + assertEquals(105, unrelated.value) + } + } + + @Test + fun asyncDistinguishesCallSitesWithIdenticalDependencies() = runTest { + // Two different `async {}` call sites passed the exact same dependency (1). Under the old + // deps-only cache key, both would hash/equal to the same SuspendCalculation and the second + // call site would silently reuse the first's result. + var resultA = -1 + var resultB = -1 + val ctx = TypedReactiveContext(this) { + resultA = async("a", 1) { 100 } + resultB = async("b", 1) { 200 } + } + ctx.startCalculation() + runCurrent() + + assertEquals(100, resultA, "first call site should get its own result") + assertEquals(200, resultB, "second call site must not have collided with the first") + + ctx.cancel() + } + + @Test + fun asyncCancelsStaleLaunchOnRerun() = runTest { + // The async's own dependency is `trigger()`'s value, so changing `trigger` gives the async + // block a new cache key each run - the old entry becomes unused and should be torn down + // (job cancelled) rather than left running to write into an orphaned calc. + val trigger = Signal(0) + var completions = 0 + val ctx = TypedReactiveContext(this) { + val t = trigger() + async("A", t) { + delay(100) + completions++ + } + } + ctx.startCalculation() + runCurrent() + assertEquals(0, completions, "first run's async should still be delaying") + + trigger.value = 1 // gives the async a new key; the stale (t=0) launch must be cancelled + runCurrent() + assertEquals(0, completions, "second run's async should also still be delaying") + + // Let the stale run's delay(100) elapse, as if nothing had cancelled it. + advanceTimeBy(150) + runCurrent() + + // Only the fresh (t=1) async should ever complete - the cancelled (t=0) one must not. + assertEquals(1, completions, "the cancelled first launch must not have completed") + + ctx.cancel() + } + + @Test + fun changingADependencyEndsThePreviousRunImmediately() = runTest { + // Under a dispatching scheduler the rerun does not begin until the scheduler gets to it, + // but the run it supersedes is finished the moment its input changed - whatever that run + // launched is now computing from a stale value. Cleanup must not wait for the rerun. + val trigger = Signal(0) + val cleanedUp = ArrayList() + val ctx = TypedReactiveContext(this) { + val t = trigger() + onRemove { cleanedUp.add(t) } + } + ctx.startCalculation() + runCurrent() + assertEquals(listOf(), cleanedUp, "the first run is still current") + + trigger.value = 1 + assertEquals( + listOf(0), cleanedUp, + "the superseded run should be torn down at the change, not when the rerun is dispatched" + ) + + runCurrent() + ctx.cancel() + } + + @Test + fun readingStateWithoutListenersDoesNotCalculate() { + val source = Signal(1) + var computeCount = 0 + val r = remember { + computeCount++ + source() + } + + // No listener added, so the Remember is lazy and has never calculated. + assertEquals(0, source.listenerCount, "precondition: no listeners before reading state") + + assertEquals(ReactiveState.notActive, r.state, "a Remember with no listeners maintains no value") + assertEquals(0, computeCount, "reading state must not calculate") + assertEquals(0, source.listenerCount, "reading state must not subscribe to sources") + + // Mutating the source must not resurrect it either. + source.value = 2 + assertEquals(0, computeCount, "mutating a source must not calculate a Remember nobody listens to") + assertEquals(0, source.listenerCount, "source must still have zero listeners after mutation") + + // Gaining a listener is what makes it calculate, and losing it stops it again. + val release = r.addListener { } + assertEquals(2, r.state.get(), "activating calculates the current value") + assertEquals(1, computeCount) + assertEquals(1, source.listenerCount) + release() + assertEquals(0, source.listenerCount, "deactivating releases the subscription") + + // Going dormant drops the value rather than reporting one that is no longer maintained. + assertEquals(ReactiveState.notActive, r.state, "a dormant Remember must not report its last value") + source.value = 3 + assertEquals(1, computeCount, "a dormant Remember must not recalculate") + + // Listening again recalculates against the source as it is now. + r.addListener { } + assertEquals(3, r.state.get(), "reactivating recalculates rather than reporting the old value") + } + + // The tests below cover reading a lazy source: because a Remember only calculates while + // something is listening, and BaseListenable activates before the new listener is in the list, + // every one of these has to subscribe before reading or it sees the pre-activation state. + + @Test + fun lensOfLazySourceGetsValueOnActivation() { + val source = Signal(1) + val lensed = remember { source() }.lens { it * 10 } + val release = lensed.addListener { } + assertEquals(10, lensed.state.get(), "activating a lens must pick up the value its source calculates") + source.value = 2 + assertEquals(20, lensed.state.get()) + release() + } + + @Test + fun onceObtainsValueFromLazySource() { + val source = Signal(1) + val r = remember { source() } + val seen = ArrayList() + testContext { + reactive { seen.add(r.once()) } + assertEquals(listOf(1), seen, "once must obtain a value from a lazy source") + source.value = 2 + source.value = 3 + assertEquals(listOf(1), seen, "once must not rerun the calculation, nor change what it reports") + } + } + + @Test + fun stateTransformSeesValueFromActivation() { + val source = Signal(1) + val r = remember { source() } + val seen = ArrayList() + testContext { + reactive { seen.add(r.state { it.ready }) } + assertEquals(listOf(true), seen, "state(get) must see the value its subscription caused") + source.value = 2 + assertEquals(listOf(true), seen, "readiness did not change, so there is nothing to rerun") + } + } + + @Test + fun onNextSuccessFiresForLazySource() { + val source = Signal(1) + val seen = ArrayList() + remember { source() }.onNextSuccess { seen.add(it) } + assertEquals(listOf(1), seen, "onNextSuccess must obtain a value from a lazy source") + source.value = 2 + assertEquals(listOf(1), seen, "onNextSuccess fires exactly once") + } + + @Test + fun writingThroughFlattenedLazySourceReachesTheTarget() = runTest { + val inner = Signal(1) + val selection = Signal>(inner) + // The outer reactive is lazy, so the write has to activate it to find the target. + val outer: Reactive> = remember { selection() } + outer.flatten().set(42) + assertEquals(42, inner.value, "a write through flatten must not be silently dropped") + } + + @Test + fun notActiveIsDistinctFromLoading() { + val source = LateInitSignal() + val r = remember { source() } + + assertEquals(ReactiveState.notActive, r.state, "nobody is maintaining this") + + val release = r.addListener { } + assertEquals( + ReactiveState.notReady, r.state, + "now somebody is maintaining it - it just doesn't have a value yet" + ) + + source.value = 1 + assertEquals(ReactiveState(1), r.state) + + release() + assertEquals(ReactiveState.notActive, r.state, "and back to nobody maintaining it") + } + + @Test + fun notActivePropagatesThroughLenses() { + val source = Signal(1) + val r = remember { source() } + val lensed = r.lens { it * 10 } + + assertTrue(lensed.state.notActive, "a lens over an unmaintained value is equally unmaintained") + + val release = lensed.addListener { } + assertEquals(ReactiveState(10), lensed.state) + release() + assertTrue(lensed.state.notActive) + } + + @Test + fun lensOfAlwaysDefiniteSourceNeedsNoListeners() { + // A Signal's value is accurate whether or not anyone is listening, so a lens over one can + // be read directly - this is what notActive being a separate state buys. + val source = Signal(1) + val lensed = source.lens { it * 10 } + assertEquals(ReactiveState(10), lensed.state) + source.value = 2 + assertEquals(ReactiveState(20), lensed.state) + } + + @Test + fun readingAValueSomebodyElseMaintainsIsFree() = runTest { + val source = Signal(1) + var computeCount = 0 + val r = remember { + computeCount++ + source() + } + val release = r.addListener { } + assertEquals(1, computeCount) + + // Someone else is keeping this current, so reading it must not re-run anything or churn + // the subscription. + assertEquals(1, r.awaitOnce()) + assertEquals(1, computeCount, "awaitOnce must not recalculate a value already being maintained") + assertEquals(1, source.listenerCount, "awaitOnce must not disturb the existing subscription") + + release() + } + + @Test + fun readingNotActiveStateThrowsSomethingExplanatory() { + val r = remember { Signal(1)() } + @Suppress("DEPRECATION") + val thrown = assertFailsWith { r.state.get() } + assertTrue(thrown.message!!.contains("listening"), "the message should say what to do about it") + } + + @Test + fun awaitOnceCalculatesAndReleases() = runTest { + val source = Signal(1) + var computeCount = 0 + val r = remember { + computeCount++ + source() + } + + assertEquals(1, r.awaitOnce(), "awaitOnce must calculate a dormant Remember") + assertEquals(1, computeCount) + assertEquals(0, source.listenerCount, "awaitOnce must not hold onto its subscription") + + source.value = 2 + assertEquals(2, r.awaitOnce(), "awaitOnce must recalculate rather than report a stale value") + assertEquals(0, source.listenerCount) + } } class VirtualDelay(val action: () -> T) {