Fix duplicate consumption of events when connection is broken

abpframework · Feb 7, 2025 · de10dcc · de10dcc
1 parent 7b293c3
commit de10dcc
Show file tree

Hide file tree

Showing 3 changed files with 27 additions and 14 deletions.
diff --git a/framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/JobQueue.cs b/framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/JobQueue.cs
@@ -122,7 +122,7 @@ public virtual void Dispose()
 
     protected virtual Task EnsureInitializedAsync()
     {
-        if (ChannelAccessor != null)
+        if (ChannelAccessor != null && ChannelAccessor.Channel.IsOpen)
         {
             return Task.CompletedTask;
         }

diff --git a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/AbpRabbitMqModule.cs b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/AbpRabbitMqModule.cs
@@ -20,6 +20,7 @@ public override void ConfigureServices(ServiceConfigurationContext context)
             foreach (var connectionFactory in options.Connections.Values)
             {
                 connectionFactory.DispatchConsumersAsync = true;
+                connectionFactory.AutomaticRecoveryEnabled = false;
             }
         });
     }

diff --git a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ConnectionPool.cs b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ConnectionPool.cs
@@ -24,21 +24,19 @@ public ConnectionPool(IOptions<AbpRabbitMqOptions> options)
     public virtual IConnection Get(string? connectionName = null)
     {
         connectionName ??= RabbitMqConnections.DefaultConnectionName;
-
+        var connectionFactory = Options.Connections.GetOrDefault(connectionName);
         try
         {
-            var lazyConnection = Connections.GetOrAdd(
-                connectionName, () => new Lazy<IConnection>(() =>
-                {
-                    var connection = Options.Connections.GetOrDefault(connectionName);
-                    var hostnames = connection.HostName.TrimEnd(';').Split(';');
-                        // Handle Rabbit MQ Cluster.
-                        return hostnames.Length == 1 ? connection.CreateConnection() : connection.CreateConnection(hostnames);
-
-                })
-            );
-
-            return lazyConnection.Value;
+            var connection = GetConnection(connectionName, connectionFactory);
+
+            if (connection.IsOpen)
+            {
+                return connection;
+            }
+
+            connection.Dispose();
+            Connections.TryRemove(connectionName, out _);
+            return GetConnection(connectionName, connectionFactory);
         }
         catch (Exception)
         {
@@ -47,6 +45,20 @@ public virtual IConnection Get(string? connectionName = null)
         }
     }
 
+    protected virtual IConnection GetConnection(string connectionName, ConnectionFactory connectionFactory)
+    {
+        return Connections.GetOrAdd(
+            connectionName, () => new Lazy<IConnection>(() =>
+            {
+                var hostnames = connectionFactory.HostName.TrimEnd(';').Split(';');
+                // Handle Rabbit MQ Cluster.
+                return hostnames.Length == 1
+                    ? connectionFactory.CreateConnection()
+                    : connectionFactory.CreateConnection(hostnames);
+            })
+        ).Value;
+    }
+
     public void Dispose()
     {
         if (_isDisposed)
-Original file line number
+Diff line change
@@ Expand Up @@
                 foreach (var connectionFactory in options.Connections.Values)
                 {
                     connectionFactory.DispatchConsumersAsync = true;
+                    connectionFactory.AutomaticRecoveryEnabled = false;
                 }
             });
         }
@@ Expand Down @@