A library for scheduling, executing and observing repeateble reactive tasks.
Kotlin Multiplatform: JVM, Android, iOS (iosX64, iosArm64, iosSimulatorArm64).
implementation("ru.kode:reactive-tasks:2.0.1")//
// Create a scheduler
//
val scheduler = Scheduler(scope = CoroutineScope(Dispatchers.IO))
//
// Register tasks
//
val fetchTaskHandle = scheduler.registerTask("fetch", body = { sampleId: String ->
val response = networkApi.fetchSampleData(sampleId) // suspending call
val result = response.data.size
result // return value
})
val commitChangesHandle = scheduler.registerTask("commit", body = { changes: Changes ->
database.store(changes)
networkApi.updateCommitDate(changes.date)
})
val searchTaskHandle = scheduler.registerTask("search", body = { query: String ->
networkApi.searchBy(query)
})
//
// Start tasks in parallel
//
val jobHandle1 = scheduler.launch(fetchTaskHandle, "sample-id-001")
val jobHandle2 = scheduler.launch(fetchTaskHandle, "sample-id-002")
val jobHandle3 = scheduler.launch(commitChangesHandle, Changes(date = LocalDate.now()))
//
// Start tasks where each new start cancels previous running (like Flow.flatMapLatest() does)
//
scheduler.launchLatest(searchTaskHandle, "Kaliningrad")
scheduler.launchLatest(searchTaskHandle, "Moscow")
scheduler.launchLatest(searchTaskHandle, "Saint-Petersburg")
//
// Start tasks as suspend-functions in some scope
//
val mainScope = CoroutineScope(Dispatchers.Main)
mainScope.launch {
val resultState = scheduler.start(fetchTaskHandle, "sample-id-003")
if (resultState.result == 3) {
scheduler.start(fetchTaskHandle, "sample-id-004")
println("all fetches done!")
}
}
//
// Add listener to observe changes
//
scheduler.addTaskStateChangeListener(object : TaskStateChangeListener {
override suspend fun onJobStateChanged(state: JobState) {
when (state.runState) {
RunState.NotStarted -> println("[${state.jobId}] not started")
RunState.Running -> println("[${state.jobId}] running")
RunState.FinishedSuccess -> println("[${state.jobId}] success result=${state.result}")
RunState.FinishedError-> {
println("[${state.jobId}] error")
state.error.printStackTrace()
}
}
}
override fun onJobCancelled(taskId: Uuid, jobId: Uuid) {
println("task is cancelled: jobId = $jobId, taskId = $taskId")
}
}
//
// Or observe one task as a Flow. Read a typed result with state.resultOf(handle): R?
//
val searchStates = scheduler.jobStates(searchTaskHandle) // replays a running job's state, never completes on cancel
val lastResult = scheduler.successResults(searchTaskHandle, replayFinished = true) // also replays the last finished result
val searchResults = scheduler.successResults(searchTaskHandle) // Flow<R>, no casts
val searchErrors = scheduler.errors(searchTaskHandle) // Flow<Throwable>
val searchFinished = scheduler.finishedStates(searchTaskHandle, replayLast = true) // finished states only
val searchProgress = scheduler.runningStates(searchTaskHandle) // Flow<Boolean>, false after a cancel
val fetchRunning = scheduler.isRunning(fetchTaskHandle)
//
// observeStateChanges completes on the first cancelled job of the task (pass completeOnJobCancel = false
// to keep observing). It is fairly simple implementation, if you need something more involved,
// check how it is implemented, use it as a reference
//
val taskStateChanges = scheduler.observeStateChanges(searchTaskHandle)
runBlocking {
taskStateChanges
.shareIn(CoroutineScope(Dispatchers.Main), SharingStarted.Lazily)
.onSubscription {
println("Subscribed!")
}
.collect {
println("Received state $it")
}
}
//
// Cancel all jobs of a certain task
//
scheduler.cancel(searchTaskHandle)
// or
runBlocking {
scheduler.cancelAndJoin(searchTaskHandle)
}
//
// Cancel a specific started job
//
scheduler.cancel(jobHandle2)
// or
runBlocking {
scheduler.cancelAndJoin(jobHandle2)
}
//
// Cancel the jobs of all tasks and all coroutines launched in scheduler.scope, the scheduler stays usable.
// 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.launchIfIdle(fetchTaskHandle, "arg")
scheduler.launchIfIdle(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: suspending `start` jobs are not counted by isRunning
//
scheduler.startIfIdle(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 { } }remo (ru.kode:remo, ru.kode.android:remo, ReactiveModel) is deprecated,
reactive-tasks replaces it. Both run suspend "jobs" and let you observe their state, results and errors;
reactive-tasks drops the ReactiveModel base class and kotlin-result, and adds launchLatest, start,
restart and iOS.
// 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:2.0.1")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>.
| remo 1.2.0 | remo 9.0.1 | reactive-tasks 2.0.1 |
|---|---|---|
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: 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 -> }) |
ReactiveModel(errorMapper = …), task(errorMapper = …), mapErrors {} |
same | map in the flow: errors(fetch).map { … }, or catch inside the task body |
val fetch = task { id: String -> api.fetch(id) } |
same | val fetch = scheduler.registerTask("fetch") { id: String -> api.fetch(id) } |
task { -> api.fetch() } (Task0) |
same | registerTask("fetch") { _: Unit -> api.fetch() }, launch(fetch, Unit) |
task { a: A, b: B -> } (Task2 / Task3) |
same | one data class argument: registerTask("t") { (a, b): Args -> } |
taskIn(watchContext) { }, WatchContext.executeInModelScope { } |
same | registerTask + launch |
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.launch(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 | scheduler.launchIfIdle(fetch, id): drops the start instead of throwing; plain launch runs jobs in parallel |
start(id, queueingStrategy = SkipNew) |
same | scheduler.launchIfIdle(fetch, id) (atomic, returns JobHandle?, null if dropped) |
start(id, SkipNew / Disallow).join() |
same | scheduler.startIfIdle(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.launchLatest(fetch, id): cancels running jobs of the task |
| re-run with the last argument | same | scheduler.relaunch(fetch) |
start(id, StartScheduled.Lazily()) |
same | flow { emit(scheduler.start(fetch, id)) }, or shareIn(…).onSubscription { scheduler.launch(fetch, id) } |
| — | — | start(fetch, id): suspends until done, returns JobState (state.resultOf(fetch): R?) |
fetch.jobFlow |
same | scheduler.jobStates(fetch): Flow<JobState> |
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> (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 {} |
// remo
class UserListModel : ReactiveModel() {
val fetch = task { sort: Sort -> api.fetch(sort) }
}
model.fetch.jobFlow.successResults().collect { render(it) }
model.fetch.start(Sort.Ascending)
// reactive-tasks
class UserListModel(private val scheduler: Scheduler) {
val fetch = scheduler.registerTask("fetch") { sort: Sort -> api.fetch(sort) }
val users: Flow<List<User>> = scheduler.successResults(fetch)
fun fetch(sort: Sort) = scheduler.launch(fetch, sort)
}