OnSubscribeRange implements OnSubscribe<Integer> { @Override public void call(final Subscriber<? super Integer> o) { o.setProducer(new RangeProducer(o, start, end)); } OnSubscribeRange(int start, int end); @Override void call(final Subscriber<? super Integer> o);  }
@Test public void testBackpressureViaRequest() { OnSubscribeRange o = new OnSubscribeRange(1, RxRingBuffer.SIZE); TestSubscriber<Integer> ts = new TestSubscriber<Integer>(); ts.assertReceivedOnNext(Collections.<Integer> emptyList()); ts.requestMore(1); o.call(ts); ts.assertReceivedOnNext(Arrays.asList(1)); ts.requestMore(2); ts.assertReceivedOnNext(Arrays.asList(1, 2, 3)); ts.requestMore(3); ts.assertReceivedOnNext(Arrays.asList(1, 2, 3, 4, 5, 6)); ts.requestMore(RxRingBuffer.SIZE); ts.assertTerminalEvent(); }
@Test public void testNoBackpressure() { ArrayList<Integer> list = new ArrayList<Integer>(RxRingBuffer.SIZE * 2); for (int i = 1; i <= RxRingBuffer.SIZE * 2 + 1; i++) { list.add(i); } OnSubscribeRange o = new OnSubscribeRange(1, list.size()); TestSubscriber<Integer> ts = new TestSubscriber<Integer>(); ts.assertReceivedOnNext(Collections.<Integer> emptyList()); ts.requestMore(Long.MAX_VALUE); o.call(ts); ts.assertReceivedOnNext(list); ts.assertTerminalEvent(); }