@Test public void emissionRequestRace2() { Worker w = Schedulers.io().createWorker(); Worker w2 = Schedulers.io().createWorker(); int m = 10000; if (Runtime.getRuntime().availableProcessors() < 3) { m = 1000; } try { for (int i = 0; i < m; i++) { final TestSubscriber<Integer> ts = TestSubscriber.create(0L); TestingDeferredScalarSubscriber ds = new TestingDeferredScalarSubscriber(ts); ds.setupDownstream(); ds.onNext(1); final AtomicInteger ready = new AtomicInteger(3); w.schedule(new Runnable() { @Override public void run() { ready.decrementAndGet(); while (ready.get() != 0) { } ts.request(1); } }); w2.schedule(new Runnable() { @Override public void run() { ready.decrementAndGet(); while (ready.get() != 0) { } ts.request(1); } }); ready.decrementAndGet(); while (ready.get() != 0) { } ds.onComplete(); ts.awaitTerminalEvent(5, TimeUnit.SECONDS); ts.assertValues(1); ts.assertNoErrors(); ts.assertComplete(); } } finally { w.dispose(); w2.dispose(); } }
@Override public void onComplete() { if (hasValue) { complete(value); } else { actual.onComplete(); } }
DeferredScalarSubscriber extends DeferredScalarSubscription<R> implements FlowableSubscriber<T> { @Override public void onComplete() { if (hasValue) { complete(value); } else { actual.onComplete(); } } }
DeferredScalarSubscriber extends DeferredScalarSubscription<R> implements FlowableSubscriber<T> { @Override public void onComplete() { if (hasValue) { complete(value); } else { actual.onComplete(); } } DeferredScalarSubscriber(Subscriber<? super R> actual); }
DeferredScalarSubscriber extends DeferredScalarSubscription<R> implements FlowableSubscriber<T> { @Override public void onComplete() { if (hasValue) { complete(value); } else { actual.onComplete(); } } DeferredScalarSubscriber(Subscriber<? super R> actual); @Override void onSubscribe(Subscription s); @Override void onError(Throwable t); @Override void onComplete(); @Override void cancel(); }
DeferredScalarSubscriber extends DeferredScalarSubscription<R> implements FlowableSubscriber<T> { @Override public void onComplete() { if (hasValue) { complete(value); } else { actual.onComplete(); } } DeferredScalarSubscriber(Subscriber<? super R> actual); @Override void onSubscribe(Subscription s); @Override void onError(Throwable t); @Override void onComplete(); @Override void cancel();  }