Automatonymous

Question

我一直在努力让 MassTransitStateMachine 正常工作，但我似乎不明白它应该如何工作。

当我尝试调用 stateMachine.RaiseEvent(instance, stateMachine.DeployApplicationRequest, request) 并结合状态机中的 .Send() Activity 时，我收到的错误（下面的完整代码）是 PayloadNotFoundException。我一直在努力追查这个问题，但一无所获。

相关代码及意图说明如下：

我收到通过网络部署应用程序的请求API post，并将请求转换为 DeployApplicationRequest。

public record DeployApplicationRequest
    {
        // State machine properties
        public Guid CorrelationId { get; init; }
        public ChatApplication ChatApplication { get; init; }
        public string ChannelId { get; init; }
        public string UserId { get; init; }


        // Command
        public DeployApplication DeployApplication { get; init; }
    }

    public enum ChatApplication
    {
        Slack,
        Teams
    }

这个请求封装了两件事：API 需要维护的状态，以便它知道如何将结果路由回请求者（状态机属性）和要发送的 Command到服务总线。

命令 DeployApplication 如下所示：

public record DeployApplication
    {
        public Guid CorrelationId { get; init;}
        public string InitiatedBy { get; init; }
        public DateTime CreatedDate { get; init; }
        public string Environment { get; init; }
        public string PullRequestId { get; init; }
    }

我创建了一个状态实例来保存请求的详细信息（例如它是通过 Slack 还是 Teams 传入的）。我不想将此信息发布到总线上：

public class DeployApplicationState : SagaStateMachineInstance
    {
        public Guid CorrelationId { get; set; }
        public string CurrentState { get; set; }

        public ChatApplication ChatApplication { get; set; }
        public string ChannelId { get; set; }
        public string UserId { get; set; }
    }

我创建了一个 StateMachine 来处理这个问题：

public class DeployApplicationStateMachine : MassTransitStateMachine<DeployApplicationState>
    {
        private readonly ILogger<DeployApplicationStateMachine> logger;
        public DeployApplicationStateMachine(ILogger<DeployApplicationStateMachine> logger)
        {
            this.logger = logger;

            InstanceState(x => x.CurrentState);

            Event(() => DeployApplicationRequest, x => x.CorrelateById(context => context.Message.CorrelationId));

            Initially(
                When(DeployApplicationRequest)
                    .Then(x => {
                        x.Instance.CorrelationId = x.Data.CorrelationId;
                        x.Instance.ChannelId = x.Data.ChannelId;
                        x.Instance.ChatApplication = x.Data.ChatApplication;
                        x.Instance.UserId = x.Data.UserId;
                    })
                    .Send(context => context.Init<DeployApplication>(context.Data.DeployApplication))
                    .TransitionTo(Submitted));
        }

        public Event<DeployApplicationRequest> DeployApplicationRequest { get; private set; }

        public State Submitted { get; private set; }
    }

为了触发初始事件（因为请求不是通过消费者而是通过控制器传入），我已将状态机注入 MassTransit 客户端，并且正在调用 RaiseEvent方法：

public class MassTransitDeployClient : IDeployClient
    {
        private readonly DeployApplicationStateMachine stateMachine;
        public MassTransitDeployClient(DeployApplicationStateMachine stateMachine)
        {
            this.stateMachine = stateMachine;
        }

        public async Task Send(DeployApplicationRequest request)
        {
            var instance = new DeployApplicationState
            {
                CorrelationId = request.CorrelationId,
                ChannelId = request.ChannelId,
                ChatApplication = request.ChatApplication,
                UserId = request.UserId
            };

            await stateMachine.RaiseEvent(instance, stateMachine.DeployApplicationRequest, request);
            // This works for sending to the bus, but I lose the state information
            //await sendEndpointProvider.Send(request.DeployApplication);
        }
    }

并且容器配置如下：

services.AddMassTransit(x =>
            {
                x.AddSagaStateMachine<DeployApplicationStateMachine, DeployApplicationState>()
                    .InMemoryRepository();

                x.UsingInMemory((context, cfg) =>
                {
                    cfg.ConfigureEndpoints(context);
                });
                
                
            });

            EndpointConvention.Map<DeployApplication>(new Uri($"queue:{typeof(DeployApplication).FullName}"));

            services.AddMassTransitHostedService();

引发事件工作得很好，如果我不在状态机中包含 .Send(...)，状态机会成功转移到Submitted。第二个我介绍 Activity 我得到 PayloadNotFoundException。我在周末尝试了大约 15 种不同的方法，但都没有成功，我希望有人可以帮助我看到我的方法的错误。

感谢阅读！顺便说一句，很棒的图书馆。我将寻找方法为今后的发展做出贡献，这是我遇到的最有用的库之一（Chris，您的 Youtube 视频非常棒）。

Answer 1

Saga 状态机不是意味着直接从控制器调用。您应该发送（或发布）一条消息，然后通过消息代理发送到 saga。

如果你想这样做，你可以改用 MassTransit Mediator，但你仍然会通过 Mediator 向 saga 发送消息，然后由 MassTransit 处理。

TL;DR - you don't use RaiseEvent with saga state machines.

Automatonymous - 使用 Send Activity 调用 RaiseEvent 时未找到负载

Automatonymous - Payload not found when calling RaiseEvent with Send Activity

masstransit