如何使用Mediator配置MassTransit以发布消息？

编程问答 2022-06-03

问题描述

我是 Masstransit 和 Mediator 的新手，我有一系列要连续执行的事件，我正在使用 Masstransit 在进程内和内存中，对于我的用例，不需要传输。

我想通过Mediator向消费者，sagas和活动发送和发布消息，我有下面的代码，但是我想通过在startup.cs中注册Masstransit来改进它：

//asp net core 3.1 Controller

 [ApiController]
 public class MyController : ControllerBase
 {    
    private readonly IProductService _productService ;
    private readonly IMediator mediator;
    public MyController(IProductService productService)
    {
       _productService = productService;
      var repository = new InMemorySagaRepository<ApiSaga>();
      mediator = Bus.Factory.CreateMediator(cfg =>
      {
          cfg.Saga<ProductSaga>(repository);
      });
    }
   
     [HttpPost]
     public async Task<IActionResult> Post([FromBody] ProductContract productContract)
     {            
         try
         {
             var result = await _productService.DoSomeThingAsync(productContract);
             await mediator.Publish<ProductSubmittedEvent>(new { CorrelationId = Guid.NewGuid(),result.Label });
         return Ok();
         }
         catch (Exception ex)
         {
             return BadRequest(ex.Message);
         }
      }
 }

//My saga
public class ProductSaga :
        ISaga,InitiatedBy<ProductSubmittedEvent>
    {
        public Guid CorrelationId { get; set; }
        public string State { get; private set; } = "Not Started";

        public Task Consume(ConsumeContext<ProductSubmittedEvent> context)
        {
            var label= context.Message.Label;
            State = "AwaitingForNextStep";
            //...
           //send next command
        }
    }

像这样可以正常工作，但这是不合适的，我想在我的startup.cs中使用Mediator配置masstransit以拥有一个适当的实例，为此，我首先使用{{ 1}}，将消息发布到IMediator并配置我的IPublishEndpoint，但无法正常工作：

Saga

我有一个startup.cs：

尝试激活“ GaaS.API.Controllers.ManageApiController”时，无法解析类型为“ Masstransit.IPublishEndpoint”的服务。

我尝试更新//startup.cs public void ConfigureServices(IServiceCollection services) { services.AddMediator(cfg => { cfg.AddSaga<ProductSaga>().InMemoryRepository(); }); } //in controller using: private readonly IPublishEndpoint _publishEndpoint; //then await _publishEndpoint.Publish<ProductSubmittedEvent>( new { CorrelationId = Guid.NewGuid(),result.Label }); ：

system.invalidOperationException

我知道了

无法将类型'Masstransit.Mediator.IMediator'隐式转换为'Masstransit.IBusControl'。

如果您有任何建议，感谢您与我分享和挑战我challenging

解决方法

在项目中配置MassTransit Mediator的正确方法是通过Startup.cs文件，您似乎已经尝试过。

public void ConfigureServices(IServiceCollection services)
{
    services.AddMediator(cfg =>
    {
        cfg.AddSaga<ProductSaga>().InMemoryRepository();
    });
}

使用中介程序，您需要依赖IMediator界面。您不能使用IPublishEndpoint或ISendEndpointProvider，因为它们是总线接口。由于您可以同时在容器中同时包含调解器和总线实例，因此在解析容器中的服务时会引起混乱。

[ApiController]
public class MyController : ControllerBase
{    
    private readonly IProductService _productService ;
    private readonly IMediator _mediator;

    public MyController(IProductService productService,IMediator mediator)
    {
        _productService = productService;
        _mediator = mediator;
    }

    [HttpPost]
    public async Task<IActionResult> Post([FromBody] ProductContract productContract)
    {            
        try
        {
            var result = await _productService.DoSomeThingAsync(productContract);
            
            await _mediator.Publish<ProductSubmittedEvent>(new { CorrelationId = NewId.NextGuid(),result.Label });

            return Ok();
        }
        catch (Exception ex)
        {
            return BadRequest(ex.Message);
        }
    }
}

如果您仅使用中介程序，并且想使用IPublishEndpoint，则可以自己将其添加到容器中并委派它。

services.AddSingleton<IPublishEndpoint>(provider => provider.GetService<IMediator>());
,

我从（优秀的）youtube video - MassTransit starting with Mediator 得到了这个，在那个示例中有一行代码

AddMediator()

我找不到。我相信以下设置提供了使代码基于该视频工作所需的一切...

            services.AddMassTransit(config =>
            {
                config.AddRequestClient<ISubmitOrder>();
                config.AddConsumersFromNamespaceContaining<SubmitOrderConsumer>();

                config.UsingInMemory(ConfigureBus);
            });

然后 ConfigureBus 是：

        private void ConfigureBus(IBusRegistrationContext context,IInMemoryBusFactoryConfigurator configurator)
        {
            configurator.ConfigureEndpoints(context);
        }

我在别处找不到这个，所以发在这里。

c#c#dependency-injectionmasstransit