QueueDrainSubscriber extends QueueDrainSubscriberPad4 implements FlowableSubscriber<T>, QueueDrain<U, V> { @Override public final long requested() { return requested.get(); } QueueDrainSubscriber(Subscriber<? super V> actual, SimplePlainQueue<U> queue); @Override final boolean cancelled(); @Override final boolean done(); @Override final boolean enter(); final boolean fastEnter(); @Override boolean accept(Subscriber<? super V> a, U v); @Override final Throwable error(); @Override final int leave(int m); @Override final long requested(); @Override final long produced(long n); final void requested(long n);  }
@Test public void unorderedFastPathRequest1() { TestSubscriber<Integer> ts = new TestSubscriber<Integer>(1); Disposable d = Disposables.empty(); QueueDrainSubscriber<Integer, Integer, Integer> qd = createUnordered(ts, d); ts.onSubscribe(new BooleanSubscription()); qd.requested(1); qd.onNext(1); ts.assertValuesOnly(1); }
@Test public void orderedFastPathRequest1() { TestSubscriber<Integer> ts = new TestSubscriber<Integer>(1); Disposable d = Disposables.empty(); QueueDrainSubscriber<Integer, Integer, Integer> qd = createOrdered(ts, d); ts.onSubscribe(new BooleanSubscription()); qd.requested(1); qd.onNext(1); ts.assertValuesOnly(1); }
@Test public void orderedSlowPathNonEmptyQueue() { TestSubscriber<Integer> ts = new TestSubscriber<Integer>(1); Disposable d = Disposables.empty(); QueueDrainSubscriber<Integer, Integer, Integer> qd = createOrdered(ts, d); ts.onSubscribe(new BooleanSubscription()); qd.queue.offer(0); qd.requested(2); qd.onNext(1); ts.assertValuesOnly(0, 1); }
@Test public void unorderedOnNextRace() { for (int i = 0; i < TestHelper.RACE_LONG_LOOPS; i++) { TestSubscriber<Integer> ts = new TestSubscriber<Integer>(1); Disposable d = Disposables.empty(); final QueueDrainSubscriber<Integer, Integer, Integer> qd = createUnordered(ts, d); ts.onSubscribe(new BooleanSubscription()); qd.requested(Long.MAX_VALUE); Runnable r1 = new Runnable() { @Override public void run() { qd.onNext(1); } }; TestHelper.race(r1, r1); ts.assertValuesOnly(1, 1); } }
@Test public void orderedOnNextRace() { for (int i = 0; i < TestHelper.RACE_LONG_LOOPS; i++) { TestSubscriber<Integer> ts = new TestSubscriber<Integer>(1); Disposable d = Disposables.empty(); final QueueDrainSubscriber<Integer, Integer, Integer> qd = createOrdered(ts, d); ts.onSubscribe(new BooleanSubscription()); qd.requested(Long.MAX_VALUE); Runnable r1 = new Runnable() { @Override public void run() { qd.onNext(1); } }; TestHelper.race(r1, r1); ts.assertValuesOnly(1, 1); } }
@Test public void unorderedFastPathReject() { TestSubscriber<Integer> ts = new TestSubscriber<Integer>(1); Disposable d = Disposables.empty(); QueueDrainSubscriber<Integer, Integer, Integer> qd = createUnorderedReject(ts, d); ts.onSubscribe(new BooleanSubscription()); qd.requested(1); qd.onNext(1); ts.assertValuesOnly(1); assertEquals(1, qd.requested()); }
@Test public void orderedFastPathReject() { TestSubscriber<Integer> ts = new TestSubscriber<Integer>(1); Disposable d = Disposables.empty(); QueueDrainSubscriber<Integer, Integer, Integer> qd = createOrderedReject(ts, d); ts.onSubscribe(new BooleanSubscription()); qd.requested(1); qd.onNext(1); ts.assertValuesOnly(1); assertEquals(1, qd.requested()); }