akka-to-monix-scheduler.scala

import java.util.concurrent.TimeUnit
import monix.execution.cancelables.MultiAssignmentCancelable
import monix.execution.{Cancelable, Scheduler}
import monix.execution.schedulers.{ExecutionModel, LocalBatchingExecutor}
import scala.concurrent.ExecutionContext
import scala.concurrent.duration.FiniteDuration

class AkkaToMonixScheduler(
  akkaScheduler: akka.actor.Scheduler,
  context: ExecutionContext,
  em: ExecutionModel = ExecutionModel.Default)
  extends Scheduler with LocalBatchingExecutor {

  def currentTimeMillis(): Long = System.currentTimeMillis()
  def executionModel: ExecutionModel = em

  def reportFailure(t: Throwable): Unit =
    context.reportFailure(t)
  protected def executeAsync(r: Runnable): Unit =
    context.execute(r)

  def scheduleOnce(initialDelay: Long, unit: TimeUnit, r: Runnable): Cancelable = {
    val c = akkaScheduler.scheduleOnce(FiniteDuration(initialDelay, unit), r)(context)
    Cancelable(() => c.cancel())
  }

  def scheduleWithFixedDelay(initialDelay: Long, delay: Long, unit: TimeUnit, r: Runnable): Cancelable = {
    val sub = MultiAssignmentCancelable()

    def loop(initialDelay: Long, delay: Long): Unit =
      sub := scheduleOnce(initialDelay, unit, new Runnable {
        def run(): Unit = {
          r.run()
          loop(delay, delay)
        }
      })

    loop(initialDelay, delay)
    sub
  }

  def scheduleAtFixedRate(initialDelay: Long, period: Long, unit: TimeUnit, r: Runnable): Cancelable = {
    val initialD = FiniteDuration(initialDelay, unit)
    val periodD = FiniteDuration(period, unit)
    // Docs says this executes at fixed rate, however I really
    // don't know if it has the indented behavior!
    val c = akkaScheduler.schedule(initialD, periodD, r)(context)
    Cancelable(() => c.cancel())
  }
}