自动完成从另一个 Observable 创建的 Single

Question

我有一个 long-运行操作，returns 代码中的一个值是我无法控制的。我需要将该值发布给需要它的东西。为此，我使用 BehaviorSubject:

var subject: Subject<Value>? = null

fun retrieveValue(): Single<Value> {

    if (subject == null) {
        subject = BehaviorSubject.create<Value>()

        someOtherThing.retrieveValueAsync { value ->
            subject.onNext(value)
        }
    }

    return subject.singleOrError()
}

这让我只执行一次操作，并将结果作为单个结果发送给所有未来感兴趣的各方。但是，它不起作用。在我调用之前，单个不会发出值：

subject.onComplete()

但这是一个问题，因为一旦主题完成，未来的东西就无法再订阅它。

从另一个 observable 缓存值并将其传递给 Single 的合适方法是什么？如果有一种方法可以让一个主题在其源可观察对象发出一个有效的值后自动完成。 Single.cache() 看起来也很有前途，但我不确定在那种情况下我将如何处理我的值异步传入的事实。

感觉好像少了什么傻事

Answer 1

这个案例有一个SingleSubject。

如果您不想在您的代码库中使用实验代码，您可以使用 ReplaySubject.createWithSize(1) 并在不丢失最后一个值的情况下调用 onComplete，然后将其转换为 Single。

自动完成从另一个 Observable 创建的 Single

Auto-complete a Single created from another Observable

rx-java2