src/main/java/io/reactivex/internal/util/QueueDrainHelper.java
public static <T> boolean postCompleteRequest(long n,
                                                  Subscriber<? super T> actual,
                                                  Queue<T> queue,
                                                  AtomicLong state,
                                                  BooleanSupplier isCancelled) {
        for (; ; ) {
            long r = state.get();

            // extract the current request amount
            long r0 = r & REQUESTED_MASK;

            // preserve COMPLETED_MASK and calculate new requested amount
            long u = (r & COMPLETED_MASK) | BackpressureHelper.addCap(r0, n);

            if (state.compareAndSet(r, u)) {
                // (complete, 0) -> (complete, n) transition then replay
                if (r == COMPLETED_MASK) {

                    postCompleteDrain(n | COMPLETED_MASK, actual, queue, state, isCancelled);

                    return true;
                }
                // (active, r) -> (active, r + n) transition then continue with requesting from upstream
                return false;
            }
        }

    }
public static <T> boolean postCompleteRequest(long n,
                                                  Subscriber<? super T> actual,
                                                  Queue<T> queue,
                                                  AtomicLong state,
                                                  BooleanSupplier isCancelled) {
        for (; ; ) {
            long r = state.get();

            // extract the current request amount
            long r0 = r & REQUESTED_MASK;

            // preserve COMPLETED_MASK and calculate new requested amount
            long u = (r & COMPLETED_MASK) | BackpressureHelper.addCap(r0, n);

            if (state.compareAndSet(r, u)) {
                // (complete, 0) -> (complete, n) transition then replay
                if (r == COMPLETED_MASK) {

                    postCompleteDrain(n | COMPLETED_MASK, actual, queue, state, isCancelled);

                    return true;
                }
                // (active, r) -> (active, r + n) transition then continue with requesting from upstream
                return false;
            }
        }

    }
public static < T > boolean postCompleteRequest ( long n , Subscriber < ? super T > actual , Queue < T > queue , AtomicLong state , BooleanSupplier isCancelled ) { for ( ; ; ) { long r = state . get ( ) ; // extract the current request amount long r0 = r & REQUESTED_MASK ; // preserve COMPLETED_MASK and calculate new requested amount long u = ( r & COMPLETED_MASK ) | BackpressureHelper . addCap ( r0 , n ) ; if ( state . compareAndSet ( r , u ) ) { // (complete, 0) -> (complete, n) transition then replay if ( r == COMPLETED_MASK ) { postCompleteDrain ( n | COMPLETED_MASK , actual , queue , state , isCancelled ) ; return true ; } // (active, r) -> (active, r + n) transition then continue with requesting from upstream return false ; } } }
Accumulates requests (not validated) and handles the completed mode draining of the queue based on the requests.

<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.

@param <T> the value type emitted
@param n the request amount, positive (not validated)
@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
@return true if the state indicates a completion state.
Accumulates requests ( not validated ) and handles the completed mode draining of the queue based on the requests .