Add zip of a collection of Observables

Author: abersnaze

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

@@ -21,6 +21,7 @@
 
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
@@ -2630,6 +2631,60 @@ public R call(T0 t0, T1 t1, T2 t2, T3 t3) {
         });
     }
 
+    /**
+     * 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 Observable
+     * @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(Collection<Observable<?>> ws, FuncN<R> reduceFunction) {
+        return create(OperationZip.zip(ws, 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
+     *            A collection of source Observable
+     * @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(Collection<Observable<?>> ws, final Object function) {
+        @SuppressWarnings({ "unchecked" })
+        final FuncN<R> _f = Functions.from(function);
+        return zip(ws, _f);
+    }
+
     /**
      * Filters an Observable by discarding any items it emits that do not meet some test.
      * <p>

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

@@ -19,6 +19,7 @@
 import static org.mockito.Mockito.*;
 
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -63,6 +64,16 @@ public static <T0, T1, T2, T3, R> Func1<Observer<R>, Subscription> zip(Observabl
         return a;
     }
 
+    @SuppressWarnings("unchecked")
+    public static <R> Func1<Observer<R>, Subscription> zip(Collection<Observable<?>> ws, FuncN<R> zipFunction) {
+        Aggregator a = new Aggregator(zipFunction);
+        for (Observable<?> w : ws) {
+            ZipObserver zipObserver = new ZipObserver(a, w);
+            a.addObserver(zipObserver);
+        }
+        return a;
+    }
+
     /*
      * ThreadSafe
      */