Interoperability with Kotlin Coroutines: conversions between coroutines (including Flow) and Reaktive
Interoperability with RxJava2 and RxJava3: conversion of sources between Reaktive and RxJava, ability to reuse RxJava's schedulers
Reaktive and the old (strict) Kotlin/Native memory model
The old (strict) Kotlin Native memory model and concurrency are very special. In general shared mutable state between threads is not allowed.
Since Reaktive supports multithreading in Kotlin Native, please read the following documents before using it:
Object detachment is relatively difficult to achieve and is very error-prone when the objects are created from outside and
are not fully managed by the library. This is why Reaktive prefers frozen state. Here are some hints:
Any callback (and any captured objects) submitted to a Scheduler will be frozen
subscribeOn freezes both its upstream source and downstream observer,
all the Disposables (upstream's and downstream's) are frozen as well,
all the values (including errors) are not frozen by the operator
observeOn freezes only its downstream observer and all the values (including errors) passed through it, plus all the Disposables,
upstream source is not frozen by the operator
Other operators that use scheduler (like debounce, timer, delay, etc.) behave same as observeOn in most of the cases
Thread local tricks to avoid freezing
Sometimes freezing is not acceptable, e.g. we might want to load some data in background and then update the UI.
Obviously UI can not be frozen. With Reaktive it is possible to achieve such a behaviour in two ways:
Use threadLocal operator:
val values = mutableListOf<Any>()
var isFinished =false
observable<Any> { emitter ->// Background job
}
.subscribeOn(ioScheduler)
.observeOn(mainScheduler)
.threadLocal()
.doOnBeforeNext { values += it } // Callback is not frozen, we can updated the mutable list
.doOnBeforeFinally { isFinished =true } // Callback is not frozen, we can change the flag
.subscribe()
Set isThreadLocal flag to true in subscribe operator:
val values = mutableListOf<Any>()
var isComplete =false
observable<Any> { emitter ->// Background job
}
.subscribeOn(ioScheduler)
.observeOn(mainScheduler)
.subscribe(
isThreadLocal =true,
onNext = { values += it }, // Callback is not frozen, we can updated the mutable list
onComplete = { isComplete =true } // Callback is not frozen, we can change the flag
)
In both cases subscription (subscribe call) must be performed on the Main thread.
Reaktive and the new (relaxed) Kotlin/Native memory model
The new (relaxed) Kotlin/Native memory model
allows passing objects between threads without freezing. When using this memory model, there is no need
to use the threadLocal operator/argument anymore. Please make sure that you also disabled freezing
as described in the documentation.
Coroutines interop
This functionality is provided by the coroutines-interop module which is published in two versions:
coroutines-interop:<version> is based on stable kotlinx.coroutines - use this variant with the stable version of coroutines and with the old (strict) memory model.
coroutines-interop:<version>-nmtc is based on work-in-progress multi-threaded kotlinx.coroutines - use this variant with either the multi-threaded version of coroutines or the new (relaxed) memory model.
Coroutines interop based on stable kotlinx.coroutines
There are few important limitations:
Neither Job nor CoroutineContext can be frozen (until release of the multi-threaded coroutines).
Because of the first limitation all xxxFromCoroutine {} builders and Flow.asObservable() converter are executed inside runBlocking block in Kotlin/Native and should be subscribed on a background Scheduler.
Consider the following example for corutines-interop:
singleFromCoroutine {
// This block will be executed inside `runBlocking` in Kotlin/Native
}
.subscribeOn(ioScheduler) // Switching to a background thread is necessary
.observeOn(mainScheduler)
.subscribe { /* Get the result here */ }
Please note that Ktor uses multi-threaded coroutines by default. If you are using Ktor, please use coroutines-interop module based on multi-threaded coroutines and proceed to the next Readme secion.
Coroutines interop based on multi-threaded kotlinx.coroutines
The multi-threaded kotlinx.coroutines variant lifts some unpleasant restrictions - both Job and CoroutineContext can be frozen.
So there is one crucial difference - all xxxFromCoroutine {} builders and Flow.asObservable() converter are executed asynchronously in all targets (including Kotlin/Native), so can be subscribed on any scheduler.
Notes:
Because multi-threaded coroutines are work-in-progress, there are possible issues.
Ktor can be used out of the box without any known limitations
Coroutines interop general limitations
Converters Scheduler.asCoroutineDispatcher() and CoroutineContext.asScheduler() are available only in JVM and JS currently.
Subscription management with DisposableScope
Reaktive provides an easy way to manage subscriptions: DisposableScope.
Take a look at the following examples:
val scope =
disposableScope {
observable.subscribeScoped(...) // Subscription will be disposed when the scope is disposed
doOnDispose {
// Will be called when the scope is disposed
}
someDisposable.scope() // `someDisposable` will be disposed when the scope is disposed
}
// At some point later
scope.dispose()
classMyPresenter(
privatevalview:MyView,
privatevallongRunningAction:Completable
) : DisposableScope by DisposableScope() {
init {
doOnDispose {
// Will be called when the presenter is disposed
}
}
funload() {
view.showProgressBar()
// Subscription will be disposed when the presenter is disposed
longRunningAction.subscribeScoped(onComplete = view::hideProgressBar)
}
}
classMyActivity : AppCompatActivity(), DisposableScope by DisposableScope() {
overridefunonCreate(savedInstanceState:Bundle?) {
super.onCreate(savedInstanceState)
MyPresenter(...).scope()
}
overridefunonDestroy() {
dispose()
super.onDestroy()
}
}
客服电话
APP下载
官方微信
请发表评论