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();  }
@Test public void completeFirst() { TestSubscriber<Integer> ts = TestSubscriber.create(0L); TestingDeferredScalarSubscriber ds = new TestingDeferredScalarSubscriber(ts); ds.setupDownstream(); ds.onNext(1); ts.assertNoValues(); ds.onComplete(); ts.assertNoValues(); ts.request(1); ts.assertValues(1); ts.assertNoErrors(); ts.assertComplete(); }
@Test public void requestFirst() { TestSubscriber<Integer> ts = TestSubscriber.create(1); TestingDeferredScalarSubscriber ds = new TestingDeferredScalarSubscriber(ts); ds.setupDownstream(); ds.onNext(1); ts.assertNoValues(); ds.onComplete(); ts.assertValues(1); ts.assertNoErrors(); ts.assertComplete(); }
@Test public void empty() { TestSubscriber<Integer> ts = TestSubscriber.create(0L); TestingDeferredScalarSubscriber ds = new TestingDeferredScalarSubscriber(ts); ds.setupDownstream(); ts.assertNoValues(); ds.onComplete(); ts.assertNoValues(); ts.assertNoErrors(); ts.assertComplete(); }
@Test public void doubleComplete() { TestSubscriber<Integer> ts = TestSubscriber.create(0); TestingDeferredScalarSubscriber ds = new TestingDeferredScalarSubscriber(ts); ds.setupDownstream(); ds.onNext(1); ts.request(1); ds.onComplete(); ds.onComplete(); ts.assertValue(1); ts.assertNoErrors(); ts.assertComplete(); }
@Test public void doubleComplete2() { TestSubscriber<Integer> ts = TestSubscriber.create(0); TestingDeferredScalarSubscriber ds = new TestingDeferredScalarSubscriber(ts); ds.setupDownstream(); ds.onNext(1); ds.onComplete(); ds.onComplete(); ts.request(1); ts.assertValue(1); ts.assertNoErrors(); ts.assertComplete(); }
@Test public void doubleRequest() { TestSubscriber<Integer> ts = TestSubscriber.create(0); TestingDeferredScalarSubscriber ds = new TestingDeferredScalarSubscriber(ts); ds.setupDownstream(); ds.onNext(1); ts.request(1); ts.request(1); ds.onComplete(); ts.assertValue(1); ts.assertNoErrors(); ts.assertComplete(); }
@Test public void emissionRequestRace() { Worker w = Schedulers.computation().createWorker(); try { for (int i = 0; i < 10000; i++) { final TestSubscriber<Integer> ts = TestSubscriber.create(0L); TestingDeferredScalarSubscriber ds = new TestingDeferredScalarSubscriber(ts); ds.setupDownstream(); ds.onNext(1); final AtomicInteger ready = new AtomicInteger(2); w.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(); } }
@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(); } }