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

Question

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

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

//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，使其具有一个适当的实例，为此我首先删除了IMediator，使用IPublishEndpoint将消息发布到Saga并配置了startup.cs，但它不像预期的那样工作：

//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.IPublishEndpoint”类型的服务，同时试图激活'GaaS.API.Controllers.ManageApiController‘。

我试图更新我的startup.cs

var repository = new InMemorySagaRepository<ApiSaga>();
            services.AddMassTransit(cfg =>
            {
                cfg.AddBus(provider =>
                {
                    return Bus.Factory.CreateMediator(x =>
                    {
                        x.Saga<ProductSaga>(repository);
                    });
                });
            });

我得到了：

无法隐式地将“MassTransit.Mediator.IMediator”类型转换为“MassTransit.IBusControl”。

如果你有什么建议的话，谢谢你和我分享和挑战

Answer 1

在您的项目中配置MassTransit中介程序的正确方法是通过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>());

Answer 2

我是从(优秀的) youtube视频- MassTransit从调解员开始获得的，在这个示例中有一行代码

AddMediator()

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

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

                config.UsingInMemory(ConfigureBus);
            });

然后，ConfigureBus是：

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

我很难在其他地方找到这个，所以我在这里发帖。