java - RxJava 2，DisposableSubscriber 释放时未调用 Flowable.doOnCancel

Question

我有一个 Flowable 的 DisposableSubscriber。 Flowable 会运行一段时间直到超时，但在某些情况下我需要提前终止它。现在我在 DisposableSubscriber 上调用 .dispose() 但 Flowable 继续发出事件。最终 Flowable 超时并调用 .doOnCancel()。

我有以下代码:

private Disposable mDisposableSubscription = null;

public void start() {
    mDisposableSubscription = getFlowable()
            .timeout(timeUntilTimeout, TimeUnit.MILLISECONDS)
            .subscribeWith(new DisposableSubscriber<T>() {
                @Override
                public void onComplete() {

                }
                @Override
                public void onError(Throwable throwable) {
                    dispose();
                }
                @Override
                public void onNext(T t) {
                    // Do something
                }
            });
}

public void stop() {
    // Log "dispose"
    mDisposableSubscription.dispose();
}

private Flowable<T> getFlowable() {
    return Flowable.create(new FlowableOnSubscribe<T>() {
        public void subscribe(FlowableEmitter<T> emitter) {
            startSomeAsyncProcess();
        }
    }).doOnCancel(new Action() {
        public void run() {
            // Log "do on cancel"
            stopSomeAsyncProcess();
        }
    });
}

在 Flowable 超时之前调用 stop() 来处理 DisposableSubscriber 意味着不再处理 Flowable 发出的事件，但 Flowable 继续发出事件并且异步进程继续运行。我的印象是调用 Flowable 下游的 .dispose() 会通过调用 .doOnCancel() 杀死 Flowable，但事实似乎并非如此。我错过了什么？

Answer 1

flowable 正在被释放，但是您没有在 Flowable.create 函数上检查它，所以发生的情况是 startSomeAsyncProcess() 忽略它并继续运行.

要解决此问题，您应该检查 emitter.isDispose() 标志以了解是否应该停止发射。

示例:

Flowable<T> getFlowable() {
    return Flowable.create(new FlowableOnSubscribe<T>() {

        public void subscribe(FlowableEmitter<T> emitter) {
            while(!emitter.isDisposed()) {
                emitter.onNext(...);
            }
        }

    });
}

如果该 startSomeAsyncProcess() 函数不允许您检查该标志，那么肯定有某种方法可以取消它。然后你可以附加一个可取消的:

Flowable<T> getFlowable() {
    return Flowable.create(new FlowableOnSubscribe<T>() {

        public void subscribe(FlowableEmitter<T> emitter) {
            startSomeAsyncProcess();
            emitter.setCancellable(() -> stopSomeAsyncProcess());
            // I don't remember if it's setCancellable() or setDisposable()
        }

    });
}

更新:方法 setCancellable(...) 和 setDisposable(...) 的行为应该相同，they just take different arguments .