raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 10

#1508 - Reducing closure wrapping in scheduler

Browse source
This commit is contained in:
Viktor Klang 2012-05-10 23:55:11 +02:00
parent cfbfa19e41
commit ed4bc1cf0b
1 changed files with 54 additions and 53 deletions
Download patch file Download diff file Expand all files Collapse all files

83
akka-actor/src/main/scala/akka/actor/Scheduler.scala
View file

@ -121,70 +121,70 @@ class DefaultScheduler(hashedWheelTimer: HashedWheelTimer,
def schedule(initialDelay: Duration, delay: Duration, receiver: ActorRef, message: Any): Cancellable = {
val continuousCancellable = new ContinuousCancellable
val task = new TimerTask with ContinuousScheduling {
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)
if (receiver.isTerminated) log.debug("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
},
initialDelay))
}
def schedule(initialDelay: Duration, delay: Duration)(f: ⇒ Unit): Cancellable = {
val continuousCancellable = new ContinuousCancellable
val task = new TimerTask with ContinuousScheduling with Runnable {
continuousCancellable.init(
hashedWheelTimer.newTimeout(
new TimerTask with ContinuousScheduling with Runnable {
def run = f
def run(timeout: HWTimeout) {
dispatcher execute this
dispatcher.execute(this)
scheduleNext(timeout, delay, continuousCancellable)
}
}
continuousCancellable.init(hashedWheelTimer.newTimeout(task, initialDelay))
continuousCancellable
},
initialDelay))
}
def schedule(initialDelay: Duration, delay: Duration, runnable: Runnable): Cancellable = {
val continuousCancellable = new ContinuousCancellable
val task = new TimerTask with ContinuousScheduling {
continuousCancellable.init(
hashedWheelTimer.newTimeout(
new TimerTask with ContinuousScheduling {
def run(timeout: HWTimeout) {
dispatcher.execute(runnable)
scheduleNext(timeout, delay, continuousCancellable)
}
}
continuousCancellable.init(hashedWheelTimer.newTimeout(task, initialDelay))
continuousCancellable
},
initialDelay))
}
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, runnable: Runnable): Cancellable =
new DefaultCancellable(
hashedWheelTimer.newTimeout(
new TimerTask() {
def run(timeout: HWTimeout): Unit = dispatcher.execute(runnable)
},
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, receiver: ActorRef, message: Any): Cancellable =
new DefaultCancellable(
hashedWheelTimer.newTimeout(
new TimerTask {
def run(timeout: HWTimeout): Unit = receiver ! message
},
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))
}
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))
private trait ContinuousScheduling { this: TimerTask ⇒
def scheduleNext(timeout: HWTimeout, delay: Duration, delegator: ContinuousCancellable) {
@ -220,8 +220,9 @@ private[akka] class ContinuousCancellable extends Cancellable {
@volatile
private var cancelled = false
private[akka] def init(initialTimeout: HWTimeout): Unit = {
private[akka] def init(initialTimeout: HWTimeout): this.type = {
delegate = initialTimeout
this
}
private[akka] def swap(newTimeout: HWTimeout): Unit = {