RxJava CompletableSubenttoFlowable/to观察不发出

Question

所以我有一个制片人/出版商，它是CompletableSubject

我想要一个供订阅者/观察者使用的只读版本，但toFLowable（）和toWatable都没有发射任何东西

我错过了什么？

kotlin prettyprint-override">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

Answer 1

正如akarnokd在评论中所说，readOnlyStream.subscribe{Item-

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

完整主题的订阅方法没有onNext方法，因为Completables不发出任何项目，只发出onError或on完整事件。