Add combineLatest overload for Collection

src/main/java/rx/Observable.java

@@ -785,6 +785,31 @@ public static <T1, T2, T3, T4, T5, T6, T7, T8, T9, R> Observable<R> combineLates
             Func9<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? super T8, ? super T9, ? extends R> combineFunction) {
         return combineLatest(Arrays.asList(o1, o2, o3, o4, o5, o6, o7, o8, o9), Functions.fromFunc(combineFunction));
     }
+    /**
+     * Combines a collection of source Observables by emitting an item that aggregates the latest values of each of
+     * the source Observables each time an item is received from any of the source Observables, where this
+     * aggregation is defined by a specified function.
+     * <dl>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code combineLatest} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     *
+     * @param <T>
+     *            the common base type of source values
+     * @param <R>
+     *            the result type
+     * @param sources
+     *            the collection of source Observables
+     * @param combineFunction
+     *            the aggregation function used to combine the items emitted by the source Observables
+     * @return an Observable that emits items that are the result of combining the items emitted by the source
+     *         Observables by means of the given aggregation function
+     * @see <a href="http://reactivex.io/documentation/operators/combinelatest.html">ReactiveX operators documentation: CombineLatest</a>
+     */
+    public static <T, R> Observable<R> combineLatest(Collection<? extends Observable<? extends T>> sources, FuncN<? extends R> combineFunction) {
+        return create(new OnSubscribeCombineLatest<T, R>(sources, combineFunction));
+    }
+
     /**
      * Combines a list of source Observables by emitting an item that aggregates the latest values of each of
      * the source Observables each time an item is received from any of the source Observables, where this

src/main/java/rx/internal/operators/OnSubscribeCombineLatest.java

@@ -16,7 +16,8 @@
 package rx.internal.operators;
 
 import java.util.BitSet;
-import java.util.List;
+import java.util.Collection;
+import java.util.Iterator;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 
@@ -42,10 +43,10 @@
  *            the result type of the combinator function
  */
 public final class OnSubscribeCombineLatest<T, R> implements OnSubscribe<R> {
-    final List<? extends Observable<? extends T>> sources;
+    final Collection<? extends Observable<? extends T>> sources;
     final FuncN<? extends R> combinator;
 
-    public OnSubscribeCombineLatest(List<? extends Observable<? extends T>> sources, FuncN<? extends R> combinator) {
+    public OnSubscribeCombineLatest(Collection<? extends Observable<? extends T>> sources, FuncN<? extends R> combinator) {
         this.sources = sources;
         this.combinator = combinator;
         if (sources.size() > RxRingBuffer.SIZE) {
@@ -62,7 +63,7 @@ public void call(final Subscriber<? super R> child) {
             return;
         }
         if (sources.size() == 1) {
-            child.setProducer(new SingleSourceProducer<T, R>(child, sources.get(0), combinator));
+            child.setProducer(new SingleSourceProducer<T, R>(child, sources.iterator().next(), combinator));
         } else {
             child.setProducer(new MultiSourceProducer<T, R>(child, sources, combinator));
         }
@@ -76,7 +77,7 @@ public void call(final Subscriber<? super R> child) {
     final static class MultiSourceProducer<T, R> implements Producer {
         private final AtomicBoolean started = new AtomicBoolean();
         private final AtomicLong requested = new AtomicLong();
-        private final List<? extends Observable<? extends T>> sources;
+        private final Collection<? extends Observable<? extends T>> sources;
         private final Subscriber<? super R> child;
         private final FuncN<? extends R> combinator;
         private final MultiSourceRequestableSubscriber<T, R>[] subscribers;
@@ -92,7 +93,7 @@ final static class MultiSourceProducer<T, R> implements Producer {
         private final AtomicLong counter = new AtomicLong();
 
         @SuppressWarnings("unchecked")
-        public MultiSourceProducer(final Subscriber<? super R> child, final List<? extends Observable<? extends T>> sources, FuncN<? extends R> combinator) {
+        public MultiSourceProducer(final Subscriber<? super R> child, final Collection<? extends Observable<? extends T>> sources, FuncN<? extends R> combinator) {
             this.sources = sources;
             this.child = child;
             this.combinator = combinator;
@@ -116,10 +117,11 @@ public void request(long n) {
                  */
                 int sizePerSubscriber = RxRingBuffer.SIZE / sources.size();
                 int leftOver = RxRingBuffer.SIZE % sources.size();
-                for (int i = 0; i < sources.size(); i++) {
-                    Observable<? extends T> o = sources.get(i);
+                Iterator<? extends Observable<? extends T>> iterator = sources.iterator();
+                for (int i = 0; iterator.hasNext(); i++) {
+                    Observable<? extends T> o = iterator.next();
                     int toRequest = sizePerSubscriber;
-                    if (i == sources.size() - 1) {
+                    if (!iterator.hasNext()) {
                         toRequest += leftOver;
                     }
                     MultiSourceRequestableSubscriber<T, R> s = new MultiSourceRequestableSubscriber<T, R>(i, toRequest, child, this);

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

@@ -26,6 +26,7 @@
 
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
@@ -477,7 +478,7 @@ public List<Object> call(Object... args) {
         };
         for (int i = 1; i <= n; i++) {
             System.out.println("test1ToNSources: " + i + " sources");
-            List<Observable<Integer>> sources = new ArrayList<Observable<Integer>>();
+            Collection<Observable<Integer>> sources = new ArrayList<Observable<Integer>>();
             List<Object> values = new ArrayList<Object>();
             for (int j = 0; j < i; j++) {
                 sources.add(Observable.just(j));
@@ -509,7 +510,7 @@ public List<Object> call(Object... args) {
         };
         for (int i = 1; i <= n; i++) {
             System.out.println("test1ToNSourcesScheduled: " + i + " sources");
-            List<Observable<Integer>> sources = new ArrayList<Observable<Integer>>();
+            Collection<Observable<Integer>> sources = new ArrayList<Observable<Integer>>();
             List<Object> values = new ArrayList<Object>();
             for (int j = 0; j < i; j++) {
                 sources.add(Observable.just(j).subscribeOn(Schedulers.io()));