@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(); }
@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();  }