reactor-core/src/main/java/reactor/core/publisher/QueueDrainSubscriber.java
static <Q, S> void drainMaxLoop(Queue<Q> q, Subscriber<? super S> a, boolean delayError,
			Disposable dispose, QueueDrainSubscriber<?, Q, S> qd) {
		int missed = 1;

		for (;;) {
			for (;;) {
				boolean d = qd.done();

				Q v = q.poll();

				boolean empty = v == null;

				if (checkTerminated(d, empty, a, delayError, q, qd)) {
					if (dispose != null) {
						dispose.dispose();
					}
					return;
				}

				if (empty) {
					break;
				}

				long r = qd.requested();
				if (r != 0L) {
					if (qd.accept(a, v)) {
						if (r != Long.MAX_VALUE) {
							qd.produced(1);
						}
					}
				} else {
					q.clear();
					if (dispose != null) {
						dispose.dispose();
					}
					a.onError(Exceptions.failWithOverflow("Could not emit value due to lack of requests."));
					return;
				}
			}

			missed = qd.leave(-missed);
			if (missed == 0) {
				break;
			}
		}
	}
static <Q, S> void drainMaxLoop(Queue<Q> q, Subscriber<? super S> a, boolean delayError,
			Disposable dispose, QueueDrainSubscriber<?, Q, S> qd) {
		int missed = 1;

		for (;;) {
			for (;;) {
				boolean d = qd.done();

				Q v = q.poll();

				boolean empty = v == null;

				if (checkTerminated(d, empty, a, delayError, q, qd)) {
					if (dispose != null) {
						dispose.dispose();
					}
					return;
				}

				if (empty) {
					break;
				}

				long r = qd.requested();
				if (r != 0L) {
					if (qd.accept(a, v)) {
						if (r != Long.MAX_VALUE) {
							qd.produced(1);
						}
					}
				} else {
					q.clear();
					if (dispose != null) {
						dispose.dispose();
					}
					a.onError(Exceptions.failWithOverflow("Could not emit value due to lack of requests."));
					return;
				}
			}

			missed = qd.leave(-missed);
			if (missed == 0) {
				break;
			}
		}
	}
Drain the queue but give up with an error if there aren't enough requests.
@param <Q> the queue value type
@param <S> the emission value type
@param q the queue
@param a the subscriber
@param delayError true if errors should be delayed after all normal items
@param dispose the disposable to call when termination happens and cleanup is necessary
@param qd the QueueDrain instance that gives status information to the drain logic