reactor-core/src/main/java/reactor/core/publisher/DrainUtils.java
public static <T, F> boolean postCompleteRequestDelayError(long n,
            Subscriber<? super T> actual,
            Queue<T> queue,
            AtomicLongFieldUpdater<F> field,
            F instance,
            BooleanSupplier isCancelled, Throwable error) {

        for (; ; ) {
            long r = field.get(instance);

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

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

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

                    postCompleteDrainDelayError(n | COMPLETED_MASK, actual, queue, field, instance, isCancelled, error);

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

    }
public static <T, F> boolean postCompleteRequestDelayError(long n,
            Subscriber<? super T> actual,
            Queue<T> queue,
            AtomicLongFieldUpdater<F> field,
            F instance,
            BooleanSupplier isCancelled, Throwable error) {

        for (; ; ) {
            long r = field.get(instance);

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

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

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

                    postCompleteDrainDelayError(n | COMPLETED_MASK, actual, queue, field, instance, isCancelled, error);

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

    }
Perform a potential post-completion request accounting.

@param <T> the output value type
@param <F> the field type holding the requested amount
@param n the request amount
@param actual the consumer of values
@param queue the queue of available values
@param field the field updater for the requested amount
@param instance the parent instance of the requested field
@param isCancelled callback to detect cancellation
@param error if not null, the error to signal after the queue has been drained
@return true if the state indicates a completion state.