Опыт использования MassTransit 3.001.11.2016 05:18

private static async Task SendSmsCommand(IBus busControl)
{
   var command = new SendSmsCommand
   {
       CommandId = Guid.NewGuid(),
       PhoneNumber = "89031112233",
       Message = "Thank you for your reply"
   };

   var endpoint = await busControl.GetSendEndpoint(AppSettings.CommandEndpoint);
   await endpoint.Send(command);
}

public interface ISendSms {
	Guid CommandId { get; }
	string PhoneNumber { get; }
	string Message { get; }
}

public interface ISmsSent {
	Guid EventId { get; }
	DateTime SentAtUtc { get; }	
}

public interface ICommand
{
}

public interface ISendSms : ICommand
{
   Guid CommandId { get; }
   string PhoneNumber { get; }
   string Message { get; }
}

class SendSmsCommand : ISendSms
{
   public Guid CommandId { get; set; }
   public string PhoneNumber { get; set; }
   public string Message { get; set; }
}

public class SendSmsConsumer : IConsumer>
{
   public SendSmsConsumer(IBusControl busControl)
   {
       _busControl = busControl;
   }

   public async Task Consume(ConsumeContext> context)
   {
       var message = context.Message;

       Console.WriteLine($"[IConsumer] Send sms command consumed");
       Console.WriteLine($"[IConsumer] CommandId: {message.CommandId}");
       Console.WriteLine($"[IConsumer] Phone number: {message.PhoneNumber}");
       Console.WriteLine($"[IConsumer] Message: {message.Message}");

       Console.Write(Environment.NewLine);
       Console.WriteLine("Публикация события: Смс сообщение отправлено");
       await _busControl.SmsSent(DateTime.UtcNow);
   }

   private readonly IBus _busControl;
}

public static class BusExtensions
{
   /// 

   /// Отправка смс сообщения
   /// 
   /// 
   /// 
   /// 
   /// 
   /// 
   public static async Task SendSms(
       this IBus bus, Uri host, string phoneNumber, string message
   )
   {
       var command = new SendSmsCommand
       {
           CommandId = Guid.NewGuid(),
           PhoneNumber = phoneNumber,
           Message = message
       };

       await bus.SendCommand(host, command);
   }

   /// 

   /// Публикация события об отправке смс сообщения
   /// 
   /// 
   /// 
   /// 
   public static async Task SmsSent(
       this IBus bus, DateTime sentAtUtc
   )
   {
       var @event = new SmsSentEvent
       {
           EventId = Guid.NewGuid(),
           SentAtUtc = sentAtUtc
       };

       await bus.PublishEvent(@event);
   }

   /// 

   /// Отправка команды
   /// 
   /// 
   /// 
   /// 
   /// 
   /// 
   private static async Task SendCommand(this IBus bus, Uri address, T command) where T : class
   {
       var endpoint = await bus.GetSendEndpoint(address);
       await endpoint.Send(command);
   }

   /// 

   /// Публикация события
   /// 
   /// 
   /// 
   /// 
   /// 
   private static async Task PublishEvent(this IBus bus, T message) where T : class
   {
       await bus.Publish(message);
   }
}

class SendSmsCommand : ISendSms
{
   public Guid CommandId { get; set; }
   public string PhoneNumber { get; set; }
   public string Message { get; set; }
}

class SmsSentEvent : ISmsSent
{
   public Guid EventId { get; set; }
   public DateTime SentAtUtc { get; set; }
}

public static IBusControl GetConfiguredFactory(IUnityContainer container)
{
   if (container == null)
   {
       throw new ArgumentNullException(nameof(container));
   }

   var control = Bus.Factory.CreateUsingRabbitMq(cfg => {
       var host = cfg.Host(AppSettings.Host, h => { });


       // cfg.ReceiveEndpoint(host, e => e.LoadFrom(container));
       cfg.ReceiveEndpoint(host, "play_with_masstransit_queue", e => e.LoadFrom(container));
   });

   control.ConnectConsumeObserver(new ConsumeObserver());
   control.ConnectReceiveObserver(new ReceiveObserver());
   control.ConnectConsumeMessageObserver(new ConsumeObserverSendSmsCommand());
   control.ConnectSendObserver(new SendObserver());
   control.ConnectPublishObserver(new PublishObserver());

   return control;
}

public class ConsumeObserver : IConsumeObserver
{
   public Task PreConsume(ConsumeContext context) where T : class
   {
       Console.WriteLine($"[ConsumeObserver] PreConsume {context.MessageId}");
       return Task.CompletedTask;
   }

   public Task PostConsume(ConsumeContext context) where T : class
   {
       Console.WriteLine($"[ConsumeObserver] PostConsume {context.MessageId}");
       return Task.CompletedTask;
   }

   public Task ConsumeFault(ConsumeContext context, Exception exception) where T : class
   {
       Console.WriteLine($"[ConsumeObserver] ConsumeFault {context.MessageId}");
       return Task.CompletedTask;
   }
}