ayan.poddar/eShopOnContainers
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Fix bug with queue names

Browse Source
This commit is contained in:
Unai Zorrilla Castro 2017-04-20 16:44:07 +02:00
parent 09a53f0137
commit 70f50a0fce
1 changed files with 18 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
src/BuildingBlocks/EventBus/EventBusRabbitMQ/EventBusRabbitMQ.cs
View File

@ -31,6 +31,7 @@ namespace Microsoft.eShopOnContainers.BuildingBlocks.EventBusRabbitMQ
= new List<Type>();
private IModel _consumerChannel;
private string _queueName;
public EventBusRabbitMQ(IRabbitMQPersisterConnection persisterConnection, ILogger<EventBusRabbitMQ> logger)
{
@ -93,7 +94,7 @@ namespace Microsoft.eShopOnContainers.BuildingBlocks.EventBusRabbitMQ
using (var channel = _persisterConnection.CreateModel())
{
channel.QueueBind(queue: channel.QueueDeclare().QueueName,
channel.QueueBind(queue: _queueName,
exchange: BROKER_NAME,
routingKey: eventName);
@ -131,9 +132,16 @@ namespace Microsoft.eShopOnContainers.BuildingBlocks.EventBusRabbitMQ
using (var channel = _persisterConnection.CreateModel())
{
channel.QueueUnbind(queue: channel.QueueDeclare().QueueName,
channel.QueueUnbind(queue: _queueName,
exchange: BROKER_NAME,
routingKey: eventName);
if (_handlers.Keys.Count == 0)
{
_queueName = string.Empty;
_consumerChannel.Close();
}
}
}
}
@ -142,7 +150,11 @@ namespace Microsoft.eShopOnContainers.BuildingBlocks.EventBusRabbitMQ
public void Dispose()
{
_consumerChannel.Dispose();
if (_consumerChannel != null)
{
_consumerChannel.Dispose();
}
_handlers.Clear();
}
@ -158,6 +170,8 @@ namespace Microsoft.eShopOnContainers.BuildingBlocks.EventBusRabbitMQ
channel.ExchangeDeclare(exchange: BROKER_NAME,
type: "direct");
_queueName = channel.QueueDeclare().QueueName;
var consumer = new EventingBasicConsumer(channel);
consumer.Received += async (model, ea) =>
{
@ -167,16 +181,10 @@ namespace Microsoft.eShopOnContainers.BuildingBlocks.EventBusRabbitMQ
await ProcessEvent(eventName, message);
};
channel.BasicConsume(queue: channel.QueueDeclare().QueueName,
channel.BasicConsume(queue: _queueName,
noAck: true,
consumer: consumer);
channel.ModelShutdown += (sender, ea) =>
{
_consumerChannel.Dispose();
_consumerChannel = CreateConsumerChannel();
};
channel.CallbackException += (sender, ea) =>
{
_consumerChannel.Dispose();