我正在测试一个使用异步的基于未来的API的actor。当将来完成时,actor使用管道模式向自己发送消息:
import akka.pattern.pipe
// ...
// somewhere in the actor's receive method
futureBasedApi.doSomething().pipeTo(self)
在我的测试中,我模拟了API,因此我通过promises控制未来的完成。但是,这与直接发送给actor的其他消息交错:
myActor ! Message("A")
promiseFromApiCall.success(Message("B"))
myActor ! Message("C")
现在我想知道如何保证演员在我的测试中接收并处理消息A和C之间的消息B,因为消息B实际上是在另一个线程中发送的,所以我无法控制演员的邮箱收到的顺序消息。
我想到了几种可能的解决方案:
我真的不喜欢这些选项,但我倾向于使用最后一个选项。还有另一种更好的方法可以在测试中强制执行某个消息顺序吗?
澄清:问题不在于如何处理可能在生产中以随机顺序接收消息的事实。控制测试中的顺序对于确保actor可以实际处理不同的消息顺序至关重要。
一个想法是在你的actor中定义一个标志,指示玩家是否收到了消息B.当玩家收到消息C时,如果标志为假,则演员可以stash消息C,然后一旦玩家收到消息B,则取消暂存。例如:
class MyActor extends Actor with Stash {
def receiveBlock(seenMsgB: Boolean, seenMsgC: Boolean): Receive = {
case MakeApiCall =>
callExternalApi().mapTo[MessageB].pipeTo(self)
case m: MessageB if seenMsgC => // assume msg C has been stashed
unstashAll()
// ...do something with msg B
become(receiveBlock(true, seenMsgC)) // true, true
case m: MessageB if !seenMsgC =>
// ...do something with message B
become(receiveBlock(true, seenMsgC)) // true, false
case m: MessageC if seenMsgB =>
// ...do something with message C
context.become(receiveBlock(seenMsgB, true)) // true, true
case m: MessageC if !seenMsgB =>
stash()
context.become(receiveBlock(seenMsgB, true)) // false, true
case ...
}
def receive = receiveBlock(false, false)
}
在阅读了很多关于akka的内容之后,我终于找到了一个更好的解决方案:用我可以在测试中观察到的一个替换actor邮箱。这样我可以等到演员在完成承诺后收到新消息。只有这样才会发送下一条消息。这个TestingMailbox
的代码在帖子的末尾给出。
更新:在Akka Typed中,这可以通过BehaviorInterceptor
非常优雅地实现。只需使用自定义拦截器包装Behavior
,该拦截器会转发所有消息和信号,但让您观察它们。下面给出了无类型Akka的邮箱解决方案。
可以像这样配置actor:
actorUnderTest = system.actorOf(Props[MyActor]).withMailbox("testing-mailbox"))
我必须通过提供配置来确保actor系统知道“testing-mailbox”:
class MyTest extends TestKit(ActorSystem("some name",
ConfigFactory.parseString("""{
testing-mailbox = {
mailbox-type = "my.package.TestingMailbox"
}
}""")))
with BeforeAndAfterAll // ... and so on
设置完成后,我可以像这样更改我的测试:
myActor ! Message("A")
val nextMessage = TestingMailbox.nextMessage(actorUnderTest)
promiseFromApiCall.success(Message("B"))
Await.ready(nextMessage, 3.seconds)
myActor ! Message("C")
使用一个小帮助方法,我甚至可以像这样写:
myActor ! Message("A")
receiveMessageAfter { promiseFromApiCall.success(Message("B")) }
myActor ! Message("C")
这是我的自定义邮箱:
import akka.actor.{ActorRef, ActorSystem}
import akka.dispatch._
import com.typesafe.config.Config
import scala.concurrent.{Future, Promise}
object TestingMailbox {
val promisesByReceiver =
scala.collection.concurrent.TrieMap[ActorRef, Promise[Any]]()
class MessageQueue extends UnboundedMailbox.MessageQueue {
override def enqueue(receiver: ActorRef, handle: Envelope): Unit = {
super.enqueue(receiver, handle)
promisesByReceiver.remove(receiver).foreach(_.success(handle.message))
}
}
def nextMessage(receiver: ActorRef): Future[Any] =
promisesByReceiver.getOrElseUpdate(receiver, Promise[Any]).future
}
class TestingMailbox extends MailboxType
with ProducesMessageQueue[TestingMailbox.MessageQueue] {
import TestingMailbox._
def this(settings: ActorSystem.Settings, config: Config) = this()
final override def create(owner: Option[ActorRef],
system: Option[ActorSystem]) =
new MessageQueue()
}
如果订购消息非常重要,你应该使用ask
(?
)返回Future
并链接它们,即使你不期待演员的任何回应。