static < T , F > boolean postCompleteDrain ( long n , Subscriber < ? super T > actual , Queue < T > queue , AtomicLongFieldUpdater < F > field , F instance , BooleanSupplier isCancelled ) { // TODO enable fast-path // if (n == -1 || n == Long.MAX_VALUE) { // for (;;) { // if (isDisposed.getAsBoolean()) { // break; // } // // T v = queue.poll(); // // if (v == null) { // actual.onComplete(); // break; // } // // actual.onNext(v); // } // // return true; // } long e = n & COMPLETED_MASK ; for ( ; ; ) { while ( e != n ) { if ( isCancelled . getAsBoolean ( ) ) { return true ; } T t = queue . poll ( ) ; if ( t == null ) { actual . onComplete ( ) ; return true ; } actual . onNext ( t ) ; e ++ ; } if ( isCancelled . getAsBoolean ( ) ) { return true ; } if ( queue . isEmpty ( ) ) { actual . onComplete ( ) ; return true ; } n = field . get ( instance ) ; if ( n == e ) { n = field . addAndGet ( instance , - ( e & REQUESTED_MASK ) ) ; if ( ( n & REQUESTED_MASK ) == 0L ) { return false ; } e = n & COMPLETED_MASK ; } } }
Drains the queue either in a pre - or post - complete state .