diff --git a/actor-testkit-typed/src/main/scala/org/apache/pekko/actor/testkit/typed/javadsl/ManualTime.scala b/actor-testkit-typed/src/main/scala/org/apache/pekko/actor/testkit/typed/javadsl/ManualTime.scala index abb91dfa6a..9f0d7c35c5 100644 --- a/actor-testkit-typed/src/main/scala/org/apache/pekko/actor/testkit/typed/javadsl/ManualTime.scala +++ b/actor-testkit-typed/src/main/scala/org/apache/pekko/actor/testkit/typed/javadsl/ManualTime.scala @@ -72,10 +72,16 @@ final class ManualTime(delegate: pekko.testkit.ExplicitlyTriggeredScheduler) { */ def timePasses(amount: Duration): Unit = delegate.timePasses(amount.toScala) + /** + * Advance the clock by the specified duration and assert that the probes receive no messages. + * + * After advancing the clock, each probe waits for the configured + * `pekko.actor.testkit.typed.expect-no-message-default` duration so that asynchronously dispatched messages can arrive. + */ @varargs def expectNoMessageFor(duration: Duration, on: TestProbe[?]*): Unit = { delegate.timePasses(duration.toScala) - on.foreach(_.expectNoMessage(Duration.ZERO)) + on.foreach(_.expectNoMessage()) } } diff --git a/actor-testkit-typed/src/main/scala/org/apache/pekko/actor/testkit/typed/scaladsl/ManualTime.scala b/actor-testkit-typed/src/main/scala/org/apache/pekko/actor/testkit/typed/scaladsl/ManualTime.scala index ae765bcbfe..c631307343 100644 --- a/actor-testkit-typed/src/main/scala/org/apache/pekko/actor/testkit/typed/scaladsl/ManualTime.scala +++ b/actor-testkit-typed/src/main/scala/org/apache/pekko/actor/testkit/typed/scaladsl/ManualTime.scala @@ -14,7 +14,7 @@ package org.apache.pekko.actor.testkit.typed.scaladsl import scala.annotation.varargs -import scala.concurrent.duration.{ Duration, FiniteDuration } +import scala.concurrent.duration.FiniteDuration import org.apache.pekko import pekko.actor.typed.ActorSystem @@ -72,10 +72,16 @@ final class ManualTime(delegate: pekko.testkit.ExplicitlyTriggeredScheduler) { */ def timePasses(amount: FiniteDuration): Unit = delegate.timePasses(amount) + /** + * Advance the clock by the specified duration and assert that the probes receive no messages. + * + * After advancing the clock, each probe waits for the configured + * `pekko.actor.testkit.typed.expect-no-message-default` duration so that asynchronously dispatched messages can arrive. + */ @varargs def expectNoMessageFor(duration: FiniteDuration, on: TestProbe[?]*): Unit = { delegate.timePasses(duration) - on.foreach(_.expectNoMessage(Duration.Zero)) + on.foreach(_.expectNoMessage()) } } diff --git a/actor-testkit-typed/src/test/scala/org/apache/pekko/actor/testkit/typed/scaladsl/ManualTimeSpec.scala b/actor-testkit-typed/src/test/scala/org/apache/pekko/actor/testkit/typed/scaladsl/ManualTimeSpec.scala new file mode 100644 index 0000000000..e9e308bd0f --- /dev/null +++ b/actor-testkit-typed/src/test/scala/org/apache/pekko/actor/testkit/typed/scaladsl/ManualTimeSpec.scala @@ -0,0 +1,122 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.pekko.actor.testkit.typed.scaladsl + +import java.time.{ Duration => JDuration } +import java.util.concurrent.{ CountDownLatch, TimeUnit } + +import scala.concurrent.duration._ + +import org.apache.pekko +import pekko.Done +import pekko.actor.testkit.typed.scaladsl.ManualTimeSpec.config +import pekko.actor.typed.DispatcherSelector +import pekko.actor.typed.scaladsl.Behaviors + +import org.scalatest.wordspec.AnyWordSpecLike + +import com.typesafe.config.ConfigFactory + +object ManualTimeSpec { + val config = ConfigFactory + .parseString(""" + manual-time-test-dispatcher { + type = Dispatcher + executor = "thread-pool-executor" + thread-pool-executor.fixed-pool-size = 1 + throughput = 1 + } + pekko.actor.testkit.typed.expect-no-message-default = 3 s + """) + .withFallback(ManualTime.config) +} + +class ManualTimeSpec extends ScalaTestWithActorTestKit(config) with AnyWordSpecLike with LogCapturing { + + private val manualTime = ManualTime() + + private def verifyDelayedMessageIsDetected(expectNoMessageFor: TestProbe[String] => Unit): Unit = { + val target = TestProbe[String]() + val timerStarted = TestProbe[Done]() + val dispatcherSelector = DispatcherSelector.fromConfig("manual-time-test-dispatcher") + + spawn( + Behaviors.withTimers[String] { timers => + timers.startSingleTimer("unexpected", 1.milli) + timerStarted.ref ! Done + Behaviors.receiveMessage { message => + target.ref ! message + Behaviors.same + } + }, + dispatcherSelector) + timerStarted.expectMessage(Done) + + val dispatcherBlocked = new CountDownLatch(1) + val releaseDispatcher = new CountDownLatch(1) + val blocker = spawn( + Behaviors.receiveMessage[Done] { _ => + dispatcherBlocked.countDown() + releaseDispatcher.await() + Behaviors.stopped + }, + dispatcherSelector) + blocker ! Done + dispatcherBlocked.await(3, TimeUnit.SECONDS) shouldBe true + + val testThread = Thread.currentThread() + val watcherDone = new CountDownLatch(1) + system.scheduler.scheduleOnce( + 2.millis, + () => { + val watcher = new Thread( + () => { + try { + while (releaseDispatcher.getCount != 0 && testThread.getState != Thread.State.TIMED_WAITING) + Thread.onSpinWait() + if (testThread.getState == Thread.State.TIMED_WAITING) + releaseDispatcher.countDown() + } finally watcherDone.countDown() + }, + "manual-time-expect-no-message-watcher") + watcher.setDaemon(true) + watcher.start() + })(system.executionContext) + + try { + intercept[AssertionError] { + expectNoMessageFor(target) + }.getMessage should include("Received unexpected message unexpected") + } finally { + releaseDispatcher.countDown() + watcherDone.await(3, TimeUnit.SECONDS) shouldBe true + } + } + + "ManualTime.expectNoMessageFor" must { + "detect a delayed message with the Scala DSL" in { + verifyDelayedMessageIsDetected(probe => manualTime.expectNoMessageFor(2.millis, probe)) + } + + "detect a delayed message with the Java DSL" in { + val javaManualTime = pekko.actor.testkit.typed.javadsl.ManualTime.get(system) + verifyDelayedMessageIsDetected(probe => + javaManualTime.expectNoMessageFor(JDuration.ofMillis(2), probe.asJava)) + } + } +}