masstransit - 让 Masstransit 使用范围生命周期

Question

我在消费者观察者中获取 IIdentityService（我自己的类）或 IServiceProvider 的相同实例时遇到问题。我有一个身份服务，我在其中为用户设置凭据，稍后在消费者管道中使用。

我已经尝试了下面的代码以及其他代码和配置更改。

_serviceProvider.GetRequiredService<IIdentityService>()
_serviceProvider.CreateScope()
 var consumerScopeProvider = _serviceProvider.GetRequiredService<IConsumerScopeProvider>();
            using (var scope = consumerScopeProvider.GetScope(context))
            {
                // this next line of code is where we must access the payload
                // using a container specific interface to get access to the
                // scoped IServiceProvider
                var serviceScope = scope.Context.GetPayload<IServiceScope>();
                var serviceProviderScoped = serviceScope.ServiceProvider;

                IIdentityService identityService = _serviceProvider.GetRequiredService<IIdentityService>();
            }

// Also tried this as Scoped
services.AddTransient<CustomConsumer>();


var busControl = Bus.Factory.CreateUsingRabbitMq(cfg =>
{
    var host = cfg.Host(new Uri(busConfiguration.Address), h =>
    {
        h.Username(busConfiguration.Username);
        h.Password(busConfiguration.Password);
    });

    cfg.ReceiveEndpoint(host, "queue_1", endpointCfg => ConfigureConsumers(endpointCfg, provider));
    cfg.UseServiceScope(provider);
});


busControl.ConnectConsumeObserver(new ConsumeObserver(provider));




public class ConsumeObserver : IConsumeObserver
{
    private readonly IServiceProvider _serviceProvider;

    public ConsumeObserver(IServiceProvider serviceProvider)
    {
        _serviceProvider = serviceProvider;
    }

    Task IConsumeObserver.PreConsume<T>(ConsumeContext<T> context)
    {
        var identityMessage = (IIdentityMessage)context.Message;
        if (identityMessage == null)
        {
            return Task.CompletedTask;
        }

        // Here I get a "Cannot resolve scoped service '' from root provider." error
        IIdentityService identityService = _serviceProvider.GetRequiredService<IIdentityService>();
        var task = identityService.SetIdentityAsync(identityMessage.Identity.TenantIdentifier, identityMessage.Identity.UserIdentifier);

        // This gets a different instance of the IIdentityService.cs service.
        //IIdentityService identityService = _serviceProvider.CreateScope().ServiceProvider.GetRequiredService<IIdentityService>();


        // called before the consumer's Consume method is called
        return task;
    }

    Task IConsumeObserver.PostConsume<T>(ConsumeContext<T> context)
    {
        // called after the consumer's Consume method is called
        // if an exception was thrown, the ConsumeFault method is called instead
        return TaskUtil.Completed;
    }

    Task IConsumeObserver.ConsumeFault<T>(ConsumeContext<T> context, Exception exception)
    {
        // called if the consumer's Consume method throws an exception
        return TaskUtil.Completed;
    }
}


// This is where I need to identity credentials
services.AddScoped<UserContext>((provider =>
{
    // This gets a different instance of IIdentityService
    var identityService = provider.GetRequiredService<IIdentityService>();
    var contextProvider = provider.GetRequiredService<IContextProvider>();

    var identity = identityService.GetIdentity();

    return contextProvider.GetContext(identity.UserId);
}));

var task = identityService.SetIdentityAsync(identityMessage.Identity.TenantIdentifier, identityMessage.Identity.UserIdentifier);

上面的身份设置应该可以在下面检索。

var identity = identityService.GetIdentity();

我得到的是空引用，因为服务提供者是不同的实例。

谁能告诉我如何通过消费者管道获取服务提供者的相同实例？

标签： masstransit