How to create caching/hot version of rx.Single?

Question

The RxJava v1.0.13 introduced new type of an Observable: rx.Single. It fits great the request-response model but lacks the standard side-effects introducing operators like doOnNext(). So, it's much harder to make multiple things happen as a result.

My idea was to replace doOnNext() with multiple subscriptions to the same Single instance. But this can cause the underlaying work to be done multiple times: once per every subscription.

Example rx.Single implementation:

private class WorkerSubscribe<SomeData>() : Single.OnSubscribe<SomeData> {
    override fun call(sub: SingleSubscriber<in SomeData>) {
        try {
            val result = fetchSomeData()
            sub.onSuccess(result)
        } catch(t: Throwable) {
            sub.onError(t)
        }
    }
}

val single = Single.create<SomeData>(WorkerSubscribe())

Usage:

single.subscribe({}, {})
single.subscribe({}, {})   // Data is fetched for the second time

Is it possible to create a instance of Single that will not fetchSomeData() multiple times even when single.subscribe() is called multiple times, but cache and return the same result?

Answer 1

You need RxJava Subject: BehaviorSubject or AsyncSubject