RXJava 取消订阅 subscribe 调用

【问题标题】：RXJava unsubscribe from the subscribe callRXJava 取消订阅 subscribe 调用
【发布时间】：2018-08-26 11:36:56
【问题描述】：

我有以下现有示例代码

val disposable = Observable.interval(interval, TimeUnit.SECONDS)
                .take((maxTimeForRequestInSecond / interval).toInt())
                .subscribe {
                    Manager.fetchAvailableSubscriptions(context, params) { status: WebService.Status, response: String? ->
                        if (status == WebService.Status.Success) {
                            // TODO stop the timer and dismiss observable!
                        }
                    }
                }

当订阅满足条件时，我如何关闭 Observable，就像我在 TODO 上一样？（我无法访问disposable）

注意：使用 RxJava 1.1.6

【问题讨论】：

标签： rx-java

【解决方案1】：

您的流有些损坏，请将fetchAvailableSubscriptions 表示为Observable<ResponseWithStatus>。这样你就没有回调中的回调，使用操作符takeUntil实现会更容易

您可以通过将Manager.fetchAvailableSubscriptions 包装成Observable.fromCallable 或Observable.create 来做到这一点

那么你的流看起来或多或少像

Disposable disposable = Observable.interval(interval, TimeUnit.SECONDS)
                .take((int)(maxTimeForRequestInSecond / interval))
                .flatMap(__ -> fetchAvailableSubscriptionsObservable())
                .takeUntil(responseWithStatus -> status == WebService.Status.Success)
                .subscribe(responseWithStatus -> {
                         // use the response 
                 });

【讨论】：