Masstransitでコマンドベースおよびイベントベースのシステムをテストする方法

Question

私は、操作が実行されたときにイベントを発生させるドメインオブジェクトに対して操作を呼び出すコマンドハンドラを持っています。対応するコマンドが送信されたときに、イベントハンドラがイベントを受け取ることをテストしたいと思います（下記を参照して、簡潔にするためにいくつかのコードは省略しています）。イベントハンドラ（MyEventConsumer.Consume）は、イベントメッセージがバス（この場合はループバックバス）にパブリッシュされても呼び出されません。何か案は？

//Test 
[TestFixture] 
public class TestSendCommandReceiveEvent 
{ 
    [Given] 
    public void installation_of_infrastructure_objects() 
    { 
     container.Register(Component.For<MyEventConsumer>().UsingFactoryMethod(() => new MyEventConsumer(_received))); 
     container.Register(
     Component.For<IServiceBus>() 
     .UsingFactoryMethod(() => ServiceBusFactory.New(x => { x.ReceiveFrom("loopback://localhost/mt_client"); x.Subscribe(conf => conf.LoadFrom(container));              }))); 
    } 

    [When] 
    public void sending_a_command() 
    { 
     var LocalBus = container.Resolve<IServiceBus>(); 
     LocalBus.Publish(new DoSomething(_aggregateId)); 
    } 
    [Then] 
    public void corresponding_event_should_be_received_by_consumer() 
    { 
     _received.WaitOne(5000).ShouldBeTrue(); 
    } 
} 
public class MyEventConsumer : Consumes<SomethingDone>.All 
{ 
    private readonly ManualResetEvent _received; 
    public MyEventConsumer(ManualResetEvent received) 
    { 
     _received = received; 
    } 
    public void Consume(SomethingDone message) 
    { 
     _received.Set(); 
    } 
} 

//Command handler 
public class DoSomethingCommandHandler : Consumes<DoSomething>.All where T:class 
{ 
    public void Consume(DoSomething message) 
    { 
     var ar = Repository.GetById<SomeAR>(message.ArId); 
     ar.DoSomething(); 
     Repository.Save(ar, Guid.NewGuid(), null); 
    } 
} 
//Domain object 
public class SomeDomainObject : AggregateBase 
{ 
    public void DoSomething() 
    { 
     RaiseEvent(new SomethingDone(Id, 1)); 
    } 
} 

Answer 1

これは私のために渡します。

// Copyright 2012 Henrik Feldt 
// 
// Licensed under the Apache License, Version 2.0 (the "License"); you may not use 
// this file except in compliance with the License. You may obtain a copy of the 
// License at 
// 
//  http://www.apache.org/licenses/LICENSE-2.0 
// 
// Unless required by applicable law or agreed to in writing, software distributed 
// under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR 
// CONDITIONS OF ANY KIND, either express or implied. See the License for the 
// specific language governing permissions and limitations under the License. 

using System; 
using System.Threading; 
using Castle.MicroKernel.Registration; 
using Castle.Windsor; 
using Magnum.Extensions; 
using Magnum.TestFramework; 
using MassTransit; 
using NUnit.Framework; 

namespace ConsoleApplication11 
{ 
    [TestFixture] 
    public class TestSendCommandReceiveEvent 
    { 
     ManualResetEventSlim _received = new ManualResetEventSlim(false); 
     IWindsorContainer _container; 

     [Given] 
     public void installation_of_infrastructure_objects() 
     { 
      _container = new WindsorContainer(); 
      _container.Register(
       Component.For<IServiceBus>() 
        .UsingFactoryMethod(() => ServiceBusFactory.New(x => 
         { 
          x.ReceiveFrom("loopback://localhost/mt_client"); 
          x.Subscribe(conf => 
           { 
            conf.Consumer(() => new MyEventConsumer(_received)); 
            conf.Consumer(() => new MyCmdConsumer()); 
           }); 
         }))); 

      when(); 
     } 

     public void when() 
     { 
      var localBus = _container.Resolve<IServiceBus>(); 
      // wait for startup 
      localBus.Endpoint.InboundTransport.Receive(c1 => c2 => { }, 1.Milliseconds()); 

      localBus.Publish(new DoSomething()); 
     } 

     [Then] 
     public void corresponding_event_should_be_received_by_consumer() 
     { 
      _received.Wait(5000).ShouldBeTrue(); 
     } 
    } 

    [Serializable] 
    public class DoSomething 
    { 
    } 

    [Serializable] 
    public class SomethingDone 
    { 
    } 

    public class MyEventConsumer : Consumes<SomethingDone>.All 
    { 
     readonly ManualResetEventSlim _received; 

     public MyEventConsumer(ManualResetEventSlim received) 
     { 
      _received = received; 
     } 

     public void Consume(SomethingDone message) 
     { 
      _received.Set(); 
     } 
    } 

    public class MyCmdConsumer : Consumes<DoSomething>.Context 
    { 
     public void Consume(IConsumeContext<DoSomething> ctx) 
     { 
      Console.WriteLine("consumed cmd"); 
      ctx.Bus.Publish(new SomethingDone()); 
     } 
    } 
} 

Answer 2

私の経験では、バスインスタンスの作成直後に、公開されたメッセージが失われてしまうという短期間があります。何らかの非同期初期化が行われている必要があります。

container.Resolve<IServiceBus>()とLocalBus.Publish(new DoSomething(_aggregateId))の間に遅延を追加してみてください。

Thread.Sleep私の場合は動作しませんでしたが、Console.ReadLine()驚いたことにでした！