merge converted to flatMap

Author: akarnokd

src/main/java/rx/Observable.java

@@ -1635,8 +1635,9 @@ public final static <T> Observable<T> merge(Iterable<? extends Observable<? exte
      *         {@code source} Observable
      * @see <a href="http://reactivex.io/documentation/operators/merge.html">ReactiveX operators documentation: Merge</a>
      */
+    @SuppressWarnings({"unchecked", "rawtypes"})
     public final static <T> Observable<T> merge(Observable<? extends Observable<? extends T>> source) {
-        return source.lift(OperatorMerge.<T>instance(false));
+        return source.flatMap((Func1)UtilityFunctions.identity());
     }
 
     /**
@@ -1664,8 +1665,9 @@ public final static <T> Observable<T> merge(Observable<? extends Observable<? ex
      * @see <a href="http://reactivex.io/documentation/operators/merge.html">ReactiveX operators documentation: Merge</a>
      */
     @Experimental
+    @SuppressWarnings({"unchecked", "rawtypes"})
     public final static <T> Observable<T> merge(Observable<? extends Observable<? extends T>> source, int maxConcurrent) {
-        return source.lift(OperatorMerge.<T>instance(false, maxConcurrent));
+        return source.flatMap((Func1)UtilityFunctions.identity(), maxConcurrent);
     }
 
     /**
@@ -1985,8 +1987,9 @@ public final static <T> Observable<T> merge(Observable<? extends T>[] sequences,
      *         {@code source} Observable
      * @see <a href="http://reactivex.io/documentation/operators/merge.html">ReactiveX operators documentation: Merge</a>
      */
+    @SuppressWarnings({"unchecked", "rawtypes"})
     public final static <T> Observable<T> mergeDelayError(Observable<? extends Observable<? extends T>> source) {
-        return source.lift(OperatorMerge.<T>instance(true));
+        return source.flatMap((Func1)UtilityFunctions.identity(), Integer.MAX_VALUE, true);
     }
     /**
      * Flattens an Observable that emits Observables into one Observable, in a way that allows an Observer to
@@ -2016,8 +2019,9 @@ public final static <T> Observable<T> mergeDelayError(Observable<? extends Obser
      * @see <a href="http://reactivex.io/documentation/operators/merge.html">ReactiveX operators documentation: Merge</a>
      */
     @Experimental
+    @SuppressWarnings({"unchecked", "rawtypes"})
     public final static <T> Observable<T> mergeDelayError(Observable<? extends Observable<? extends T>> source, int maxConcurrent) {
-        return source.lift(OperatorMerge.<T>instance(true, maxConcurrent));
+        return source.flatMap((Func1)UtilityFunctions.identity(), maxConcurrent, true);
     }
 
     /**
@@ -4614,7 +4618,7 @@ public final Observable<T> firstOrDefault(T defaultValue, Func1<? super T, Boole
      * @see <a href="http://reactivex.io/documentation/operators/flatmap.html">ReactiveX operators documentation: FlatMap</a>
      */
     public final <R> Observable<R> flatMap(Func1<? super T, ? extends Observable<? extends R>> func) {
-        return merge(map(func));
+        return lift(OperatorMerge.instance(false, Integer.MAX_VALUE, func));
     }
 
     /**
@@ -4642,7 +4646,67 @@ public final <R> Observable<R> flatMap(Func1<? super T, ? extends Observable<? e
      */
     @Beta
     public final <R> Observable<R> flatMap(Func1<? super T, ? extends Observable<? extends R>> func, int maxConcurrent) {
-        return merge(map(func), maxConcurrent);
+        return flatMap(func, maxConcurrent, false);
+    }
+    
+    /**
+     * Returns an Observable that emits items based on 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, while limiting the maximum number of concurrent
+     * subscriptions to these Observables.
+     * <p>
+     * <!-- <img width="640" height="310" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/flatMap.png" alt=""> -->
+     * <dl>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code flatMap} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     * 
+     * @param func
+     *            a function that, when applied to an item emitted by the source Observable, returns an
+     *            Observable
+     * @param maxConcurrent
+     *         the maximum number of Observables that may be subscribed to concurrently
+     * @param delayErrors
+     *         should the errors delayed until all sources have terminated in some way
+     * @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="http://reactivex.io/documentation/operators/flatmap.html">ReactiveX operators documentation: FlatMap</a>
+     * @since (if this graduates from Experimental/Beta to supported, replace this parenthetical with the release number)
+     */
+    @Beta
+    public final <R> Observable<R> flatMap(Func1<? super T, ? extends Observable<? extends R>> func, int maxConcurrent, boolean delayErrors) {
+        return lift(OperatorMerge.instance(delayErrors, maxConcurrent, func));
+    }
+
+    /**
+     * Returns an Observable that emits items based on 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, while limiting the maximum number of concurrent
+     * subscriptions to these Observables.
+     * <p>
+     * <!-- <img width="640" height="310" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/flatMap.png" alt=""> -->
+     * <dl>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code flatMap} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     * 
+     * @param func
+     *            a function that, when applied to an item emitted by the source Observable, returns an
+     *            Observable
+     * @param maxConcurrent
+     *         the maximum number of Observables that may be subscribed to concurrently
+     * @param delayErrors
+     *         should the errors delayed until all sources have terminated in some way
+     * @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="http://reactivex.io/documentation/operators/flatmap.html">ReactiveX operators documentation: FlatMap</a>
+     * @since (if this graduates from Experimental/Beta to supported, replace this parenthetical with the release number)
+     */
+    @Beta
+    public final <R> Observable<R> flatMap(Func1<? super T, ? extends Observable<? extends R>> func, boolean delayErrors) {
+        return lift(OperatorMerge.instance(delayErrors, Integer.MAX_VALUE, func));
     }
 
     /**
@@ -4791,7 +4855,7 @@ public final <U, R> Observable<R> flatMap(final Func1<? super T, ? extends Obser
      * @see <a href="http://reactivex.io/documentation/operators/flatmap.html">ReactiveX operators documentation: FlatMap</a>
      */
     public final <R> Observable<R> flatMapIterable(Func1<? super T, ? extends Iterable<? extends R>> collectionSelector) {
-        return merge(map(OperatorMapPair.convertSelector(collectionSelector)));
+        return flatMap(OperatorMapPair.convertSelector(collectionSelector));
     }
 
     /**