@Test
public void testBackpressureFastSlow() {

    ConnectableObservable<Integer> is = Observable.range(1, RxRingBuffer.SIZE * 2).publish();
    Observable<Integer> fast = is.observeOn(Schedulers.computation()).doOnCompleted(new Action0() {
        @Override
        public void call() {
            System.out.println("^^^^^^^^^^^^^ completed FAST");
        }
    });
    Observable<Integer> slow = is.observeOn(Schedulers.computation()).map(new Func1<Integer, Integer>() {
        int c = 0;
        @Override
        public Integer call(Integer i) {
            if (c == 0) {
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                }
            }
            c++;
            return i;
        }
    }).doOnCompleted(new Action0() {
        @Override
        public void call() {
            System.out.println("^^^^^^^^^^^^^ completed SLOW");
        }
    });
    TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
    Observable.merge(fast, slow).subscribe(ts);
    is.connect();
    ts.awaitTerminalEvent();
    ts.assertNoErrors();
    assertEquals(RxRingBuffer.SIZE * 4, ts.getOnNextEvents().size());
}