java - 如果调用 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;
}