Reaktive Documentation

DisposableScope provides an easy way to manage the lifecycle of multiple subscriptions. When the scope is disposed, all subscriptions and disposables associated with it are automatically cleaned up.

You can use the DisposableScope interface via delegation to integrate it directly into your classes (like Presenters or Activities).

Usage Example

val scope = disposableScope {
    observable.subscribeScoped(...) // Disposed when scope is disposed

    doOnDispose { /* Called when scope is disposed */ }

    someDisposable.scope() // someDisposable is disposed when scope is disposed
}

// Later
scope.dispose()

Integration Example

class MyPresenter(...) : DisposableScope by DisposableScope() {
    fun load() {
        // Subscription will be disposed when the presenter is disposed
        longRunningAction.subscribeScoped(onComplete = view::hideProgressBar)
    }
}

class MyActivity : AppCompatActivity(), DisposableScope by DisposableScope() {
    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        MyPresenter(...).scope()
    }

    override fun onDestroy() {
        dispose()
        super.onDestroy()
    }
}

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()

While there is no official interop module, you can bridge Reaktive wrappers to RxSwift by implementing extension methods on RxSwift.Observable, RxSwift.Single, RxSwift.Maybe, and RxSwift.Completable. These extensions use the .create operator to map Reaktive subscription events to RxSwift observer events.

import RxSwift
import YourKotlinFramework

extension RxSwift.Observable where Element : AnyObject {
    static func from(_ observable: ObservableWrapper<Element>) -> RxSwift.Observable<Element> {
        return RxSwift.Observable<Element>.create { observer in
            let disposable = observable.subscribe(
                onError: { observer.onError(KotlinError($0)) },
                onComplete: observer.onCompleted,
                onNext: observer.onNext
            )

            return Disposables.create(with: disposable.dispose)
        }
    }
}

extension RxSwift.Single where Element : AnyObject {
    static func from(_ single: SingleWrapper<Element>) -> RxSwift.Single<Element> {
        return RxSwift.Single<Element>.create { observer in
            let disposable = single.subscribe(
                onError: { observer(.failure(KotlinError($0))) },
                onSuccess: { observer(.success($0)) }
            )

            return Disposables.create(with: disposable.dispose)
        }
    }
}

extension RxSwift.Maybe where Element : AnyObject {
    static func from(_ maybe: MaybeWrapper<Element>) -> RxSwift.Maybe<Element> {
        return RxSwift.Maybe<Element>.create { observer in
            let disposable = maybe.subscribe(
                onError: { observer(.error(KotlinError($0))) },
                onComplete: { observer(.completed) },
                onSuccess: { observer(.success($0)) }
            )

            return Disposables.create(with: disposable.dispose)
        }
    }
}

extension RxSwift.Completable {
    static func from(_ completable: CompletableWrapper) -> RxSwift.Completable {
        return RxSwift.Completable.create { observer in
            let disposable = completable.subscribe(
                onError: { observer(.error(KotlinError($0))) },
                onComplete: { observer(.completed) }
            )

            return Disposables.create(with: disposable.dispose)
        }
    }
}

struct KotlinError : Error {
    let throwable: KotlinThrowable

    init (_ throwable: KotlinThrowable) {
        self.throwable = throwable
    }
}

Reaktive is distributed via Maven Central as several multiplatform modules. Depending on your needs, you can include the core library, annotations, testing utilities, or interoperability modules for Coroutines and RxJava.

Add the following to your build.gradle file, replacing <version> with the desired version number:

kotlin {
    sourceSets {
        commonMain {
            dependencies {
                implementation 'com.badoo.reaktive:reaktive:<version>'
                implementation 'com.badoo.reaktive:reaktive-annotations:<version>'
                implementation 'com.badoo.reaktive:coroutines-interop:<version>' // For interop with coroutines
                implementation 'com.badoo.reaktive:rxjava2-interop:<version>' // For interop with RxJava v2
                implementation 'com.badoo.reaktive:rxjava3-interop:<version>' // For interop with RxJava v3
            }
        }

        commonTest {
            dependencies {
                implementation 'com.badoo.reaktive:reaktive-testing:<version>'
            }
        }
    }
}

The Plugin API allows you to decorate Reaktive sources (Observables, Singles, Maybes, Completables). To create a plugin, implement the ReaktivePlugin interface and register it using registerReaktivePlugin.

object MyPlugin : ReaktivePlugin {
    override fun <T> onAssembleObservable(observable: Observable<T>): Observable<T> = 
        // Return a decorated Observable

    override fun <T> onAssembleSingle(single: Single<T>): Single<T> = TODO()
    override fun <T> onAssembleMaybe(maybe: Maybe<T>): Maybe<T> = TODO()
    override fun onAssembleCompletable(completable: Completable): Completable = TODO()
}

object MyPlugin : ReaktivePlugin {
    override fun <T> onAssembleObservable(observable: Observable<T>): Observable<T> =
        object : Observable<T> {
            private val traceException = TraceException()

            override fun subscribe(observer: ObservableObserver<T>) {
                observable.subscribe(
                    object : ObservableObserver<T> by observer {
                        override fun onError(error: Throwable) {
                            observer.onError(error, traceException)
                        }
                    }
                )
            }
        }

    override fun <T> onAssembleSingle(single: Single<T>): Single<T> =
        TODO("Similar to onAssembleSingle")

    override fun <T> onAssembleMaybe(maybe: Maybe<T>): Maybe<T> =
        TODO("Similar to onAssembleSingle")

    override fun onAssembleCompletable(completable: Completable): Completable =
        TODO("Similar to onAssembleSingle")

    private fun ErrorCallback.onError(error: Throwable, traceException: TraceException) {
        if (error.suppressedExceptions.lastOrNull() !is TraceException) {
            error.addSuppressed(traceException)
        }
        onError(error)
    }

    private class TraceException : Exception()
}

Because Kotlin interface generics are not exported to Swift, you cannot use Observable<T>, Single<T>, etc., directly in Swift. Instead, you must use the provided wrap() extension functions in your Kotlin code to convert these sources into wrapper classes that Swift can consume.

Available Wrappers:

• Observable<T>.wrap() $\rightarrow$ ObservableWrapper<T>
• BehaviorObservable<T>.wrap() $\rightarrow$ BehaviorObservableWrapper<T> (also works for BehaviorSubject)
• Single<T>.wrap() $\rightarrow$ SingleWrapper<T>
• Maybe<T>.wrap() $\rightarrow$ MaybeWrapper<T>
• Completable.wrap() $\rightarrow$ CompletableWrapper

class SharedDataSource {
    fun load(): SingleWrapper<String> =
        singleFromFunction {
            // A long running operation
            "A result"
        }
            .subscribeOn(ioScheduler)
            .observeOn(mainScheduler)
            .wrap()
}

class SharedViewModel {
    private val _state = BehaviorSubject(State())
    val state: BehaviorObservableWrapper<State> = _state.wrap()
}