MassTransit 的 IConsumeObserver 应该如何使用？

Question

我在使用 MassTransit3 中的 IConsumeObserver 时遇到问题。我写了一个非常简单的观察器，它只是在控制台前打印一条消息并使用 post，但消息没有显示，而且似乎代码永远不会被调用（其中的断点没有命中）。

public class ConsumeObserver : IConsumeObserver
{
    public async Task ConsumeFault<T>(ConsumeContext<T> context, Exception exception) where T : class
    {
        await Console.Out.WriteLineAsync("ConsumeObserver.ConsumeFault");
    }

    public async Task PostConsume<T>(ConsumeContext<T> context) where T : class
    {
        await Console.Out.WriteLineAsync("ConsumeObserver.PostConsume");
    }

    public async Task PreConsume<T>(ConsumeContext<T> context) where T : class
    {
        await Console.Out.WriteLineAsync("ConsumeObserver.PreConsume");
    }
}

奇怪的是，我已经实现了一个类似的接收观察器，它确实像我预期的那样被调用。

public class ReceiveObserver : IReceiveObserver
{
    public async Task PostReceive(ReceiveContext context)
    {
        await Console.Out.WriteLineAsync("ReceiveObserver.PostReceive");
    }

    //Other methods omitted for brevity        
}

我像这样将观察者连接到总线：

        var bus = Bus.Factory.CreateUsingRabbitMq(cfg =>
        {
            var host = cfg.Host(new Uri(ConfigurationManager.AppSettings["RabbitMQHost"]), h => { });

            cfg.ReceiveEndpoint(host, "Subscriber", e =>
            {
                e.Consumer<SomethingHappenedConsumer>();
                e.UseRetry(Retry.Interval(2, TimeSpan.FromSeconds(1)));
            });
        });

        //The callbacks in this observer get called...
        bus.ConnectReceiveObserver(new ReceiveObserver());

        //...but not in these two observers
        bus.ConnectConsumeObserver(new ConsumeObserver());
        bus.ConnectConsumeMessageObserver(new ConsumeSomethingHappenedObserver());

感谢任何有助于理解的帮助。我已将完整的源代码放在 GitHub.

Answer 1

目前 ConsumeObserver 必须在每个接收端点（使用 .13）注册。因此，通过将观察者连接到总线，接收端点使用的消息是不可见的，除非观察者也连接到接收端点。

为了与 ReceiveObserver 保持一致，我将更新总线代码，以便其他观察者也连接到接收端点。

MassTransit 的 IConsumeObserver 应该如何使用？

How should MassTransit's IConsumeObserver be used?

masstransit