add rx.lang.scala.concurrency.Schedulers

Author: samuelgruetter

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

@@ -21,7 +21,7 @@ import rx.lang.scala._
 import scala.concurrent.duration._
 import org.junit.{Before, Test, Ignore}
 import org.junit.Assert._
-import rx.lang.scala.concurrency.NewThreadScheduler
+import rx.lang.scala.concurrency.Schedulers
 import rx.lang.scala.util.Timestamped
 import java.io.IOException
 
@@ -169,10 +169,10 @@ class RxScalaDemo extends JUnitSuite {
 
   @Test def schedulersExample() {
     val o = Observable.interval(100 millis).take(8)
-    o.observeOn(NewThreadScheduler).subscribe(
+    o.observeOn(Schedulers.newThread).subscribe(
         i => println(s"${i}a (on thread #${Thread.currentThread().getId()})")
     )
-    o.observeOn(NewThreadScheduler).subscribe(
+    o.observeOn(Schedulers.newThread).subscribe(
         i => println(s"${i}b (on thread #${Thread.currentThread().getId()})")
     )
     waitFor(o)

language-adaptors/rxjava-scala/src/main/scala/rx/lang/scala/concurrency/Schedulers.scala

@@ -0,0 +1,61 @@
+package rx.lang.scala.concurrency
+
+import rx.Scheduler
+import java.util.concurrent.Executor
+import java.util.concurrent.ScheduledExecutorService
+
+/**
+ * Factory methods for creating Schedulers.
+ */
+object Schedulers {
+
+  /**
+   * Returns a [[rx.lang.scala.Scheduler]] that executes work immediately on the current thread.
+   */
+  def immediate: Scheduler = rx.concurrency.Schedulers.immediate()
+
+  /**
+   * Returns a [[rx.lang.scala.Scheduler]] that queues work on the current thread to be executed after the current work completes.
+   */
+  def currentThread: Scheduler = rx.concurrency.Schedulers.currentThread()
+
+  /**
+   * Returns a [[rx.lang.scala.Scheduler]] that creates a new {@link Thread} for each unit of work.
+   */
+  def newThread: Scheduler = rx.concurrency.Schedulers.newThread
+
+  /**
+   * Returns a [[rx.lang.scala.Scheduler]] that queues work on an [[java.util.concurrent.Executor]].
+   * 
+   * Note that this does not support scheduled actions with a delay.
+   */
+  def executor(executor: Executor): Scheduler = rx.concurrency.Schedulers.executor(executor)
+
+  /**
+   * Returns a [[rx.lang.scala.Scheduler]] that queues work on an [[java.util.concurrent.ScheduledExecutorService]].
+   */
+  def executor(executor: ScheduledExecutorService): Scheduler = rx.concurrency.Schedulers.executor(executor)
+
+  /**
+   * Returns a [[rx.lang.scala.Scheduler]] intended for computational work.
+   * 
+   * The implementation is backed by a [[java.util.concurrent.ScheduledExecutorService]] thread-pool sized to the number of CPU cores.
+   *
+   * This can be used for event-loops, processing callbacks and other computational work.
+   * 
+   * Do not perform IO-bound work on this scheduler. Use [[rx.lang.scala.concurrency.Schedulers.threadPoolForIO]] instead.
+   */
+  def threadPoolForComputation: Scheduler = rx.concurrency.Schedulers.threadPoolForComputation()
+  
+  /**
+   * [[rx.lang.scala.Scheduler]] intended for IO-bound work.
+   * 
+   * The implementation is backed by an [[java.util.concurrent.Executor]] thread-pool that will grow as needed.
+   * 
+   * This can be used for asynchronously performing blocking IO.
+   * 
+   * Do not perform computational work on this scheduler. Use [[rx.lang.scala.concurrency.Schedulers.threadPoolForComputation]] instead.
+   */
+  def threadPoolForIO: Scheduler = rx.concurrency.Schedulers.threadPoolForIO()
+
+}

language-adaptors/rxjava-scala/src/main/scala/rx/lang/scala/concurrency/package.scala

@@ -18,17 +18,11 @@ package rx.lang.scala
 import rx.concurrency.CurrentThreadScheduler
 
 package object concurrency {
-  /*
-  TODO
-  rx.concurrency.CurrentThreadScheduler
-  rx.concurrency.ExecutorScheduler
-  rx.concurrency.ImmediateScheduler
-  rx.concurrency.NewThreadScheduler
-  rx.concurrency.Schedulers
-  rx.concurrency.TestScheduler
- */
+  // These classes are not exposed to Scala users, but are accessible through
+  // rx.lang.scala.concurrency.Schedulers:
 
-  lazy val CurrentThreadScheduler = rx.concurrency.CurrentThreadScheduler.getInstance()
-  lazy val NewThreadScheduler = rx.concurrency.NewThreadScheduler.getInstance()
-
+  // rx.concurrency.CurrentThreadScheduler
+  // rx.concurrency.ExecutorScheduler
+  // rx.concurrency.ImmediateScheduler
+  // rx.concurrency.NewThreadScheduler
 }