2017-05-11 13:44:38 +02:00
|
|
|
|
using Autofac;
|
|
|
|
|
using Microsoft.eShopOnContainers.BuildingBlocks.EventBus;
|
2017-05-08 13:59:59 +02:00
|
|
|
|
using Microsoft.eShopOnContainers.BuildingBlocks.EventBus.Abstractions;
|
2017-03-16 13:30:01 +01:00
|
|
|
|
using Microsoft.eShopOnContainers.BuildingBlocks.EventBus.Events;
|
2019-03-04 19:08:14 +00:00
|
|
|
|
using Microsoft.eShopOnContainers.BuildingBlocks.EventBus.Extensions;
|
2017-04-20 10:53:17 +02:00
|
|
|
|
using Microsoft.Extensions.Logging;
|
2017-03-10 13:17:16 +01:00
|
|
|
|
using Newtonsoft.Json;
|
2017-05-08 13:59:59 +02:00
|
|
|
|
using Newtonsoft.Json.Linq;
|
2017-04-20 10:53:17 +02:00
|
|
|
|
using Polly;
|
|
|
|
|
using Polly.Retry;
|
2017-03-09 15:56:34 +01:00
|
|
|
|
using RabbitMQ.Client;
|
|
|
|
|
using RabbitMQ.Client.Events;
|
2017-04-20 10:53:17 +02:00
|
|
|
|
using RabbitMQ.Client.Exceptions;
|
2017-03-09 15:56:34 +01:00
|
|
|
|
using System;
|
2017-04-20 10:53:17 +02:00
|
|
|
|
using System.Net.Sockets;
|
2017-03-09 15:56:34 +01:00
|
|
|
|
using System.Text;
|
2017-03-10 18:34:58 +01:00
|
|
|
|
using System.Threading.Tasks;
|
2017-03-09 15:56:34 +01:00
|
|
|
|
|
2017-03-16 13:30:01 +01:00
|
|
|
|
namespace Microsoft.eShopOnContainers.BuildingBlocks.EventBusRabbitMQ
|
2017-03-09 15:56:34 +01:00
|
|
|
|
{
|
2017-03-17 15:57:57 +01:00
|
|
|
|
public class EventBusRabbitMQ : IEventBus, IDisposable
|
2017-03-09 15:56:34 +01:00
|
|
|
|
{
|
2017-04-20 10:53:17 +02:00
|
|
|
|
const string BROKER_NAME = "eshop_event_bus";
|
2017-03-10 13:17:16 +01:00
|
|
|
|
|
2017-05-08 13:59:59 +02:00
|
|
|
|
private readonly IRabbitMQPersistentConnection _persistentConnection;
|
2017-04-20 10:53:17 +02:00
|
|
|
|
private readonly ILogger<EventBusRabbitMQ> _logger;
|
2017-05-08 13:59:59 +02:00
|
|
|
|
private readonly IEventBusSubscriptionsManager _subsManager;
|
2017-05-11 13:44:38 +02:00
|
|
|
|
private readonly ILifetimeScope _autofac;
|
|
|
|
|
private readonly string AUTOFAC_SCOPE_NAME = "eshop_event_bus";
|
2017-10-12 09:25:01 +01:00
|
|
|
|
private readonly int _retryCount;
|
2017-04-20 10:53:17 +02:00
|
|
|
|
|
|
|
|
|
private IModel _consumerChannel;
|
2017-04-20 16:44:07 +02:00
|
|
|
|
private string _queueName;
|
2017-04-20 10:53:17 +02:00
|
|
|
|
|
2017-05-11 13:44:38 +02:00
|
|
|
|
public EventBusRabbitMQ(IRabbitMQPersistentConnection persistentConnection, ILogger<EventBusRabbitMQ> logger,
|
2017-11-10 17:37:18 +01:00
|
|
|
|
ILifetimeScope autofac, IEventBusSubscriptionsManager subsManager, string queueName = null, int retryCount = 5)
|
2017-03-09 15:56:34 +01:00
|
|
|
|
{
|
2017-05-08 13:59:59 +02:00
|
|
|
|
_persistentConnection = persistentConnection ?? throw new ArgumentNullException(nameof(persistentConnection));
|
2017-04-20 10:53:17 +02:00
|
|
|
|
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
|
2017-05-08 13:59:59 +02:00
|
|
|
|
_subsManager = subsManager ?? new InMemoryEventBusSubscriptionsManager();
|
2017-11-10 17:37:18 +01:00
|
|
|
|
_queueName = queueName;
|
2017-04-20 10:53:17 +02:00
|
|
|
|
_consumerChannel = CreateConsumerChannel();
|
2017-05-11 13:44:38 +02:00
|
|
|
|
_autofac = autofac;
|
2017-10-12 09:25:01 +01:00
|
|
|
|
_retryCount = retryCount;
|
2017-05-08 13:59:59 +02:00
|
|
|
|
_subsManager.OnEventRemoved += SubsManager_OnEventRemoved;
|
2017-03-09 15:56:34 +01:00
|
|
|
|
}
|
2017-03-17 15:57:57 +01:00
|
|
|
|
|
2017-05-08 13:59:59 +02:00
|
|
|
|
private void SubsManager_OnEventRemoved(object sender, string eventName)
|
|
|
|
|
{
|
|
|
|
|
if (!_persistentConnection.IsConnected)
|
|
|
|
|
{
|
|
|
|
|
_persistentConnection.TryConnect();
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
using (var channel = _persistentConnection.CreateModel())
|
|
|
|
|
{
|
|
|
|
|
channel.QueueUnbind(queue: _queueName,
|
|
|
|
|
exchange: BROKER_NAME,
|
|
|
|
|
routingKey: eventName);
|
|
|
|
|
|
|
|
|
|
if (_subsManager.IsEmpty)
|
|
|
|
|
{
|
|
|
|
|
_queueName = string.Empty;
|
|
|
|
|
_consumerChannel.Close();
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
2017-04-20 10:53:17 +02:00
|
|
|
|
|
2017-03-15 18:42:47 -07:00
|
|
|
|
public void Publish(IntegrationEvent @event)
|
2017-03-09 15:56:34 +01:00
|
|
|
|
{
|
2017-05-08 13:59:59 +02:00
|
|
|
|
if (!_persistentConnection.IsConnected)
|
2017-04-20 10:53:17 +02:00
|
|
|
|
{
|
2017-05-08 13:59:59 +02:00
|
|
|
|
_persistentConnection.TryConnect();
|
2017-04-20 10:53:17 +02:00
|
|
|
|
}
|
2017-04-17 15:00:53 +02:00
|
|
|
|
|
2017-04-20 10:53:17 +02:00
|
|
|
|
var policy = RetryPolicy.Handle<BrokerUnreachableException>()
|
|
|
|
|
.Or<SocketException>()
|
2017-10-12 09:25:01 +01:00
|
|
|
|
.WaitAndRetry(_retryCount, retryAttempt => TimeSpan.FromSeconds(Math.Pow(2, retryAttempt)), (ex, time) =>
|
2017-04-20 10:53:17 +02:00
|
|
|
|
{
|
2019-01-31 13:32:27 +00:00
|
|
|
|
_logger.LogWarning(ex, "Could not publish event: {EventId} after {Timeout}s ({ExceptionMessage})", @event.Id, $"{time.TotalSeconds:n1}", ex.Message);
|
2017-04-20 10:53:17 +02:00
|
|
|
|
});
|
|
|
|
|
|
2019-05-07 21:29:37 +01:00
|
|
|
|
var eventName = @event.GetType().Name;
|
|
|
|
|
|
|
|
|
|
_logger.LogTrace("Creating RabbitMQ channel to publish event: {EventId} ({EventName})", @event.Id, eventName);
|
|
|
|
|
|
2017-05-08 13:59:59 +02:00
|
|
|
|
using (var channel = _persistentConnection.CreateModel())
|
2017-03-10 13:17:16 +01:00
|
|
|
|
{
|
2017-04-20 10:53:17 +02:00
|
|
|
|
|
2019-05-07 21:29:37 +01:00
|
|
|
|
_logger.LogTrace("Declaring RabbitMQ exchange to publish event: {EventId}", @event.Id);
|
|
|
|
|
|
|
|
|
|
channel.ExchangeDeclare(exchange: BROKER_NAME, type: "direct");
|
2017-03-09 15:56:34 +01:00
|
|
|
|
|
2017-04-20 10:53:17 +02:00
|
|
|
|
var message = JsonConvert.SerializeObject(@event);
|
2017-03-09 15:56:34 +01:00
|
|
|
|
var body = Encoding.UTF8.GetBytes(message);
|
|
|
|
|
|
2017-04-20 10:53:17 +02:00
|
|
|
|
policy.Execute(() =>
|
|
|
|
|
{
|
2018-01-18 17:29:23 +01:00
|
|
|
|
var properties = channel.CreateBasicProperties();
|
|
|
|
|
properties.DeliveryMode = 2; // persistent
|
|
|
|
|
|
2019-05-07 21:29:37 +01:00
|
|
|
|
_logger.LogTrace("Publishing event to RabbitMQ: {EventId}", @event.Id);
|
|
|
|
|
|
|
|
|
|
channel.BasicPublish(
|
|
|
|
|
exchange: BROKER_NAME,
|
|
|
|
|
routingKey: eventName,
|
|
|
|
|
mandatory: true,
|
|
|
|
|
basicProperties: properties,
|
|
|
|
|
body: body);
|
2017-04-20 10:53:17 +02:00
|
|
|
|
});
|
2017-03-09 15:56:34 +01:00
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
2017-05-11 13:44:38 +02:00
|
|
|
|
public void SubscribeDynamic<TH>(string eventName)
|
|
|
|
|
where TH : IDynamicIntegrationEventHandler
|
2017-03-09 15:56:34 +01:00
|
|
|
|
{
|
2019-03-04 19:08:14 +00:00
|
|
|
|
_logger.LogInformation("Subscribing to dynamic event {EventName} with {EventHandler}", eventName, typeof(TH).GetGenericTypeName());
|
|
|
|
|
|
2017-05-08 13:59:59 +02:00
|
|
|
|
DoInternalSubscription(eventName);
|
2017-05-11 13:44:38 +02:00
|
|
|
|
_subsManager.AddDynamicSubscription<TH>(eventName);
|
2019-03-15 21:03:42 +08:00
|
|
|
|
StartBasicConsume();
|
2017-05-08 13:59:59 +02:00
|
|
|
|
}
|
2017-04-20 10:53:17 +02:00
|
|
|
|
|
2017-05-11 13:44:38 +02:00
|
|
|
|
public void Subscribe<T, TH>()
|
2017-05-08 13:59:59 +02:00
|
|
|
|
where T : IntegrationEvent
|
|
|
|
|
where TH : IIntegrationEventHandler<T>
|
|
|
|
|
{
|
|
|
|
|
var eventName = _subsManager.GetEventKey<T>();
|
|
|
|
|
DoInternalSubscription(eventName);
|
2019-03-04 19:08:14 +00:00
|
|
|
|
|
|
|
|
|
_logger.LogInformation("Subscribing to event {EventName} with {EventHandler}", eventName, typeof(TH).GetGenericTypeName());
|
|
|
|
|
|
2017-05-11 13:44:38 +02:00
|
|
|
|
_subsManager.AddSubscription<T, TH>();
|
2019-03-15 21:03:42 +08:00
|
|
|
|
StartBasicConsume();
|
2017-05-08 13:59:59 +02:00
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
private void DoInternalSubscription(string eventName)
|
|
|
|
|
{
|
|
|
|
|
var containsKey = _subsManager.HasSubscriptionsForEvent(eventName);
|
|
|
|
|
if (!containsKey)
|
2017-03-09 15:56:34 +01:00
|
|
|
|
{
|
2017-05-08 13:59:59 +02:00
|
|
|
|
if (!_persistentConnection.IsConnected)
|
2017-04-20 10:53:17 +02:00
|
|
|
|
{
|
2017-05-08 13:59:59 +02:00
|
|
|
|
_persistentConnection.TryConnect();
|
2017-04-20 10:53:17 +02:00
|
|
|
|
}
|
|
|
|
|
|
2017-05-08 13:59:59 +02:00
|
|
|
|
using (var channel = _persistentConnection.CreateModel())
|
2017-04-20 10:53:17 +02:00
|
|
|
|
{
|
2017-04-20 16:44:07 +02:00
|
|
|
|
channel.QueueBind(queue: _queueName,
|
2017-04-20 10:53:17 +02:00
|
|
|
|
exchange: BROKER_NAME,
|
|
|
|
|
routingKey: eventName);
|
|
|
|
|
}
|
2017-03-09 15:56:34 +01:00
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
2017-05-08 13:59:59 +02:00
|
|
|
|
public void Unsubscribe<T, TH>()
|
|
|
|
|
where T : IntegrationEvent
|
2019-03-04 19:08:14 +00:00
|
|
|
|
where TH : IIntegrationEventHandler<T>
|
2017-03-09 15:56:34 +01:00
|
|
|
|
{
|
2019-03-04 19:08:14 +00:00
|
|
|
|
var eventName = _subsManager.GetEventKey<T>();
|
|
|
|
|
|
|
|
|
|
_logger.LogInformation("Unsubscribing from event {EventName}", eventName);
|
|
|
|
|
|
2017-05-08 13:59:59 +02:00
|
|
|
|
_subsManager.RemoveSubscription<T, TH>();
|
|
|
|
|
}
|
2017-04-20 10:53:17 +02:00
|
|
|
|
|
2017-05-08 13:59:59 +02:00
|
|
|
|
public void UnsubscribeDynamic<TH>(string eventName)
|
2017-05-11 13:44:38 +02:00
|
|
|
|
where TH : IDynamicIntegrationEventHandler
|
2017-05-08 13:59:59 +02:00
|
|
|
|
{
|
|
|
|
|
_subsManager.RemoveDynamicSubscription<TH>(eventName);
|
2017-03-10 13:17:16 +01:00
|
|
|
|
}
|
2017-03-09 15:56:34 +01:00
|
|
|
|
|
2017-03-17 15:57:57 +01:00
|
|
|
|
public void Dispose()
|
|
|
|
|
{
|
2017-04-20 16:44:07 +02:00
|
|
|
|
if (_consumerChannel != null)
|
|
|
|
|
{
|
|
|
|
|
_consumerChannel.Dispose();
|
|
|
|
|
}
|
2017-05-08 13:59:59 +02:00
|
|
|
|
|
|
|
|
|
_subsManager.Clear();
|
2017-03-17 15:57:57 +01:00
|
|
|
|
}
|
|
|
|
|
|
2019-03-15 21:03:42 +08:00
|
|
|
|
private void StartBasicConsume()
|
|
|
|
|
{
|
2019-05-07 21:29:37 +01:00
|
|
|
|
_logger.LogTrace("Starting RabbitMQ basic consume");
|
|
|
|
|
|
2019-03-15 21:03:42 +08:00
|
|
|
|
if (_consumerChannel != null)
|
|
|
|
|
{
|
2019-04-02 15:36:20 +01:00
|
|
|
|
var consumer = new AsyncEventingBasicConsumer(_consumerChannel);
|
2019-03-15 21:03:42 +08:00
|
|
|
|
|
2019-04-02 15:36:20 +01:00
|
|
|
|
consumer.Received += Consumer_Received;
|
2019-03-15 21:03:42 +08:00
|
|
|
|
|
2019-04-02 15:36:20 +01:00
|
|
|
|
_consumerChannel.BasicConsume(
|
|
|
|
|
queue: _queueName,
|
|
|
|
|
autoAck: false,
|
|
|
|
|
consumer: consumer);
|
2019-03-15 21:03:42 +08:00
|
|
|
|
}
|
|
|
|
|
else
|
|
|
|
|
{
|
2019-04-02 15:36:20 +01:00
|
|
|
|
_logger.LogError("StartBasicConsume can't call on _consumerChannel == null");
|
2019-03-15 21:03:42 +08:00
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
2019-04-02 15:36:20 +01:00
|
|
|
|
private async Task Consumer_Received(object sender, BasicDeliverEventArgs eventArgs)
|
|
|
|
|
{
|
|
|
|
|
var eventName = eventArgs.RoutingKey;
|
|
|
|
|
var message = Encoding.UTF8.GetString(eventArgs.Body);
|
|
|
|
|
|
|
|
|
|
try
|
|
|
|
|
{
|
|
|
|
|
if (message.ToLowerInvariant().Contains("throw-fake-exception"))
|
|
|
|
|
{
|
|
|
|
|
throw new InvalidOperationException($"Fake exception requested: \"{message}\"");
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
await ProcessEvent(eventName, message);
|
|
|
|
|
}
|
|
|
|
|
catch (Exception ex)
|
|
|
|
|
{
|
|
|
|
|
_logger.LogWarning(ex, "----- ERROR Processing message \"{Message}\"", message);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// Even on exception we take the message off the queue.
|
|
|
|
|
// in a REAL WORLD app this should be handled with a Dead Letter Exchange (DLX).
|
|
|
|
|
// For more information see: https://www.rabbitmq.com/dlx.html
|
|
|
|
|
_consumerChannel.BasicAck(eventArgs.DeliveryTag, multiple: false);
|
|
|
|
|
}
|
|
|
|
|
|
2017-04-20 10:53:17 +02:00
|
|
|
|
private IModel CreateConsumerChannel()
|
2017-03-10 13:17:16 +01:00
|
|
|
|
{
|
2017-05-08 13:59:59 +02:00
|
|
|
|
if (!_persistentConnection.IsConnected)
|
2017-03-10 13:17:16 +01:00
|
|
|
|
{
|
2017-05-08 13:59:59 +02:00
|
|
|
|
_persistentConnection.TryConnect();
|
2017-03-22 14:23:25 +01:00
|
|
|
|
}
|
2017-03-09 15:56:34 +01:00
|
|
|
|
|
2019-05-07 21:29:37 +01:00
|
|
|
|
_logger.LogTrace("Creating RabbitMQ consumer channel");
|
|
|
|
|
|
2017-05-08 13:59:59 +02:00
|
|
|
|
var channel = _persistentConnection.CreateModel();
|
2017-03-10 13:17:16 +01:00
|
|
|
|
|
2017-04-20 10:53:17 +02:00
|
|
|
|
channel.ExchangeDeclare(exchange: BROKER_NAME,
|
2019-04-02 15:36:20 +01:00
|
|
|
|
type: "direct");
|
2017-03-22 14:23:25 +01:00
|
|
|
|
|
2017-11-10 17:37:18 +01:00
|
|
|
|
channel.QueueDeclare(queue: _queueName,
|
|
|
|
|
durable: true,
|
|
|
|
|
exclusive: false,
|
|
|
|
|
autoDelete: false,
|
|
|
|
|
arguments: null);
|
|
|
|
|
|
2017-04-20 10:53:17 +02:00
|
|
|
|
channel.CallbackException += (sender, ea) =>
|
|
|
|
|
{
|
2019-05-07 21:29:37 +01:00
|
|
|
|
_logger.LogWarning(ea.Exception, "Recreating RabbitMQ consumer channel");
|
|
|
|
|
|
2017-04-20 10:53:17 +02:00
|
|
|
|
_consumerChannel.Dispose();
|
|
|
|
|
_consumerChannel = CreateConsumerChannel();
|
2019-03-15 21:03:42 +08:00
|
|
|
|
StartBasicConsume();
|
2017-04-20 10:53:17 +02:00
|
|
|
|
};
|
|
|
|
|
|
|
|
|
|
return channel;
|
2017-03-09 15:56:34 +01:00
|
|
|
|
}
|
2017-03-14 14:15:34 +01:00
|
|
|
|
|
|
|
|
|
private async Task ProcessEvent(string eventName, string message)
|
|
|
|
|
{
|
2019-05-07 21:29:37 +01:00
|
|
|
|
_logger.LogTrace("Processing RabbitMQ event: {EventName}", eventName);
|
|
|
|
|
|
2017-05-08 13:59:59 +02:00
|
|
|
|
if (_subsManager.HasSubscriptionsForEvent(eventName))
|
2017-03-14 14:15:34 +01:00
|
|
|
|
{
|
2017-05-11 13:44:38 +02:00
|
|
|
|
using (var scope = _autofac.BeginLifetimeScope(AUTOFAC_SCOPE_NAME))
|
2017-03-14 14:15:34 +01:00
|
|
|
|
{
|
2017-05-11 13:44:38 +02:00
|
|
|
|
var subscriptions = _subsManager.GetHandlersForEvent(eventName);
|
|
|
|
|
foreach (var subscription in subscriptions)
|
2017-05-08 13:59:59 +02:00
|
|
|
|
{
|
2017-05-11 13:44:38 +02:00
|
|
|
|
if (subscription.IsDynamic)
|
2019-03-04 19:08:14 +00:00
|
|
|
|
{
|
2017-05-11 13:44:38 +02:00
|
|
|
|
var handler = scope.ResolveOptional(subscription.HandlerType) as IDynamicIntegrationEventHandler;
|
2018-10-11 17:16:31 +02:00
|
|
|
|
if (handler == null) continue;
|
2017-05-11 13:44:38 +02:00
|
|
|
|
dynamic eventData = JObject.Parse(message);
|
2019-06-17 22:00:19 +01:00
|
|
|
|
|
|
|
|
|
await Task.Yield();
|
2017-05-11 13:44:38 +02:00
|
|
|
|
await handler.Handle(eventData);
|
|
|
|
|
}
|
|
|
|
|
else
|
|
|
|
|
{
|
2018-10-11 17:16:31 +02:00
|
|
|
|
var handler = scope.ResolveOptional(subscription.HandlerType);
|
|
|
|
|
if (handler == null) continue;
|
2017-05-11 13:44:38 +02:00
|
|
|
|
var eventType = _subsManager.GetEventTypeByName(eventName);
|
|
|
|
|
var integrationEvent = JsonConvert.DeserializeObject(message, eventType);
|
|
|
|
|
var concreteType = typeof(IIntegrationEventHandler<>).MakeGenericType(eventType);
|
2019-06-17 22:00:19 +01:00
|
|
|
|
|
|
|
|
|
await Task.Yield();
|
2017-05-11 13:44:38 +02:00
|
|
|
|
await (Task)concreteType.GetMethod("Handle").Invoke(handler, new object[] { integrationEvent });
|
|
|
|
|
}
|
2017-05-08 13:59:59 +02:00
|
|
|
|
}
|
2017-03-14 14:15:34 +01:00
|
|
|
|
}
|
|
|
|
|
}
|
2019-05-07 21:29:37 +01:00
|
|
|
|
else
|
|
|
|
|
{
|
|
|
|
|
_logger.LogWarning("No subscription for RabbitMQ event: {EventName}", eventName);
|
|
|
|
|
}
|
2017-03-14 14:15:34 +01:00
|
|
|
|
}
|
2017-03-09 15:56:34 +01:00
|
|
|
|
}
|
|
|
|
|
}
|