FlowableFromIterable extends Flowable<T> { public static <T> void subscribe(Subscriber<? super T> s, Iterator<? extends T> it) { boolean hasNext; try { hasNext = it.hasNext(); } catch (Throwable e) { Exceptions.throwIfFatal(e); EmptySubscription.error(e, s); return; } if (!hasNext) { EmptySubscription.complete(s); return; } if (s instanceof ConditionalSubscriber) { s.onSubscribe(new IteratorConditionalSubscription<T>( (ConditionalSubscriber<? super T>)s, it)); } else { s.onSubscribe(new IteratorSubscription<T>(s, it)); } } FlowableFromIterable(Iterable<? extends T> source); @Override void subscribeActual(Subscriber<? super T> s); static void subscribe(Subscriber<? super T> s, Iterator<? extends T> it);  }
@Test public void fusionWithConcatMap() { TestSubscriber<Integer> ts = new TestSubscriber<Integer>(); Flowable.fromIterable(Arrays.asList(1, 2, 3, 4)).concatMap( new Function<Integer, Flowable <Integer>>() { @Override public Flowable<Integer> apply(Integer v) { return Flowable.range(v, 2); } }).subscribe(ts); ts.assertValues(1, 2, 2, 3, 3, 4, 4, 5); ts.assertNoErrors(); ts.assertComplete(); }
@Test public void fusedAPICalls() { Flowable.fromIterable(Arrays.asList(1, 2, 3)) .subscribe(new FlowableSubscriber<Integer>() { @Override public void onSubscribe(Subscription s) { @SuppressWarnings("unchecked") QueueSubscription<Integer> qs = (QueueSubscription<Integer>)s; assertFalse(qs.isEmpty()); try { assertEquals(1, qs.poll().intValue()); } catch (Exception ex) { throw new AssertionError(ex); } assertFalse(qs.isEmpty()); qs.clear(); List<Throwable> errors = TestHelper.trackPluginErrors(); try { qs.request(-99); TestHelper.assertError(errors, 0, IllegalArgumentException.class, "n > 0 required but it was -99"); } finally { RxJavaPlugins.reset(); } } @Override public void onNext(Integer t) { } @Override public void onError(Throwable t) { } @Override public void onComplete() { } }); }
@Test public void requestRaceConditional() { for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { final TestSubscriber<Integer> ts = new TestSubscriber<Integer>(0L); Runnable r = new Runnable() { @Override public void run() { ts.request(1); } }; Flowable.fromIterable(Arrays.asList(1, 2, 3, 4)) .filter(Functions.alwaysTrue()) .subscribe(ts); TestHelper.race(r, r); } }
@Test public void requestRaceConditional2() { for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { final TestSubscriber<Integer> ts = new TestSubscriber<Integer>(0L); Runnable r = new Runnable() { @Override public void run() { ts.request(1); } }; Flowable.fromIterable(Arrays.asList(1, 2, 3, 4)) .filter(Functions.alwaysFalse()) .subscribe(ts); TestHelper.race(r, r); } }
@Test public void requestCancelConditionalRace() { for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { final TestSubscriber<Integer> ts = new TestSubscriber<Integer>(0L); Runnable r1 = new Runnable() { @Override public void run() { ts.request(1); } }; Runnable r2 = new Runnable() { @Override public void run() { ts.cancel(); } }; Flowable.fromIterable(Arrays.asList(1, 2, 3, 4)) .filter(Functions.alwaysTrue()) .subscribe(ts); TestHelper.race(r1, r2); } }
@Test public void requestCancelConditionalRace2() { for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { final TestSubscriber<Integer> ts = new TestSubscriber<Integer>(0L); Runnable r1 = new Runnable() { @Override public void run() { ts.request(Long.MAX_VALUE); } }; Runnable r2 = new Runnable() { @Override public void run() { ts.cancel(); } }; Flowable.fromIterable(Arrays.asList(1, 2, 3, 4)) .filter(Functions.alwaysTrue()) .subscribe(ts); TestHelper.race(r1, r2); } }
@Test public void requestCancelRace() { for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { final TestSubscriber<Integer> ts = new TestSubscriber<Integer>(0L); Runnable r1 = new Runnable() { @Override public void run() { ts.request(1); } }; Runnable r2 = new Runnable() { @Override public void run() { ts.cancel(); } }; Flowable.fromIterable(Arrays.asList(1, 2, 3, 4)) .subscribe(ts); TestHelper.race(r1, r2); } }
@Test public void requestCancelRace2() { for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { final TestSubscriber<Integer> ts = new TestSubscriber<Integer>(0L); Runnable r1 = new Runnable() { @Override public void run() { ts.request(Long.MAX_VALUE); } }; Runnable r2 = new Runnable() { @Override public void run() { ts.cancel(); } }; Flowable.fromIterable(Arrays.asList(1, 2, 3, 4)) .subscribe(ts); TestHelper.race(r1, r2); } }
@Test public void fusionRejected() { TestSubscriber<Integer> ts = SubscriberFusion.newTest(QueueFuseable.ASYNC); Flowable.fromIterable(Arrays.asList(1, 2, 3)) .subscribe(ts); SubscriberFusion.assertFusion(ts, QueueFuseable.NONE) .assertResult(1, 2, 3); }
@Test public void fusionClear() { Flowable.fromIterable(Arrays.asList(1, 2, 3)) .subscribe(new FlowableSubscriber<Integer>() { @Override public void onSubscribe(Subscription d) { @SuppressWarnings("unchecked") QueueSubscription<Integer> qd = (QueueSubscription<Integer>)d; qd.requestFusion(QueueFuseable.ANY); try { assertEquals(1, qd.poll().intValue()); } catch (Throwable ex) { fail(ex.toString()); } qd.clear(); try { assertNull(qd.poll()); } catch (Throwable ex) { fail(ex.toString()); } } @Override public void onNext(Integer value) { } @Override public void onError(Throwable e) { } @Override public void onComplete() { } }); }
@Test public void hasNextCancels() { final TestSubscriber<Integer> ts = new TestSubscriber<Integer>(); Flowable.fromIterable(new Iterable<Integer>() { @Override public Iterator<Integer> iterator() { return new Iterator<Integer>() { int count; @Override public boolean hasNext() { if (++count == 2) { ts.cancel(); } return true; } @Override public Integer next() { return 1; } @Override public void remove() { throw new UnsupportedOperationException(); } }; } }) .subscribe(ts); ts.assertValue(1) .assertNoErrors() .assertNotComplete(); }