CompletableCache extends Completable implements CompletableObserver { @Override public void onError(Throwable e) { error = e; for (InnerCompletableCache inner : observers.getAndSet(TERMINATED)) { if (!inner.get()) { inner.actual.onError(e); } } } CompletableCache(CompletableSource source); @Override void onSubscribe(Disposable d); @Override void onError(Throwable e); @Override void onComplete();  }
@Test public void crossDisposeOnError() { PublishSubject<Integer> ps = PublishSubject.create(); final TestObserver<Void> to1 = new TestObserver<Void>(); final TestObserver<Void> to2 = new TestObserver<Void>() { @Override public void onError(Throwable ex) { super.onError(ex); to1.cancel(); } }; Completable c = ps.ignoreElements().cache(); c.subscribe(to2); c.subscribe(to1); ps.onError(new TestException()); to1.assertEmpty(); to2.assertFailure(TestException.class); }