@Test public void testPostCompleteAfterCancellation() { AssertSubscriber<Integer> subscriber = new AssertSubscriber<>(1); ArrayDeque<Integer> queue = new ArrayDeque<>(); AtomicLong state = new AtomicLong(); Subscription subscription = mock(Subscription.class); subscriber.onSubscribe(subscription); queue.offer(1); state.getAndIncrement(); subscriber.cancel(); QueueDrainHelper.postComplete(subscriber, queue, state, subscriber::isCancelled); subscriber .assertSubscribed() .assertHasNotReceivedAnyItem() .assertNotTerminated(); }
public static <T> void postComplete(Subscriber<? super T> downstream, Queue<T> queue, AtomicLong requested, BooleanSupplier isCancelled) { if (queue.isEmpty()) { downstream.onComplete(); return; } if (postCompleteDrain(requested.get(), downstream, queue, requested, isCancelled)) { return; } for (;;) { long r = requested.get(); if ((r & COMPLETED_MASK) != 0L) { return; } long u = r | COMPLETED_MASK; if (requested.compareAndSet(r, u)) { if (r != 0L) { postCompleteDrain(u, downstream, queue, requested, isCancelled); } return; } } }
DrainUtils { public static <T> void postComplete(Subscriber<? super T> downstream, Queue<T> queue, AtomicLong requested, BooleanSupplier isCancelled) { if (queue.isEmpty()) { downstream.onComplete(); return; } if (postCompleteDrain(requested.get(), downstream, queue, requested, isCancelled)) { return; } for (;;) { long r = requested.get(); if ((r & COMPLETED_MASK) != 0L) { return; } long u = r | COMPLETED_MASK; if (requested.compareAndSet(r, u)) { if (r != 0L) { postCompleteDrain(u, downstream, queue, requested, isCancelled); } return; } } } }
DrainUtils { public static <T> void postComplete(Subscriber<? super T> downstream, Queue<T> queue, AtomicLong requested, BooleanSupplier isCancelled) { if (queue.isEmpty()) { downstream.onComplete(); return; } if (postCompleteDrain(requested.get(), downstream, queue, requested, isCancelled)) { return; } for (;;) { long r = requested.get(); if ((r & COMPLETED_MASK) != 0L) { return; } long u = r | COMPLETED_MASK; if (requested.compareAndSet(r, u)) { if (r != 0L) { postCompleteDrain(u, downstream, queue, requested, isCancelled); } return; } } } private  DrainUtils(); }
DrainUtils { public static <T> void postComplete(Subscriber<? super T> downstream, Queue<T> queue, AtomicLong requested, BooleanSupplier isCancelled) { if (queue.isEmpty()) { downstream.onComplete(); return; } if (postCompleteDrain(requested.get(), downstream, queue, requested, isCancelled)) { return; } for (;;) { long r = requested.get(); if ((r & COMPLETED_MASK) != 0L) { return; } long u = r | COMPLETED_MASK; if (requested.compareAndSet(r, u)) { if (r != 0L) { postCompleteDrain(u, downstream, queue, requested, isCancelled); } return; } } } private  DrainUtils(); static boolean postCompleteRequest(long n,
            Subscriber<? super T> downstream,
            Queue<T> queue,
            AtomicLong requested,
            BooleanSupplier isCancelled); static void postComplete(Subscriber<? super T> downstream,
            Queue<T> queue,
            AtomicLong requested,
            BooleanSupplier isCancelled); }
DrainUtils { public static <T> void postComplete(Subscriber<? super T> downstream, Queue<T> queue, AtomicLong requested, BooleanSupplier isCancelled) { if (queue.isEmpty()) { downstream.onComplete(); return; } if (postCompleteDrain(requested.get(), downstream, queue, requested, isCancelled)) { return; } for (;;) { long r = requested.get(); if ((r & COMPLETED_MASK) != 0L) { return; } long u = r | COMPLETED_MASK; if (requested.compareAndSet(r, u)) { if (r != 0L) { postCompleteDrain(u, downstream, queue, requested, isCancelled); } return; } } } private  DrainUtils(); static boolean postCompleteRequest(long n,
            Subscriber<? super T> downstream,
            Queue<T> queue,
            AtomicLong requested,
            BooleanSupplier isCancelled); static void postComplete(Subscriber<? super T> downstream,
            Queue<T> queue,
            AtomicLong requested,
            BooleanSupplier isCancelled);  }