QueueDrainHelper { public static <T, U> void drainMaxLoop(SimplePlainQueue<T> q, Subscriber<? super U> a, boolean delayError, Disposable dispose, QueueDrain<T, U> qd) { int missed = 1; for (;;) { for (;;) { boolean d = qd.done(); T v = q.poll(); boolean empty = v == null; if (checkTerminated(d, empty, a, delayError, q, qd)) { if (dispose != null) { dispose.dispose(); } return; } if (empty) { break; } long r = qd.requested(); if (r != 0L) { if (qd.accept(a, v)) { if (r != Long.MAX_VALUE) { qd.produced(1); } } } else { q.clear(); if (dispose != null) { dispose.dispose(); } a.onError(new MissingBackpressureException("Could not emit value due to lack of requests.")); return; } } missed = qd.leave(-missed); if (missed == 0) { break; } } } private  QueueDrainHelper(); static void drainMaxLoop(SimplePlainQueue<T> q, Subscriber<? super U> a, boolean delayError,
            Disposable dispose, QueueDrain<T, U> qd); static boolean checkTerminated(boolean d, boolean empty,
            Subscriber<?> s, boolean delayError, SimpleQueue<?> q, QueueDrain<T, U> qd); static void drainLoop(SimplePlainQueue<T> q, Observer<? super U> a, boolean delayError, Disposable dispose, ObservableQueueDrain<T, U> qd); static boolean checkTerminated(boolean d, boolean empty,
            Observer<?> s, boolean delayError, SimpleQueue<?> q, Disposable disposable, ObservableQueueDrain<T, U> qd); static SimpleQueue<T> createQueue(int capacityHint); static void request(Subscription s, int prefetch); static boolean postCompleteRequest(long n,
                                                  Subscriber<? super T> actual,
                                                  Queue<T> queue,
                                                  AtomicLong state,
                                                  BooleanSupplier isCancelled); static void postComplete(Subscriber<? super T> actual,
                                        Queue<T> queue,
                                        AtomicLong state,
                                        BooleanSupplier isCancelled);  }
@Test public void drainMaxLoopMissingBackpressure() { TestSubscriber<Integer> ts = new TestSubscriber<Integer>(); ts.onSubscribe(new BooleanSubscription()); QueueDrain<Integer, Integer> qd = new QueueDrain<Integer, Integer>() { @Override public boolean cancelled() { return false; } @Override public boolean done() { return false; } @Override public Throwable error() { return null; } @Override public boolean enter() { return true; } @Override public long requested() { return 0; } @Override public long produced(long n) { return 0; } @Override public int leave(int m) { return 0; } @Override public boolean accept(Subscriber<? super Integer> a, Integer v) { return false; } }; SpscArrayQueue<Integer> q = new SpscArrayQueue<Integer>(32); q.offer(1); QueueDrainHelper.drainMaxLoop(q, ts, false, null, qd); ts.assertFailure(MissingBackpressureException.class); }
@Test public void drainMaxLoopMissingBackpressureWithResource() { TestSubscriber<Integer> ts = new TestSubscriber<Integer>(); ts.onSubscribe(new BooleanSubscription()); QueueDrain<Integer, Integer> qd = new QueueDrain<Integer, Integer>() { @Override public boolean cancelled() { return false; } @Override public boolean done() { return false; } @Override public Throwable error() { return null; } @Override public boolean enter() { return true; } @Override public long requested() { return 0; } @Override public long produced(long n) { return 0; } @Override public int leave(int m) { return 0; } @Override public boolean accept(Subscriber<? super Integer> a, Integer v) { return false; } }; SpscArrayQueue<Integer> q = new SpscArrayQueue<Integer>(32); q.offer(1); Disposable d = Disposables.empty(); QueueDrainHelper.drainMaxLoop(q, ts, false, d, qd); ts.assertFailure(MissingBackpressureException.class); assertTrue(d.isDisposed()); }
@Test public void drainMaxLoopDontAccept() { TestSubscriber<Integer> ts = new TestSubscriber<Integer>(); ts.onSubscribe(new BooleanSubscription()); QueueDrain<Integer, Integer> qd = new QueueDrain<Integer, Integer>() { @Override public boolean cancelled() { return false; } @Override public boolean done() { return false; } @Override public Throwable error() { return null; } @Override public boolean enter() { return true; } @Override public long requested() { return 1; } @Override public long produced(long n) { return 0; } @Override public int leave(int m) { return 0; } @Override public boolean accept(Subscriber<? super Integer> a, Integer v) { return false; } }; SpscArrayQueue<Integer> q = new SpscArrayQueue<Integer>(32); q.offer(1); QueueDrainHelper.drainMaxLoop(q, ts, false, null, qd); ts.assertEmpty(); }