@Test public void postCompleteEmpty() { TestSubscriber<Integer> ts = new TestSubscriber<Integer>(); ArrayDeque<Integer> queue = new ArrayDeque<Integer>(); AtomicLong state = new AtomicLong(); BooleanSupplier isCancelled = new BooleanSupplier() { @Override public boolean getAsBoolean() throws Exception { return false; } }; ts.onSubscribe(new BooleanSubscription()); QueueDrainHelper.postComplete(ts, queue, state, isCancelled); ts.assertResult(); }
public static <T> void postComplete(Subscriber<? super T> actual, Queue<T> queue, AtomicLong state, BooleanSupplier isCancelled) { if (queue.isEmpty()) { actual.onComplete(); return; } if (postCompleteDrain(state.get(), actual, queue, state, isCancelled)) { return; } for (; ; ) { long r = state.get(); if ((r & COMPLETED_MASK) != 0L) { return; } long u = r | COMPLETED_MASK; if (state.compareAndSet(r, u)) { if (r != 0L) { postCompleteDrain(u, actual, queue, state, isCancelled); } return; } } }
QueueDrainHelper { public static <T> void postComplete(Subscriber<? super T> actual, Queue<T> queue, AtomicLong state, BooleanSupplier isCancelled) { if (queue.isEmpty()) { actual.onComplete(); return; } if (postCompleteDrain(state.get(), actual, queue, state, isCancelled)) { return; } for (; ; ) { long r = state.get(); if ((r & COMPLETED_MASK) != 0L) { return; } long u = r | COMPLETED_MASK; if (state.compareAndSet(r, u)) { if (r != 0L) { postCompleteDrain(u, actual, queue, state, isCancelled); } return; } } } }
QueueDrainHelper { public static <T> void postComplete(Subscriber<? super T> actual, Queue<T> queue, AtomicLong state, BooleanSupplier isCancelled) { if (queue.isEmpty()) { actual.onComplete(); return; } if (postCompleteDrain(state.get(), actual, queue, state, isCancelled)) { return; } for (; ; ) { long r = state.get(); if ((r & COMPLETED_MASK) != 0L) { return; } long u = r | COMPLETED_MASK; if (state.compareAndSet(r, u)) { if (r != 0L) { postCompleteDrain(u, actual, queue, state, isCancelled); } return; } } } private  QueueDrainHelper(); }
QueueDrainHelper { public static <T> void postComplete(Subscriber<? super T> actual, Queue<T> queue, AtomicLong state, BooleanSupplier isCancelled) { if (queue.isEmpty()) { actual.onComplete(); return; } if (postCompleteDrain(state.get(), actual, queue, state, isCancelled)) { return; } for (; ; ) { long r = state.get(); if ((r & COMPLETED_MASK) != 0L) { return; } long u = r | COMPLETED_MASK; if (state.compareAndSet(r, u)) { if (r != 0L) { postCompleteDrain(u, actual, queue, state, isCancelled); } return; } } } 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); }
QueueDrainHelper { public static <T> void postComplete(Subscriber<? super T> actual, Queue<T> queue, AtomicLong state, BooleanSupplier isCancelled) { if (queue.isEmpty()) { actual.onComplete(); return; } if (postCompleteDrain(state.get(), actual, queue, state, isCancelled)) { return; } for (; ; ) { long r = state.get(); if ((r & COMPLETED_MASK) != 0L) { return; } long u = r | COMPLETED_MASK; if (state.compareAndSet(r, u)) { if (r != 0L) { postCompleteDrain(u, actual, queue, state, isCancelled); } return; } } } 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);  }