feat: add RabbitMQ tests and improve channel pool handling

framework/Volo.Abp.abpmdl

@@ -478,6 +478,10 @@
       "path": "test/Volo.Abp.ObjectMapping.Tests/Volo.Abp.ObjectMapping.Tests.abppkg",
       "folder": "test"
     },
+    "Volo.Abp.RabbitMQ.Tests": {
+      "path": "test/Volo.Abp.RabbitMQ.Tests/Volo.Abp.RabbitMQ.Tests.abppkg",
+      "folder": "test"
+    },
     "Volo.Abp.Ddd.Application.Contracts": {
       "path": "src/Volo.Abp.Ddd.Application.Contracts/Volo.Abp.Ddd.Application.Contracts.abppkg",
       "folder": "src"

framework/Volo.Abp.slnx

@@ -240,6 +240,7 @@
     <Project Path="test/Volo.Abp.MultiTenancy.Tests/Volo.Abp.MultiTenancy.Tests.csproj" />
     <Project Path="test/Volo.Abp.ObjectExtending.Tests/Volo.Abp.ObjectExtending.Tests.csproj" />
     <Project Path="test/Volo.Abp.ObjectMapping.Tests/Volo.Abp.ObjectMapping.Tests.csproj" />
+    <Project Path="test/Volo.Abp.RabbitMQ.Tests/Volo.Abp.RabbitMQ.Tests.csproj" />
     <Project Path="test/Volo.Abp.RemoteServices.Tests/Volo.Abp.RemoteServices.Tests.csproj" />
     <Project Path="test/Volo.Abp.Security.Tests/Volo.Abp.Security.Tests.csproj" />
     <Project Path="test/Volo.Abp.Serialization.Tests/Volo.Abp.Serialization.Tests.csproj" />

framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ChannelPool.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Concurrent;
+using System.Collections.Generic;
 using System.Diagnostics;
 using System.Linq;
 using System.Threading;
@@ -63,14 +64,22 @@ public class ChannelPool : IChannelPool, ISingletonDependency
 
         if (poolItem.Channel.IsClosed)
         {
-            await poolItem.DisposeAsync();
-            Channels.TryRemove(channelName, out _);
-
             using (await Semaphore.LockAsync())
             {
-                if (Channels.TryGetValue(channelName, out var existingChannelPoolItem3))
+                if (Channels.TryGetValue(channelName, out var currentChannelPoolItem) &&
+                    ReferenceEquals(currentChannelPoolItem, poolItem))
+                {
+                    // This caller is the first to notice the cached channel is closed; rebuild it.
+                    await poolItem.DisposeAsync();
+                    Channels.TryRemove(new KeyValuePair<string, ChannelPoolItem>(channelName, poolItem));
+
+                    poolItem = new ChannelPoolItem(await CreateChannelAsync(channelName, connectionName));
+                    Channels.TryAdd(channelName, poolItem);
+                }
+                else if (currentChannelPoolItem != null)
                 {
-                    poolItem = existingChannelPoolItem3;
+                    // Another caller already rebuilt the channel; use the fresh one.
+                    poolItem = currentChannelPoolItem;
                 }
                 else
                 {

framework/test/Volo.Abp.RabbitMQ.Tests/Volo.Abp.RabbitMQ.Tests.abppkg

@@ -0,0 +1,3 @@
+{
+  "role": "lib.test"
+}

framework/test/Volo.Abp.RabbitMQ.Tests/Volo.Abp.RabbitMQ.Tests.csproj

@@ -0,0 +1,16 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <Import Project="..\..\..\common.test.props" />
+
+  <PropertyGroup>
+    <TargetFramework>net10.0</TargetFramework>
+    <RootNamespace />
+  </PropertyGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\AbpTestBase\AbpTestBase.csproj" />
+    <ProjectReference Include="..\..\src\Volo.Abp.RabbitMQ\Volo.Abp.RabbitMQ.csproj" />
+    <PackageReference Include="Microsoft.NET.Test.Sdk" />
+  </ItemGroup>
+
+</Project>

framework/test/Volo.Abp.RabbitMQ.Tests/Volo/Abp/RabbitMQ/ChannelPool_Tests.cs

@@ -0,0 +1,331 @@
+#nullable enable
+
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using NSubstitute;
+using RabbitMQ.Client;
+using Shouldly;
+using Xunit;
+
+namespace Volo.Abp.RabbitMQ;
+
+public class ChannelPool_Tests
+{
+    private static readonly TimeSpan RaceTimeout = TimeSpan.FromSeconds(5);
+
+    // ---- Regression tests for issue #25310 / PR #25311 --------------------
+
+    /// <summary>
+    /// After a RabbitMQ restart the cached channel becomes closed. Two callers
+    /// that both observe the stale poolItem must not deadlock while the pool
+    /// replaces it with a fresh channel.
+    /// </summary>
+    [Fact]
+    public async Task AcquireAsync_Should_Not_Hang_When_Channel_Closed_With_Concurrent_Callers()
+    {
+        var fixture = BuildConnectionPool();
+        var channelPool = new TestChannelPool(fixture.Pool);
+
+        using (await channelPool.AcquireAsync("q"))
+        {
+        }
+
+        fixture.Channel1.IsClosed.Returns(true);
+        fixture.Channel1
+            .When(x => x.DisposeAsync())
+            .Do(_ => Thread.Sleep(300));
+
+        using var barrier = new Barrier(2);
+
+        var firstCaller = Task.Run(async () =>
+        {
+            barrier.SignalAndWait();
+            using var accessor = await channelPool.AcquireAsync("q");
+        });
+
+        var secondCaller = Task.Run(async () =>
+        {
+            barrier.SignalAndWait();
+            await Task.Delay(50);
+            using var accessor = await channelPool.AcquireAsync("q");
+        });
+
+        var completed = Task.WhenAll(firstCaller, secondCaller);
+        var winner = await Task.WhenAny(completed, Task.Delay(RaceTimeout));
+
+        winner.ShouldBe(
+            completed,
+            $"AcquireAsync is hanging on a stale poolItem. firstCaller={firstCaller.Status}, secondCaller={secondCaller.Status}");
+    }
+
+    /// <summary>
+    /// When several callers race through the closed-channel recovery path, only one new
+    /// channel must be created and the dictionary must keep exactly the channel that callers
+    /// receive — otherwise the replacement leaks on the broker side.
+    /// </summary>
+    [Fact]
+    public async Task AcquireAsync_Should_Create_Only_One_Replacement_Channel_When_Many_Callers_Race()
+    {
+        var fixture = BuildConnectionPool();
+        var channelPool = new TestChannelPool(fixture.Pool);
+
+        using (await channelPool.AcquireAsync("q"))
+        {
+        }
+
+        fixture.Channel1.IsClosed.Returns(true);
+
+        const int callerCount = 8;
+        using var barrier = new Barrier(callerCount);
+        var tasks = new Task<IChannel>[callerCount];
+        for (var i = 0; i < callerCount; i++)
+        {
+            tasks[i] = Task.Run(async () =>
+            {
+                barrier.SignalAndWait();
+                using var accessor = await channelPool.AcquireAsync("q");
+                return accessor.Channel;
+            });
+        }
+
+        var channels = await Task.WhenAll(tasks);
+
+        foreach (var channel in channels)
+        {
+            channel.ShouldBe(fixture.Channel2, "every caller must receive the single replacement channel");
+        }
+
+        fixture.CreateChannelCalls.ShouldBe(
+            2,
+            $"exactly one initial + one replacement channel should be created, but got {fixture.CreateChannelCalls}");
+    }
+
+    // ---- Behavior guard-rails: make sure the refactor preserves core semantics ----
+
+    [Fact]
+    public async Task AcquireAsync_Should_Return_Channel_From_ConnectionPool()
+    {
+        var fixture = BuildConnectionPool();
+        var channelPool = new TestChannelPool(fixture.Pool);
+
+        using var accessor = await channelPool.AcquireAsync("q");
+
+        accessor.Channel.ShouldBe(fixture.Channel1);
+        accessor.Name.ShouldBe("q");
+        fixture.CreateChannelCalls.ShouldBe(1);
+    }
+
+    [Fact]
+    public async Task AcquireAsync_Should_Reuse_Cached_Channel_On_Subsequent_Calls()
+    {
+        var fixture = BuildConnectionPool();
+        var channelPool = new TestChannelPool(fixture.Pool);
+
+        using (var first = await channelPool.AcquireAsync("q"))
+        {
+            first.Channel.ShouldBe(fixture.Channel1);
+        }
+        using (var second = await channelPool.AcquireAsync("q"))
+        {
+            second.Channel.ShouldBe(fixture.Channel1);
+        }
+        using (var third = await channelPool.AcquireAsync("q"))
+        {
+            third.Channel.ShouldBe(fixture.Channel1);
+        }
+
+        fixture.CreateChannelCalls.ShouldBe(1, "a healthy cached channel must be reused");
+    }
+
+    [Fact]
+    public async Task AcquireAsync_Should_Keep_Separate_PoolItems_For_Different_ChannelNames()
+    {
+        var fixture = BuildConnectionPool();
+        var channelPool = new TestChannelPool(fixture.Pool);
+
+        using var a = await channelPool.AcquireAsync("queue-a");
+        using var b = await channelPool.AcquireAsync("queue-b");
+
+        a.Channel.ShouldBe(fixture.Channel1);
+        b.Channel.ShouldBe(fixture.Channel2);
+        a.Name.ShouldBe("queue-a");
+        b.Name.ShouldBe("queue-b");
+        fixture.CreateChannelCalls.ShouldBe(2);
+    }
+
+    [Fact]
+    public async Task AcquireAsync_Should_Serialize_Concurrent_Callers_On_Same_Channel()
+    {
+        var fixture = BuildConnectionPool();
+        var channelPool = new TestChannelPool(fixture.Pool);
+
+        // First caller holds the channel for a while.
+        var first = await channelPool.AcquireAsync("q");
+        var firstAcquiredAt = DateTime.UtcNow;
+
+        // Second caller should block until the first one releases.
+        var secondTask = Task.Run(async () =>
+        {
+            var accessor = await channelPool.AcquireAsync("q");
+            return (accessor, acquiredAt: DateTime.UtcNow);
+        });
+
+        await Task.Delay(100);
+        secondTask.IsCompleted.ShouldBeFalse("second caller must block while the channel is held");
+
+        first.Dispose();
+        var firstReleasedAt = DateTime.UtcNow;
+
+        var completed = await Task.WhenAny(secondTask, Task.Delay(RaceTimeout));
+        completed.ShouldBe(secondTask, "second caller must be unblocked after release");
+
+        var (secondAccessor, secondAcquiredAt) = await secondTask;
+        secondAccessor.Channel.ShouldBe(fixture.Channel1, "the cached channel is reused");
+        secondAcquiredAt.ShouldBeGreaterThanOrEqualTo(firstReleasedAt);
+
+        secondAccessor.Dispose();
+        _ = firstAcquiredAt;
+        fixture.CreateChannelCalls.ShouldBe(1, "channel is reused, never recreated");
+    }
+
+    [Fact]
+    public async Task AcquireAsync_Should_Not_Rebuild_When_Cached_Channel_Is_Healthy()
+    {
+        var fixture = BuildConnectionPool();
+        var channelPool = new TestChannelPool(fixture.Pool);
+
+        using (await channelPool.AcquireAsync("q"))
+        {
+        }
+        // Channel stays open, so subsequent calls must not go through the rebuild branch.
+        using (await channelPool.AcquireAsync("q"))
+        {
+        }
+
+        await fixture.Channel1.DidNotReceive().DisposeAsync();
+        fixture.CreateChannelCalls.ShouldBe(1);
+    }
+
+    [Fact]
+    public async Task AcquireAsync_Should_Propagate_Exception_When_CreateChannel_Fails()
+    {
+        var channel = Substitute.For<IChannel>();
+        channel.IsClosed.Returns(false);
+
+        var connection = Substitute.For<IConnection>();
+        var attempts = 0;
+        connection
+            .CreateChannelAsync(Arg.Any<CreateChannelOptions?>(), Arg.Any<CancellationToken>())
+            .Returns(_ =>
+            {
+                var n = Interlocked.Increment(ref attempts);
+                if (n == 1)
+                {
+                    throw new InvalidOperationException("broker down");
+                }
+                return Task.FromResult(channel);
+            });
+
+        var connectionPool = Substitute.For<IConnectionPool>();
+        connectionPool.GetAsync(Arg.Any<string?>()).Returns(Task.FromResult(connection));
+
+        var channelPool = new TestChannelPool(connectionPool);
+
+        await Should.ThrowAsync<InvalidOperationException>(() => channelPool.AcquireAsync("q"));
+
+        // Subsequent call should be able to succeed (broker came back).
+        using var accessor = await channelPool.AcquireAsync("q");
+        accessor.Channel.ShouldBe(channel);
+        attempts.ShouldBe(2);
+    }
+
+    [Fact]
+    public async Task AcquireAsync_Should_Throw_After_Pool_Disposed()
+    {
+        var fixture = BuildConnectionPool();
+        var channelPool = new TestChannelPool(fixture.Pool);
+
+        await channelPool.DisposeAsync();
+
+        await Should.ThrowAsync<ObjectDisposedException>(() => channelPool.AcquireAsync("q"));
+    }
+
+    [Fact]
+    public async Task AcquireAsync_Should_Only_Create_One_Channel_Even_When_First_Callers_Race()
+    {
+        var fixture = BuildConnectionPool();
+        var channelPool = new TestChannelPool(fixture.Pool);
+
+        const int callerCount = 8;
+        using var barrier = new Barrier(callerCount);
+        var tasks = new Task<IChannel>[callerCount];
+        for (var i = 0; i < callerCount; i++)
+        {
+            tasks[i] = Task.Run(async () =>
+            {
+                barrier.SignalAndWait();
+                using var accessor = await channelPool.AcquireAsync("q");
+                return accessor.Channel;
+            });
+        }
+
+        var channels = await Task.WhenAll(tasks);
+
+        foreach (var channel in channels)
+        {
+            channel.ShouldBe(fixture.Channel1, "the first-creation semaphore must serialize initial creation");
+        }
+        fixture.CreateChannelCalls.ShouldBe(1);
+    }
+
+    // ---- Fixture ---------------------------------------------------------
+
+    private static ConnectionPoolFixture BuildConnectionPool()
+    {
+        var channel1 = Substitute.For<IChannel>();
+        channel1.IsClosed.Returns(false);
+
+        var channel2 = Substitute.For<IChannel>();
+        channel2.IsClosed.Returns(false);
+
+        var fixture = new ConnectionPoolFixture(channel1, channel2);
+
+        var connection = Substitute.For<IConnection>();
+        connection
+            .CreateChannelAsync(Arg.Any<CreateChannelOptions?>(), Arg.Any<CancellationToken>())
+            .Returns(_ =>
+            {
+                var n = Interlocked.Increment(ref fixture.CreateChannelCallsField);
+                return n == 1 ? Task.FromResult(channel1) : Task.FromResult(channel2);
+            });
+
+        var connectionPool = Substitute.For<IConnectionPool>();
+        connectionPool.GetAsync(Arg.Any<string?>()).Returns(Task.FromResult(connection));
+
+        fixture.Pool = connectionPool;
+        return fixture;
+    }
+
+    private sealed class ConnectionPoolFixture
+    {
+        public IConnectionPool Pool { get; set; } = default!;
+        public IChannel Channel1 { get; }
+        public IChannel Channel2 { get; }
+        public int CreateChannelCallsField;
+        public int CreateChannelCalls => CreateChannelCallsField;
+
+        public ConnectionPoolFixture(IChannel channel1, IChannel channel2)
+        {
+            Channel1 = channel1;
+            Channel2 = channel2;
+        }
+    }
+
+    private sealed class TestChannelPool : ChannelPool
+    {
+        public TestChannelPool(IConnectionPool connectionPool) : base(connectionPool)
+        {
+        }
+    }
+}