Как создать CreateUsingInMemory() для работы с MassTransit?

Question

Вот мой весь код. Я думаю, что тест должен пройти, но он терпит неудачу. Я (безуспешно) пытался использовать некоторые из перегрузок до Consumer.

using MassTransit; 
using NUnit.Framework; 
using System.Threading.Tasks; 

namespace MassTransitTests 
{ 
    public class Message 
    { 
    } 

    public class MessageConsumer : IConsumer<Message> 
    { 
     public static int ConsumedCount 
     { 
      get; 
      private set; 
     } 

     public Task Consume(ConsumeContext<Message> context) 
     { 
      ConsumedCount++; 
      return Task.FromResult(0); 
     } 
    } 

    [TestFixture] 
    public class MassTransitTest 
    { 
     [Test] 
     public async Task BasicTestAsync() 
     { 
      // Arrange 
      var control = Bus.Factory.CreateUsingInMemory(configure => 
      { 
       configure.ReceiveEndpoint("myQueue", endpoint => 
       { 
        endpoint.Consumer<MessageConsumer>(); 
       }); 
      }); 

      // Act 
      using (var handle = control.Start()) 
      { 
       await control.Publish(new Message()); 
       await control.Publish(new Message()); 
      } 

      // Assert 
      Assert.That(MessageConsumer.ConsumedCount, Is.EqualTo(2)); 
     } 
    } 
} 

documentation их показывает это, именно то, что я делаю:

var busControl = Bus.Factory.CreateUsingInMemory(cfg => 
{ 
    cfg.ReceiveEndpoint("queue_name", ep => 
    { 
     //configure the endpoint 
    }) 
}); 

Что я делаю неправильно/что мне необходимо изменить в моем Arrange/Act, чтобы получить мой Assert к Работа?

Answer 1

После копаться их испытаний, я обнаружил, что мне не хватает:

[1] Вам нужно *, чтобы awaitBusHandle.Ready, что я не делал. * (тест работает без этого - по крайней мере, первый раз, когда я побежал, но это может быть просто состояние гонки работает в мою пользу ....)

[2] Вызовы Publish, по-видимому, когда закончить автобус получил сообщение, которое я угадываю, а не когда обработчики/потребители сообщения завершили свою работу. Поэтому вам нужно уведомить вызывающий код, который обработчики завершили, если это то, что вы тестируете. Вот один из способов сделать это - используйте TaskCompletionSource<T> (аналогично тому, что я нашел в их кодовой базе). Очевидно, что я не был совершенным в моей безопасности потоков и мой lock использование немного кувалдой-Эск, но это иллюстрирует пункт:

using MassTransit; 
using NUnit.Framework; 
using System.Threading.Tasks; 

namespace MassTransitTests 
{ 
    public class Message 
    { 
    } 

    public class MessageConsumer : IConsumer<Message> 
    {   
     public static int TargetConsumedCount 
     { 
      get { return _targetConsumedCount; } 
      set 
      { 
       lock (_lock) 
       { 
        _targetConsumedCount = value; 
        CheckTargetReached(); 
       } 
      } 
     } 

     private static void CheckTargetReached() 
     { 
      if (_consumedCount >= TargetConsumedCount) 
      { 
       _targetReached.SetResult(true); 
      } 
     } 

     public static Task<bool> TargetReached { get; private set; } 

     private static int _consumedCount; 
     private static int _targetConsumedCount; 
     private static TaskCompletionSource<bool> _targetReached; 
     private static object _lock; 

     static MessageConsumer() 
     { 
      _lock = new object(); 
      _targetReached = new TaskCompletionSource<bool>(); 
      TargetReached = _targetReached.Task; 
     } 

     public Task Consume(ConsumeContext<Message> context) 
     { 
      lock (_lock) 
      { 
       _consumedCount++; 
       CheckTargetReached(); 
      } 
      return Task.FromResult(0); 
     } 
    } 

    [TestFixture] 
    public class MassTransitTest 
    { 
     [Test] 
     public async Task BasicTestAsync() 
     { 
      // Arrange 
      var control = Bus.Factory.CreateUsingInMemory(configure => 
      { 
       configure.ReceiveEndpoint("myQueue", endpoint => 
       { 
        endpoint.Consumer<MessageConsumer>(); 
       }); 
      }); 

      using (var handle = control.Start()) 
      { 
       await handle.Ready; // [1] 

       // Act 
       await control.Publish(new Message()); 
       await control.Publish(new Message()); 

       // Assert 
       MessageConsumer.TargetConsumedCount = 2; 
       await MessageConsumer.TargetReached; // [2] 
      }    
     } 
    } 
}