@Test public void crossCancelOnSuccess() { final TestSubscriber<Integer> ts = new TestSubscriber<Integer>(); PublishProcessor<Integer> pp = PublishProcessor.create(); Maybe<Integer> source = pp.singleElement().cache(); source.subscribe(new Consumer<Integer>() { @Override public void accept(Integer v) throws Exception { ts.cancel(); } }); source.toFlowable().subscribe(ts); pp.onNext(1); pp.onComplete(); ts.assertEmpty(); }
@SuppressWarnings("unchecked") @Override public void onComplete() { for (CacheDisposable<T> inner : observers.getAndSet(TERMINATED)) { if (!inner.isDisposed()) { inner.actual.onComplete(); } } }
MaybeCache extends Maybe<T> implements MaybeObserver<T> { @SuppressWarnings("unchecked") @Override public void onComplete() { for (CacheDisposable<T> inner : observers.getAndSet(TERMINATED)) { if (!inner.isDisposed()) { inner.actual.onComplete(); } } } }
MaybeCache extends Maybe<T> implements MaybeObserver<T> { @SuppressWarnings("unchecked") @Override public void onComplete() { for (CacheDisposable<T> inner : observers.getAndSet(TERMINATED)) { if (!inner.isDisposed()) { inner.actual.onComplete(); } } } @SuppressWarnings("unchecked")  MaybeCache(MaybeSource<T> source); }
MaybeCache extends Maybe<T> implements MaybeObserver<T> { @SuppressWarnings("unchecked") @Override public void onComplete() { for (CacheDisposable<T> inner : observers.getAndSet(TERMINATED)) { if (!inner.isDisposed()) { inner.actual.onComplete(); } } } @SuppressWarnings("unchecked")  MaybeCache(MaybeSource<T> source); @Override void onSubscribe(Disposable d); @SuppressWarnings("unchecked") @Override void onSuccess(T value); @SuppressWarnings("unchecked") @Override void onError(Throwable e); @SuppressWarnings("unchecked") @Override void onComplete(); }
MaybeCache extends Maybe<T> implements MaybeObserver<T> { @SuppressWarnings("unchecked") @Override public void onComplete() { for (CacheDisposable<T> inner : observers.getAndSet(TERMINATED)) { if (!inner.isDisposed()) { inner.actual.onComplete(); } } } @SuppressWarnings("unchecked")  MaybeCache(MaybeSource<T> source); @Override void onSubscribe(Disposable d); @SuppressWarnings("unchecked") @Override void onSuccess(T value); @SuppressWarnings("unchecked") @Override void onError(Throwable e); @SuppressWarnings("unchecked") @Override void onComplete();  }