-
Notifications
You must be signed in to change notification settings - Fork 3.6k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Scheduler that manually advances time (#24150)
First version for review. More tests, docs and Java examples still TBD.
- Loading branch information
Showing
7 changed files
with
181 additions
and
4 deletions.
There are no files selected for viewing
37 changes: 37 additions & 0 deletions
37
akka-actor-typed-tests/src/test/scala/akka/actor/typed/ManualTimerSpec.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,37 @@ | ||
package akka.actor.typed | ||
|
||
import scala.concurrent.duration._ | ||
|
||
import akka.actor.typed.scaladsl.Behaviors | ||
import akka.testkit.typed.TestKit | ||
import akka.testkit.typed.scaladsl.{ ManualTime, TestProbe } | ||
import org.scalatest.WordSpecLike | ||
|
||
class ManualTimerSpec extends TestKit() with ManualTime with WordSpecLike { | ||
//#manual-scheduling-simple | ||
"A timer" must { | ||
"schedule non-repeated ticks" in { | ||
case object Tick | ||
case object Tock | ||
|
||
val probe = TestProbe[Tock.type]() | ||
val behv = Behaviors.withTimers[Tick.type] { timer ⇒ | ||
timer.startSingleTimer("T", Tick, 10.millis) | ||
Behaviors.immutable { (ctx, Tick) ⇒ | ||
probe.ref ! Tock | ||
Behaviors.same | ||
} | ||
} | ||
|
||
val ref = spawn(behv) | ||
|
||
scheduler.timePasses(9.millis) | ||
probe.expectNoMessage(Duration.Zero) | ||
|
||
scheduler.timePasses(2.millis) | ||
probe.expectMsg(Tock) | ||
probe.expectNoMessage(Duration.Zero) | ||
} | ||
} | ||
//#manual-scheduling-simple | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
16 changes: 16 additions & 0 deletions
16
akka-testkit-typed/src/main/scala/akka/testkit/typed/ExplicitlyTriggeredScheduler.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,16 @@ | ||
package akka.testkit.typed | ||
|
||
import java.util.concurrent.ThreadFactory | ||
|
||
import akka.event.LoggingAdapter | ||
import com.typesafe.config.Config | ||
|
||
import scala.concurrent.duration.FiniteDuration | ||
|
||
class ExplicitlyTriggeredScheduler(config: Config, log: LoggingAdapter, tf: ThreadFactory) extends akka.testkit.ExplicitlyTriggeredScheduler(config, log, tf) { | ||
def timePasses(amount: FiniteDuration) = { | ||
val newTime = currentTime.get + amount.toMillis | ||
executeTasks(newTime) | ||
currentTime.set(newTime) | ||
} | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
14 changes: 14 additions & 0 deletions
14
akka-testkit-typed/src/main/scala/akka/testkit/typed/scaladsl/ManualTime.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,14 @@ | ||
package akka.testkit.typed.scaladsl | ||
|
||
import com.typesafe.config.{ Config, ConfigFactory } | ||
|
||
import akka.testkit.typed._ | ||
|
||
trait ManualTime extends TestKitMixin { self: TestKit ⇒ | ||
override def mixedInConfig: Config = | ||
ConfigFactory | ||
.parseString("""akka.scheduler.implementation = "akka.testkit.typed.ExplicitlyTriggeredScheduler"""") | ||
.withFallback(super.mixedInConfig) | ||
|
||
override val scheduler: ExplicitlyTriggeredScheduler = self.system.scheduler.asInstanceOf[ExplicitlyTriggeredScheduler] | ||
} |
86 changes: 86 additions & 0 deletions
86
akka-testkit/src/main/scala/akka/testkit/ExplicitlyTriggeredScheduler.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,86 @@ | ||
package akka.testkit | ||
|
||
import java.util.concurrent.ThreadFactory | ||
import java.util.concurrent.ConcurrentHashMap | ||
import java.util.concurrent.atomic.AtomicLong | ||
|
||
import scala.annotation.tailrec | ||
import scala.collection.JavaConverters._ | ||
import scala.concurrent.ExecutionContext | ||
import scala.concurrent.duration.{ Duration, FiniteDuration } | ||
|
||
import com.typesafe.config.Config | ||
|
||
import akka.actor.{ ActorSystem, Cancellable, Scheduler } | ||
import akka.event.LoggingAdapter | ||
|
||
/** | ||
* For testing: scheduler that does not look at the clock, but must be progressed manually by calling `timePasses`. | ||
* | ||
* This is not entirely realistic: jobs will be executed on the test thread instead of using the `ExecutionContext`, but does | ||
* allow for faster and less timing-sensitive specs.. | ||
*/ | ||
class ExplicitlyTriggeredScheduler(config: Config, log: LoggingAdapter, tf: ThreadFactory) extends Scheduler { | ||
|
||
case class Item(time: Long, interval: Option[FiniteDuration], runnable: Runnable) | ||
|
||
val currentTime = new AtomicLong() | ||
val scheduled = new ConcurrentHashMap[Item, Unit]() | ||
|
||
override def schedule(initialDelay: FiniteDuration, interval: FiniteDuration, runnable: Runnable)(implicit executor: ExecutionContext): Cancellable = | ||
schedule(initialDelay, Some(interval), runnable) | ||
|
||
override def scheduleOnce(delay: FiniteDuration, runnable: Runnable)(implicit executor: ExecutionContext): Cancellable = | ||
schedule(delay, None, runnable) | ||
|
||
def timePasses(amount: FiniteDuration)(implicit system: ActorSystem) = { | ||
// TODO double-check if we really want/need dilation here | ||
val newTime = currentTime.get + amount.dilated.toMillis | ||
executeTasks(newTime) | ||
currentTime.set(newTime) | ||
} | ||
|
||
@tailrec | ||
private[testkit] final def executeTasks(runTo: Long): Unit = { | ||
scheduled | ||
.keySet | ||
.asScala | ||
.filter(_.time <= runTo) | ||
.toList | ||
.sortBy(_.time) | ||
.headOption match { | ||
case Some(task) ⇒ | ||
currentTime.set(task.time) | ||
task.runnable.run() | ||
scheduled.remove(task) | ||
task.interval.foreach(i ⇒ scheduled.put(task.copy(time = task.time + i.toMillis), ())) | ||
|
||
// running the runnable might have scheduled new events | ||
executeTasks(runTo) | ||
case _ ⇒ // Done | ||
} | ||
} | ||
|
||
private def schedule(initialDelay: FiniteDuration, interval: Option[FiniteDuration], runnable: Runnable)(implicit executor: ExecutionContext): Cancellable = { | ||
val item = Item(currentTime.get + initialDelay.toMillis, interval, runnable) | ||
scheduled.put(item, ()) | ||
|
||
if (initialDelay == Duration.Zero) | ||
executeTasks(currentTime.get) | ||
|
||
new Cancellable { | ||
var cancelled = false | ||
|
||
override def cancel(): Boolean = { | ||
val before = scheduled.size | ||
scheduled.remove(item) | ||
cancelled = true | ||
before > scheduled.size | ||
} | ||
|
||
override def isCancelled: Boolean = cancelled | ||
} | ||
} | ||
|
||
override def maxFrequency: Double = 42 | ||
} |