dependency-injection - 使用依赖注入(inject)设置 saga 时出现公共(public)交通错误

Question

我有一个 saga，我正在尝试使用需要 retrofit 客户端的 masstransit 运行，我想将客户端依赖项注入(inject)到 saga 中。

我正在使用 MassTransit 5.2.3 和 MassTransit.Extensions.DependencyInjection 5.2.3 进行如下设置:

...
var serviceCollection = new ServiceCollection();
serviceCollection.AddRefitClient<IClient>(...).ConfigureHttpClient(...);
serviceCollection.AddMassTransit(c =>
{
    c.AddSaga<MySaga>();
});    
serviceCollection.AddScoped<MySaga>();
serviceCollection.AddSingleton<ISagaRepository<MySaga>, MessageSessionSagaRepository<MySaga>>(x => new MessageSessionSagaRepository<MySaga>());

var serviceProvider = serviceCollection.BuildServiceProvider();
var bus = Bus.Factory.CreateUsingAzureServiceBus(cfg =>
{
    var host = ...;
    ...
    cfg.ReceiveEndpoint(host, "MyQueue", e =>
    {   
        ... 
        e.Saga<MySaga>(serviceProvider);
    });
});

bus.Start();
...

传奇的代码是:

class MySaga :
    ISaga,
    InitiatedBy<IStep1>,
    Orchestrates<IStep2>
{
    private readonly IClient _client;

    public Guid CorrelationId { get; set; }

    public MySaga(IClient client)
    {
        _client = client;
    }

    public async Task Consume(ConsumeContext<IStep1> context) {...}
    public async Task Consume(ConsumeContext<IStep2> context) {...}
}

这会导致“无法为 MyNamespace.MySaga 创建传奇连接器”异常，内部异常为“ConfigurationException:传奇 MyNamespace.MySaga 必须具有默认构造函数和可写 CorrelationId 属性或具有单个 Guid 参数的构造函数分配 CorrelationId"

Answer 1

MassTransit 对 sagas 的关注点已转移到状态机 sagas，因此总体上 Automatonymous sagas 是首选。对于“经典”sagas，错误消息中会给出要求。您可以使用 Injection_Specs.cs 中的代码

var refitClient = ...;

var serviceCollection = new ServiceCollection();
serviceCollection.AddMassTransit(c =>
{
    c.AddSaga<MySaga>();
});    
serviceCollection.AddScoped<MySaga>();
serviceCollection.AddSingleton<ISagaRepository<MySaga>, MessageSessionSagaRepository<MySaga>>(x => new MessageSessionSagaRepository<MySaga>());

var serviceProvider = serviceCollection.BuildServiceProvider();
var bus = Bus.Factory.CreateUsingAzureServiceBus(cfg =>
{
    var host = ...;
    ...
    cfg.ReceiveEndpoint(host, "MyQueue", e =>
    {   
        ... 
        e.Saga<MySaga>(serviceProvider, 
            x => x.UseExecute(ctx => ctx.Saga.Client = refitClient));
    });
});

bus.Start();

当然，您需要在 saga 类中拥有 Client 公共(public)属性，并且您不会使用构造函数注入(inject)。您还需要错误消息告诉您的那两个构造函数:

class MySaga :
    ISaga,
    InitiatedBy<IStep1>,
    Orchestrates<IStep2>
{
    public IClient Client { get; set; }

    public Guid CorrelationId { get; set; }

    public MySaga()
    {
    }

    public MySaga(Guid correlationId)
    {
        CorrelationId = correlationId;
    }

    public async Task Consume(ConsumeContext<IStep1> context) {...}
    public async Task Consume(ConsumeContext<IStep2> context) {...}
}

设计考虑

我必须补充的一件事是，MassTransit 中的 Saga 接近流程管理器模式。这意味着 sagas 除了编排逻辑之外不应该有任何逻辑，并且除了处理消息之外不应该做任何其他事情。所以，我不建议在 sagas 中使用 REST API 客户端之类的东西。如果 saga 需要获取一些数据来决定消息路由，它应该使用消息来获取该数据。