CompletableCache extends Completable implements CompletableObserver { @Override public void onComplete() { for (InnerCompletableCache inner : observers.getAndSet(TERMINATED)) { if (!inner.get()) { inner.actual.onComplete(); } } } CompletableCache(CompletableSource source); @Override void onSubscribe(Disposable d); @Override void onError(Throwable e); @Override void onComplete();  }
@Test public void crossDispose() { PublishSubject<Integer> ps = PublishSubject.create(); final TestObserver<Void> to1 = new TestObserver<Void>(); final TestObserver<Void> to2 = new TestObserver<Void>() { @Override public void onComplete() { super.onComplete(); to1.cancel(); } }; Completable c = ps.ignoreElements().cache(); c.subscribe(to2); c.subscribe(to1); ps.onComplete(); to1.assertEmpty(); to2.assertResult(); }
@Test public void dispose() { PublishSubject<Integer> ps = PublishSubject.create(); Completable c = ps.ignoreElements().cache(); assertFalse(ps.hasObservers()); TestObserver<Void> to1 = c.test(); assertTrue(ps.hasObservers()); to1.cancel(); assertTrue(ps.hasObservers()); TestObserver<Void> to2 = c.test(); TestObserver<Void> to3 = c.test(); to3.cancel(); TestObserver<Void> to4 = c.test(true); to3.cancel(); ps.onComplete(); to1.assertEmpty(); to2.assertResult(); to3.assertEmpty(); to4.assertEmpty(); }
@Test public void subscribeRace() { for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { PublishSubject<Integer> ps = PublishSubject.create(); final Completable c = ps.ignoreElements().cache(); final TestObserver<Void> to1 = new TestObserver<Void>(); final TestObserver<Void> to2 = new TestObserver<Void>(); Runnable r1 = new Runnable() { @Override public void run() { c.subscribe(to1); } }; Runnable r2 = new Runnable() { @Override public void run() { c.subscribe(to2); } }; TestHelper.race(r1, r2); ps.onComplete(); to1.assertResult(); to2.assertResult(); } }
@Test public void subscribeDisposeRace() { for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { PublishSubject<Integer> ps = PublishSubject.create(); final Completable c = ps.ignoreElements().cache(); final TestObserver<Void> to1 = c.test(); final TestObserver<Void> to2 = new TestObserver<Void>(); Runnable r1 = new Runnable() { @Override public void run() { to1.cancel(); } }; Runnable r2 = new Runnable() { @Override public void run() { c.subscribe(to2); } }; TestHelper.race(r1, r2); ps.onComplete(); to1.assertEmpty(); to2.assertResult(); } }