Masstransit saga обработка ошибок - PullRequest
Новая
       38

Masstransit saga обработка ошибок

0 голосов
/ 18 января 2019

Я пытаюсь создать простой сценарий обработки ошибок саги, используя masstransit и azure sb, и получаю следующее, работающее для получателя.

В данный момент сага выдает ошибку на одном из шагов и после нескольких попыток отправляется в очередь _error, как и ожидалось.

В зависимости от шага, который вызывает ошибку в саге, сообщение будет обработано Step1FaultConsumer или Step2FaultConsumer . Я хотел узнать, есть ли лучший способ обработки ошибок из всей саги из одного места, независимо от шага, который вызывает ошибку.

Я пытаюсь сделать это без использования конечного автомата, если это возможно.

Program.cs 

 var bus = Bus.Factory.CreateUsingAzureServiceBus(cfg =>
        {
            var host = cfg.Host(new Uri("HostAddress"),
                h =>
                {
                    h.TransportType = TransportType.AmqpWebSockets;
                    h.OperationTimeout = TimeSpan.FromSeconds(5);
                    h.RetryLimit = 5;
                    h.TokenProvider = TokenProvider.CreateSharedAccessSignatureTokenProvider("RootManageSharedAccessKey", "SharedAccessKey");
                });

            cfg.RequiresSession = true;
            cfg.UseRetry(x => x.Exponential(5, TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(10), TimeSpan.FromSeconds(2)));
            cfg.ReceiveEndpoint(host, "QueueName", e =>
            {
                e.RequiresSession = true;
                e.Saga<SagaConsumer>(new MessageSessionSagaRepository<SagaConsumer>());
                e.Consumer<Step1FaultConsumer>();
                e.Consumer<Step2FaultConsumer>();
            });
        });

saga.cs 

public class SagaConsumer:
    ISaga,
    InitiatedBy<ISagaStep1>,
    Orchestrates<ISagaStep2>
{
    public Guid CorrelationId { get; set; }

    public async Task Consume(ConsumeContext<ISagaStep1> context)
    {
        if(context.Message.Input1 == "000")
             throw new Exception("Saga Step 1 Exception");
        await context.Send<ISagaStep2>(context.DestinationAddress, new { Input2 = $"2-{context.Message.Input1}" });
    }

    public async Task Consume(ConsumeContext<ISagaStep2> context)
    {
        throw new Exception("Saga Step 2 Exception");
    }
}

public class Step1FaultConsumer:
    IConsumer<Fault<ISagaStep1>>
{
    public async Task Consume(ConsumeContext<Fault<ISagaStep1>> context)
    {
        // Handle Error
    }
}

public class Step2FaultConsumer:
    IConsumer<Fault<ISagaStep2>>
{
    public async Task Consume(ConsumeContext<Fault<ISagaStep2>> context)
    {
        // Handle Error
    }
}

1 Ответ

0 голосов
/ 18 января 2019

Вы можете обновить своего потребителя так, чтобы он обрабатывал оба типа ошибок через два интерфейса IConsumer<T> и два метода Consume.

Я бы также предложил использовать отдельную конечную точку очереди / получения для потребителя ошибок.), чтобы избежать засорения пропускной способности вашей конечной точки саги.

...