xUnit测试下MassTransit消费者触发异常及Autofac集成问题咨询

问题1：仅使用Autofac配置时，状态机/消费者消费断言失败 #

原因 #

MassTransit测试套件的Harness体系原生基于Microsoft.Extensions.DependencyInjection设计，单独在Autofac中注册的消费者、状态机没有和测试总线的配置链路打通，导致组件没有挂载到测试用的InMemory总线上，自然不会消费消息。

解决方案 #

无需维护两套独立的容器注册逻辑，用Autofac官方的Populate扩展合并DI注册即可消除冗余：

// 1. 先通过ServiceCollection注册MassTransit测试相关组件
var services = new ServiceCollection();
services.AddMassTransitTestHarness(cfg =>
{
    // 注册状态机、消费者
    cfg.AddSagaStateMachine<ExampleStateMachine, ExampleState>().InMemoryRepository();
    cfg.AddConsumer<ExampleConsumer1>();
    cfg.AddConsumer<ExampleConsumer2>();
    cfg.UsingInMemory((ctx, busCfg) => busCfg.ConfigureEndpoints(ctx));
});

// 2. 将ServiceCollection的注册合并到Autofac容器
var builder = new ContainerBuilder();
builder.Populate(services);
var container = builder.Build();

// 3. 从Autofac容器解析出ServiceProvider，统一获取所有Harness
var serviceProvider = container.Resolve<IServiceProvider>();
var testHarness = serviceProvider.GetTestHarness();
var sagaHarness = serviceProvider.GetSagaStateMachineHarness<ExampleStateMachine, ExampleState>();
var consumer1Harness = serviceProvider.GetConsumerHarness<ExampleConsumer1>();
var consumer2Harness = serviceProvider.GetConsumerHarness<ExampleConsumer2>();

问题2：查询Saga实例后，后续发布的消息无法被消费 #

原因 #

sagaHarness.Sagas是带跟踪逻辑的可等待集合，直接同步遍历/查询该集合时，会隐式锁定Harness的消息跟踪上下文，或是未等待之前的消息全部处理完成就执行后续操作，导致新发布的消息没有被正确路由、跟踪。

解决方案 #

两种方案二选一即可：

1. 使用Harness提供的异步查询API，不要直接同步遍历Sagas集合：

// 等待Saga创建完成
Assert.True(await sagaHarness.Created.Any(x => x.CorrelationId == correlationId));
// 异步查询Saga实例
var saga = (await sagaHarness.Sagas.FirstOrDefault(x => x.CorrelationId == correlationId))?.Saga;
Assert.Equal("Progress state machine", saga.Message);

// 后续发布消息、断言逻辑正常执行
await testHarness.Bus.Publish(new ExampleMessage2
{
    CorrelationId = correlationId,
    Message = "Trigger consumer"
});
// 发布后等待总线空闲再断言
await testHarness.InactivityTask;
Assert.True(await consumer2Harness.Consumed.Any<ExampleMessage2>(m => m.Context.CorrelationId == correlationId));

2. 查询前显式等待总线空闲，确保所有之前的消息处理完毕：

// 等待所有已发布消息处理完成，总线进入空闲状态
await testHarness.InactivityTask;

// 再执行Saga查询逻辑
var state = sagaHarness.Sagas.Select(s => s.CorrelationId == correlationId).Single();
Assert.Equal("Progress state machine", state.Saga.Message);

// 后续发布、断言逻辑正常执行
await testHarness.Bus.Publish(new ExampleMessage2 { CorrelationId = correlationId, Message = "Trigger consumer" });
await testHarness.InactivityTask;
Assert.True(await consumer2Harness.Consumed.Any<ExampleMessage2>(m => m.Context.CorrelationId == correlationId));

内容的提问来源于stack exchange，提问作者Jez