1.4.0: startIfIdle, startSuspendedIfIdle, bindTo, results

CHANGELOG.md

@@ -1,5 +1,21 @@
 # Changelog
 
+## 1.4.0 - 2026-11-11
+
+### Added
+
+- `Scheduler.startIfIdle(handle, argument, tag)` and `startIfIdle(handle)` for `Unit` tasks: starts a job unless one is
+  running. Atomic with respect to other `startIfIdle` calls (two concurrent calls start one job). Returns the
+  `JobHandle`, or null if dropped. Replaces remo `SkipNew` / `Disallow`.
+- `Scheduler.startSuspendedIfIdle(handle, argument, tag)` and `startSuspendedIfIdle(handle)`: runs `startSuspended`
+  unless a job is running, returns the `JobState`, or null (without waiting) if dropped.
+- `Scheduler.bindTo(scope)`: calls `cancelAll()` when the job of `scope` completes.
+- `Scheduler.results(handle, replayLast = true)`: `Flow<kotlin.Result<R>>` of finished jobs.
+
+### Changed
+
+- `kotlinx-atomicfu` is now an `implementation` dependency of all targets, it was iOS only.
+
 ## 1.3.0 - 2026-11-11
 
 ### Added

README.md

@@ -7,7 +7,7 @@ A library for scheduling, executing and observing repeateble reactive tasks.
 Kotlin Multiplatform: JVM, Android, iOS (`iosX64`, `iosArm64`, `iosSimulatorArm64`).
 
 ```kotlin
-implementation("ru.kode:reactive-tasks:1.3.0")
+implementation("ru.kode:reactive-tasks:1.4.0")
 ```
 
 # How to use
