Scheduling a task at a fixed time of the day with Akka

Question

I am a beginner with Akka. I need to schedule a task each day at a fixed time of the day, say 8AM. What I know how to do is scheduling a task periodically, for instance

import akka.util.duration._

scheduler.schedule(0 seconds, 10 minutes) {
  doSomething()
}

What is the simplest way to schedule tasks at fixed times of the day in Akka?

A small parenthesis

It is easy to do what I want just using this feature. A toy implementation would look like

scheduler.schedule(0 seconds, 24 hours) {
  val now = computeTimeOfDay()
  val delay = desiredTime - now

  scheduler.scheduleOnce(delay) {
    doSomething()
  }
}

It is not difficult, but I introduced a little race condition. In fact, consider what happens if I launch this just before 8AM. The external closure will start, but by the time I compute delay we may be after 8AM. This means that the internal closure - which should execute right away - will be postponed to tomorrow, thereby skipping execution for one day.

There are ways to fix this race condition: for instance I could perform the check every 12 hours, and instead of scheduling the task right away, sending it to an actor that will not accept more than one task at a time.

But probably, this already exist in Akka or some extension.

Answer 1

Write once, run everyday

val GatherStatisticsPeriod = 24 hours

private[this] val scheduled = new AtomicBoolean(false)

def calcBeforeMidnight: Duration = { 
  // TODO implement 
} 

def preRestart(reason: Throwable, message: Option[Any]) {
  self ! GatherStatisticsScheduled(scheduled.get)
  super.preRestart(reason, message)
}

def schedule(period: Duration, who: ActorRef) = 
  ServerRoot.actorSystem.scheduler
    .scheduleOnce(period)(who ! GatherStatisticsTick)

def receive = {

  case StartServer(nodeName) => 
    sender ! ServerStarted(nodeName)
    if (scheduled.compareAndSet(false, true)) 
      schedule(calcBeforeMidnight, self)

  case GatherStatisticsTick =>
    stats.update
    scheduled.set(true)
    schedule(GatherStatisticsPeriod, self) 

  case GatherStatisticsScheduled(isScheduled) =>
    if (isScheduled && scheduled.compareAndSet(false, isScheduled))
      schedule(calcBeforeMidnight, self)

}

I believe that Akka's scheduler handles restarts internally, one way or another. I used non-persistent way of sending a message to self - actually no strict guarantee of delivery. Also, ticks may vary, so GatherStatisticsPeriod might be a function.

Answer 2

To use this kind of scheduling in Akka, you would have to roll your own or maybe use Quartz, either through Akka Camel or this prototype quartz for akka.

If you don't need anything fancy and extremely accurate, then I would just calculate the delay to the desired first time and use that as the start delay to the schedule call, and trust the interval.

Answer 3

Let's say you want to run your task every day at 13 pm.

import scala.concurrent.duration._
import java.time.LocalTime

val interval = 24.hours
val delay = {
  val time = LocalTime.of(13, 0).toSecondOfDay
  val now = LocalTime.now().toSecondOfDay
  val fullDay = 60 * 60 * 24
  val difference = time - now
  if (difference < 0) {
    fullDay + difference
  } else {
    time - now
  }
}.seconds

system.scheduler.schedule(delay, interval)(doSomething())

Also remember that server timezone may be different from yours.

Answer 4

Just to add another way to achieve it, this can be done using Akka Streams by ticking a message and filtering on time.

Source
  .tick(0.seconds, 2.seconds, "hello") // emits "hello" every two seconds
  .filter(_ => {
    val now = LocalDateTime.now.getSecond
    now > 20 && now < 30 // will let through only if the timing is right.
  })
  .runForeach(n => println("final sink received " + n))