public static <T> void subscribe(ObservableSource<? extends T> o, Observer<? super T> observer) {
        final BlockingQueue<Object> queue = new LinkedBlockingQueue<Object>();

        BlockingObserver<T> bs = new BlockingObserver<T>(queue);
        observer.onSubscribe(bs);

        o.subscribe(bs);
        for (;;) {
            if (bs.isDisposed()) {
                break;
            }
            Object v = queue.poll();
            if (v == null) {
                try {
                    v = queue.take();
                } catch (InterruptedException ex) {
                    bs.dispose();
                    observer.onError(ex);
                    return;
                }
            }
            if (bs.isDisposed()
                    || o == BlockingObserver.TERMINATED
                    || NotificationLite.acceptFull(v, observer)) {
                break;
            }
        }
    }
public static <T> void subscribe(ObservableSource<? extends T> o, Observer<? super T> observer) {
        final BlockingQueue<Object> queue = new LinkedBlockingQueue<Object>();

        BlockingObserver<T> bs = new BlockingObserver<T>(queue);
        observer.onSubscribe(bs);

        o.subscribe(bs);
        for (;;) {
            if (bs.isDisposed()) {
                break;
            }
            Object v = queue.poll();
            if (v == null) {
                try {
                    v = queue.take();
                } catch (InterruptedException ex) {
                    bs.dispose();
                    observer.onError(ex);
                    return;
                }
            }
            if (bs.isDisposed()
                    || o == BlockingObserver.TERMINATED
                    || NotificationLite.acceptFull(v, observer)) {
                break;
            }
        }
    }
public static < T > void subscribe ( ObservableSource < ? extends T > o , Observer < ? super T > observer ) { final BlockingQueue < Object > queue = new LinkedBlockingQueue < Object > ( ) ; BlockingObserver < T > bs = new BlockingObserver < T > ( queue ) ; observer . onSubscribe ( bs ) ; o . subscribe ( bs ) ; for ( ; ; ) { if ( bs . isDisposed ( ) ) { break ; } Object v = queue . poll ( ) ; if ( v == null ) { try { v = queue . take ( ) ; } catch ( InterruptedException ex ) { bs . dispose ( ) ; observer . onError ( ex ) ; return ; } } if ( bs . isDisposed ( ) || o == BlockingObserver . TERMINATED || NotificationLite . acceptFull ( v , observer ) ) { break ; } } }
Subscribes to the source and calls the Observer methods on the current thread.
<p>
@param o the source ObservableSource
The call to dispose() is composed through.
@param observer the subscriber to forward events and calls to in the current thread
@param <T> the value type
Subscribes to the source and calls the Observer methods on the current thread . <p >