fixed OutOfMemoryError in RxScala

src/main/scala/org/zalando/benchmarks/ComputationFollowedByAsyncPublishing.scala

@@ -40,8 +40,7 @@ class ComputationFollowedByAsyncPublishing {
   @Benchmark def bmParallelFutures(): Unit =  futures benchmark  1          // <=  44 threads ("Live peak" in JVisualVM)
   @Benchmark def bmBlocking():        Unit = blocking benchmark 64          // <= 549 threads ("Live peak" in JVisualVM)
   @Benchmark def bmStreams():         Unit =  streams benchmark  1          // <=  52 threads ("Live peak" in JVisualVM)
-
-//@Benchmark def bmRxScala():         Unit =       rx benchmark             // blows up with OutOfMemoryError :(
+  @Benchmark def bmRxScala():         Unit =        rx benchmark 1          // <= 50 threads
 
   @TearDown def tearDown(): Unit = system.terminate()
 }

src/main/scala/org/zalando/benchmarks/RxScala.scala

@@ -1,21 +1,29 @@
 package org.zalando.benchmarks
 
+import java.util.concurrent.{ExecutorService, Executors}
+
 import akka.actor.ActorSystem
 import rx.lang.scala.Observable
-import rx.lang.scala.schedulers.{ComputationScheduler, ExecutionContextScheduler}
+
+import scala.concurrent.{ExecutionContext, Future}
 
 class RxScala(system: ActorSystem) {
   import ComputationFollowedByAsyncPublishing._
 
-  def benchmark: Unit = {
-    // looks nice, not sure if correct, blows up the heap
-    Observable
-      .from(1 to numTasks map Job)
-      .subscribeOn(ComputationScheduler())
-      .map(Computer compute)
-      .subscribeOn(ExecutionContextScheduler(system dispatcher))
-      .flatMap(1024, r => Observable.from(Publisher publish (r, system))(system dispatcher))
-      .foldLeft(0) { case (s, r) => s + computeResult(r) }
-      .foreach(println)
+  def benchmark(coreFactor: Int): Unit = {
+    val executor: ExecutorService = Executors.newCachedThreadPool()
+    implicit val ec = ExecutionContext.fromExecutor(executor)
+    try {
+      Observable
+        .from(1 to numTasks)
+        .map(Job)
+        .flatMap(numWorkers(coreFactor), job => Observable.from(Future(Computer compute job)))
+        .flatMap(r => Observable.from(Publisher publish (r, system))(system dispatcher))
+        .foldLeft(0) { case (s, r) => s + computeResult(r) }
+        .toBlocking
+        .foreach(println)
+    } finally {
+      executor.shutdown()
+    }
   }
 }