RxJava CompletableSubject toFlowable/toObservable 不发射答案

【问题标题】：RxJava CompletableSubject toFlowable/ toObservable doesn't emitRxJava CompletableSubject toFlowable/toObservable 不发射
【发布时间】：2021-04-05 21:44:26
【问题描述】：

所以我有一个 CompletableSubject 的制作者/发布者

我想要一个只读版本供订阅者/观察者使用但toFLowable() 和 toObservable 都没有发射任何东西

我错过了什么？

fun main() {
    val publisher = CompletableSubject.create()

    val readOnlyStream = publisher.toFlowable<Any>()

    println("1 ${publisher.hasComplete()}")
    readOnlyStream.subscribe { item ->
        println("yay, got it $item")
    }

    println("2 ${publisher.hasComplete()}")
    publisher.onComplete()
    println("3 ${publisher.hasComplete()}")

    Thread.sleep(3000L)
}

输出：

1 false
2 false
3 true

我发现了奇怪的解决方法

fun main() {
    val publisher = CompletableSubject.create()

    val readOnlyStream = publisher.toSingle {
        "doneee"
    }

    println("1 ${publisher.hasComplete()}")
    readOnlyStream.subscribe { item ->
        println("yay, got it $item")
    }

    println("2 ${publisher.hasComplete()}")
    publisher.onComplete()
    println("3 ${publisher.hasComplete()}")

    Thread.sleep(3000L)
}

输出：

1 false
2 false
yay, got it doneee
3 true

【问题讨论】：

CompletableSubject 只能这样终止并且没有 onNext 的概念。因此，您必须在其终止时注入一个值。 toSingle 这样做。
@akarnokd，谢谢。我的一个团队成员正在做completableSubject.subscribe {}，这意味着无论发布者如何，消费者/观察者都可能错误地导致终止。我不禁觉得它与其他类型的主题相当不一致

标签： kotlin rx-java rx-java2

【解决方案1】：

正如 akarnokd 在 cmets 中所说，readOnlyStream.subscribe { item -> 正在为项目添加回调，并且从未发出项目。然而，这并不意味着 onComplete 事件没有传播到 Flowable。 onComplete 事件可以这样订阅：

readOnlyStream.subscribe(
    {}, // onNext, called for each item
    {}, // onError, called once when there is an error
    { println("yay, got onComplete") }, // onComplete, called once when completed
)

completableSubject 的 subscribe 方法没有 onNext 方法，因为 Completables 不会发出任何项目，只会发出 onError 或 onComplete 事件。

【讨论】：