OnSubscribeFromIterable implements OnSubscribe<T> { @Override public void call(final Subscriber<? super T> o) { final Iterator<? extends T> it = is.iterator(); o.setProducer(new IterableProducer<T>(o, it)); } OnSubscribeFromIterable(Iterable<? extends T> iterable); @Override void call(final Subscriber<? super T> o);  }
@Test public void testBackpressureViaRequest() { ArrayList<Integer> list = new ArrayList<Integer>(RxRingBuffer.SIZE); for (int i = 1; i <= RxRingBuffer.SIZE + 1; i++) { list.add(i); } OnSubscribeFromIterable<Integer> o = new OnSubscribeFromIterable<Integer>(list); 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(list.size()); ts.assertTerminalEvent(); }
@Test public void testNoBackpressure() { OnSubscribeFromIterable<Integer> o = new OnSubscribeFromIterable<Integer>(Arrays.asList(1, 2, 3, 4, 5)); TestSubscriber<Integer> ts = new TestSubscriber<Integer>(); ts.assertReceivedOnNext(Collections.<Integer> emptyList()); ts.requestMore(Long.MAX_VALUE); o.call(ts); ts.assertReceivedOnNext(Arrays.asList(1, 2, 3, 4, 5)); ts.assertTerminalEvent(); }
@Test public void testSubscribeMultipleTimes() { OnSubscribeFromIterable<Integer> o = new OnSubscribeFromIterable<Integer>(Arrays.asList(1, 2, 3)); TestSubscriber<Integer> ts = new TestSubscriber<Integer>(); o.call(ts); ts.assertReceivedOnNext(Arrays.asList(1, 2, 3)); ts = new TestSubscriber<Integer>(); o.call(ts); ts.assertReceivedOnNext(Arrays.asList(1, 2, 3)); ts = new TestSubscriber<Integer>(); o.call(ts); ts.assertReceivedOnNext(Arrays.asList(1, 2, 3)); }