从Saga调度命令时未调用事件处理程序

如何解决从Saga调度命令时未调用事件处理程序

我有一个usersHttpService: accept(code: string,user: UserInfo) { return this.http.post(`${this.env.MY URL}/mon/accepting/${code}/accept`,user); }，CommandHandler和EventSourcingHandler链，当直接分派命令EventHandler时，它们都按预期被调用。如果从Foo或公用EventHandler中分派同一命令，则不会导致调用同一Saga。

不幸的是，我无法透露实际的代码，但这是发生的情况的说明：

未调用的EventHandler仅将事件作为参数，没有带注释的EventHandler或类似参数。它负责写入读取模型，@MetaData需要继续执行该读取模型。此时，此过程失败，因为尚未更新读取模型，因为尚未调用@SagaEventHandler。

这里可能是什么问题？使用Axon 4.3.1。

配置：

EventHandler

简短的通话链：

通过HTTP请求接收到初始命令。

/*
 * Copyright (c) 2020 divine Cloud Services Ltd.
 * All rights reserved
 *
 * All information contained herein is and remains the property of
 * divine Cloud Services Ltd. The intellectual and technical concepts
 * contained herein are proprietary to divine Cloud Services Ltd. and
 * may be protected by trade secret or copyright law. Dissemination
 * of this information or reproduction of this material is strictly prohibited
 * unless prior written permission is obtained from divine Cloud Services Ltd.
 */

package engineering.divine.infrastructure.conf;

import engineering.divine.artefact.model.meta.ArtefactDefinition;
import engineering.divine.artefact.ports.ArtefactDefinitionRepository;
import engineering.divine.infrastructure.persistence.BinarySerialiser;
import engineering.divine.infrastructure.persistence.DefaultArtefactDefinitionRepository;
import engineering.divine.infrastructure.persistence.graph.GraphEventStorageEngine;
import engineering.divine.infrastructure.persistence.graph.database.GraphManager;
import engineering.divine.infrastructure.security.interceptors.AuditCommandDispatchInterceptor;
import engineering.divine.integration.github.GitHubCommandHandler;
import engineering.divine.integration.github.GitHubQueryHandler;
import org.axonframework.commandhandling.CommandBus;
import org.axonframework.commandhandling.CommandMessage;
import org.axonframework.commandhandling.SimpleCommandBus;
import org.axonframework.common.transaction.TransactionManager;
import org.axonframework.config.Configurer;
import org.axonframework.config.EventProcessingConfigurer;
import org.axonframework.eventhandling.PropagatingErrorHandler;
import org.axonframework.eventsourcing.EventCountSnapshotTriggerDefinition;
import org.axonframework.eventsourcing.EventSourcingRepository;
import org.axonframework.eventsourcing.SnapshotTriggerDefinition;
import org.axonframework.eventsourcing.Snapshotter;
import org.axonframework.eventsourcing.eventstore.EmbeddedEventStore;
import org.axonframework.eventsourcing.eventstore.EventStorageEngine;
import org.axonframework.eventsourcing.eventstore.EventStore;
import org.axonframework.messaging.interceptors.BeanValidationInterceptor;
import org.axonframework.modelling.command.Repository;
import org.axonframework.queryhandling.QueryBus;
import org.axonframework.queryhandling.QueryMessage;
import org.axonframework.queryhandling.SimpleQueryBus;
import org.axonframework.spring.eventsourcing.SpringAggregateSnapshotterFactoryBean;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;

import java.util.concurrent.Executors;

@SuppressWarnings("unchecked")
@Configuration
public class AxonConfig implements ApplicationContextAware {

    private ApplicationContext applicationContext;

    @Value("${engineering.divine.snapshot.threshold:30}")
    private int snapshotThreshold;

    @Bean
    public CommandBus commandBus(TransactionManager transactionManager,AuditCommandDispatchInterceptor auditCommandDispatchInterceptor,BeanValidationInterceptor<CommandMessage<?>> validationInterceptor) {

        SimpleCommandBus commandBus = SimpleCommandBus.builder()
                .transactionManager(transactionManager)
                .build();

        commandBus.registerDispatchInterceptor(auditCommandDispatchInterceptor);
        commandBus.registerDispatchInterceptor(validationInterceptor);

        return commandBus;
    }

    @Bean
    public QueryBus queryBus(TransactionManager transactionManager,BeanValidationInterceptor<QueryMessage<?,?>> validationInterceptor) {
        SimpleQueryBus queryBus = SimpleQueryBus.builder()
                .transactionManager(transactionManager)
                .build();

        queryBus.registerDispatchInterceptor(validationInterceptor);

        return queryBus;
    }

    @Bean
    @Primary
    EventStore eventStore(GraphManager graphManager,BinarySerialiser serialiser) {
        EventStorageEngine storageEngine = GraphEventStorageEngine
                .builder()
                .graphManager(graphManager)
                .serialiser(serialiser)
                .build();

        return EmbeddedEventStore.builder()
                .storageEngine(storageEngine)
                .build();
    }

    @Bean
    Snapshotter snapshotter(EventStore eventStore) {
        SpringAggregateSnapshotterFactoryBean factoryBean = new SpringAggregateSnapshotterFactoryBean();
        factoryBean.setApplicationContext(applicationContext);
        factoryBean.setEventStore(eventStore);
        factoryBean.setExecutor(Executors.newSingleThreadExecutor());
        return factoryBean.getObject();
    }

