### Question:
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);  }### Answer:
@Test public void testPostCompleteWithNoItems() { AssertSubscriber<Integer> subscriber = AssertSubscriber.create(10); ArrayDeque<Integer> queue = new ArrayDeque<>(); AtomicLong state = new AtomicLong(); Subscription subscription = mock(Subscription.class); subscriber.onSubscribe(subscription); DrainUtils.postComplete(subscriber, queue, state, () -> false); subscriber.assertHasNotReceivedAnyItem() .assertCompletedSuccessfully(); }
@Test public void testPostCompleteWithRequest() { AssertSubscriber<Integer> subscriber = AssertSubscriber.create(10); ArrayDeque<Integer> queue = new ArrayDeque<>(); AtomicLong state = new AtomicLong(); Subscription subscription = mock(Subscription.class); subscriber.onSubscribe(subscription); queue.offer(1); state.getAndIncrement(); DrainUtils.postComplete(subscriber, queue, state, () -> false); subscriber .assertCompletedSuccessfully() .assertReceived(1); }
@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(); }
@Test public void testPostCompleteCancelledAfterReceptionOfTheFirstItem() { final AssertSubscriber<Integer> subscriber = new AssertSubscriber<Integer>(10) { @Override public void onNext(Integer item) { super.onNext(item); cancel(); } }; ArrayDeque<Integer> queue = new ArrayDeque<>(); AtomicLong state = new AtomicLong(); Subscription subscription = mock(Subscription.class); subscriber.onSubscribe(subscription); queue.offer(1); state.getAndIncrement(); DrainUtils.postComplete(subscriber, queue, state, subscriber::isCancelled); subscriber .assertNotTerminated() .assertReceived(1) .assertHasNotFailed(); }
@Test public void testPostCompleteAlreadyComplete() { AssertSubscriber<Integer> subscriber = new AssertSubscriber<>(1); Queue<Integer> q = new ArrayDeque<>(); q.offer(1); AtomicLong state = new AtomicLong(DrainUtils.COMPLETED_MASK); DrainUtils.postComplete(subscriber, q, state, () -> false); }