Merge remote-tracking branch 'origin/master' into wip-1581-patterns-ask

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

@@ -13,6 +13,11 @@
 package akka.actor
 
 import akka.util.Duration
+import org.jboss.netty.akka.util.{ Timer, TimerTask, HashedWheelTimer, Timeout ⇒ HWTimeout }
+import akka.event.LoggingAdapter
+import akka.dispatch.MessageDispatcher
+import java.io.Closeable
+
 //#scheduler
 /**
  * An Akka scheduler service. This one needs one special behavior: if
@@ -108,3 +113,149 @@ trait Cancellable {
   def isCancelled: Boolean
 }
 //#cancellable
+
+/**
+ * Scheduled tasks (Runnable and functions) are executed with the supplied dispatcher.
+ * Note that dispatcher is by-name parameter, because dispatcher might not be initialized
+ * when the scheduler is created.
+ *
+ * The HashedWheelTimer used by this class MUST throw an IllegalStateException
+ * 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 {
+
+  def schedule(initialDelay: Duration, delay: Duration, receiver: ActorRef, message: Any): Cancellable = {
+    val continuousCancellable = new ContinuousCancellable
+    val task = 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.warning("Could not reschedule message to be sent because receiving actor has been terminated.")
+        } else {
+          scheduleNext(timeout, delay, continuousCancellable)
+        }
+      }
+    }
+    continuousCancellable.init(hashedWheelTimer.newTimeout(task, initialDelay))
+    continuousCancellable
+  }
+
+  def schedule(initialDelay: Duration, delay: Duration)(f: ⇒ Unit): Cancellable = {
+    val continuousCancellable = new ContinuousCancellable
+    val task = new TimerTask with ContinuousScheduling with Runnable {
+      def run = f
+      def run(timeout: HWTimeout) {
+        dispatcher execute this
+        scheduleNext(timeout, delay, continuousCancellable)
+      }
+    }
+    continuousCancellable.init(hashedWheelTimer.newTimeout(task, initialDelay))
+    continuousCancellable
+  }
+
+  def schedule(initialDelay: Duration, delay: Duration, runnable: Runnable): Cancellable = {
+    val continuousCancellable = new ContinuousCancellable
+    val task = new TimerTask with ContinuousScheduling {
+      def run(timeout: HWTimeout) {
+        dispatcher.execute(runnable)
+        scheduleNext(timeout, delay, continuousCancellable)
+      }
+    }
+    continuousCancellable.init(hashedWheelTimer.newTimeout(task, initialDelay))
+    continuousCancellable
+  }
+
+  def scheduleOnce(delay: Duration, runnable: Runnable): Cancellable = {
+    val task = new TimerTask() {
+      def run(timeout: HWTimeout) { dispatcher.execute(runnable) }
+    }
+    new DefaultCancellable(hashedWheelTimer.newTimeout(task, delay))
+  }
+
+  def scheduleOnce(delay: Duration, receiver: ActorRef, message: Any): Cancellable = {
+    val task = new TimerTask {
+      def run(timeout: HWTimeout) {
+        receiver ! message
+      }
+    }
+    new DefaultCancellable(hashedWheelTimer.newTimeout(task, delay))
+  }
+
+  def scheduleOnce(delay: Duration)(f: ⇒ Unit): Cancellable = {
+    val task = new TimerTask {
+      def run(timeout: HWTimeout) {
+        dispatcher.execute(new Runnable { def run = f })
+      }
+    }
+    new DefaultCancellable(hashedWheelTimer.newTimeout(task, delay))
+  }
+
+  private trait ContinuousScheduling { this: TimerTask ⇒
+    def scheduleNext(timeout: HWTimeout, delay: Duration, delegator: ContinuousCancellable) {
+      try {
+        delegator.swap(timeout.getTimer.newTimeout(this, delay))
+      } catch {
+        case _: IllegalStateException ⇒ // stop recurring if timer is stopped
+      }
+    }
+  }
+
+  private def execDirectly(t: HWTimeout): Unit = {
+    try t.getTask.run(t) catch {
+      case e: InterruptedException ⇒ throw e
+      case e: Exception            ⇒ log.error(e, "exception while executing timer task")
+    }
+  }
+
+  def close() = {
+    import scala.collection.JavaConverters._
+    hashedWheelTimer.stop().asScala foreach execDirectly
+  }
+}
+
+/**
+ * Wrapper of a [[org.jboss.netty.akka.util.Timeout]] that delegates all
+ * methods. Needed to be able to cancel continuous tasks,
+ * since they create new Timeout for each tick.
+ */
+private[akka] class ContinuousCancellable extends Cancellable {
+  @volatile
+  private var delegate: HWTimeout = _
+  @volatile
+  private var cancelled = false
+
+  private[akka] def init(initialTimeout: HWTimeout): Unit = {
+    delegate = initialTimeout
+  }
+
+  private[akka] def swap(newTimeout: HWTimeout): Unit = {
+    val wasCancelled = isCancelled
+    delegate = newTimeout
+    if (wasCancelled || isCancelled) cancel()
+  }
+
+  def isCancelled(): Boolean = {
+    // delegate is initially null, but this object will not be exposed to the world until after init
+    cancelled || delegate.isCancelled()
+  }
+
+  def cancel(): Unit = {
+    // the underlying Timeout will not become cancelled once the task has been started to run,
+    // therefore we keep a flag here to make sure that rescheduling doesn't occur when cancelled
+    cancelled = true
+    // delegate is initially null, but this object will not be exposed to the world until after init
+    delegate.cancel()
+  }
+}
+
+class DefaultCancellable(val timeout: HWTimeout) extends Cancellable {
+  def cancel() {
+    timeout.cancel()
+  }
+
+  def isCancelled: Boolean = {
+    timeout.isCancelled
+  }
+}