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 .