Axon 4：从不同线程应用事件时未触发EventSourcingHandler

Question

外部服务使用异步客户端（vertx tcp client+rxjava)，因此响应是在不同于创建聚合实例的线程中给出的。

我想在给定服务结果的情况下应用一个事件，但它不起作用，因为aggregateLifecycle.apply()调用在不同的线程上...

如何“转移”聚合的范围？

@Slf4j
@Aggregate
@NoArgsConstructor
public class MyAggregate {

    @AggregateIdentifier
    private String id;

    @CommandHandler
    public MyAggregate(CreationCommand creationCommand) {
        Single.just("some data")
                .observeOn(Schedulers.computation()) // <- comment this line and the test pass, uncomment and it fail because apply is on another thread ?
                .subscribe((s, throwable) -> apply(new AggregateCreatedEvent(creationCommand.getId())));
    }

    @EventSourcingHandler
    public void on(AggregateCreatedEvent event) {
        this.id = event.getId();
    }
}

@Value class CreationCommand { String id; }
@Value class AggregateCreatedEvent { String id;}

public class MyAggregateTest {

    AggregateTestFixture<MyAggregate> testFixture = new AggregateTestFixture<>(MyAggregate.class);

    @Test
    public void test() {
        testFixture.givenNoPriorActivity()
                .when(new CreationCommand("123"))
                .expectEvents(new AggregateCreatedEvent("123"));
    }
}

以下是我得到的错误：

java.lang.IllegalStateException: Cannot request current Scope if none is active

Answer 1

事件必须在管理该工作单元的线程中应用，在本例中是CommandHandler。Axon为异步操作提供了自己的机制。commandBus异步接受命令，事件处理器异步处理事件。异步实现CommandHandler也不会有什么好处，而且在任何情况下，目前都不支持它。

应用事件所需的所有数据通常应在命令或聚合状态中可用，而不是来自其他外部源。

这可能是您希望命令处理程序的样子：

@CommandHandler
public MyAggregate(CreationCommand creationCommand) {
    apply(new AggregateCreatedEvent(creationCommand.getId());
}