使用v8测试消费者TestServer

Question

通过使用所有DI服务创建整个应用程序，我尝试用MassTransit v8来测试TestServer使用者。我已经创建了TestServer实例，并替换了一些依赖项，但我最关心的是向使用者添加MasstransitTestHarness。在测试中，消息被正确地发送和使用，但是我不明白为什么我的使用者不能使用消息。当我定义了所有DI服务时，TestStartup继承了我的应用程序启动。

下面是我的代码示例：

TestServer程序

public class TestProgram
{
    private const string API_URL = "http://localhost:5010";

    public TestServer Server { get; set; }

    public void SetUpTestServer()
    {
        var host = CreateWebHostBuilder();

        Server = new TestServer(host);
    }

    private IWebHostBuilder CreateWebHostBuilder()
    {
        return new WebHostBuilder()
            .ConfigureTestServices(serviceCollection =>
            {
                serviceCollection.AddMassTransitTestHarness(cfg =>
                {
                    cfg.AddConsumer<MyConsumer>();

                    cfg.UsingInMemory((provider, config) =>
                    {
                        config.ReceiveEndpoint("MyMessageQueue",
                            e =>
                            {
                                e.Batch<IMyMessage>(b =>
                                {
                                    b.Consumer<MyConsumer, IMyMessage>(provider);
                                });
                            });
                    });
                });
            })
            .UseStartup<TestStartup>()
            .UseUrls(API_URL)
            .UseEnvironment("Test");
    }
}

单次试验

public async Task Test()
{
    EndpointConvention.Map<IMyMessage>(new Uri($"queue:MyMessageQueue"));

    var program = new TestProgram();
    program.SetUpTestServer();

    var harness = program.Server.Services.GetTestHarness();
    await harness.Start();

    try
    {
        await harness.Bus.Send<IMyMessage>(new MyMessage());

        // It's OK
        Assert.IsTrue(await harness.Sent.Any<IMyMessage>());
        Assert.IsTrue(await harness.Consumed.Any<IMyMessage>());

        var consumer = harness.GetConsumerHarness<MyConsumer>();
        // It's wrong
        Assert.That(await consumer.Consumed.Any<IMyMessage>());
    }
    finally
    {
        await harness.Stop();
    }
}

有人知道为什么我的消费者不消费发送的信息吗？

Answer 1

您的配置很奇怪，您应该遵循配置指南：

private IWebHostBuilder CreateWebHostBuilder()
{
    return new WebHostBuilder()
        .ConfigureTestServices(serviceCollection =>
        {
            serviceCollection.AddMassTransitTestHarness(cfg =>
            {
                cfg.AddConsumer<MyConsumer>(c =>
                    c.Options<BatchOptions>(o => o.SetMessageLimit(5).SetTimeLimit(2000)))
                    .Endpoint(e => e.Name = "MyMessageQueue");;
            });
        })
        .UseStartup<TestStartup>()
        .UseUrls(API_URL)
        .UseEnvironment("Test");
}

而且，使用者不使用IMyMessage，它使用Batch<IMyMessage>，而且我不完全确定使用批处理消息是否可以在线束中看到，我必须检查并确认。

此外，作为比较的一个完整测试：

[TestFixture]
public class When_batch_limit_is_reached
{
    [Test]
    public async Task Should_deliver_the_batch_to_the_consumer()
    {
        await using var provider = new ServiceCollection()
            .AddMassTransitTestHarness(x =>
            {
                x.AddConsumer<TestBatchConsumer>();
            })
            .BuildServiceProvider(true);

        var harness = provider.GetTestHarness();

        await harness.Start();

        await harness.Bus.PublishBatch(new[] { new BatchItem(), new BatchItem() });

        Assert.That(await harness.Consumed.SelectAsync<BatchItem>().Take(2).Count(), Is.EqualTo(2));

        Assert.That(await harness.GetConsumerHarness<TestBatchConsumer>().Consumed.Any<Batch<BatchItem>>(), Is.True);

        Assert.That(await harness.Published.Any<BatchResult>(x => x.Context.Message.Count == 2), Is.True);
    }
}