Come testare un sistema basato su comandi ed eventi con Masstransit

Question

Ho un gestore di comandi che richiama un'operazione su un oggetto dominio che a sua volta genera un evento quando l'operazione è stata eseguita. Mi piacerebbe testare che un gestore di eventi riceve l'evento quando il comando corrispondente è stato inviato (vedi sotto, alcuni codici omessi per brevità). Il gestore di eventi (MyEventConsumer.Consume) non viene mai richiamato anche se il messaggio di evento è pubblicato sul bus (bus di loopback in questo caso). Qualche idea?

//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

Questo passa per me:

// 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

Nella mia esperienza, c'è un breve periodo di tempo, subito dopo la creazione dell'istanza del bus, durante il quale vengono persi tutti i messaggi pubblicati. Deve essere una sorta di inizializzazione asincronica in corso.

Provare ad aggiungere un ritardo tra container.Resolve<IServiceBus>() e LocalBus.Publish(new DoSomething(_aggregateId)).

Thread.Sleep non ha funzionato nel mio caso, ma un Console.ReadLine() ha fatto sorprendentemente!