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

Question

我是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，使其具有一个合适的实例，为此，我首先删除了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:

在尝试激活"GaaS.Eneneneba API.Controllers.ManageApiController".时，无法解析类型为"MassTransit.IPublishEndpoint"的服务

我尝试更新我的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 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>());

Answer 2

我从(出色的)youtube视频-MassTransit中了解到这一点，从Mediator开始，在该示例中有一行代码

AddMediator()

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

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

则CCD_ 14为

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

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