@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(); }
@Override public void onComplete() { for (InnerCompletableCache inner : observers.getAndSet(TERMINATED)) { if (!inner.get()) { inner.actual.onComplete(); } } }
CompletableCache extends Completable implements CompletableObserver { @Override public void onComplete() { for (InnerCompletableCache inner : observers.getAndSet(TERMINATED)) { if (!inner.get()) { inner.actual.onComplete(); } } } }
CompletableCache extends Completable implements CompletableObserver { @Override public void onComplete() { for (InnerCompletableCache inner : observers.getAndSet(TERMINATED)) { if (!inner.get()) { inner.actual.onComplete(); } } } CompletableCache(CompletableSource source); }
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(); }
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();  }