    @Bean
    SnapshotTriggerDefinition snapshotTriggerDefinition(Snapshotter snapshotter) {
        return new EventCountSnapshotTriggerDefinition(snapshotter,snapshotThreshold);
    }

    @Bean
    ArtefactDefinitionRepository artefactDefinitionRepository(EventStore eventStore) {
        Repository<ArtefactDefinition> delegate = EventSourcingRepository
                .builder(ArtefactDefinition.class)
                .eventStore(eventStore)
                .build();

        return new DefaultArtefactDefinitionRepository(delegate);
    }

    @Bean
    BeanValidationInterceptor<CommandMessage<?>> beanValidationCommandInterceptor() {
        return new BeanValidationInterceptor();
    }

    @Bean
    BeanValidationInterceptor<QueryMessage<?,?>> beanValidationQueryInterceptor() {
        return new BeanValidationInterceptor();
    }

    @Autowired
    public void configure(EventProcessingConfigurer configurer) {
        configurer.usingSubscribingEventProcessors();
        configurer.registerDefaultListenerInvocationErrorHandler(configuration -> PropagatingErrorHandler.instance());
    }

    @Autowired
    public void configure(Configurer configurer,GitHubCommandHandler commandHandler,GitHubQueryHandler queryHandler) {
        configurer.registerCommandHandler(c -> commandHandler);
        configurer.registerQueryHandler(c -> queryHandler);
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}

命令A产生一个新的@RestController public class Controller { ... @RequestMapping("/doA") public CompletableFuture<Void> doA(@RequestBody CommandA cmdA) { ... // Do something else before cmdA CommandX cmdX = new CommandX(); commandGateway.sendAndWait(cmdX); ... commandGateway.sendAndWait(cmdA); } }。

@Aggregate

@Aggregate public class MyFooAggregate { @CommandHandler public MyFooAggregate(CommandA cmd,MetaData metaData) { apply(new FooCreated(),metaData); } ... }接获@Saga事件并发出新命令。没有针对sagas的独特配置类。

FooCreated

不是集合的专用命令处理程序负责命令B。它使用接口@Saga public class MySaga { // (end will not be the same as start in the future) @StartSaga @EndSaga @SagaEventHandler(associationProperty = "id") public void handleFooCreated(FooCreated event,@MetaDataValue("valueX") String valueX,CommandGateway commandGateway) { CommandB cmdB = new CommandB(); ... Map<String,Object> meta = new HashMap<>(); meta.put("valueX",valueX); CommandMessage<CommandB> cmdMsg = GenericCommandMessage.asCommandMessage(cmdB); cmdMsg.withMetaData(meta); // because I don't need to wait for the result,I'll just "send" it commandGateway.send(cmdMsg); } }，实现为该接口调度另一个命令。该处理程序已在配置中注册（请参见MyProvider）。

AxonConfig

@Component public class CommandHandler { MyProvider provider; @CommandHandler public void doB(CommandB cmdB) { ... provider.provideMeWithSomething(); ... } }的实现最终调度了命令C。

MyProvider

命令C产生一个新的@Component public class MyDefaultProvider implements MyProvider { @Override public Object provideMeWithSomething() { CommandC cmdC = new CommandC(); commandGateway.sendAndWait(cmdC); ... } }。创建此聚合所调度的事件是@Aggregate中正在处理的事件，但找不到到达@EventSourcingHandler的事件。

@EventHandler

同一类@Aggregate public class MyBarAggregate { @CommandHandler public MyBarAggregate(CommandC cmdC) { apply(new BarCreated()); } ... }中的@EventSourcingHandler被调用，但不是负责编写读取模型的公用MyBarAggregate。

@EventHandler

但是，@Aggregate public class MyBarAggregate { ... @EventSourcingHandler public handleBarCreated(BarCreated event) { // DOES get called ... } } @Component public class EventHandler { @EventHandler public handleBarCreated(BarCreated event) { // DOES NOT get called ... } }被直接调度时（例如，从CommandC内部），两个事件处理程序都会被调用。其他@RestController确实会被调用，只有在这种特定情况下才有此特定实例。

解决方法

首先，我在考虑您使用的EventProcessor的类型，试图找出问题所在。如配置所示（感谢添加），您正在使用SubscribingEventProcessor。

从本质上讲，这意味着发布命令（在本例中为Saga）的同一线程（至少应该）调用了@EventHandler注释方法，因为这就是SubscribingEventProcessor的工作方式。尤其是如果您正在执行CommandGateway#sendAndWait(Object)，这意味着锁定Saga的线程还将锁定聚集以处理命令，并将负责向SubscribingEventProcessor的事件提供相同的事件。所说的事件处理程序（未被调用）。

因此，第二，您如何调度该命令？不能共享确切的代码很好，但是sample / pseudocode应该可行吧？第三，传奇如何配置？只是具有@Saga构造型的常规Spring自动配置，还是有独特的配置类？

说实话，我仍然很难理解为什么没有调用负责SubscribingEventProcessor注释功能的@EventHandler。您可以确认是否未调用属于此SubscribingEventProcessor的所有事件处理程序，或者仅仅是该事件处理类吗？

从Saga调度命令时未调用事件处理程序

如何解决从Saga调度命令时未调用事件处理程序

解决方法

相关推荐