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

// TODO enable fast-path
//        if (n == -1 || n == Long.MAX_VALUE) {
//            for (;;) {
//                if (isDisposed.getAsBoolean()) {
//                    break;
//                }
//
//                T v = queue.poll();
//
//                if (v == null) {
//                    actual.onComplete();
//                    break;
//                }
//
//                actual.onNext(v);
//            }
//
//            return true;
//        }

		long e = n & COMPLETED_MASK;

		for (; ; ) {

			while (e != n) {
				if (isCancelled.getAsBoolean()) {
					return true;
				}

				T t = queue.poll();

				if (t == null) {
					actual.onComplete();
					return true;
				}

				actual.onNext(t);
				e++;
			}

			if (isCancelled.getAsBoolean()) {
				return true;
			}

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

			n = field.get(instance);

			if (n == e) {

				n = field.addAndGet(instance, -(e & REQUESTED_MASK));

				if ((n & REQUESTED_MASK) == 0L) {
					return false;
				}

				e = n & COMPLETED_MASK;
			}
		}

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

// TODO enable fast-path
//        if (n == -1 || n == Long.MAX_VALUE) {
//            for (;;) {
//                if (isDisposed.getAsBoolean()) {
//                    break;
//                }
//
//                T v = queue.poll();
//
//                if (v == null) {
//                    actual.onComplete();
//                    break;
//                }
//
//                actual.onNext(v);
//            }
//
//            return true;
//        }

		long e = n & COMPLETED_MASK;

		for (; ; ) {

			while (e != n) {
				if (isCancelled.getAsBoolean()) {
					return true;
				}

				T t = queue.poll();

				if (t == null) {
					actual.onComplete();
					return true;
				}

				actual.onNext(t);
				e++;
			}

			if (isCancelled.getAsBoolean()) {
				return true;
			}

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

			n = field.get(instance);

			if (n == e) {

				n = field.addAndGet(instance, -(e & REQUESTED_MASK));

				if ((n & REQUESTED_MASK) == 0L) {
					return false;
				}

				e = n & COMPLETED_MASK;
			}
		}

	}
Drains the queue either in a pre- or post-complete state.

@param n 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
@return true if the queue was completely drained or the drain process was cancelled