MergeMap, ConcatMap, SwitchMap

- flattening map implementations that better fit mental models and use cases
- keeping flatMap as alias to mergeMap since flatMap is common … (though flatMap still confuses people)
- deprecate mapMany as alias to flatMap and mergeMap

Author: benjchristensen

rxjava-core/src/main/java/rx/Observable.java

@@ -3653,7 +3653,7 @@ public Observable<Observable<T>> window(long timespan, long timeshift, TimeUnit
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Combining-Observables#zip">RxJava Wiki: zip()</a>
      */
     public static <R> Observable<R> zip(Observable<? extends Observable<?>> ws, final FuncN<? extends R> zipFunction) {
-        return ws.toList().mapMany(new Func1<List<? extends Observable<?>>, Observable<? extends R>>() {
+        return ws.toList().mergeMap(new Func1<List<? extends Observable<?>>, Observable<? extends R>>() {
             @Override
             public Observable<R> call(List<? extends Observable<?>> wsList) {
                 return create(OperationZip.zip(wsList, zipFunction));
@@ -3891,7 +3891,64 @@ public Observable<T> finallyDo(Action0 action) {
      * @see #mapMany(Func1)
      */
     public <R> Observable<R> flatMap(Func1<? super T, ? extends Observable<? extends R>> func) {
-        return mapMany(func);
+        return mergeMap(func);
+    }
+    
+    /**
+     * Creates a new Observable by applying a function that you supply to each
+     * item emitted by the source Observable, where that function returns an
+     * Observable, and then merging those resulting Observables and emitting the
+     * results of this merger.
+     * <p>
+     * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/flatMap.png">
+     * <p>
+     * Note: {@code mapMany} and {@code flatMap} are equivalent.
+     * 
+     * @param func a function that, when applied to an item emitted by the
+     *             source Observable, returns an Observable
+     * @return an Observable that emits the result of applying the
+     *         transformation function to each item emitted by the source
+     *         Observable and merging the results of the Observables obtained
+     *         from this transformation.
+     * @see <a href="https://github.com/Netflix/RxJava/wiki/Transforming-Observables#mapmany-or-flatmap-and-mapmanydelayerror">RxJava Wiki: flatMap()</a>
+     * @see #flatMap(Func1)
+     */
+    public <R> Observable<R> mergeMap(Func1<? super T, ? extends Observable<? extends R>> func) {
+        return merge(map(func));
+    }
+    
+    /**
+     * Creates a new Observable by applying a function that you supply to each
+     * item emitted by the source Observable, where that function returns an
+     * Observable, and then concatting those resulting Observables and emitting the
+     * results of this concat.
+     * <p>
+     * 
+     * @param func a function that, when applied to an item emitted by the
+     *             source Observable, returns an Observable
+     * @return an Observable that emits the result of applying the
+     *         transformation function to each item emitted by the source
+     *         Observable and concatting the results of the Observables obtained
+     *         from this transformation.
+     */
+    public <R> Observable<R> concatMap(Func1<? super T, ? extends Observable<? extends R>> func) {
+        return concat(map(func));
+    }
+    
+    /**
+     * Creates a new Observable by applying a function that you supply to each
+     * item emitted by the source Observable resulting in an Observable of Observables.
+     * <p>
+     * Then a {@link #switchLatest(Observable)} / {@link #switchOnNext(Observable)} is applied.
+     * 
+     * @param func a function that, when applied to an item emitted by the
+     *             source Observable, returns an Observable
+     * @return an Observable that emits the result of applying the
+     *         transformation function to each item emitted by the source
+     *         Observable and then switch
+     */
+    public <R> Observable<R> switchMap(Func1<? super T, ? extends Observable<? extends R>> func) {
+        return switchOnNext(map(func));
     }
 
     /**
@@ -3965,9 +4022,10 @@ public <R> Observable<R> mapWithIndex(Func2<? super T, Integer, ? extends R> fun
      *         from this transformation.
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Transforming-Observables#mapmany-or-flatmap-and-mapmanydelayerror">RxJava Wiki: mapMany()</a>
      * @see #flatMap(Func1)
+     * @deprecated
      */
     public <R> Observable<R> mapMany(Func1<? super T, ? extends Observable<? extends R>> func) {
-        return create(OperationMap.mapMany(this, func));
+        return mergeMap(func);
     }
 
     /**

rxjava-core/src/main/java/rx/operators/OperationMap.java

@@ -77,26 +77,6 @@ public Subscription onSubscribe(Observer<? super R> observer) {
         };
     }
 
-    /**
-     * Accepts a sequence of observable sequences and a transformation function. Returns a flattened sequence that is the result of
-     * applying the transformation function to each item in the sequence of each observable sequence.
-     * <p>
-     * The closure should return an Observable which will then be merged.
-     * 
-     * @param sequence
-     *            the input sequence.
-     * @param func
-     *            a function to apply to each item in the sequence.
-     * @param <T>
-     *            the type of the input sequence.
-     * @param <R>
-     *            the type of the output sequence.
-     * @return a sequence that is the result of applying the transformation function to each item in the input sequence.
-     */
-    public static <T, R> OnSubscribeFunc<R> mapMany(Observable<? extends T> sequence, Func1<? super T, ? extends Observable<? extends R>> func) {
-        return OperationMerge.merge(Observable.create(map(sequence, func)));
-    }
-
     /**
      * An observable sequence that is the result of applying a transformation to each item in an input sequence.
      * 

rxjava-core/src/test/java/rx/operators/OperationMapTest.java

@@ -118,7 +118,7 @@ public void testMapMany() {
         Observable<Integer> ids = Observable.from(1, 2);
 
         /* now simulate the behavior to take those IDs and perform nested async calls based on them */
-        Observable<String> m = Observable.create(mapMany(ids, new Func1<Integer, Observable<String>>() {
+        Observable<String> m = ids.flatMap(new Func1<Integer, Observable<String>>() {
 
             @Override
             public Observable<String> call(Integer id) {
@@ -143,7 +143,7 @@ public String call(Map<String, String> map) {
                 }));
             }
 
-        }));
+        });
         m.subscribe(stringObserver);
 
         verify(stringObserver, never()).onError(any(Throwable.class));
@@ -166,7 +166,7 @@ public void testMapMany2() {
 
         Observable<Observable<Map<String, String>>> observable = Observable.from(observable1, observable2);
 
-        Observable<String> m = Observable.create(mapMany(observable, new Func1<Observable<Map<String, String>>, Observable<String>>() {
+        Observable<String> m = observable.flatMap(new Func1<Observable<Map<String, String>>, Observable<String>>() {
 
             @Override
             public Observable<String> call(Observable<Map<String, String>> o) {
@@ -179,7 +179,7 @@ public String call(Map<String, String> map) {
                 }));
             }
 
-        }));
+        });
         m.subscribe(stringObserver);
 
         verify(stringObserver, never()).onError(any(Throwable.class));