Merge pull request #987 from dotnet-architecture/fix/888-rabbitmq-message-processing-problem

Use AsyncEventingBasicConsumer in RabbitMQ

Author: mvelosop

src/BuildingBlocks/EventBus/EventBusRabbitMQ/EventBusRabbitMQ.cs

@@ -178,27 +178,46 @@ private void StartBasicConsume()
         {
             if (_consumerChannel != null)
             {
-                var consumer = new EventingBasicConsumer(_consumerChannel);
-                consumer.Received += async (model, ea) =>
-                {
-                    var eventName = ea.RoutingKey;
-                    var message = Encoding.UTF8.GetString(ea.Body);
-
-                    await ProcessEvent(eventName, message);
+                var consumer = new AsyncEventingBasicConsumer(_consumerChannel);
 
-                    _consumerChannel.BasicAck(ea.DeliveryTag, multiple: false);
-                };
+                consumer.Received += Consumer_Received;
 
-                _consumerChannel.BasicConsume(queue: _queueName,
-                                     autoAck: false,
-                                     consumer: consumer);
+                _consumerChannel.BasicConsume(
+                    queue: _queueName,
+                    autoAck: false,
+                    consumer: consumer);
             }
             else
             {
-                _logger.LogError("StartBasicConsume can not call on _consumerChannelCreated == false");
+                _logger.LogError("StartBasicConsume can't call on _consumerChannel == null");
             }
         }
 
+        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);
+        }
+
         private IModel CreateConsumerChannel()
         {
             if (!_persistentConnection.IsConnected)
@@ -209,7 +228,7 @@ private IModel CreateConsumerChannel()
             var channel = _persistentConnection.CreateModel();
 
             channel.ExchangeDeclare(exchange: BROKER_NAME,
-                                 type: "direct");
+                                    type: "direct");
 
             channel.QueueDeclare(queue: _queueName,
                                  durable: true,

src/Services/Basket/Basket.API/Startup.cs

@@ -105,7 +105,8 @@ public IServiceProvider ConfigureServices(IServiceCollection services)
 
                     var factory = new ConnectionFactory()
                     {
-                        HostName = Configuration["EventBusConnection"]
+                        HostName = Configuration["EventBusConnection"],
+                        DispatchConsumersAsync = true
                     };
 
                     if (!string.IsNullOrEmpty(Configuration["EventBusUserName"]))

src/Services/Catalog/Catalog.API/Startup.cs

@@ -297,7 +297,8 @@ public static IServiceCollection AddIntegrationServices(this IServiceCollection
 
                     var factory = new ConnectionFactory()
                     {
-                        HostName = configuration["EventBusConnection"]
+                        HostName = configuration["EventBusConnection"],
+                        DispatchConsumersAsync = true
                     };
 
                     if (!string.IsNullOrEmpty(configuration["EventBusUserName"]))

src/Services/Location/Locations.API/Startup.cs

@@ -77,7 +77,8 @@ public IServiceProvider ConfigureServices(IServiceCollection services)
 
                     var factory = new ConnectionFactory()
                     {
-                        HostName = Configuration["EventBusConnection"]
+                        HostName = Configuration["EventBusConnection"],
+                        DispatchConsumersAsync = true
                     };
 
                     if (!string.IsNullOrEmpty(Configuration["EventBusUserName"]))

src/Services/Marketing/Marketing.API/Startup.cs

@@ -101,7 +101,8 @@ public IServiceProvider ConfigureServices(IServiceCollection services)
 
                     var factory = new ConnectionFactory()
                     {
-                        HostName = Configuration["EventBusConnection"]
+                        HostName = Configuration["EventBusConnection"],
+                        DispatchConsumersAsync = true
                     };
 
                     if (!string.IsNullOrEmpty(Configuration["EventBusUserName"]))

src/Services/Ordering/Ordering.API/Startup.cs

@@ -305,7 +305,8 @@ public static IServiceCollection AddCustomIntegrations(this IServiceCollection s
 
                     var factory = new ConnectionFactory()
                     {
-                        HostName = configuration["EventBusConnection"]
+                        HostName = configuration["EventBusConnection"],
+                        DispatchConsumersAsync = true
                     };
 
                     if (!string.IsNullOrEmpty(configuration["EventBusUserName"]))

src/Services/Ordering/Ordering.BackgroundTasks/Startup.cs

@@ -68,7 +68,8 @@ public IServiceProvider ConfigureServices(IServiceCollection services)
 
                     var factory = new ConnectionFactory()
                     {
-                        HostName = Configuration["EventBusConnection"]
+                        HostName = Configuration["EventBusConnection"],
+                        DispatchConsumersAsync = true
                     };
 
                     if (!string.IsNullOrEmpty(Configuration["EventBusUserName"]))

src/Services/Ordering/Ordering.SignalrHub/Startup.cs

@@ -80,7 +80,8 @@ public IServiceProvider ConfigureServices(IServiceCollection services)
 
                     var factory = new ConnectionFactory()
                     {
-                        HostName = Configuration["EventBusConnection"]
+                        HostName = Configuration["EventBusConnection"],
+                        DispatchConsumersAsync = true
                     };
 
                     if (!string.IsNullOrEmpty(Configuration["EventBusUserName"]))

src/Services/Payment/Payment.API/Startup.cs

@@ -58,7 +58,8 @@ public IServiceProvider ConfigureServices(IServiceCollection services)
                     var logger = sp.GetRequiredService<ILogger<DefaultRabbitMQPersistentConnection>>();
                     var factory = new ConnectionFactory()
                     {
-                        HostName = Configuration["EventBusConnection"]
+                        HostName = Configuration["EventBusConnection"],
+                        DispatchConsumersAsync = true
                     };
 
                     if (!string.IsNullOrEmpty(Configuration["EventBusUserName"]))

src/Services/Webhooks/Webhooks.API/Startup.cs

@@ -320,7 +320,8 @@ public static IServiceCollection AddIntegrationServices(this IServiceCollection
 
                     var factory = new ConnectionFactory()
                     {
-                        HostName = configuration["EventBusConnection"]
+                        HostName = configuration["EventBusConnection"],
+                        DispatchConsumersAsync = true
                     };
 
                     if (!string.IsNullOrEmpty(configuration["EventBusUserName"]))