src/main/java/io/reactivex/internal/util/QueueDrainHelper.java
public static <T, U> void drainMaxLoop(SimplePlainQueue<T> q, Subscriber<? super U> a, boolean delayError,
            Disposable dispose, QueueDrain<T, U> qd) {
        int missed = 1;

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

                T 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(new MissingBackpressureException("Could not emit value due to lack of requests."));
                    return;
                }
            }

            missed = qd.leave(-missed);
            if (missed == 0) {
                break;
            }
        }
    }
public static <T, U> void drainMaxLoop(SimplePlainQueue<T> q, Subscriber<? super U> a, boolean delayError,
            Disposable dispose, QueueDrain<T, U> qd) {
        int missed = 1;

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

                T 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(new MissingBackpressureException("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 <T> the queue value type
@param <U> 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