@Test
public void testBackpressure() {

    final AtomicLong requested = new AtomicLong();
    Flowable<Integer> source = Flowable.range(1, 1000)
            .doOnRequest(new LongConsumer() {
                @Override
                public void accept(long t) {
                    requested.addAndGet(t);
                }
            });
    ConnectableFlowable<Integer> cf = source.replay();
    TestSubscriber<Integer> ts1 = new TestSubscriber<Integer>(10L);
    TestSubscriber<Integer> ts2 = new TestSubscriber<Integer>(90L);
    cf.subscribe(ts1);
    cf.subscribe(ts2);
    ts2.request(10);
    cf.connect();
    ts1.assertValueCount(10);
    ts1.assertNotTerminated();
    ts2.assertValueCount(100);
    ts2.assertNotTerminated();
    Assert.assertEquals(100, requested.get());
}