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