src/main/java/io/reactivex/internal/util/QueueDrainHelper.java
static <T> boolean postCompleteDrain(long n,
                                         Subscriber<? super T> actual,
                                         Queue<T> queue,
                                         AtomicLong state,
                                         BooleanSupplier isCancelled) {

// TODO enable fast-path
//        if (n == -1 || n == Long.MAX_VALUE) {
//            for (;;) {
//                if (isCancelled.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(isCancelled)) {
                    return true;
                }

                T t = queue.poll();

                if (t == null) {
                    actual.onComplete();
                    return true;
                }

                actual.onNext(t);
                e++;
            }

            if (isCancelled(isCancelled)) {
                return true;
            }

            if (queue.isEmpty()) {
                actual.onComplete();
                return true;
            }

            n = state.get();

            if (n == e) {

                n = state.addAndGet(-(e & REQUESTED_MASK));

                if ((n & REQUESTED_MASK) == 0L) {
                    return false;
                }

                e = n & COMPLETED_MASK;
            }
        }

    }
static <T> boolean postCompleteDrain(long n,
                                         Subscriber<? super T> actual,
                                         Queue<T> queue,
                                         AtomicLong state,
                                         BooleanSupplier isCancelled) {

// TODO enable fast-path
//        if (n == -1 || n == Long.MAX_VALUE) {
//            for (;;) {
//                if (isCancelled.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(isCancelled)) {
                    return true;
                }

                T t = queue.poll();

                if (t == null) {
                    actual.onComplete();
                    return true;
                }

                actual.onNext(t);
                e++;
            }

            if (isCancelled(isCancelled)) {
                return true;
            }

            if (queue.isEmpty()) {
                actual.onComplete();
                return true;
            }

            n = state.get();

            if (n == e) {

                n = state.addAndGet(-(e & REQUESTED_MASK));

                if ((n & REQUESTED_MASK) == 0L) {
                    return false;
                }

                e = n & COMPLETED_MASK;
            }
        }

    }
static < T > boolean postCompleteDrain ( long n , Subscriber < ? super T > actual , Queue < T > queue , AtomicLong state , BooleanSupplier isCancelled ) { // TODO enable fast-path //        if (n == -1 || n == Long.MAX_VALUE) { //            for (;;) { //                if (isCancelled.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 ( isCancelled ) ) { return true ; } T t = queue . poll ( ) ; if ( t == null ) { actual . onComplete ( ) ; return true ; } actual . onNext ( t ) ; e ++ ; } if ( isCancelled ( isCancelled ) ) { return true ; } if ( queue . isEmpty ( ) ) { actual . onComplete ( ) ; return true ; } n = state . get ( ) ; if ( n == e ) { n = state . addAndGet ( - ( e & REQUESTED_MASK ) ) ; if ( ( n & REQUESTED_MASK ) == 0L ) { return false ; } e = n & COMPLETED_MASK ; } } }
Drains the queue based on the outstanding requests in post-completed mode (only!).

@param n the current request amount
@param actual the target Subscriber to send events to
@param queue the queue to drain if in the post-complete state
@param state holds the request amount and the post-completed flag
@param isCancelled a supplier that returns true if the drain has been cancelled
@return true if the queue was completely drained or the drain process was cancelled
Drains the queue based on the outstanding requests in post - completed mode ( only! ) .