如果调用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和氯氟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;
}