MassTransit：コンシューマにコールバックキューを作成する

Question

MassTransitを初めて使用しても、チュートリアルプロジェクトのいくつかでまだ遊んでいます。私はたぶん20分間走るサービスを持っています。それが終わったら何かをする必要があります。私はリクエスト/レスポンスのパターンに従うことを望んでいないので、スレッドを保持して応答を待っています。他のオプションは、仕事が終わったときに消費者が公開するための別のキューを作成することだと思います。私はこの投稿を見てきました：MassTransit3 how to make request from consumerしかし、私はこれを実装する方法がわかりません。再びthisチュートリアルからの私のプロジェクト、次のようになります。

出版社：

static void Main(string[] args) 
{ 
    var bus = Bus.Factory.CreateUsingRabbitMq(x => 
     x.Host(new Uri("rabbitmq://localhost/"), h => {})); 
    var busHandle = bus.Start(); 
    var text = ""' 
    Console.WriteLine("Publisher"); 
    while(text != "quit") 
    { 
     Console.Write("Enter a message: "); 
     text = Console.ReadLine(); 

     var message = new SomethingHappenedMessage() 
     { 
      What = text, 
      When = DateTime.Now 
     } 
     bus.Publish(message); 
    } 
    busHandle.Stop(); 
} 

購読者：

static void Main(string[] args) 
{ 
    var bus = Bus.Factory.CreateUsingRabbitMq(x => 
    { 
     var host = x.Host(new Uri("rabbitmq://localhost"), h => {}); 
     x.ReceiveEndpoint(host, "MtPubSubExample_TestSubscriber", e => 
      e.Consumer<SomethingHappenedConsumer>()); 
    }); 
    Console.WriteLine("Subscriber"); 
    var busHandle = bus.Start(); 
    Console.ReadKey(); 
    busHandle.Stop(); 
} 

消費者：

私はAを作成するに行くかどう

class SomethingHappenedConsumer : IConsumer<ISomethingHappened> 
{ 
    public Task Consume(ConsumeContext<ISomethingHappened> context) 
    { 
     Console.Write("TXT: " + context.Message.What); 
     Console.Write(" SENT: " + context.Message.When); 
     Console.Write(" PROCESSED: " + DateTime.Now); 
     Console.WriteLine(" (" + System.Threading.Thread.CurrentThread.ManagedThreadId + ")"); 
     return Task.FromResult(0); 
    } 
} 

コンシューマのコールバックキュー

Answer 1

お客様のコンシューマでは、Bus.Publish(new ResponseMessage());（またはあなたの回答とは何ですか）だけで、パブリッシャーにそのメッセージタイプのコンシューマーを登録させるようにしてください。パブリッシャーはキューにバインドされているようには見えないので、キュー名を作成してキューにバインドするだけです。

Answer 2

もう一度@Travisにお問い合わせいただきありがとうございます。私は将来誰にでも結びついた最終的なコードを見せたいだけでした。メッセージは、レスポンスで面白く見えますが、正しくパブリッシャーにポストバックしています。

出版社：

static void Main(string[] args) 
{ 
    var bus = Bus.Factory.CreateUsingRabbitMq(x => 
    { 
     var host = x.Host(new Uri("rabbitmq://localhost/"), h => { }); 
     x.ReceiveEndpoint(host, "MtPubSubExample_TestPublisher", e => 
      e.Consumer<ResponseConsumer>()); 
    }); 
    var busHandle = bus.Start(); 
    var text = ""; 
    Console.WriteLine("Publisher"); 
    while(text != "quit") 
    { 
     Console.Write("Enter a message: "); 
     text = Console.ReadLine(); 

     var message = new SomethingHappenedMessage() 
     { 
      What = text, 
      When = DateTime.Now 
     }; 
     bus.Publish(message); 
    } 

    busHandle.Stop(); 
} 

レスポンス消費者：

class ResponseConsumer : IConsumer<IResponse> 
{ 
    public Task Consume(ConsumeContext<IResponse> context) 
    { 
     Console.WriteLine("RESPONSE MESSAGE: " + context.Message.Message); 
     return Task.FromResult(0); 
    } 
} 

購読者：

static void Main(string[] args) 
{ 
    var bus = Bus.Factory.CreateUsingRabbitMq(x => 
    { 
     var host = x.Host(new Uri("rabbitmq://localhost/"), h => { }); 
     x.ReceiveEndpoint(host, "MtPubSubExample_TestSubscriber", e => 
      e.Consumer<SomethingHappenedConsumer>()); 
    }); 
    Console.WriteLine("Subscriber"); 
    var busHandle = bus.Start(); 
    Console.ReadKey(); 
    busHandle.Stop(); 
} 

加入者の消費者：

class SomethingHappenedConsumer : IConsumer<ISomethingHappened> 
{ 
    private IBusControl bus = Bus.Factory.CreateUsingRabbitMq(x => 
     x.Host(new Uri("rabbitmq://localhost/"), h => { })); 

    public Task Consume(ConsumeContext<ISomethingHappened> context) 
    { 
     var now = DateTime.Now; 
     Console.Write("TXT: " + context.Message.What); 
     Console.Write(" SENT: " + context.Message.When); 
     Console.Write(" PROCESSED: " + now); 
     Console.WriteLine(" (" + System.Threading.Thread.CurrentThread.ManagedThreadId + ")"); 

     var response = new ResponseMessage() 
     { 
      Message = "The request was processed at " + now 
     }; 

     bus.Publish(response); 
     return Task.FromResult(0); 
    } 
}