unit-testing - 生成的 akka 持久性 actor 测试事件

Question

根据 CQRS 命令的定义，可以/应该验证并在最后甚至拒绝(如果验证未通过)。作为命令验证的一部分，我检查是否真的需要状态转换。因此，让我们举一个简单的虚拟示例:actor 处于状态 A。向 actor 发送一个命令以转换到状态 B。该命令得到验证，并在结束时生成 StateBUpdated 事件。然后发送完全相同的命令以过渡到状态 B。再次对命令进行验证，并且在验证过程中决定不会生成任何事件(因为我们已经处于状态 B)并且只响应命令已处理并且一切正常好的。这是一种幂等性。

尽管如此，我很难(单元)测试它。持久化 actor 的常规单元测试看起来像是向 actor 发送命令，然后重新启动 actor 并检查状态是否持久化。我想测试是否向 Actor 发送命令以检查生成了多少事件。如何做到这一点？

谢谢

Answer 1

我们在开发基于 akka 持久性的内部 CQRS 框架时遇到了这个问题。我们的解决方案是使用持久性查询(https://doc.akka.io/docs/akka/2.5/scala/persistence-query.html)。如果您还没有使用过它，它是一个查询接口(interface)，日志插件可以选择性地实现它，并且可以用作 CQRS 系统中的读取端。

为了您的测试目的，该方法将是 eventsByPersistenceId，它将为您提供一个 akka 流源，其中包含一个参与者保留的所有事件。源可以折叠成事件列表，例如:

public CompletableFuture<List<Message<?>>> getEventsForIdAsync(String id, FiniteDuration readTimeout) {
  return ((EventsByPersistenceIdQuery)readJournal).eventsByPersistenceId(id, 0L, Long.MAX_VALUE)
      .takeWithin(readTimeout)
      .map(eventEnvelope -> (Message<?>)eventEnvelope.event())
      .<List<Message<?>>>runFold(
          new ArrayList<Message<?>>(),
          (list, event) -> {
            list.add(event);
            return list;
          }, materializer)
      .toCompletableFuture();
}

抱歉，如果上面看起来臃肿，我们使用 Java，所以如果你习惯了 Scala，它确实很难看。获取 readJournal 非常简单:

ReadJournal readJournal = PersistenceQuery.lookup().get(actorSystem)
    .getReadJournalFor(InMemoryReadJournal.class, InMemoryReadJournal.Identifier())

您可以看到我们使用了 akka.persistence.inmemory 插件，因为它最适合测试，但任何实现 Persistence Query API 的插件都可以。

我们实际上在我们的框架内做了一个类似 BDD 的测试 API，所以一个典型的测试是这样的:

fixture
  .given("ID1", event(new AccountCreated("ID1", "John Smith")))
  .when(command(new AddAmount("ID1", 2.0)))
  .then("ID1", eventApplied(new AmountAdded("ID1", 2.0)))
  .test();

如您所见，我们还处理在给定子句中设置先前事件的情况，以及可能处理多个 persistenceIds(我们使用 ClusterSharding)。