Merge pull request #34 from FoundatioFx/feature/rabbitmq7

Upgrade to RabbitMQ 7

Foundatio.RabbitMQ.sln

@@ -12,6 +12,7 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution
 		README.md = README.md
 		docker-compose.yml = docker-compose.yml
 		tests\Directory.Build.props = tests\Directory.Build.props
+		build\Dockerfile = build\Dockerfile
 	EndProjectSection
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Foundatio.RabbitMQ", "src\Foundatio.RabbitMQ\Foundatio.RabbitMQ.csproj", "{EAE3607D-73A1-4D02-BDAA-24A37DDA15CB}"

build/Dockerfile

@@ -1,4 +1,4 @@
-FROM rabbitmq:3.10-management
+FROM rabbitmq:4.0.3-management
 
-COPY rabbitmq_delayed_message_exchange-3.10.0.ez /opt/rabbitmq/plugins
+COPY rabbitmq_delayed_message_exchange-4.0.2.ez /opt/rabbitmq/plugins
 RUN rabbitmq-plugins enable rabbitmq_delayed_message_exchange

build/common.props

@@ -1,7 +1,7 @@
 <Project ToolsVersion="15.0">
 
   <PropertyGroup>
-    <TargetFrameworks>netstandard2.0</TargetFrameworks>
+    <TargetFrameworks>netstandard2.1</TargetFrameworks>
     <Product>Foundatio</Product>
     <Description>Pluggable foundation blocks for building distributed apps.</Description>
     <PackageProjectUrl>https://github.com/FoundatioFx/Foundatio.RabbitMQ</PackageProjectUrl>

docker-compose.yml

@@ -1,6 +1,6 @@
 services:
   rabbitmq:
-    image: rabbitmq:3.10-management
+    image: rabbitmq:4.0.3-management
     ports:
       - "5672:5672"
       - "8080:15672" # management ui - login with guest:guest

src/Foundatio.RabbitMQ/Extensions/TaskExtensions.cs

@@ -25,4 +25,10 @@ public static ConfiguredTaskAwaitable<TResult> AnyContext<TResult>(this Awaitabl
     {
         return task.ConfigureAwait(continueOnCapturedContext: false);
     }
+
+    [DebuggerStepThrough]
+    public static ConfiguredValueTaskAwaitable AnyContext(this ValueTask task)
+    {
+        return task.ConfigureAwait(continueOnCapturedContext: false);
+    }
 }

src/Foundatio.RabbitMQ/Foundatio.RabbitMQ.csproj

@@ -1,6 +1,6 @@
 <Project Sdk="Microsoft.NET.Sdk">
   <ItemGroup>
-    <PackageReference Include="RabbitMQ.Client" Version="6.8.1" />
+    <PackageReference Include="RabbitMQ.Client" Version="7.0.0" />
 
     <PackageReference Include="Foundatio" Version="11.0.5" Condition="'$(ReferenceFoundatioSource)' == '' OR '$(ReferenceFoundatioSource)' == 'false'" />
     <ProjectReference Include="..\..\..\Foundatio\src\Foundatio\Foundatio.csproj" Condition="'$(ReferenceFoundatioSource)' == 'true'" />

tests/Foundatio.RabbitMQ.Tests/Messaging/RabbitMqMessageBusDelayedExchangeTests.cs

@@ -0,0 +1,8 @@
+using Xunit.Abstractions;
+
+namespace Foundatio.RabbitMQ.Tests.Messaging;
+
+public class RabbitMqMessageBusDelayedExchangeTests : RabbitMqMessageBusTestBase
+{
+    public RabbitMqMessageBusDelayedExchangeTests(ITestOutputHelper output) : base("amqp://localhost:5673", output) { }
+}

tests/Foundatio.RabbitMQ.Tests/Messaging/RabbitMqMessageBusTestBase.cs

