Merge pull request #3718 from ginbalin/1.x

new method concatMapIterable #3713

Author: akarnokd

src/main/java/rx/Observable.java

@@ -3889,6 +3889,29 @@ public final <R> Observable<R> concatMap(Func1<? super T, ? extends Observable<?
         return concat(map(func));
     }
 
+    /**
+     * Returns an Observable that concatenate each item emitted by the source Observable with the values in an
+     * Iterable corresponding to that item that is generated by a selector.
+     * <p>
+     * 
+     * <dl>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code concatMapIterable} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     * 
+     * @param <R>
+     *            the type of item emitted by the resulting Observable
+     * @param collectionSelector
+     *            a function that returns an Iterable sequence of values for when given an item emitted by the
+     *            source Observable
+     * @return an Observable that emits the results of concatenating the items emitted by the source Observable with
+     *         the values in the Iterables corresponding to those items, as generated by {@code collectionSelector}
+     * @see <a href="http://reactivex.io/documentation/operators/flatmap.html">ReactiveX operators documentation: FlatMap</a>
+     */
+    public final <R> Observable<R> concatMapIterable(Func1<? super T, ? extends Iterable<? extends R>> collectionSelector) {
+        return concat(map(OperatorMapPair.convertSelector(collectionSelector)));
+    }
+    
     /**
      * Returns an Observable that emits the items emitted from the current Observable, then the next, one after
      * the other, without interleaving them.

src/test/java/rx/internal/operators/OperatorConcatTest.java

@@ -82,6 +82,34 @@ public void testConcatWithList() {
 
         verify(observer, times(7)).onNext(anyString());
     }
+    
+    @Test
+    public void testConcatMapIterable() {
+        @SuppressWarnings("unchecked")
+        Observer<String> observer = mock(Observer.class);
+
+        final String[] l = { "a", "b", "c", "d", "e" };
+        
+        Func1<List<String>,List<String>> identity = new Func1<List<String>, List<String>>() {
+			@Override
+			public List<String> call(List<String> t) {
+				return t;
+			}
+		};
+
+        final Observable<List<String>> listObs = Observable.just(Arrays.asList(l));
+        final Observable<String> concatMap = listObs.concatMapIterable(identity);
+        
+        concatMap.subscribe(observer);
+
+        InOrder inOrder = inOrder(observer);
+        inOrder.verify(observer, times(1)).onNext("a");
+        inOrder.verify(observer, times(1)).onNext("b");
+        inOrder.verify(observer, times(1)).onNext("c");
+        inOrder.verify(observer, times(1)).onNext("d");
+        inOrder.verify(observer, times(1)).onNext("e");
+        inOrder.verify(observer, times(1)).onCompleted();
+    }
 
     @Test
     public void testConcatObservableOfObservables() {