src/main/java/io/reactivex/internal/operators/flowable/FlowableBlockingSubscribe.java
public static <T> void subscribe(Publisher<? extends T> o, Subscriber<? super T> subscriber) {
        final BlockingQueue<Object> queue = new LinkedBlockingQueue<Object>();

        BlockingSubscriber<T> bs = new BlockingSubscriber<T>(queue);

        o.subscribe(bs);

        try {
            for (;;) {
                if (bs.isCancelled()) {
                    break;
                }
                Object v = queue.poll();
                if (v == null) {
                    if (bs.isCancelled()) {
                        break;
                    }
                    BlockingHelper.verifyNonBlocking();
                    v = queue.take();
                }
                if (bs.isCancelled()) {
                    break;
                }
                if (v == BlockingSubscriber.TERMINATED
                        || NotificationLite.acceptFull(v, subscriber)) {
                    break;
                }
            }
        } catch (InterruptedException e) {
            bs.cancel();
            subscriber.onError(e);
        }
    }
public static <T> void subscribe(Publisher<? extends T> o, Subscriber<? super T> subscriber) {
        final BlockingQueue<Object> queue = new LinkedBlockingQueue<Object>();

        BlockingSubscriber<T> bs = new BlockingSubscriber<T>(queue);

        o.subscribe(bs);

        try {
            for (;;) {
                if (bs.isCancelled()) {
                    break;
                }
                Object v = queue.poll();
                if (v == null) {
                    if (bs.isCancelled()) {
                        break;
                    }
                    BlockingHelper.verifyNonBlocking();
                    v = queue.take();
                }
                if (bs.isCancelled()) {
                    break;
                }
                if (v == BlockingSubscriber.TERMINATED
                        || NotificationLite.acceptFull(v, subscriber)) {
                    break;
                }
            }
        } catch (InterruptedException e) {
            bs.cancel();
            subscriber.onError(e);
        }
    }
public static < T > void subscribe ( Publisher < ? extends T > o , Subscriber < ? super T > subscriber ) { final BlockingQueue < Object > queue = new LinkedBlockingQueue < Object > ( ) ; BlockingSubscriber < T > bs = new BlockingSubscriber < T > ( queue ) ; o . subscribe ( bs ) ; try { for ( ; ; ) { if ( bs . isCancelled ( ) ) { break ; } Object v = queue . poll ( ) ; if ( v == null ) { if ( bs . isCancelled ( ) ) { break ; } BlockingHelper . verifyNonBlocking ( ) ; v = queue . take ( ) ; } if ( bs . isCancelled ( ) ) { break ; } if ( v == BlockingSubscriber . TERMINATED || NotificationLite . acceptFull ( v , subscriber ) ) { break ; } } } catch ( InterruptedException e ) { bs . cancel ( ) ; subscriber . onError ( e ) ; } }
Subscribes to the source and calls the Subscriber methods on the current thread.
<p>
@param o the source publisher
The cancellation and backpressure is composed through.
@param subscriber the subscriber to forward events and calls to in the current thread
@param <T> the value type
Subscribes to the source and calls the Subscriber methods on the current thread . <p >