Merge pull request #501 from benjchristensen/parallelMerge

ParallelMerge Operator

Author: benjchristensen

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

@@ -64,6 +64,7 @@
 import rx.operators.OperationOnErrorReturn;
 import rx.operators.OperationOnExceptionResumeNextViaObservable;
 import rx.operators.OperationParallel;
+import rx.operators.OperationParallelMerge;
 import rx.operators.OperationRetry;
 import rx.operators.OperationSample;
 import rx.operators.OperationScan;
@@ -4052,6 +4053,24 @@ public <R> Observable<R> parallel(final Func1<Observable<T>, Observable<R>> f, f
         return OperationParallel.parallel(this, f, s);
     }
 
+    
+    /**
+     * Merges an <code>Observable<Observable<T>></code> to <code>Observable<Observable<T>></code>
+     * with number of inner Observables as defined by <code>parallelObservables</code>.
+     * <p>
+     * For example, if the original <code>Observable<Observable<T>></code> has 100 Observables to be emitted and <code>parallelObservables</code>
+     * is defined as 8, the 100 will be grouped onto 8 output Observables.
+     * <p>
+     * This is a mechanism for efficiently processing N number of Observables on a smaller N number of resources (typically CPU cores).
+     * 
+     * @param parallelObservables
+     *            the number of Observables to merge into.
+     * @return an Observable of Observables constrained to number defined by <code>parallelObservables</code>.
+     */
+    public static <T> Observable<Observable<T>> parallelMerge(Observable<Observable<T>> source, int parallelObservables) {
+        return OperationParallelMerge.parallelMerge(source, parallelObservables);
+    }
+    
     /**
      * Returns a {@link ConnectableObservable}, which waits until its
      * {@link ConnectableObservable#connect connect} method is called before it

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

@@ -0,0 +1,50 @@
+/**
+ * Copyright 2013 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx.operators;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+import rx.Observable;
+import rx.observables.GroupedObservable;
+import rx.util.functions.Func1;
+
+public class OperationParallelMerge {
+
+    public static <T> Observable<Observable<T>> parallelMerge(final Observable<Observable<T>> source, final int num) {
+
+        return source.groupBy(new Func1<Observable<T>, Integer>() {
+            final AtomicLong rollingCount = new AtomicLong();
+
+            @Override
+            public Integer call(Observable<T> o) {
+                return (int) rollingCount.incrementAndGet() % num;
+            }
+        }).map(new Func1<GroupedObservable<Integer, Observable<T>>, Observable<T>>() {
+
+            /**
+             * Safe to cast from GroupedObservable to Observable so suppressing warning
+             */
+            @SuppressWarnings("unchecked")
+            @Override
+            public Observable<T> call(GroupedObservable<Integer, Observable<T>> o) {
+                return (Observable<T>) o;
+            }
+
+        });
+
+    }
+
+}

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

@@ -0,0 +1,49 @@
+/**
+ * Copyright 2013 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx.operators;
+
+import static org.junit.Assert.*;
+
+import java.util.List;
+
+import org.junit.Test;
+
+import rx.Observable;
+import rx.subjects.PublishSubject;
+
+public class OperationParallelMergeTest {
+
+    @Test
+    public void testParallelMerge() {
+        PublishSubject<String> p1 = PublishSubject.<String> create();
+        PublishSubject<String> p2 = PublishSubject.<String> create();
+        PublishSubject<String> p3 = PublishSubject.<String> create();
+        PublishSubject<String> p4 = PublishSubject.<String> create();
+
+        Observable<Observable<String>> fourStreams = Observable.<Observable<String>> from(p1, p2, p3, p4);
+
+        Observable<Observable<String>> twoStreams = OperationParallelMerge.parallelMerge(fourStreams, 2);
+        Observable<Observable<String>> threeStreams = OperationParallelMerge.parallelMerge(fourStreams, 3);
+
+        List<? super Observable<String>> fourList = fourStreams.toList().toBlockingObservable().last();
+        List<? super Observable<String>> threeList = threeStreams.toList().toBlockingObservable().last();
+        List<? super Observable<String>> twoList = twoStreams.toList().toBlockingObservable().last();
+
+        assertEquals(4, fourList.size());
+        assertEquals(3, threeList.size());
+        assertEquals(2, twoList.size());
+    }
+}