Как следует использовать IConsumeObserver MassTransit?

Question

У меня возникли проблемы с использованием IConsumeObserver в MassTransit3. Я написал очень простой наблюдатель, который просто печатает сообщение на консоль pre и 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, я обновляю код шины, чтобы другие наблюдатели также подключались к принимающим оконечным точкам.