scala - Akka:测试监控\死神守望

Question

在我的场景中，我有 2 个 Actor :

1. watchee(我使用TestProbe)
2. watcher(Watcher 包装到 TestActorRef 中以公开我在测试中跟踪的一些内部 state)<

观察者应该在 watchee 死亡时采取一些行动。

这是我到目前为止编写的完整测试用例:

class TempTest(_system: ActorSystem) extends TestKit(_system) with ImplicitSender with FunSuiteLike with Matchers with BeforeAndAfterAll {

  def this() = this(ActorSystem("TempTest"))

  override def afterAll {
    TestKit.shutdownActorSystem(system)
  }

  class WatcherActor(watchee: ActorRef) extends Actor {

    var state = "initial"
    context.watch(watchee)

    override def receive: Receive = {
      case "start" =>
        state = "start"
      case _: Terminated =>
        state = "terminated"
    }

  }

  test("example") {
    val watchee = TestProbe()
    val watcher = TestActorRef[WatcherActor](Props(new WatcherActor(watchee.ref)))

    assert(watcher.underlyingActor.state === "initial")

    watcher ! "start" // "start" will be sent and handled by watcher synchronously
    assert(watcher.underlyingActor.state === "start")

    system.stop(watchee.ref) // will cause Terminated to be sent and handled asynchronously by watcher
    Thread.sleep(100) // what is the best way to avoid blocking here?
    assert(watcher.underlyingActor.state === "terminated")
  }

}

现在，由于所有相关参与者都使用 CallingThreadDispatcher(所有 Akka 的测试助手都是使用带有 .withDispatcher(CallingThreadDispatcher.Id) 的 Prop 构建的)我可以放心地假设当这语句返回:

watcher ! "start"

...“开始”消息已由 WatchingActor 处理，因此我可以根据 watcher.underlyingActor.state

进行断言

但是，根据我的观察，当我使用 system.stop 或通过向其发送 Kill 停止 watchee 时，Terminated 作为 watchee 死亡的副作用产生的消息在另一个线程中异步执行。

不是解决方案是停止 watchee，阻塞线程一段时间，然后验证 Watcher 状态，但我想知道我该怎么做这是正确的方法(即如何确保在杀死 Actor 后它的观察者收到并处理 Terminated 消息表明它已经死亡)？

Answer 1

解决此问题的一种方法是在您的测试中引入另一个观察者，该观察者也观察 watchee。另一个观察者是 TestProbe，它允许我们对其执行断言，从而消除您看到的计时问题。一、修改后的测试代码:

 val watchee = TestProbe()
 val watcher = TestActorRef[WatcherActor](Props(new WatcherActor(watchee.ref)))
 val probeWatcher = TestProbe()
 probeWatcher watch watchee.ref

 assert(watcher.underlyingActor.state === "initial")

 watcher ! "start" // "start" will be sent and handled by watcher synchronously
 assert(watcher.underlyingActor.state === "start")

 system.stop(watchee.ref) // will cause Terminated to be sent and handled asynchronously by watcher
 probeWatcher.expectTerminated(watchee.ref)
 assert(watcher.underlyingActor.state === "terminated")

所以你可以看到我用以下几行介绍了额外的观察者:

val probeWatcher = TestProbe()
probeWatcher watch watchee.ref

然后，在代码的后面，在对您来说失败的最终断言之前，我使用了另一个断言，让我知道已停止 actor 的 Terminated 消息已正确分发:

probeWatcher.expectTerminated(watchee.ref)

当代码移过这一行时，我可以确定被测试的 watcher 也收到了它的终止消息，随后的断言将通过。

编辑

正如 OP 所指出的，此代码存在一定程度的不确定性。另一种可能的解决方案是将测试代码中停止 actor 的行更改为:

watcher.underlyingActor.context.stop(watchee.ref)

通过使用 TestActorRef 的 context 我相信 Terminated 将全部通过 CallingThreadDispatcher 传送并且因此是完全同步的。我在一个循环中对此进行了测试，它对我有效超过 1000 次迭代。

现在我想也许是因为我正在使用期待 Terminated 的同一个 actor 执行 stop 可能有一个优化来交付 Terminated 为那个 scanario 自己，所以我还用一个完全不同的 Actor 测试了它，如下所示:

class FooActor extends Actor{
  def receive = {
    case _ =>
  }

然后在测试代码中:

val foo = TestActorRef(new FooActor)

在停止时:

foo.underlyingActor.context.stop(watchee.ref)

这也按预期工作。