等待Mass Transit传奇完成

Question

我正在尝试创建一个向调用者返回一些结果的传奇，就像请求/响应模式一样。如果我调用Send方法，我可以启动传奇，但不能通过提交请求来启动传奇。

因此，saga逻辑运行正常，但它不会向客户端返回任何内容。

或者提交请求由其消费者处理并向客户端返回响应，但从不启动传奇。

更新：masstransit deferred respond in sagas的答案似乎不适用于我的问题有两个原因：

1）我无法通过调用Request方法启动saga;

2）如果我调用Send方法发送请求，并且稍后发送响应，则调用者线程不会等待响应返回，然后继续下一行代码;

[更新结束]

请找到完整的代码here。以下是更相关的片段：

这是传奇类：

public class MySaga : MassTransitStateMachine<MySagaState>
    {
        public static Uri address = new Uri($"loopback://localhost/req_resp_saga");

        public Event<IStartSagaCommand> StartSaga { get; private set; }

        public Request<MySagaState, MyRequest, MyResponse> SomeRequest { get; private set; }

        public MySaga()
        {
            InstanceState(s => s.CurrentState);

            Event(() => StartSaga,
                cc =>
                    cc.CorrelateBy(state => state.Data, context => context.Message.Data)
                        .SelectId(context => Guid.NewGuid()));

            Request(() => SomeRequest, x => x.NullableCorrelationId, cfg =>
            {
                cfg.ServiceAddress = address;
                cfg.SchedulingServiceAddress = address;
                cfg.Timeout = TimeSpan.FromSeconds(30);
            });

            Initially(
                When(StartSaga)
                    .Then(context =>
                    {
                        context.Instance.Data = context.Data.Data;
                    })
                    .ThenAsync(
                        context => Console.Out.WriteLineAsync($"Saga started: " +
                                                              $" {context.Data.Data} received"))
                    .Request(SomeRequest, context => new MyRequest() { CorrelationId = context.Instance.CorrelationId, RequestMessage = "Please do this" })
                    .TransitionTo(SomeRequest.Pending)
                    .ThenAsync(context => Console.Out.WriteLineAsync($"Transition completed: " +
                                                                     $" {(context.Instance.CurrentState == SomeRequest.Pending ? "pending" : "done")} received"))
                    //.Then(context =>
                    //{
                    //    var endpoint = context.GetSendEndpoint(address).GetAwaiter().GetResult();
                    //    endpoint.Send(new MyResponse() { CorrelationId = context.Instance.CorrelationId, ResponseMessage = "Your wish is my command" });
                    //})
            );

            During(SomeRequest.Pending,
                When(SomeRequest.Completed)
                    .ThenAsync(
                        context => Console.Out.WriteLineAsync($"Saga ended: " +
                                                              $" {context.Data.ResponseMessage} received"))
                    .Finalize()
            );
        }
    }

这开始了传奇，但不等待它完成并回应：

            var address = new Uri($"loopback://localhost/req_resp_saga");
            var endPoint = bus.GetSendEndpoint(address)
                .Result;

            endPoint.Send<IStartSagaCommand>(new { Data = "Hello World!!" });

这等待回应，但根本不涉及传奇：

var address = new Uri($"loopback://localhost/req_resp_saga");
            var requestClient = new MessageRequestClient<MyRequest, MyResponse>(bus, address, TimeSpan.FromSeconds(30));
            var response = requestClient.Request(new MyRequest() { CorrelationId = Guid.NewGuid(), RequestMessage = "Please do this" })
                .GetAwaiter()
                .GetResult();

我怎样才能让调用者启动传奇并等待它完成并用它的响应做一些事情？

Answer 1

我自己找到了解决方案。

问题是我对如何通过请求调用触发传奇感到困惑。我以为我必须申报一个

Request<in TInstance, TRequest, TResponse>

（Automatonimous）

那不适合我。

我曾经开始传奇的事件有它自己的界面

Event<IStartSaga>

这与我在调用Request方法时使用的不一样

var requestClient = new MessageRequestClient<MyRequest, MyResponse>(bus, address, TimeSpan.FromSeconds(30));
            var response = requestClient.Request(new MyRequest() { CorrelationId = Guid.NewGuid(), RequestMessage = "Please do this" })
                .GetAwaiter()
                .GetResult();

所以解决方法是将事件的声明更改为

Event<MyRequest>

现在每当我用MyResquest消息调用Request时，saga就会启动。呼叫者等待传奇的响应。

我做了一些其他更改来清理代码并将其推送到github。