QueueDrainHelper { public static <T, U> boolean checkTerminated(boolean d, boolean empty, Subscriber<?> s, boolean delayError, SimpleQueue<?> q, QueueDrain<T, U> qd) { if (qd.cancelled()) { q.clear(); return true; } if (d) { if (delayError) { if (empty) { Throwable err = qd.error(); if (err != null) { s.onError(err); } else { s.onComplete(); } return true; } } else { Throwable err = qd.error(); if (err != null) { q.clear(); s.onError(err); return true; } else if (empty) { s.onComplete(); return true; } } } return false; } 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 checkTerminatedDelayErrorEmpty() { 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); QueueDrainHelper.checkTerminated(true, true, ts, true, q, qd); ts.assertResult(); }
@Test public void checkTerminatedDelayErrorNonEmpty() { 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); QueueDrainHelper.checkTerminated(true, false, ts, true, q, qd); ts.assertEmpty(); }
@Test public void checkTerminatedDelayErrorEmptyError() { 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 new TestException(); } @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); QueueDrainHelper.checkTerminated(true, true, ts, true, q, qd); ts.assertFailure(TestException.class); }
@Test public void checkTerminatedNonDelayErrorError() { 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 new TestException(); } @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); QueueDrainHelper.checkTerminated(true, false, ts, false, q, qd); ts.assertFailure(TestException.class); }
@Test public void observerCheckTerminatedDelayErrorEmpty() { TestObserver<Integer> to = new TestObserver<Integer>(); to.onSubscribe(Disposables.empty()); ObservableQueueDrain<Integer, Integer> qd = new ObservableQueueDrain<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 int leave(int m) { return 0; } @Override public void accept(Observer<? super Integer> a, Integer v) { } }; SpscArrayQueue<Integer> q = new SpscArrayQueue<Integer>(32); QueueDrainHelper.checkTerminated(true, true, to, true, q, null, qd); to.assertResult(); }
@Test public void observerCheckTerminatedDelayErrorEmptyResource() { TestObserver<Integer> to = new TestObserver<Integer>(); to.onSubscribe(Disposables.empty()); ObservableQueueDrain<Integer, Integer> qd = new ObservableQueueDrain<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 int leave(int m) { return 0; } @Override public void accept(Observer<? super Integer> a, Integer v) { } }; SpscArrayQueue<Integer> q = new SpscArrayQueue<Integer>(32); Disposable d = Disposables.empty(); QueueDrainHelper.checkTerminated(true, true, to, true, q, d, qd); to.assertResult(); assertTrue(d.isDisposed()); }
@Test public void observerCheckTerminatedDelayErrorNonEmpty() { TestObserver<Integer> to = new TestObserver<Integer>(); to.onSubscribe(Disposables.empty()); ObservableQueueDrain<Integer, Integer> qd = new ObservableQueueDrain<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 int leave(int m) { return 0; } @Override public void accept(Observer<? super Integer> a, Integer v) { } }; SpscArrayQueue<Integer> q = new SpscArrayQueue<Integer>(32); QueueDrainHelper.checkTerminated(true, false, to, true, q, null, qd); to.assertEmpty(); }
@Test public void observerCheckTerminatedDelayErrorEmptyError() { TestObserver<Integer> to = new TestObserver<Integer>(); to.onSubscribe(Disposables.empty()); ObservableQueueDrain<Integer, Integer> qd = new ObservableQueueDrain<Integer, Integer>() { @Override public boolean cancelled() { return false; } @Override public boolean done() { return false; } @Override public Throwable error() { return new TestException(); } @Override public boolean enter() { return true; } @Override public int leave(int m) { return 0; } @Override public void accept(Observer<? super Integer> a, Integer v) { } }; SpscArrayQueue<Integer> q = new SpscArrayQueue<Integer>(32); QueueDrainHelper.checkTerminated(true, true, to, true, q, null, qd); to.assertFailure(TestException.class); }
@Test public void observerCheckTerminatedNonDelayErrorError() { TestObserver<Integer> to = new TestObserver<Integer>(); to.onSubscribe(Disposables.empty()); ObservableQueueDrain<Integer, Integer> qd = new ObservableQueueDrain<Integer, Integer>() { @Override public boolean cancelled() { return false; } @Override public boolean done() { return false; } @Override public Throwable error() { return new TestException(); } @Override public boolean enter() { return true; } @Override public int leave(int m) { return 0; } @Override public void accept(Observer<? super Integer> a, Integer v) { } }; SpscArrayQueue<Integer> q = new SpscArrayQueue<Integer>(32); QueueDrainHelper.checkTerminated(true, false, to, false, q, null, qd); to.assertFailure(TestException.class); }
@Test public void observerCheckTerminatedNonDelayErrorErrorResource() { TestObserver<Integer> to = new TestObserver<Integer>(); to.onSubscribe(Disposables.empty()); ObservableQueueDrain<Integer, Integer> qd = new ObservableQueueDrain<Integer, Integer>() { @Override public boolean cancelled() { return false; } @Override public boolean done() { return false; } @Override public Throwable error() { return new TestException(); } @Override public boolean enter() { return true; } @Override public int leave(int m) { return 0; } @Override public void accept(Observer<? super Integer> a, Integer v) { } }; SpscArrayQueue<Integer> q = new SpscArrayQueue<Integer>(32); Disposable d = Disposables.empty(); QueueDrainHelper.checkTerminated(true, false, to, false, q, d, qd); to.assertFailure(TestException.class); assertTrue(d.isDisposed()); }