add parallel and example

Author: samuelgruetter

language-adaptors/rxjava-scala/src/main/scala/rx/lang/scala/Observable.scala

@@ -1674,6 +1674,34 @@ class Observable[+T](val asJava: rx.Observable[_ <: T])
   def toBlockingObservable: BlockingObservable[T] = {
     new BlockingObservable[T](asJava.toBlockingObservable())
   }
+
+  /**
+   * Perform work in parallel by sharding an {@code Observable<T>} on a {@link Schedulers#threadPoolForComputation()} {@link Scheduler} and return an {@code Observable<R>} with the output.
+   *
+   * @param f
+   *            a {@link Func1} that applies Observable operators to {@code Observable<T>} in parallel and returns an {@code Observable<R>}
+   * @return an Observable with the output of the function executed on a {@link Scheduler}
+   */
+  def parallel[R](f: Observable[T] => Observable[R]): Observable[R] = {
+    val fJava: Func1[rx.Observable[T], rx.Observable[R]] =
+      (jo: rx.Observable[T]) => f(Observable[T](jo)).asJava.asInstanceOf[rx.Observable[R]]
+    Observable[R](asJava.asInstanceOf[rx.Observable[T]].parallel[R](fJava))
+  }
+
+  /**
+   * Perform work in parallel by sharding an {@code Observable<T>} on a {@link Scheduler} and return an {@code Observable<R>} with the output.
+   *
+   * @param f
+   *            a {@link Func1} that applies Observable operators to {@code Observable<T>} in parallel and returns an {@code Observable<R>}
+   * @param s
+   *            a {@link Scheduler} to perform the work on.
+   * @return an Observable with the output of the {@link Func1} executed on a {@link Scheduler}
+   */
+  def parallel[R](f: Observable[T] => Observable[R], scheduler: Scheduler): Observable[R] = {
+    val fJava: Func1[rx.Observable[T], rx.Observable[R]] =
+      (jo: rx.Observable[T]) => f(Observable[T](jo)).asJava.asInstanceOf[rx.Observable[R]]
+    Observable[R](asJava.asInstanceOf[rx.Observable[T]].parallel[R](fJava, scheduler))
+  }  
 
   def withFilter(p: T => Boolean): WithFilter[T] = {
     new WithFilter[T](p, asJava)

language-adaptors/rxjava-scala/src/main/scala/rx/lang/scala/examples/RxScalaDemo.scala

@@ -346,6 +346,28 @@ class RxScalaDemo extends JUnitSuite {
     assertEquals(10, Observable(-1, 0, 1).filter(condition).firstOrElse(10).toBlockingObservable.single)
   }
 
+  def square(x: Int): Int = {
+    println(s"$x*$x is being calculated on thread ${Thread.currentThread().getId()}")
+    Thread.sleep(100) // calculating a square is heavy work :)
+    x*x
+  }
+  
+  def work(o1: Observable[Int]): Observable[String] = {
+    println(s"map() is being called on thread ${Thread.currentThread().getId()}")
+    o1.map(i => s"The square of $i is ${square(i)}")
+  }
+  
+  @Test def parallelExample() {  
+    val t0 = System.currentTimeMillis()
+    Observable(1 to 10).parallel(work(_)).toBlockingObservable.foreach(println(_))
+    println(s"Work took ${System.currentTimeMillis()-t0} ms")
+  }
+  
+  @Test def exampleWithoutParallel() {
+    val t0 = System.currentTimeMillis()
+    work(Observable(1 to 10)).toBlockingObservable.foreach(println(_))
+    println(s"Work took ${System.currentTimeMillis()-t0} ms")
+  }
 
   def output(s: String): Unit = println(s)
 

language-adaptors/rxjava-scala/src/test/scala/rx/lang/scala/CompletenessTest.scala

@@ -40,6 +40,8 @@ class CompletenessTest extends JUnitSuite {
       "onErrorResumeNext(Observable[_ <: T])" -> "onErrorResumeNext(Observable[U])",
       "onErrorReturn(Func1[Throwable, _ <: T])" -> "onErrorReturn(Throwable => U)",
       "onExceptionResumeNext(Observable[_ <: T])" -> "onExceptionResumeNext(Observable[U])",
+      "parallel(Func1[Observable[T], Observable[R]])" -> "parallel(Observable[T] => Observable[R])",
+      "parallel(Func1[Observable[T], Observable[R]], Scheduler)" -> "parallel(Observable[T] => Observable[R], Scheduler)",
       "reduce(Func2[T, T, T])" -> "reduce((U, U) => U)",
       "reduce(R, Func2[R, _ >: T, R])" -> "fold(R)((R, T) => R)",
       "scan(Func2[T, T, T])" -> unnecessary,
@@ -169,7 +171,7 @@ class CompletenessTest extends JUnitSuite {
   def javaMethodSignatureToScala(s: String): String = {
     s.replaceAllLiterally("Long, TimeUnit", "Duration")
      .replaceAll("Action0", "() => Unit")
-     // nested [] can't be parsed with regex, but we still do it, please forgive us ;-)
+     // nested [] can't be parsed with regex, so these will have to be added manually
      .replaceAll("Action1\\[([^]]*)\\]", "$1 => Unit")
      .replaceAll("Action2\\[([^]]*), ([^]]*)\\]", "($1, $2) => Unit")
      .replaceAll("Func0\\[([^]]*)\\]", "() => $1")