RxJava2 从监听器创建 Flowable 并在最后删除监听器答案

【问题标题】：RxJava2 create Flowable from listener and remove listener at the endRxJava2 从监听器创建 Flowable 并在最后删除监听器
【发布时间】：2017-02-26 10:40:21
【问题描述】：

我的用例与将 RxJava2 与 Firebase 数据库结合使用有关。

我有一个DatabaseReference，我可以向它注册值侦听器。我可以将它转换成这样的流：

disposable = Flowable.create<DataSnapshot>({ s ->
            dbRef.addValueEventListener(object : ValueEventListener {
                override fun onCancelled(p0: DatabaseError) {...}

                override fun onDataChange(value: DataSnapshot) {
                    s.onNext(value)
                }
            })
        }, BackpressureStrategy.BUFFER)
        .subscribe(...)

我希望能够在处置一次性用品时移除监听器。知道我该怎么做吗？

我看到在 rxjava 1 中可能有 this possibility，但它在 rxjava2 中不可用

【问题讨论】：

标签： java android rx-java2

【解决方案1】：

使用 RxJava2，您需要使用 setCancellable() 方法，并将您的侦听器删除代码放在那里。
这很像 RxJava1 中的 Emitter.setCancellation()，在使用 Observable.fromEmitter() 创建 Observable 时。

也请注意akarnokd 关于取消：
“但请注意，除非创建逻辑放弃调度程序（通过终止或异步），否则取消逻辑可能永远不会因为同池活锁而执行。” (RxJava 2: always unsubscribe on the .subscribeOn(..) scheduler?)

【讨论】：