From ba4c0bd3d68dfcfc0e8e92c015904115f3f4408f Mon Sep 17 00:00:00 2001 From: maliming Date: Thu, 23 Apr 2026 11:59:59 +0800 Subject: [PATCH] feat: add RabbitMQ tests and improve channel pool handling --- framework/Volo.Abp.abpmdl | 4 + framework/Volo.Abp.slnx | 1 + .../Volo/Abp/RabbitMQ/ChannelPool.cs | 19 +- .../Volo.Abp.RabbitMQ.Tests.abppkg | 3 + .../Volo.Abp.RabbitMQ.Tests.csproj | 16 + .../Volo/Abp/RabbitMQ/ChannelPool_Tests.cs | 331 ++++++++++++++++++ 6 files changed, 369 insertions(+), 5 deletions(-) create mode 100644 framework/test/Volo.Abp.RabbitMQ.Tests/Volo.Abp.RabbitMQ.Tests.abppkg create mode 100644 framework/test/Volo.Abp.RabbitMQ.Tests/Volo.Abp.RabbitMQ.Tests.csproj create mode 100644 framework/test/Volo.Abp.RabbitMQ.Tests/Volo/Abp/RabbitMQ/ChannelPool_Tests.cs diff --git a/framework/Volo.Abp.abpmdl b/framework/Volo.Abp.abpmdl index afc52ca801..0935fedade 100644 --- a/framework/Volo.Abp.abpmdl +++ b/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" diff --git a/framework/Volo.Abp.slnx b/framework/Volo.Abp.slnx index 26d462fb4f..df59da0515 100644 --- a/framework/Volo.Abp.slnx +++ b/framework/Volo.Abp.slnx @@ -240,6 +240,7 @@ + diff --git a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ChannelPool.cs b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ChannelPool.cs index c06c0b6a95..424c1600ad 100644 --- a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ChannelPool.cs +++ b/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(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 { diff --git a/framework/test/Volo.Abp.RabbitMQ.Tests/Volo.Abp.RabbitMQ.Tests.abppkg b/framework/test/Volo.Abp.RabbitMQ.Tests/Volo.Abp.RabbitMQ.Tests.abppkg new file mode 100644 index 0000000000..64c1552e37 --- /dev/null +++ b/framework/test/Volo.Abp.RabbitMQ.Tests/Volo.Abp.RabbitMQ.Tests.abppkg @@ -0,0 +1,3 @@ +{ + "role": "lib.test" +} diff --git a/framework/test/Volo.Abp.RabbitMQ.Tests/Volo.Abp.RabbitMQ.Tests.csproj b/framework/test/Volo.Abp.RabbitMQ.Tests/Volo.Abp.RabbitMQ.Tests.csproj new file mode 100644 index 0000000000..0c4b2a2260 --- /dev/null +++ b/framework/test/Volo.Abp.RabbitMQ.Tests/Volo.Abp.RabbitMQ.Tests.csproj @@ -0,0 +1,16 @@ + + + + + + net10.0 + + + + + + + + + + diff --git a/framework/test/Volo.Abp.RabbitMQ.Tests/Volo/Abp/RabbitMQ/ChannelPool_Tests.cs b/framework/test/Volo.Abp.RabbitMQ.Tests/Volo/Abp/RabbitMQ/ChannelPool_Tests.cs new file mode 100644 index 0000000000..3fe52ae563 --- /dev/null +++ b/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 -------------------- + + /// + /// 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. + /// + [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}"); + } + + /// + /// 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. + /// + [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[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(); + channel.IsClosed.Returns(false); + + var connection = Substitute.For(); + var attempts = 0; + connection + .CreateChannelAsync(Arg.Any(), Arg.Any()) + .Returns(_ => + { + var n = Interlocked.Increment(ref attempts); + if (n == 1) + { + throw new InvalidOperationException("broker down"); + } + return Task.FromResult(channel); + }); + + var connectionPool = Substitute.For(); + connectionPool.GetAsync(Arg.Any()).Returns(Task.FromResult(connection)); + + var channelPool = new TestChannelPool(connectionPool); + + await Should.ThrowAsync(() => 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(() => 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[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(); + channel1.IsClosed.Returns(false); + + var channel2 = Substitute.For(); + channel2.IsClosed.Returns(false); + + var fixture = new ConnectionPoolFixture(channel1, channel2); + + var connection = Substitute.For(); + connection + .CreateChannelAsync(Arg.Any(), Arg.Any()) + .Returns(_ => + { + var n = Interlocked.Increment(ref fixture.CreateChannelCallsField); + return n == 1 ? Task.FromResult(channel1) : Task.FromResult(channel2); + }); + + var connectionPool = Substitute.For(); + connectionPool.GetAsync(Arg.Any()).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) + { + } + } +}