@@ -0,0 +1,203 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using Foundatio.AsyncEx;
+using Foundatio.Messaging;
+using Foundatio.Tests.Extensions;
+using Foundatio.Tests.Messaging;
+using Microsoft.Extensions.Logging;
+using Xunit;
+using Xunit.Abstractions;
+
+namespace Foundatio.RabbitMQ.Tests.Messaging;
+
+public abstract class RabbitMqMessageBusTestBase(string connectionString, ITestOutputHelper output) : MessageBusTestBase(output)
+{
+    private readonly string _topic = $"test_topic_{DateTime.UtcNow.Ticks}";
+
+    protected override IMessageBus GetMessageBus(Func<SharedMessageBusOptions, SharedMessageBusOptions> config = null)
+    {
+        return new RabbitMQMessageBus(o =>
+        {
+            o.ConnectionString(connectionString);
+            o.LoggerFactory(Log);
+
+            config?.Invoke(o.Target);
+
+            return o;
+        });
+    }
+
+    [Fact]
+    public override Task CanUseMessageOptionsAsync()
+    {
+        return base.CanUseMessageOptionsAsync();
+    }
+
+    [Fact]
+    public override Task CanSendMessageAsync()
+    {
+        return base.CanSendMessageAsync();
+    }
+
+    [Fact]
+    public override Task CanHandleNullMessageAsync()
+    {
+        return base.CanHandleNullMessageAsync();
+    }
+
+    [Fact]
+    public override Task CanSendDerivedMessageAsync()
+    {
+        return base.CanSendDerivedMessageAsync();
+    }
+
+    [Fact]
+    public override Task CanSendMappedMessageAsync()
+    {
+        return base.CanSendMappedMessageAsync();
+    }
+
+    [Fact]
+    public override Task CanSendDelayedMessageAsync()
+    {
+        return base.CanSendDelayedMessageAsync();
+    }
+
+    [Fact]
+    public override Task CanSubscribeConcurrentlyAsync()
+    {
+        return base.CanSubscribeConcurrentlyAsync();
+    }
+
+    [Fact]
+    public override Task CanReceiveMessagesConcurrentlyAsync()
+    {
+        return base.CanReceiveMessagesConcurrentlyAsync();
+    }
+
+    [Fact]
+    public override Task CanSendMessageToMultipleSubscribersAsync()
+    {
+        return base.CanSendMessageToMultipleSubscribersAsync();
+    }
+
+    [Fact]
+    public override Task CanTolerateSubscriberFailureAsync()
+    {
+        return base.CanTolerateSubscriberFailureAsync();
+    }
+
+    [Fact]
+    public override Task WillOnlyReceiveSubscribedMessageTypeAsync()
+    {
+        return base.WillOnlyReceiveSubscribedMessageTypeAsync();
+    }
+
+    [Fact]
+    public override Task WillReceiveDerivedMessageTypesAsync()
+    {
+        return base.WillReceiveDerivedMessageTypesAsync();
+    }
+
+    [Fact]
+    public override Task CanSubscribeToAllMessageTypesAsync()
+    {
+        return base.CanSubscribeToAllMessageTypesAsync();
+    }
+
+    [Fact]
+    public override Task CanSubscribeToRawMessagesAsync()
+    {
+        return base.CanSubscribeToRawMessagesAsync();
+    }
+
+    [Fact]
+    public override Task CanCancelSubscriptionAsync()
+    {
+        return base.CanCancelSubscriptionAsync();
+    }
+
+    [Fact]
+    public override Task WontKeepMessagesWithNoSubscribersAsync()
+    {
+        return base.WontKeepMessagesWithNoSubscribersAsync();
+    }
+
+    [Fact]
+    public override Task CanReceiveFromMultipleSubscribersAsync()
+    {
+        return base.CanReceiveFromMultipleSubscribersAsync();
+    }
+
+    [Fact]
+    public override void CanDisposeWithNoSubscribersOrPublishers()
+    {
+        base.CanDisposeWithNoSubscribersOrPublishers();
+    }
+
+    [Fact]
+    public async Task CanPersistAndNotLoseMessages()
+    {
+        var messageBus1 = new RabbitMQMessageBus(o => o
+            .ConnectionString(connectionString)
+            .LoggerFactory(Log)
+            .SubscriptionQueueName($"{_topic}-offline")
+            .IsSubscriptionQueueExclusive(false)
+            .SubscriptionQueueAutoDelete(false)
+            .AcknowledgementStrategy(AcknowledgementStrategy.Automatic));
+
+        var countdownEvent = new AsyncCountdownEvent(1);
+        var cts = new CancellationTokenSource();
+        await messageBus1.SubscribeAsync<SimpleMessageA>(msg =>
+        {
+            _logger.LogInformation("[Subscriber1] Got message: {Message}", msg.Data);
+            countdownEvent.Signal();
+        }, cts.Token);
+
+        await messageBus1.PublishAsync(new SimpleMessageA { Data = "Audit message 1" });
+        await countdownEvent.WaitAsync(TimeSpan.FromSeconds(5));
+        Assert.Equal(0, countdownEvent.CurrentCount);
+        await cts.CancelAsync();
+
+        await messageBus1.PublishAsync(new SimpleMessageA { Data = "Audit message 2" });
+
+        cts = new CancellationTokenSource();
+        countdownEvent.AddCount(1);
+        await messageBus1.SubscribeAsync<SimpleMessageA>(msg =>
+        {
+            _logger.LogInformation("[Subscriber2] Got message: {Message}", msg.Data);
+            countdownEvent.Signal();
+        }, cts.Token);
+        await countdownEvent.WaitAsync(TimeSpan.FromSeconds(5));
+        Assert.Equal(0, countdownEvent.CurrentCount);
+        await cts.CancelAsync();
+
+        await messageBus1.PublishAsync(new SimpleMessageA { Data = "Audit offline message 1" });
+        await messageBus1.PublishAsync(new SimpleMessageA { Data = "Audit offline message 2" });
+        await messageBus1.PublishAsync(new SimpleMessageA { Data = "Audit offline message 3" });
+
+        await messageBus1.DisposeAsync();
+
+        var messageBus2 = new RabbitMQMessageBus(o => o
+            .ConnectionString(connectionString)
+            .LoggerFactory(Log)
+            .SubscriptionQueueName($"{_topic}-offline")
+            .IsSubscriptionQueueExclusive(false)
+            .SubscriptionQueueAutoDelete(false)
+            .AcknowledgementStrategy(AcknowledgementStrategy.Automatic));
+
+        cts = new CancellationTokenSource();
+        countdownEvent.AddCount(4);
+        await messageBus2.SubscribeAsync<SimpleMessageA>(msg =>
+        {
+            _logger.LogInformation("[Subscriber3] Got message: {Message}", msg.Data);
+            countdownEvent.Signal();
+        }, cts.Token);
+        await messageBus2.PublishAsync(new SimpleMessageA { Data = "Another audit message 4" });
+        await countdownEvent.WaitAsync(TimeSpan.FromSeconds(5));
+        Assert.Equal(0, countdownEvent.CurrentCount);
+
+        await messageBus2.DisposeAsync();
+    }
+}