如果调用 doOnSubscribe() 返回的对象，RxJava2 dispose() 不起作用

Question

我无法理解为什么以下代码不起作用。我做错了什么还是 RxJava2 实现中的某种错误？

private Disposable savedDisposable;

@Test
public void test() {
    final TestObserver<Integer> observer = new TestObserver<>();

    Observable<Integer> t = Observable.just(10)
            .delay(100, TimeUnit.MILLISECONDS)
            .doOnSubscribe(disposable -> savedDisposable = disposable);

    t.subscribe(observer);

    savedDisposable.dispose();  //this doesn't work
    //observer.dispose();       //this works  

    assertTrue(observer.isDisposed());
}

Answer 1

回答发布的问题：

您在中间处理，因此末端 Disposable 无法知道其上游已被处理，因为 dispose() 调用总是向上游移动。

DisposableObserver、ResourceObserver、subscribeWith 和 lambda-subscribe() 方法会在最后得到一个 Disposable 对象您可以通过 dispose().

进行处理

不过，在问题列表中，OP 希望 Observer 和 Disposable 出现在消费者类型上，并发现这可以通过受约束的泛型来实现，例如：

public static <T, K extends Observer<T> & Disposable> K subscribe(
        Observable<T> o, K observer) {
    o.subscribe(observer);
    return observer;
}

如果调用 doOnSubscribe() 返回的对象，RxJava2 dispose() 不起作用

RxJava2 dispose() doesn't work if called on the object returned by doOnSubscribe()

java

rx-java

rx-java2