Bind implementation of Parallel

Author: benjchristensen

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

@@ -71,7 +71,7 @@
 import rx.operators.OperationOnErrorResumeNextViaObservable;
 import rx.operators.OperationOnErrorReturn;
 import rx.operators.OperationOnExceptionResumeNextViaObservable;
-import rx.operators.OperationParallel;
+import rx.operators.OperatorParallel;
 import rx.operators.OperationParallelMerge;
 import rx.operators.OperationRepeat;
 import rx.operators.OperationReplay;
@@ -5550,7 +5550,7 @@ public final Observable<T> onExceptionResumeNext(final Observable<? extends T> r
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Observable-Utility-Operators#parallel">RxJava Wiki: parallel()</a>
      */
     public final <R> Observable<R> parallel(Func1<Observable<T>, Observable<R>> f) {
-        return OperationParallel.parallel(this, f);
+        return bind(new OperatorParallel<T, R>(f, Schedulers.computation()));
     }
 
     /**
@@ -5569,7 +5569,7 @@ public final <R> Observable<R> parallel(Func1<Observable<T>, Observable<R>> f) {
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Observable-Utility-Operators#parallel">RxJava Wiki: parallel()</a>
      */
     public final <R> Observable<R> parallel(final Func1<Observable<T>, Observable<R>> f, final Scheduler s) {
-        return OperationParallel.parallel(this, f, s);
+        return bind(new OperatorParallel<T, R>(f, s));
     }
 
     /**

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

@@ -0,0 +1,62 @@
+/**
+ * Copyright 2014 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 rx.Observable;
+import rx.Operator;
+import rx.Scheduler;
+import rx.observables.GroupedObservable;
+import rx.util.functions.Func1;
+
+/**
+ * Identifies unit of work that can be executed in parallel on a given Scheduler.
+ */
+public final class OperatorParallel<T, R> implements Func1<Operator<? super R>, Operator<? super T>> {
+
+    private final Scheduler scheduler;
+    private final Func1<Observable<T>, Observable<R>> f;
+
+    public OperatorParallel(Func1<Observable<T>, Observable<R>> f, Scheduler scheduler) {
+        this.scheduler = scheduler;
+        this.f = f;
+    }
+
+    @Override
+    public Operator<? super T> call(Operator<? super R> op) {
+
+        Func1<Operator<? super GroupedObservable<Integer, T>>, Operator<? super T>> groupBy = new OperatorGroupBy<Integer, T>(new Func1<T, Integer>() {
+
+            int i = 0;
+
+            @Override
+            public Integer call(T t) {
+                return i++ % scheduler.degreeOfParallelism();
+            }
+
+        });
+
+        Func1<Operator<? super Observable<R>>, Operator<? super GroupedObservable<Integer, T>>> map = new OperatorMap<GroupedObservable<Integer, T>, Observable<R>>(new Func1<GroupedObservable<Integer, T>, Observable<R>>() {
+
+            @Override
+            public Observable<R> call(GroupedObservable<Integer, T> g) {
+                return f.call(g.observeOn(scheduler));
+            }
+        });
+
+        // bind together operators
+        return groupBy.call(map.call(new OperatorMerge().call(op)));
+    }
+}

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

@@ -25,7 +25,7 @@
 import rx.util.functions.Action1;
 import rx.util.functions.Func1;
 
-public class OperationParallelTest {
+public class OperatorParallelTest {
 
     @Test
     public void testParallel() {