src/main/java/io/reactivex/internal/util/QueueDrainHelper.java
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;
            // (active, r) -> (complete, r) transition
            if (state.compareAndSet(r, u)) {
                // if the requested amount was non-zero, drain the queue
                if (r != 0L) {
                    postCompleteDrain(u, actual, queue, state, isCancelled);
                }

                return;
            }
        }

    }
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;
            // (active, r) -> (complete, r) transition
            if (state.compareAndSet(r, u)) {
                // if the requested amount was non-zero, drain the queue
                if (r != 0L) {
                    postCompleteDrain(u, actual, queue, state, isCancelled);
                }

                return;
            }
        }

    }
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 ; // (active, r) -> (complete, r) transition if ( state . compareAndSet ( r , u ) ) { // if the requested amount was non-zero, drain the queue if ( r != 0L ) { postCompleteDrain ( u , actual , queue , state , isCancelled ) ; } return ; } } }
Signals the completion of the main sequence and switches to post-completion replay mode.

<p>
Don't modify the queue after calling this method!

<p>
Post-completion backpressure handles the case when a source produces values based on
requests when it is active but more values are available even after its completion.
In this case, the onComplete() can't just emit the contents of the queue but has to
coordinate with the requested amounts. This requires two distinct modes: active and
completed. In active mode, requests flow through and the queue is not accessed but
in completed mode, requests no-longer reach the upstream but help in draining the queue.
<p>
The algorithm utilizes the most significant bit (bit 63) of a long value (AtomicLong) since
request amount only goes up to Long.MAX_VALUE (bits 0-62) and negative values aren't
allowed.

@param <T> the value type emitted
@param actual the target Subscriber to send events to
@param queue the queue to drain if in the post-complete state
@param state holds the request amount and the post-completed flag
@param isCancelled a supplier that returns true if the drain has been cancelled
Signals the completion of the main sequence and switches to post - completion replay mode .