如何创建rx的缓存/热版本。单身？

Question

RxJava v1.0.13引入了一种新的可观测的类型：rx。单身。它非常适合请求-响应模型，但缺乏引入像doOnNext（）这样的运算符的标准副作用。因此，很难让多种事情发生。

我的想法是用同一个单一实例的多个订阅替换doOnNext（）。但这可能会导致基础工作进行多次：每次订阅一次。

例如rx。单一实施：

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

用法：

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

是否可以创建Single的实例，即使它是Single，也不会多次获取SomeData（）。subscribe（）被多次调用，但是缓存并返回相同的结果吗？

Answer 1

请检查cache（）操作符。它应该缓存可观测的的发射，并将其复制到后续的订户s。

Answer 2

我只是需要类似的行为，并找到了一些解决方案。

您可以将Single转换为Observableapplycache（），然后将其转换回Single。

yourSingle.toObservable().cacheWithInitialCapacity(1).toSingle()

我使用cacheWithInitialCapacity（1）而不仅仅是cache（）作为优化-单个永远不会发出多个项目。

它也是好主意提供变压器实现

public class SingleUtils {

    public static <T> Single.Transformer<T, T> cached() {
        return single -> single.toObservable()
            .cacheWithInitialCapacity(1)
            .toSingle();
    }
}

因此，您可以在任何需要的地方使用缓存，只需调用

yourSingle.compose(SingleUtils.cached())

编辑：从rxJava 1.2.2开始，添加了(https://github.com/ReactiveX/RxJava/releases/tag/v1.2.2)

就是这样实现的(https://github.com/ReactiveX/RxJava/pull/4757)

Answer 3

您需要RxJava主题：BehaviorSubject或AsyncSubject