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;
			}
		}
	}
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
Drain the queue but give up with an error if there aren t enough requests .