reactor-core/src/main/java/reactor/core/publisher/DrainUtils.java
public static <T, F> void postCompleteDelayError(CoreSubscriber<? super T> actual,
            Queue<T> queue,
            AtomicLongFieldUpdater<F> field,
            F instance,
            BooleanSupplier isCancelled,
		    @Nullable Throwable error) {

        if (queue.isEmpty()) {
            if (error == null) {
                actual.onComplete();
            } else {
                actual.onError(error);
            }
            return;
        }

        if (postCompleteDrainDelayError(field.get(instance), actual, queue, field, instance, isCancelled, error)) {
            return;
        }

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

            if ((r & COMPLETED_MASK) != 0L) {
                return;
            }

            long u = r | COMPLETED_MASK;
            // (active, r) -> (complete, r) transition
            if (field.compareAndSet(instance, r, u)) {
                // if the requested amount was non-zero, drain the queue
                if (r != 0L) {
                    postCompleteDrainDelayError(u, actual, queue, field, instance, isCancelled, error);
                }

                return;
            }
        }
    }
public static <T, F> void postCompleteDelayError(CoreSubscriber<? super T> actual,
            Queue<T> queue,
            AtomicLongFieldUpdater<F> field,
            F instance,
            BooleanSupplier isCancelled,
		    @Nullable Throwable error) {

        if (queue.isEmpty()) {
            if (error == null) {
                actual.onComplete();
            } else {
                actual.onError(error);
            }
            return;
        }

        if (postCompleteDrainDelayError(field.get(instance), actual, queue, field, instance, isCancelled, error)) {
            return;
        }

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

            if ((r & COMPLETED_MASK) != 0L) {
                return;
            }

            long u = r | COMPLETED_MASK;
            // (active, r) -> (complete, r) transition
            if (field.compareAndSet(instance, r, u)) {
                // if the requested amount was non-zero, drain the queue
                if (r != 0L) {
                    postCompleteDrainDelayError(u, actual, queue, field, instance, isCancelled, error);
                }

                return;
            }
        }
    }
Tries draining the queue if the source just completed.

@param <T> the output value type
@param <F> the field type holding the requested 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