java - Collection 后订阅不起作用

Question

我正在尝试 RxJava。我需要一个 Observable ，它产生 HashSet<String> 。进入Observable我想成为阿贝尔把Pair<String, Boolean>以false的方式 boolean 值删除 String来自结果 HashSet 的 key 。这是我所拥有的代码 fragment :

private val selectionSubject = ReplaySubject.create<Pair<String, Boolean>>()
init {
    selectionSubject.onNext(Pair("dd", false))
    selectionSubject
        .collect({HashSet<String>()}, {dest, value -> collectSelection(dest, value)})
        .subscribe { t1, t2 -> Log.d(TAG, t1.toString())}
}

private fun collectSelection(dest: HashSet<String>, value: Pair<String, Boolean>): HashSet<String> {
    if (value.second) {
        dest.add(value.first)
    } else {
        dest.remove(value.first)
    }
    Log.d(TAG, "collectSelection, ${dest.toString()}")
    return dest
}

在日志中我可以看到collectSelection被调用，但我的订阅监听器没有被调用。

如何解决？

Answer 1

collect 在发出值之前等待来自上述流的 onComplete 事件。在您的情况下，ReplaySubject永远不会结束，因此不会发出任何值。

如果不知道 selectionSubject 的上下文，我无法为您的问题提供解决方案，例如，它是否必须保持开放状态？如果 onNext 调用有限，您可以使用 .take(X)。如果它必须保持打开状态，则不应依赖收集，而应将项目添加到 HashSet 中，例如 .doOnNext