Operator forIterable 2

[akarnokd]

Split from #636

This operator behaves differently than concat(Iterable). It allows projecting a potentially infinite sequence onto Observables concatenated one after each other. The input Iterable sequence is not traversed upfront like concat(Iterable) and concat(Observable) would do.

[cloudbees-pull-request-builder]

RxJava-pull-requests #602 FAILURE
Looks like there's a problem with this pull request

[benjchristensen]

I believe what you want is accomplished with concatMap that was recently added.

        Observable<Integer> result = Observable.from(keys).concatMap(new Func1<Integer, Observable<Integer>>() {

            @Override
            public Observable<Integer> call(Integer t1) {
                return Observable.from(1, 2, 3).subscribeOn(Schedulers.currentThread());
            }});

Here is a unit test from your code converted to this and it passes:

    @Test
    public void testForIterableLongSequence() {
        int n = 250;
        List<Integer> keys = new ArrayList<Integer>(n);
        List<Integer> expected = new ArrayList<Integer>(3 * n);
        for (int i = 1; i <= n; i++) {
            keys.add(n);
            for (int j = 1; j <= 3; j++) {
                expected.add(j);
            }
        }

        Observable<Integer> result = Observable.from(keys).concatMap(new Func1<Integer, Observable<Integer>>() {

            @Override
            public Observable<Integer> call(Integer t1) {
                return Observable.from(1, 2, 3).subscribeOn(Schedulers.currentThread());
            }});

        Observer<Object> o = mock(Observer.class);
        InOrder inOrder = inOrder(o);

        result.subscribe(o);

        for (Integer i : expected) {
            inOrder.verify(o, times(1)).onNext(i);
        }
        inOrder.verify(o, times(1)).onCompleted();
        inOrder.verifyNoMoreInteractions();
        verify(o, never()).onError(any(Throwable.class));

    }

[akarnokd]

True. Since the newer ConcatMap does the same thing, this PR can be skipped.