如何使用公交的IConsumeObserver？

Question

我在MassTransit3中使用MassTransit3时遇到了麻烦。我编写了一个非常简单的观察者，它只在使用前后将消息打印到控制台，但是消息没有显示，而且代码似乎永远不会被调用(未命中其中的断点)。

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保持一致，我将更新总线代码，以便其他观察者也连接到接收端点。