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();  }
@Test public void onlineSuccess() { PublishProcessor<Integer> pp = PublishProcessor.create(); Maybe<Integer> source = pp.singleElement().cache(); assertFalse(pp.hasSubscribers()); assertNotNull(((MaybeCache<Integer>)source).source.get()); TestObserver<Integer> to = source.test(); assertNull(((MaybeCache<Integer>)source).source.get()); assertTrue(pp.hasSubscribers()); source.test(true).assertEmpty(); to.assertEmpty(); pp.onNext(1); pp.onComplete(); to.assertResult(1); source.test().assertResult(1); source.test(true).assertEmpty(); }
@Test public void onlineComplete() { PublishProcessor<Integer> pp = PublishProcessor.create(); Maybe<Integer> source = pp.singleElement().cache(); assertFalse(pp.hasSubscribers()); assertNotNull(((MaybeCache<Integer>)source).source.get()); TestObserver<Integer> to = source.test(); assertNull(((MaybeCache<Integer>)source).source.get()); assertTrue(pp.hasSubscribers()); source.test(true).assertEmpty(); to.assertEmpty(); pp.onComplete(); to.assertResult(); source.test().assertResult(); source.test(true).assertEmpty(); }
@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(); }
@Test public void crossCancelOnComplete() { final TestSubscriber<Integer> ts = new TestSubscriber<Integer>(); PublishProcessor<Integer> pp = PublishProcessor.create(); Maybe<Integer> source = pp.singleElement().cache(); source.subscribe(Functions.emptyConsumer(), Functions.emptyConsumer(), new Action() { @Override public void run() throws Exception { ts.cancel(); } }); source.toFlowable().subscribe(ts); pp.onComplete(); ts.assertEmpty(); }