@@ -134,6 +134,31 @@ runBlocking {
 // Cancelling scheduler.scope itself makes the scheduler unusable for good
 //
 scheduler.cancelAll()
+
+//
+// Start only if no job of the task is running, atomically (two concurrent calls start one job).
+// Returns the JobHandle, or null if the call was dropped
+//
+scheduler.startIfIdle(fetchTaskHandle, "arg")
+scheduler.startIfIdle(unitTaskHandle)
+
+//
+// Same, but suspends until the started job is done and returns its JobState, null if the call was dropped.
+// Does not wait for an already running job. Not atomic: startSuspended jobs are not counted by isRunning
+//
+scheduler.startSuspendedIfIdle(fetchTaskHandle, "arg")
+
+//
+// Cancel all jobs (cancelAll) whenever the given scope completes. The scheduler stays usable.
+// Every call adds one more such callback, there is no unbinding
+//
+scheduler.bindTo(viewModelScope)
+
+//
+// Outcomes of finished jobs as kotlin.Result: success holds the result, failure holds the error.
+// replayLast = true (default) replays the last finished job first
+//
+scheduler.results(fetchTaskHandle).collect { result -> result.onSuccess { } .onFailure { } }
 ```
 
 # Migrating from remo
@@ -147,18 +172,18 @@ reactive-tasks drops the `ReactiveModel` base class and `kotlin-result`, and add
 // before
 implementation("ru.kode:remo:1.2.0") // or ru.kode:remo:9.0.1, ru.kode.android:remo:9.1.0
 // after
-implementation("ru.kode:reactive-tasks:1.3.0")
+implementation("ru.kode:reactive-tasks:1.4.0")
 ```
 
 remo 1.2.0 and 9.x differ only in how a model gets its scope; tasks and `JobFlow` are the same.
 Below `scheduler` is a `Scheduler`, `fetch` is a `TaskHandle<A, R>`.
 
 ## API map
 
-| remo 1.2.0 | remo 9.0.1 | reactive-tasks 1.3.0 |
+| remo 1.2.0 | remo 9.0.1 | reactive-tasks 1.4.0 |
 |---|---|---|
 | `class M : ReactiveModel(dispatcher)` | `class M(scope) : ReactiveModel(scope)` | `class M(scheduler: Scheduler)`, no base class |
-| `model.start(parentScope)`, `Job.cancel()` to stop, can be started again with a new scope | constructor `scope`, cancel it to stop | `Scheduler(scope)`, cancel `scope` to stop for good; to stop and start again: `parentScope.coroutineContext.job.invokeOnCompletion { scheduler.cancelAll() }` on each start |
+| `model.start(parentScope)`, `Job.cancel()` to stop, can be started again with a new scope | constructor `scope`, cancel it to stop | `Scheduler(scope)`, cancel `scope` to stop for good; to stop and start again: `scheduler.bindTo(parentScope)` on each start |
 | `onPostStart()` | `init {}` (`onPostStart` deprecated) | 1.x: a `start(parentScope)` function of the model, called where `model.start` was (runs on every start); 9.x: `init {}` |
 | `scope.launch {}` inside the model | same | `scheduler.scope.launch {}` |
 | `uncaughtExceptions: Flow<Throwable>` | same | `Scheduler(onUncaughtException = { taskId, taskName, e -> })` |
@@ -170,8 +195,9 @@ Below `scheduler` is a `Scheduler`, `fetch` is a `TaskHandle<A, R>`.
 | one `WatchContext` shared by several tasks | same | `merge(jobStates(a), jobStates(b))`, or one task with a sealed argument |
 | `fetch.start(id)`: `Job` | same | `scheduler.start(fetch, id)`: `JobHandle` |
 | `job.cancel()` | same | `scheduler.cancel(jobHandle)`, `scheduler.cancel(fetch)` (all jobs of the task) |
-| `start(id, queueingStrategy = Disallow)` (default, throws if a job runs) | same | no equivalent, `start` runs jobs in parallel |
-| `start(id, queueingStrategy = SkipNew)` | same | `if (!scheduler.isRunning(fetch)) scheduler.start(fetch, id)` |
+| `start(id, queueingStrategy = Disallow)` (default, throws if a job runs) | same | `scheduler.startIfIdle(fetch, id)`: drops the start instead of throwing; plain `start` runs jobs in parallel |
+| `start(id, queueingStrategy = SkipNew)` | same | `scheduler.startIfIdle(fetch, id)` (atomic, returns `JobHandle?`, null if dropped) |
+| `start(id, SkipNew / Disallow).join()` | same | `scheduler.startSuspendedIfIdle(fetch, id)`: runs and waits if idle, returns `JobState?` (null, without waiting, if a job runs) |
 | restart a running job with a new argument | same | `scheduler.startLatest(fetch, id)`: cancels running jobs of the task |
 | re-run with the last argument | same | `scheduler.restart(fetch)` |
 | `start(id, StartScheduled.Lazily())` | same | `flow { emit(scheduler.startSuspended(fetch, id)) }`, or `shareIn(…).onSubscription { scheduler.start(fetch, id) }` |
@@ -180,8 +206,8 @@ Below `scheduler` is a `Scheduler`, `fetch` is a `TaskHandle<A, R>`.
 | `fetch.jobFlow.state`: `Idle` / `Running` (progress) | same | `scheduler.runningStates(fetch)`: `Flow<Boolean>`, current value on subscribe, `false` after a cancel |
 | `fetch.jobFlow.successResults()` | same | `scheduler.successResults(fetch)`: `Flow<R>` |
 | `fetch.jobFlow.errors()` | same | `scheduler.errors(fetch)`: `Flow<Throwable>` |
-| `fetch.jobFlow.results()`: `Result<R, Throwable>` | same | `jobStates(fetch)`: `JobState` with `runState`, `error`, `resultOf(fetch)`: `R?` |
-| `results(replayLast = true)` | same | `scheduler.finishedStates(fetch, replayLast = true)`: replays the last finished job even while a new one runs |
+| `fetch.jobFlow.results()`: `Result<R, Throwable>` (kotlin-result) | same | `scheduler.results(fetch)`: `Flow<kotlin.Result<R>>` (`replayLast = true` by default), or `jobStates(fetch)`: `JobState` with `runState`, `error`, `resultOf(fetch)`: `R?` |
+| `results(replayLast = true)` | same | `scheduler.results(fetch, replayLast = true)`, or `finishedStates(fetch, replayLast = true)` for `JobState`: replays the last finished job even while a new one runs |
 | `successResults(replayLast = true)`, `errors(replayLast = true)` | same | `successResults(fetch, replayFinished = true)`, `errors(fetch, replayFinished = true)` |
 | `mapSuccessResults {}` | same | `successResults(fetch).map {}` |
 

gradle.properties

@@ -11,7 +11,7 @@ kotlin.mpp.enableCInteropCommonization=true
 android.useAndroidX=true
 android.nonTransitiveRClass=true
 #Publishing
-versionName=1.3.0
+versionName=1.4.0
 pomGroupId=ru.kode
 pomName=Reactive Tasks
 pomDescription=Kotlin Multiplatform library for scheduling, executing and observing reactive tasks

library/build.gradle.kts

@@ -29,16 +29,13 @@ kotlin {
     sourceSets {
         commonMain.dependencies {
             api(libs.kotlin.coroutines)
+            implementation(libs.kotlinx.atomicfu)
         }
 
         commonTest.dependencies {
             implementation(libs.kotlin.test)
             implementation(libs.kotlin.coroutines.test)
         }
-
-        iosMain.dependencies {
-            implementation(libs.kotlinx.atomicfu)
-        }
     }
 }
 

library/src/commonMain/kotlin/ru/kode/reactivetasks/ReactiveTasks.kt

@@ -1,5 +1,7 @@
 package ru.kode.reactivetasks
 
+import kotlinx.atomicfu.locks.SynchronizedObject
+import kotlinx.atomicfu.locks.synchronized
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.CoroutineExceptionHandler
@@ -222,6 +224,8 @@ class Scheduler(
      */
     internal val finishedTaskState: MutableMap<Uuid, JobState> = createConcurrentMutableMap()
 
+    private val startIfIdleLock = SynchronizedObject()
+
     private val stateChangeListeners: MutableList<TaskStateChangeListener> = createConcurrentMutableList()
 
     /**
@@ -321,6 +325,53 @@ class Scheduler(
         return executeWrappedBody(handle, jobId, argument, tag)
     }
 
+    /**
+     * Starts a new job for [handle] unless a job started with [start], [startLatest] or [restart] is still running
+     * ([isRunning]), in which case the call is dropped. The check and the start are one atomic step with respect to
+     * other [startIfIdle] calls, so two concurrent calls start exactly one job. Calls to [start], [startLatest] and
+     * [restart] are not serialized with it.
+     * Replaces remo's `SkipNew` / `Disallow`.
+     *
+     * @return the started job, or null if the call was dropped
+     */
+    fun <A> startIfIdle(handle: TaskHandle<A, *>, argument: A, tag: Any? = null): JobHandle? {
+        return synchronized(startIfIdleLock) {
+            if (isRunning(handle)) null else startInternal(handle, argument, tag, cancelPrevious = false)
+        }
+    }
+
+    /**
+     * Same as `startIfIdle(handle, Unit)`
+     */
+    fun startIfIdle(handle: TaskHandle<Unit, *>): JobHandle? = startIfIdle(handle, Unit)
+
+    /**
+     * Runs [startSuspended] unless a job started with [start], [startLatest] or [restart] is still running
+     * ([isRunning]): then the call is dropped and returns null immediately, it does not wait for the running job.
+     * Never throws for errors of the task body, they are returned in the [JobState].
+     * Unlike [startIfIdle] this is not atomic: jobs started with [startSuspended] are not counted by [isRunning], so
+     * concurrent calls can all run. Replaces remo's `start().join()` with `SkipNew` / `Disallow`.
+     *
+     * @return the state of the finished job, or null if the call was dropped
+     */
+    suspend fun <A, R> startSuspendedIfIdle(handle: TaskHandle<A, R>, argument: A, tag: Any? = null): JobState? {
+        return if (isRunning(handle)) null else startSuspended(handle, argument, tag)
+    }
+
+    /**
+     * Same as `startSuspendedIfIdle(handle, Unit)`
+     */
+    suspend fun startSuspendedIfIdle(handle: TaskHandle<Unit, *>): JobState? = startSuspendedIfIdle(handle, Unit)
+
+    /**
+     * Calls [cancelAll] when the job of [scope] completes (normally or by cancellation), or right away if it
+     * already has. The scheduler stays usable afterwards. Not single-use: every call registers one more
+     * [cancelAll] for the completion of the given scope, there is no unbinding.
+     */
+    fun bindTo(scope: CoroutineScope) {
+        scope.coroutineContext.job.invokeOnCompletion { cancelAll() }
+    }
+
     private fun <A> startInternal(handle: TaskHandle<A, *>, argument: A, tag: Any?, cancelPrevious: Boolean): JobHandle {
         val jobId = Uuid.random()
         argumentCache.put(jobId, argument as Any?)
@@ -546,6 +597,21 @@ fun Scheduler.runningStates(handle: TaskHandle<*, *>): Flow<Boolean> {
         .distinctUntilChanged()
 }
 
+/**
+ * Emits the outcome of each finished job of the task represented by [handle] as a [Result]: success holds the
+ * job's result (see [resultOf]), failure holds its error. Cancelled jobs emit nothing. See [finishedStates]
+ *
+ * @param replayLast when true (the default), the most recently finished job is replayed first, even while another
+ * job of the task is running
+ */
+fun <R> Scheduler.results(handle: TaskHandle<*, R>, replayLast: Boolean = true): Flow<Result<R>> {
+    @Suppress("UNCHECKED_CAST")
+    return finishedStates(handle, replayLast).map { state ->
+        if (state.runState == RunState.FinishedSuccess) Result.success(state.resultOf(handle) as R)
+        else Result.failure(checkNotNull(state.error))
+    }
+}
+
 /**
  * Emits results of successfully finished jobs of the task represented by [handle].
  *

library/src/commonTest/kotlin/ru/kode/reactivetasks/SchedulerTest.kt

@@ -3,9 +3,18 @@
 package ru.kode.reactivetasks
 
 import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.async
+import kotlinx.coroutines.awaitAll
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.flow.toList
+import kotlinx.coroutines.withContext
 import kotlinx.coroutines.currentCoroutineContext
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.launch
@@ -503,5 +512,108 @@ class SchedulerTest {
         assertEquals(10, ticks)
     }
 
+    // endregion
+    // region startIfIdle, startSuspendedIfIdle, bindTo, results
+
+    @Test
+    fun startIfIdleDropsStartWhileRunning() = runTest {
+        val scheduler = testScheduler()
+        val task = scheduler.registerTask("t") { _: Int -> delay(100) }
+
+        val first = scheduler.startIfIdle(task, 1)
+        val second = scheduler.startIfIdle(task, 2)
+        runCurrent()
+
+        assertTrue(first != null)
+        assertNull(second)
+        advanceUntilIdle()
+        assertTrue(scheduler.startIfIdle(task, 3) != null)
+    }
+
+    @Test
+    fun startIfIdleUnitOverloadStartsWithUnit() = runTest {
+        val scheduler = testScheduler()
+        val listener = RecordingListener().also(scheduler::addTaskStateChangeListener)
+        val task = scheduler.registerTask("t") { _: Unit -> }
+
+        scheduler.startIfIdle(task)
+        advanceUntilIdle()
+
+        assertEquals(Unit, listener.states.last().argument)
+        assertEquals(RunState.FinishedSuccess, listener.states.last().runState)
+    }
+
+    @Test
+    fun concurrentStartIfIdleStartsExactlyOneJob() = runTest {
+        val scheduler = Scheduler(CoroutineScope(SupervisorJob()), Dispatchers.Default)
+        val gate = CompletableDeferred<Unit>()
+        val task = scheduler.registerTask("t") { _: Int -> gate.await() }
+
+        val started = withContext(Dispatchers.Default) {
+            (1..200).map { async { scheduler.startIfIdle(task, it) } }.awaitAll()
+        }
+
+        assertEquals(1, started.count { it != null })
+        gate.complete(Unit)
+        scheduler.cancelAll()
+    }
+
+    @Test
+    fun startSuspendedIfIdleRunsWhenIdleAndDropsWhileRunning() = runTest {
+        val scheduler = testScheduler()
+        val task = scheduler.registerTask("t") { x: Int -> x * 2 }
+        val slow = scheduler.registerTask("slow") { _: Int -> delay(100) }
+
+        val state = scheduler.startSuspendedIfIdle(task, 21)
+        assertEquals(RunState.FinishedSuccess, state?.runState)
+        assertEquals(42, state?.resultOf(task))
+
+        scheduler.start(slow, 1)
+        runCurrent()
+        assertNull(scheduler.startSuspendedIfIdle(slow, 2))
+    }
+
+    @Test
+    fun bindToCancelsAllWhenScopeCompletesAndSchedulerStaysUsable() = runTest {
+        val scheduler = testScheduler()
+        val task = scheduler.registerTask("t") { _: Int -> delay(1_000) }
+        val bound = Job()
+        scheduler.bindTo(CoroutineScope(bound))
+
+        val job = scheduler.start(task, 1)
+        runCurrent()
+        assertTrue(scheduler.isRunning(task))
+
+        bound.cancel()
+        runCurrent()
+
+        assertTrue(job.job.isCancelled)
+        assertTrue(!scheduler.isRunning(task))
+        assertTrue(scheduler.startIfIdle(task, 2) != null)
+    }
+
+    @Test
+    fun resultsEmitsSuccessAndFailureAndReplaysLast() = runTest {
+        val scheduler = testScheduler()
+        val boom = IllegalStateException("boom")
+        val task = scheduler.registerTask("t") { x: Int -> if (x < 0) throw boom else x }
+
+        scheduler.startSuspended(task, 1)
+        scheduler.startSuspended(task, -1)
+
+        // replayLast is true by default: only the most recent finished job is replayed
+        val replayed = scheduler.results(task).first()
+        assertSame(boom, replayed.exceptionOrNull())
+
+        val live = mutableListOf<Result<Int>>()
+        val collector = launch { scheduler.results(task, replayLast = false).toList(live) }
+        runCurrent()
+        scheduler.startSuspended(task, 7)
+        runCurrent()
+        collector.cancel()
+
+        assertEquals(listOf(7), live.map { it.getOrThrow() })
+    }
+
     // endregion
 }