|
@ -80,13 +80,16 @@ namespace Microsoft.eShopOnContainers.BuildingBlocks.EventBusRabbitMQ |
|
|
_logger.LogWarning(ex, "Could not publish event: {EventId} after {Timeout}s ({ExceptionMessage})", @event.Id, $"{time.TotalSeconds:n1}", ex.Message); |
|
|
_logger.LogWarning(ex, "Could not publish event: {EventId} after {Timeout}s ({ExceptionMessage})", @event.Id, $"{time.TotalSeconds:n1}", ex.Message); |
|
|
}); |
|
|
}); |
|
|
|
|
|
|
|
|
|
|
|
var eventName = @event.GetType().Name; |
|
|
|
|
|
|
|
|
|
|
|
_logger.LogTrace("Creating RabbitMQ channel to publish event: {EventId} ({EventName})", @event.Id, eventName); |
|
|
|
|
|
|
|
|
using (var channel = _persistentConnection.CreateModel()) |
|
|
using (var channel = _persistentConnection.CreateModel()) |
|
|
{ |
|
|
{ |
|
|
var eventName = @event.GetType() |
|
|
|
|
|
.Name; |
|
|
|
|
|
|
|
|
|
|
|
channel.ExchangeDeclare(exchange: BROKER_NAME, |
|
|
|
|
|
type: "direct"); |
|
|
|
|
|
|
|
|
_logger.LogTrace("Declaring RabbitMQ exchange to publish event: {EventId}", @event.Id); |
|
|
|
|
|
|
|
|
|
|
|
channel.ExchangeDeclare(exchange: BROKER_NAME, type: "direct"); |
|
|
|
|
|
|
|
|
var message = JsonConvert.SerializeObject(@event); |
|
|
var message = JsonConvert.SerializeObject(@event); |
|
|
var body = Encoding.UTF8.GetBytes(message); |
|
|
var body = Encoding.UTF8.GetBytes(message); |
|
@ -96,11 +99,14 @@ namespace Microsoft.eShopOnContainers.BuildingBlocks.EventBusRabbitMQ |
|
|
var properties = channel.CreateBasicProperties(); |
|
|
var properties = channel.CreateBasicProperties(); |
|
|
properties.DeliveryMode = 2; // persistent
|
|
|
properties.DeliveryMode = 2; // persistent
|
|
|
|
|
|
|
|
|
channel.BasicPublish(exchange: BROKER_NAME, |
|
|
|
|
|
routingKey: eventName, |
|
|
|
|
|
mandatory: true, |
|
|
|
|
|
basicProperties: properties, |
|
|
|
|
|
body: body); |
|
|
|
|
|
|
|
|
_logger.LogTrace("Publishing event to RabbitMQ: {EventId}", @event.Id); |
|
|
|
|
|
|
|
|
|
|
|
channel.BasicPublish( |
|
|
|
|
|
exchange: BROKER_NAME, |
|
|
|
|
|
routingKey: eventName, |
|
|
|
|
|
mandatory: true, |
|
|
|
|
|
basicProperties: properties, |
|
|
|
|
|
body: body); |
|
|
}); |
|
|
}); |
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
@ -176,6 +182,8 @@ namespace Microsoft.eShopOnContainers.BuildingBlocks.EventBusRabbitMQ |
|
|
|
|
|
|
|
|
private void StartBasicConsume() |
|
|
private void StartBasicConsume() |
|
|
{ |
|
|
{ |
|
|
|
|
|
_logger.LogTrace("Starting RabbitMQ basic consume"); |
|
|
|
|
|
|
|
|
if (_consumerChannel != null) |
|
|
if (_consumerChannel != null) |
|
|
{ |
|
|
{ |
|
|
var consumer = new AsyncEventingBasicConsumer(_consumerChannel); |
|
|
var consumer = new AsyncEventingBasicConsumer(_consumerChannel); |
|
@ -225,6 +233,8 @@ namespace Microsoft.eShopOnContainers.BuildingBlocks.EventBusRabbitMQ |
|
|
_persistentConnection.TryConnect(); |
|
|
_persistentConnection.TryConnect(); |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
_logger.LogTrace("Creating RabbitMQ consumer channel"); |
|
|
|
|
|
|
|
|
var channel = _persistentConnection.CreateModel(); |
|
|
var channel = _persistentConnection.CreateModel(); |
|
|
|
|
|
|
|
|
channel.ExchangeDeclare(exchange: BROKER_NAME, |
|
|
channel.ExchangeDeclare(exchange: BROKER_NAME, |
|
@ -238,6 +248,8 @@ namespace Microsoft.eShopOnContainers.BuildingBlocks.EventBusRabbitMQ |
|
|
|
|
|
|
|
|
channel.CallbackException += (sender, ea) => |
|
|
channel.CallbackException += (sender, ea) => |
|
|
{ |
|
|
{ |
|
|
|
|
|
_logger.LogWarning(ea.Exception, "Recreating RabbitMQ consumer channel"); |
|
|
|
|
|
|
|
|
_consumerChannel.Dispose(); |
|
|
_consumerChannel.Dispose(); |
|
|
_consumerChannel = CreateConsumerChannel(); |
|
|
_consumerChannel = CreateConsumerChannel(); |
|
|
StartBasicConsume(); |
|
|
StartBasicConsume(); |
|
@ -248,6 +260,8 @@ namespace Microsoft.eShopOnContainers.BuildingBlocks.EventBusRabbitMQ |
|
|
|
|
|
|
|
|
private async Task ProcessEvent(string eventName, string message) |
|
|
private async Task ProcessEvent(string eventName, string message) |
|
|
{ |
|
|
{ |
|
|
|
|
|
_logger.LogTrace("Processing RabbitMQ event: {EventName}", eventName); |
|
|
|
|
|
|
|
|
if (_subsManager.HasSubscriptionsForEvent(eventName)) |
|
|
if (_subsManager.HasSubscriptionsForEvent(eventName)) |
|
|
{ |
|
|
{ |
|
|
using (var scope = _autofac.BeginLifetimeScope(AUTOFAC_SCOPE_NAME)) |
|
|
using (var scope = _autofac.BeginLifetimeScope(AUTOFAC_SCOPE_NAME)) |
|
@ -274,6 +288,10 @@ namespace Microsoft.eShopOnContainers.BuildingBlocks.EventBusRabbitMQ |
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
|
|
|
|
else |
|
|
|
|
|
{ |
|
|
|
|
|
_logger.LogWarning("No subscription for RabbitMQ event: {EventName}", eventName); |
|
|
|
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
|
} |