手动停止 takeUntil() 内的 Observable.timer()

【问题标题】：Stop Observable.timer() inside takeUntil() manually手动停止 takeUntil() 内的 Observable.timer()
【发布时间】：2021-10-03 16:11:38
【问题描述】：

案例：在接下来的 5 秒内观察到一些项目，但我也想手动终止获取项目。

我有这个代码


private var disposable: Disposable? = null

...

observableThatHasToBeAliveAllTime
    .switchMap {
        observableThatEmitsItemOver5SecsWhenUpperObsEmits()
            .takeUntil(
                Observable.timer(5, TimeUnit.SECONDS)
                    .also { disposable = it.subscribe() }
            )
            .switchMap { /* some work */ }
    }
    .subscribe { /* handle result */ }

在其他地方我打电话给disposable?.dispose，但 takeUntil() 继续工作

我做错了什么？

【问题讨论】：

标签： android observable rx-java rx-java2

【解决方案1】：

我做错了什么？

it.subscribe() 订阅timer 与takeUntil 完全分开并且没有任何效果。此外，takeUntil 需要来自其他来源的 onNext 或 onComplete 信号，因此 dispose 也无法按预期工作。

您可以使用单独的主题和另一个 takeUntil 来停止流程：

var stop = PublishSubject.create<Object>()

observableThatHasToBeAliveAllTime
    .switchMap {
        observableThatEmitsItemOver5SecsWhenUpperObsEmits()
            .takeUntil(
                Observable.timer(5, TimeUnit.SECONDS)
            )
            .takeUntil(stop)
            .switchMap { /* some work */ }
    }
    .subscribe { /* handle result */ }

stop.onNext("Stop!");

【讨论】：