我如何配置作用域筛选器以使其对使用者和活动均有效

问题描述

我已经设法为使用者配置了范围服务和范围过滤器，这意味着我可以在实现IFilter<ConsumeContext<T>>的过滤器中为范围服务设置值，并向UseConsumeFilter注册过滤器。筛选器会在我的范围内服务中设置一个值，然后范围内服务可以注入到我的使用者中，并且仍然具有该值。

我尝试使用IFilter<ExecuteContext<TArguments>>进行活动，并向UseExecuteActivityFilter注册过滤器。

在ActivActivity中无法获得ExecuteActivityContext中设置的值。我认为它们成为两个不同的DI范围。我将分享我的活动和使用者实现中的代码，也许活动一中缺少一些内容。我只是尝试保留重要的部分，因此，如果某处存在非法语法，则是我试图为SO清除代码。

这是我以错误的方式使用DI还是与DI进行活动有关的东西？我尝试遵循masstransits网站上的“范围过滤器”文档。我在.net core 3.1和masstransit 7.0.4上。

用于测试的范围服务

//Interface
public interface IContextService
{
    string TenantId { get; set; }       
}
//DI registration
services.AddScoped<IContextService,ContextService>();

活动配置，此操作不起作用

//Filter
public class RetreiveContextExecuteFilter<TArguments> : IFilter<ExecuteContext<TArguments>>
       where TArguments : class
    {
        public IContextService _contextService { get; }

        public RetreiveContextExecuteFilter(IContextService contextService)
        {
            _contextService = contextService;
        }

        public async Task Send(ExecuteContext<TArguments> context,IPipe<ExecuteContext<TArguments>> next)
        {
            _contextService.tenantId = "test-tenant";

            await next.Send(context);
        }

        public void Probe(ProbeContext context)
        {
            var scope = context.CreateFilterScope("testcontextinformation");
        }
    }

//Activity
public class ExampleActivity
    : IExecuteActivity<ExampleActivityArguments>
    {
        private readonly IContextService _contextService;

        public ExampleActivity(IContextService contextService)
        {
            _contextService = contextService;
        }

        public async Task<ExecutionResult> Execute(ExecuteContext<ExampleActivityArguments> context)
        {
            var tenant = _contextService.tenantId; //Empty
        }
    }

//DI
services.AddMasstransit(cfg =>
    {
    cfg.AddActivitiesFromNamespaceContaining<ExampleActivity>();
    services.TryAddSingleton(KebabCaseEndpointNameFormatter.Instance);                       
    cfg.UsingRabbitMq(ConfigureBus);
});

private static void ConfigureBus(IBusRegistrationContext context,IRabbitMqBusFactoryConfigurator configurator)
{
    configurator.ConfigureEndpoints(context);
    configurator.UseExecuteActivityFilter(typeof(RetreiveContextExecuteFilter<>),context);
}

消费者配置，可以使用

//Filter deFinition
public class RetreiveContextConsumeFilter<T> : IFilter<ConsumeContext<T>>
    where T : class
{
    public IContextService _contextService { get; }

    public RetreiveContextConsumeFilter(IContextService contextService)
    {
        _contextService = contextService;
    }

    public Task Send(ConsumeContext<T> context,IPipe<ConsumeContext<T>> next)
    {
        _contextService.TenantId = "test tenant";
        return next.Send(context);
    }

    public void Probe(ProbeContext context)
    {
        context.CreateFilterScope("contextinformation");
    }
}

//Consumer
public class ExampleConsumer
: IConsumer<ExampleEvent>
{
    private readonly IContextService _contextService;

    public ExampleConsumer(IContextService contextService)
    {
        _contextService = contextService;
    }

    public async Task Consume(ConsumeContext<ExampleEvent> context)
    {
         var id = _contextService.TenantId(); //Correct value
    }
}
//DI
services.AddMasstransit(cfg =>
{              
    cfg.AddConsumersFromNamespaceContaining<ExampleConsumer>();                 
    services.TryAddSingleton(KebabCaseEndpointNameFormatter.Instance);                                    
    cfg.UsingRabbitMq(ConfigureBus);
});

private static void ConfigureBus(IBusRegistrationContext context,IRabbitMqBusFactoryConfigurator configurator)
{
    configurator.ConfigureEndpoints(context);
    configurator.UseConsumeFilter(typeof(RetreiveContextConsumeFilter<>),context);
}

解决方法

首先猜测是您的配置顺序不正确。 MassTransit构建管道，并且您要在过滤器之前配置端点，这将使过滤器在端点之后 运行。那是我的猜测。

将使用者更改为：

configurator.UseConsumeFilter(typeof(RetreiveContextConsumeFilter<>),context);
configurator.ConfigureEndpoints(context);

将活动更改为：

configurator.UseExecuteActivityFilter(typeof(RetreiveContextExecuteFilter<>),context);
configurator.ConfigureEndpoints(context);