masstransit deferred respond in sagas

问题:

I am investigating using sagas in mass transit to orchestrate activities across several services. The lifetime of the saga is short - less than 2 seconds if all goes well.

For my use case, i would like to use the request/respond approach, whereby the client requests a command, the saga handles that command, goes through some state changes as messages are received and eventually responds to the first command that initiated the saga, at which point the client receives the response and can display the result of the saga.

From what i can see, by this point, the context is no longer aware of the initial request. How can I reply to a message that was received in this way? Is there something i can persist to the saga data when handling the first event, and use that to reply later on?

回答1:

Currently, the saga state machine can only do immediate response like this:

// client
var response = await client.Request(requestMessage);

// saga
During(SomeState,
    When(RequestReceived)
        .Then(...)
        .Respond(c => MakeResponseMessage(c))
        .TransitionTo(Whatever)
)

So you can respond when handling a request.

If you want to respond to something you received before, you will have to craft the request/response conversation yourself. I mean that you will have to have decoupled response, so you need to send a message and have a full-blown consumer for the reply message. This will be completely asynchronous business.

回答2:

Thanks Alexey. I have realised that I can store the ResponseAddress and RequestId from the original message on the saga, and then construct a Send() later on.

Getting the response details from the original request

MassTransit.EntityFrameworkIntegration.Saga.EntityFramework

    SagaConsumeContext<TSagaData, TMessage> payload;
    if (ctx.TryGetPayload(out payload))
    {                
       ResponseAddress = payload.ResponseAddress;
       RequestId = payload.RequestId ;                          
    }

Sending the response

var responseEndpoint = await ctx.GetSendEndpoint(responseAddress);
await responseEndpoint.Send(message, c => c.RequestId = requestId);