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

		if (queue.isEmpty()) {
			actual.onComplete();
			return;
		}

		if (postCompleteDrain(field.get(instance), actual, queue, field, instance, isCancelled)) {
			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) {
					postCompleteDrain(u, actual, queue, field, instance, isCancelled);
				}

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

		if (queue.isEmpty()) {
			actual.onComplete();
			return;
		}

		if (postCompleteDrain(field.get(instance), actual, queue, field, instance, isCancelled)) {
			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) {
					postCompleteDrain(u, actual, queue, field, instance, isCancelled);
				}

				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 holding available values
@param field the field updater holding the requested amount
@param instance the parent instance of the requested field
@param isCancelled callback to detect cancellation