Making it possible/mandatory to signal which ExecutionContext will actually execute something scheduled

akka-actor/src/main/scala/akka/actor/Scheduler.scala

@@ -12,6 +12,7 @@ import java.io.Closeable
 import java.util.concurrent.atomic.AtomicReference
 import scala.annotation.tailrec
 import akka.util.internal._
+import concurrent.ExecutionContext
 
 //#scheduler
 /**
@@ -36,7 +37,7 @@ trait Scheduler {
     initialDelay: Duration,
     frequency: Duration,
     receiver: ActorRef,
-    message: Any): Cancellable
+    message: Any)(implicit executor: ExecutionContext): Cancellable
 
   /**
    * Schedules a function to be run repeatedly with an initial delay and a
@@ -47,7 +48,7 @@ trait Scheduler {
    * Scala API
    */
   def schedule(
-    initialDelay: Duration, frequency: Duration)(f: ⇒ Unit): Cancellable
+    initialDelay: Duration, frequency: Duration)(f: ⇒ Unit)(implicit executor: ExecutionContext): Cancellable
 
   /**
    * Schedules a function to be run repeatedly with an initial delay and
@@ -58,7 +59,7 @@ trait Scheduler {
    * Java API
    */
   def schedule(
-    initialDelay: Duration, frequency: Duration, runnable: Runnable): Cancellable
+    initialDelay: Duration, frequency: Duration, runnable: Runnable)(implicit executor: ExecutionContext): Cancellable
 
   /**
    * Schedules a Runnable to be run once with a delay, i.e. a time period that
@@ -66,7 +67,7 @@ trait Scheduler {
    *
    * Java & Scala API
    */
-  def scheduleOnce(delay: Duration, runnable: Runnable): Cancellable
+  def scheduleOnce(delay: Duration, runnable: Runnable)(implicit executor: ExecutionContext): Cancellable
 
   /**
    * Schedules a message to be sent once with a delay, i.e. a time period that has
@@ -74,7 +75,7 @@ trait Scheduler {
    *
    * Java & Scala API
    */
-  def scheduleOnce(delay: Duration, receiver: ActorRef, message: Any): Cancellable
+  def scheduleOnce(delay: Duration, receiver: ActorRef, message: Any)(implicit executor: ExecutionContext): Cancellable
 
   /**
    * Schedules a function to be run once with a delay, i.e. a time period that has
@@ -82,7 +83,7 @@ trait Scheduler {
    *
    * Scala API
    */
-  def scheduleOnce(delay: Duration)(f: ⇒ Unit): Cancellable
+  def scheduleOnce(delay: Duration)(f: ⇒ Unit)(implicit executor: ExecutionContext): Cancellable
 }
 //#scheduler
 
