Adding zip(Ob<Ob<?>>)

ReactiveX · Jun 17, 2013 · ea16550 · ea16550
1 parent 8345dc7
commit ea16550
Showing 1 changed file with 61 additions and 2 deletions.
diff --git a/rxjava-core/src/main/java/rx/Observable.java b/rxjava-core/src/main/java/rx/Observable.java
@@ -2647,7 +2647,66 @@ public R call(T0 t0, T1 t1, T2 t2, T3 t3) {
      * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/zip.png">
      * 
      * @param ws
-     *            A collection of source Observable
+     *            An Observable of source Observables
+     * @param reduceFunction
+     *            a function that, when applied to an item emitted by each of the source
+     *            Observables, results in an item that will be emitted by the resulting Observable
+     * @return an Observable that emits the zipped results
+     */
+    public static <R> Observable<R> zip(Observable<Observable<?>> ws, final FuncN<R> reduceFunction) {
+        return ws.toList().mapMany(new Func1<List<Observable<?>>, Observable<R>>() {
+            @Override
+            public Observable<R> call(List<Observable<?>> wsList) {
+                return create(OperationZip.zip(wsList, reduceFunction));
+            }
+        });
+    }
+
+    /**
+     * Returns an Observable that emits the results of a function of your choosing applied to
+     * combinations of four items emitted, in sequence, by four other Observables.
+     * <p>
+     * <code>zip</code> applies this function in strict sequence, so the first item emitted by the
+     * new Observable will be the result of the function applied to the first item emitted by
+     * all of the Observalbes; the second item emitted by the new Observable will be the result of
+     * the function applied to the second item emitted by each of those Observables; and so forth.
+     * <p>
+     * The resulting <code>Observable<R></code> returned from <code>zip</code> will invoke
+     * <code>onNext</code> as many times as the number of <code>onNext</code> invocations of the
+     * source Observable that emits the fewest items.
+     * <p>
+     * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/zip.png">
+     * 
+     * @param ws
+     *            An Observable of source Observables
+     * @param function
+     *            a function that, when applied to an item emitted by each of the source
+     *            Observables, results in an item that will be emitted by the resulting Observable
+     * @return an Observable that emits the zipped results
+     */
+    public static <R> Observable<R> zip(Observable<Observable<?>> ws, final Object function) {
+        @SuppressWarnings({ "unchecked" })
+        final FuncN<R> _f = Functions.from(function);
+        return zip(ws, _f);
+    }
+
+    /**
+     * Returns an Observable that emits the results of a function of your choosing applied to
+     * combinations of four items emitted, in sequence, by four other Observables.
+     * <p>
+     * <code>zip</code> applies this function in strict sequence, so the first item emitted by the
+     * new Observable will be the result of the function applied to the first item emitted by
+     * all of the Observalbes; the second item emitted by the new Observable will be the result of
+     * the function applied to the second item emitted by each of those Observables; and so forth.
+     * <p>
+     * The resulting <code>Observable<R></code> returned from <code>zip</code> will invoke
+     * <code>onNext</code> as many times as the number of <code>onNext</code> invokations of the
+     * source Observable that emits the fewest items.
+     * <p>
+     * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/zip.png">
+     * 
+     * @param ws
+     *            A collection of source Observables
      * @param reduceFunction
      *            a function that, when applied to an item emitted by each of the source
      *            Observables, results in an item that will be emitted by the resulting Observable
@@ -2673,7 +2732,7 @@ public static <R> Observable<R> zip(Collection<Observable<?>> ws, FuncN<R> reduc
      * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/zip.png">
      * 
      * @param ws
-     *            A collection of source Observable
+     *            A collection of source Observables
      * @param function
      *            a function that, when applied to an item emitted by each of the source
      *            Observables, results in an item that will be emitted by the resulting Observable