收集后订阅不起作用答案

【问题标题】：Subscribe after collect doesn't work收集后订阅不起作用
【发布时间】：2018-01-27 18:36:42
【问题描述】：

我正在试验 RxJava。我需要一个Observable，它会产生一个HashSet<String>。在Observable 中，我想将Pair<String, Boolean> 放入false 布尔值从结果HashSet 中删除String 键的方式。这是我所拥有的代码sn-p：

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 被调用，但我的订阅侦听器没有。

如何解决？

【问题讨论】：

标签： java android kotlin rx-java rx-java2

【解决方案1】：

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

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

【讨论】：

它必须保持打开状态。我希望能够不断地在那里放置新项目并获得 HashSet 作为结果。所以，每当我添加一个新项目时，新的 HashSet 必须以反应方式生成。
在这种情况下，在每个值之后重新创建集合会浪费大量内存和计算能力。 HashSet 应该在类范围而不是函数范围内。使用 1 PublishSubject 作为 Pair 并在它的 onSubscribe.onNext 中将值添加到`HashSet。使用另一个 Publish/BehaviorSubject 将 HashSet 提供给其他订阅者，并在第一个主题中有新对时发送新的更新。
是的，将 Set 存储在类级别是一种显而易见的方法。我猜它也可以通过自定义主题来完成。但我虽然有 smth 内置的，它允许实现相同的目标。如果没有更好的答案，我会接受你的。谢谢！