@@ -118,76 +119,61 @@ trait Cancellable {
  * if it does not enqueue a task. Once a task is queued, it MUST be executed or
  * returned from stop().
  */
-class DefaultScheduler(hashedWheelTimer: HashedWheelTimer,
-                       log: LoggingAdapter,
-                       dispatcher: ⇒ MessageDispatcher) extends Scheduler with Closeable {
-
-  override def schedule(initialDelay: Duration, delay: Duration, receiver: ActorRef, message: Any): Cancellable = {
+class DefaultScheduler(hashedWheelTimer: HashedWheelTimer, log: LoggingAdapter) extends Scheduler with Closeable {
+  override def schedule(initialDelay: Duration,
+                        delay: Duration,
+                        receiver: ActorRef,
+                        message: Any)(implicit executor: ExecutionContext): Cancellable = {
     val continuousCancellable = new ContinuousCancellable
     continuousCancellable.init(
       hashedWheelTimer.newTimeout(
         new TimerTask with ContinuousScheduling {
           def run(timeout: HWTimeout) {
-            receiver ! message
-            // Check if the receiver is still alive and kicking before reschedule the task
-            if (receiver.isTerminated) log.debug("Could not reschedule message to be sent because receiving actor has been terminated.")
-            else scheduleNext(timeout, delay, continuousCancellable)
+            executor execute new Runnable {
+              override def run = {
+                receiver ! message
+                // Check if the receiver is still alive and kicking before reschedule the task
+                if (receiver.isTerminated) log.debug("Could not reschedule message to be sent because receiving actor {} has been terminated.", receiver)
+                else scheduleNext(timeout, delay, continuousCancellable)
+              }
+            }
           }
         },
         initialDelay))
   }
 
-  override def schedule(initialDelay: Duration, delay: Duration)(f: ⇒ Unit): Cancellable = {
-    val continuousCancellable = new ContinuousCancellable
-    continuousCancellable.init(
-      hashedWheelTimer.newTimeout(
-        new TimerTask with ContinuousScheduling with Runnable {
-          def run = f
-          def run(timeout: HWTimeout) {
-            dispatcher.execute(this)
-            scheduleNext(timeout, delay, continuousCancellable)
-          }
-        },
-        initialDelay))
-  }
+  override def schedule(initialDelay: Duration,
+                        delay: Duration)(f: ⇒ Unit)(implicit executor: ExecutionContext): Cancellable =
+    schedule(initialDelay, delay, new Runnable { override def run = f })
 
-  override def schedule(initialDelay: Duration, delay: Duration, runnable: Runnable): Cancellable = {
+  override def schedule(initialDelay: Duration,
+                        delay: Duration,
+                        runnable: Runnable)(implicit executor: ExecutionContext): Cancellable = {
     val continuousCancellable = new ContinuousCancellable
     continuousCancellable.init(
       hashedWheelTimer.newTimeout(
         new TimerTask with ContinuousScheduling {
-          def run(timeout: HWTimeout) {
-            dispatcher.execute(runnable)
-            scheduleNext(timeout, delay, continuousCancellable)
-          }
+          override def run(timeout: HWTimeout): Unit = executor.execute(new Runnable {
+            override def run = {
+              runnable.run()
+              scheduleNext(timeout, delay, continuousCancellable)
+            }
+          })
         },
         initialDelay))
   }
 
-  override def scheduleOnce(delay: Duration, runnable: Runnable): Cancellable =
+  override def scheduleOnce(delay: Duration, runnable: Runnable)(implicit executor: ExecutionContext): Cancellable =
     new DefaultCancellable(
       hashedWheelTimer.newTimeout(
-        new TimerTask() {
-          def run(timeout: HWTimeout): Unit = dispatcher.execute(runnable)
-        },
+        new TimerTask() { def run(timeout: HWTimeout): Unit = executor.execute(runnable) },
         delay))
 
-  override def scheduleOnce(delay: Duration, receiver: ActorRef, message: Any): Cancellable =
-    new DefaultCancellable(
-      hashedWheelTimer.newTimeout(
-        new TimerTask {
-          def run(timeout: HWTimeout): Unit = receiver ! message
-        },
-        delay))
+  override def scheduleOnce(delay: Duration, receiver: ActorRef, message: Any)(implicit executor: ExecutionContext): Cancellable =
+    scheduleOnce(delay, new Runnable { override def run = receiver ! message })
 
-  override def scheduleOnce(delay: Duration)(f: ⇒ Unit): Cancellable =
-    new DefaultCancellable(
-      hashedWheelTimer.newTimeout(
-        new TimerTask with Runnable {
-          def run = f
-          def run(timeout: HWTimeout): Unit = dispatcher.execute(this)
-        },
-        delay))
+  override def scheduleOnce(delay: Duration)(f: ⇒ Unit)(implicit executor: ExecutionContext): Cancellable =
+    scheduleOnce(delay, new Runnable { override def run = f })
 
   private trait ContinuousScheduling { this: TimerTask ⇒
     def scheduleNext(timeout: HWTimeout, delay: Duration, delegator: ContinuousCancellable) {