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