如何使用 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() 却出人意料地起作用了！