Add auto-create-topics support to RabbitMQ provider as well.

runemalm · runemalm · commit cb26cd23d759 · 2025-03-08T08:51:33.000+07:00
diff --git a/docs/configuration.rst b/docs/configuration.rst
@@ -49,7 +49,8 @@ An example configuration in `appsettings.json`:
          "Port": 5672,
          "Username": "guest",
          "Password": "guest",
-         "VirtualHost": "/"
+         "VirtualHost": "/",
+         "AutoCreateTopics": true
        },
        "Kafka": {
          "BootstrapServers": "localhost:9092",
@@ -141,14 +142,18 @@ OpenDDD.NET supports multiple messaging providers:
        port: 5672,
        username: "guest",
        password: "guest",
-       virtualHost: "/"
+       virtualHost: "/",
+       autoCreateTopics: true
    );
 
 **Kafka**:
 
 .. code-block:: csharp
 
-   options.UseKafka("localhost:9092");
+   options.UseKafka(
+       "localhost:9092",
+       autoCreateTopics: true
+   );
 
 **Azure Service Bus**:
 
diff --git a/docs/userguide.rst b/docs/userguide.rst
@@ -599,7 +599,8 @@ Add the following configuration to your `appsettings.json` file to customize Ope
         "Port": 5672,
         "Username": "guest",
         "Password": "guest",
-        "VirtualHost": "/"
+        "VirtualHost": "/",
+        "AutoCreateTopics": true
       },
       "Kafka": {
         "BootstrapServers": "localhost:9092",
diff --git a/samples/Bookstore/src/Bookstore/appsettings.json b/samples/Bookstore/src/Bookstore/appsettings.json
@@ -32,7 +32,8 @@
       "Port": 5672,
       "Username": "guest",
       "Password": "guest",
-      "VirtualHost": "/"
+      "VirtualHost": "/",
+      "AutoCreateTopics": true
     },
     "Kafka": {
       "BootstrapServers": "localhost:9092",
diff --git a/src/OpenDDD.Tests/Integration/Infrastructure/Events/RabbitMq/RabbitMqMessagingProviderTests.cs b/src/OpenDDD.Tests/Integration/Infrastructure/Events/RabbitMq/RabbitMqMessagingProviderTests.cs
@@ -1,4 +1,5 @@
 ﻿using System.Collections.Concurrent;
+using FluentAssertions;
 using Microsoft.Extensions.Logging;
 using Xunit.Abstractions;
 using OpenDDD.Infrastructure.Events.RabbitMq;
@@ -38,7 +39,11 @@ public RabbitMqMessagingProviderTests(ITestOutputHelper testOutputHelper)
             };
 
             _consumerFactory = new RabbitMqConsumerFactory(_logger);
-            _messagingProvider = new RabbitMqMessagingProvider(_connectionFactory, _consumerFactory, _logger);
+            _messagingProvider = new RabbitMqMessagingProvider(
+                _connectionFactory, 
+                _consumerFactory, 
+                autoCreateTopics: true,
+                _logger);
         }
 
         public async Task InitializeAsync()
@@ -126,29 +131,65 @@ private async Task CleanupExchangesAndQueuesAsync()
         public async Task AutoCreateTopic_ShouldCreateTopicOnSubscribe_WhenSettingEnabled()
         {
             // Arrange
-            await VerifyExchangeAndQueueDoNotExist();
+            var topicName = $"test-topic-{Guid.NewGuid()}";
+            var consumerGroup = "test-consumer-group";
+
+            var messagingProvider = new RabbitMqMessagingProvider(
+                _connectionFactory,
+                _consumerFactory,
+                autoCreateTopics: true, // Auto-create enabled
+                _logger);
+
+            var exchangeExistsBefore = await ExchangeExistsAsync(topicName, _cts.Token);
+            exchangeExistsBefore.Should().BeFalse("The exchange should not exist before subscribing.");
 
             // Act
-            await _messagingProvider.SubscribeAsync(_testTopic, _testConsumerGroup, (msg, token) => Task.CompletedTask);
+            await messagingProvider.SubscribeAsync(topicName, consumerGroup, async (msg, token) =>
+                await Task.CompletedTask, _cts.Token);
+            
+            var timeout = TimeSpan.FromSeconds(30);
+            var pollingInterval = TimeSpan.FromMilliseconds(500);
+            var startTime = DateTime.UtcNow;
 
-            // Assert
-            try
+            bool exchangeExists = false;
+            while (DateTime.UtcNow - startTime < timeout)
             {
-                await _channel!.ExchangeDeclarePassiveAsync(_testTopic, CancellationToken.None);
-            }
-            catch (OperationInterruptedException)
-            {
-                Assert.Fail($"Exchange '{_testTopic}' does not exist.");
+                if (await ExchangeExistsAsync(topicName, _cts.Token))
+                {
+                    exchangeExists = true;
+                    break;
+                }
+                await Task.Delay(pollingInterval, _cts.Token);
             }
 
-            try
-            {
-                await _channel!.QueueDeclarePassiveAsync($"{_testConsumerGroup}.{_testTopic}", CancellationToken.None);
-            }
-            catch (OperationInterruptedException)
+            // Assert
+            exchangeExists.Should().BeTrue("RabbitMQ should create the exchange automatically when subscribing.");
+        }
+
+        [Fact]
+        public async Task AutoCreateTopic_ShouldNotCreateTopicOnSubscribe_WhenSettingDisabled()
+        {
+            // Arrange
+            var topicName = $"test-topic-{Guid.NewGuid()}";
+            var consumerGroup = "test-consumer-group";
+
+            var messagingProvider = new RabbitMqMessagingProvider(
+                _connectionFactory,
+                _consumerFactory,
+                autoCreateTopics: false, // Auto-create disabled
+                _logger);
+            
+            var exchangeExistsBefore = await ExchangeExistsAsync(topicName, _cts.Token);
+            exchangeExistsBefore.Should().BeFalse("The exchange should not exist before subscribing.");
+
+            // Act & Assert
+            var exception = await Assert.ThrowsAsync<InvalidOperationException>(async () =>
             {
-                Assert.Fail($"Queue '{_testConsumerGroup}.{_testTopic}' does not exist.");
-            }
+                await messagingProvider.SubscribeAsync(topicName, consumerGroup, async (msg, token) =>
+                    await Task.CompletedTask, _cts.Token);
+            });
+
+            exception.Message.Should().Contain($"Topic '{topicName}' does not exist.");
         }
         
         [Fact]
@@ -357,5 +398,27 @@ async Task MessageHandler(string msg, CancellationToken token)
                 Assert.InRange(count, minAllowed, maxAllowed);
             }
         }
+        
+        private async Task<bool> ExchangeExistsAsync(string exchange, CancellationToken cancellationToken)
+        {
+            try
+            {
+                // Use a temporary channel to check exchange existence, since old one might have stale topic data
+                using var tempChannel = await _connection!.CreateChannelAsync(null, cancellationToken);
+                await tempChannel.ExchangeDeclarePassiveAsync(exchange, cancellationToken);
+
+                return true;
+            }
+            catch (OperationInterruptedException ex) when (ex.ShutdownReason?.ReplyCode == 404)
+            {
+                _logger.LogDebug("Exchange '{Exchange}' does not exist yet.", exchange);
+                return false;
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError(ex, "Unexpected error while checking if exchange '{Exchange}' exists.");
+                throw;
+            }
+        }
     }
 }
diff --git a/src/OpenDDD.Tests/Unit/Infrastructure/Events/RabbitMq/RabbitMqMessagingProviderTests.cs b/src/OpenDDD.Tests/Unit/Infrastructure/Events/RabbitMq/RabbitMqMessagingProviderTests.cs
@@ -26,6 +26,7 @@ public RabbitMqMessagingProviderTests()
             _provider = new RabbitMqMessagingProvider(
                 _mockConnectionFactory.Object,
                 _mockConsumerFactory.Object,
+                autoCreateTopics: true,
                 _mockLogger.Object
             );
         }
@@ -42,7 +43,7 @@ public void Constructor_ShouldThrowException_WhenDependenciesAreNull(
             var mockLogger = logger is null ? null! : _mockLogger.Object;
 
             Assert.Throws<ArgumentNullException>(() =>
-                new RabbitMqMessagingProvider(mockConnectionFactory, mockConsumerFactory, mockLogger));
+                new RabbitMqMessagingProvider(mockConnectionFactory, mockConsumerFactory, autoCreateTopics: true, mockLogger));
         }
 
         [Theory]
diff --git a/src/OpenDDD/API/Extensions/OpenDddServiceCollectionExtensions.cs b/src/OpenDDD/API/Extensions/OpenDddServiceCollectionExtensions.cs
@@ -323,7 +323,11 @@ private static void AddRabbitMq(this IServiceCollection services)
                 
                 var consumerFactory = new RabbitMqConsumerFactory(logger);
 
-                return new RabbitMqMessagingProvider(connectionFactory, consumerFactory, logger);
+                return new RabbitMqMessagingProvider(
+                    connectionFactory, 
+                    consumerFactory,
+                    rabbitMqOptions.AutoCreateTopics,
+                    logger);
             });
         }
         
diff --git a/src/OpenDDD/API/Options/OpenDddOptions.cs b/src/OpenDDD/API/Options/OpenDddOptions.cs
@@ -62,7 +62,13 @@ public OpenDddOptions UseInMemoryMessaging()
             return this;
         }
 
-        public OpenDddOptions UseRabbitMq(string hostName, int port, string username, string password, string virtualHost = "/")
+        public OpenDddOptions UseRabbitMq(
+            string hostName, 
+            int port, 
+            string username, 
+            string password, 
+            string virtualHost = "/", 
+            bool autoCreateTopics = true)
         {
             MessagingProvider = "RabbitMq";
             RabbitMq = new OpenDddRabbitMqOptions
@@ -71,12 +77,13 @@ public OpenDddOptions UseRabbitMq(string hostName, int port, string username, st
                 Port = port,
                 Username = username,
                 Password = password,
-                VirtualHost = virtualHost
+                VirtualHost = virtualHost,
+                AutoCreateTopics = autoCreateTopics
             };
             return this;
         }
 
-        public OpenDddOptions UseKafka(string bootstrapServers)
+        public OpenDddOptions UseKafka(string bootstrapServers, bool autoCreateTopics = true)
         {
             MessagingProvider = "Kafka";
             Kafka = new OpenDddKafkaOptions { BootstrapServers = bootstrapServers };
diff --git a/src/OpenDDD/Infrastructure/Events/RabbitMq/Options/OpenDddRabbitMqOptions.cs b/src/OpenDDD/Infrastructure/Events/RabbitMq/Options/OpenDddRabbitMqOptions.cs
@@ -7,5 +7,6 @@ public class OpenDddRabbitMqOptions
         public string Username { get; set; } = "guest";
         public string Password { get; set; } = "guest";
         public string VirtualHost { get; set; } = "/";
+        public bool AutoCreateTopics { get; set; } = true;
     }
 }
diff --git a/src/OpenDDD/Infrastructure/Events/RabbitMq/RabbitMqMessagingProvider.cs b/src/OpenDDD/Infrastructure/Events/RabbitMq/RabbitMqMessagingProvider.cs
@@ -4,6 +4,7 @@
 using OpenDDD.Infrastructure.Events.Base;
 using OpenDDD.Infrastructure.Events.RabbitMq.Factories;
 using RabbitMQ.Client;
+using RabbitMQ.Client.Exceptions;
 
 namespace OpenDDD.Infrastructure.Events.RabbitMq
 {
@@ -12,17 +13,20 @@ public class RabbitMqMessagingProvider : IMessagingProvider, IAsyncDisposable
         private readonly IConnectionFactory _connectionFactory;
         private readonly IRabbitMqConsumerFactory _consumerFactory;
         private readonly ILogger<RabbitMqMessagingProvider> _logger;
+        private readonly bool _autoCreateTopics;
         private IConnection? _connection;
         private IChannel? _channel;
         private readonly ConcurrentDictionary<string, RabbitMqSubscription> _subscriptions = new();
         
         public RabbitMqMessagingProvider(
             IConnectionFactory factory,
             IRabbitMqConsumerFactory consumerFactory,
+            bool autoCreateTopics,
             ILogger<RabbitMqMessagingProvider> logger)
         {
             _connectionFactory = factory ?? throw new ArgumentNullException(nameof(factory));
             _consumerFactory = consumerFactory ?? throw new ArgumentNullException(nameof(consumerFactory));
+            _autoCreateTopics = autoCreateTopics;
             _logger = logger ?? throw new ArgumentNullException(nameof(logger));
         }
 
@@ -41,6 +45,22 @@ public async Task<ISubscription> SubscribeAsync(string topic, string consumerGro
 
             if (_channel is null) throw new InvalidOperationException("RabbitMQ channel is not available.");
 
+            bool exchangeExists = await ExchangeExistsAsync(topic, cancellationToken);
+
+            if (!exchangeExists)
+            {
+                if (_autoCreateTopics)
+                {
+                    await _channel.ExchangeDeclareAsync(topic, ExchangeType.Fanout, durable: true, autoDelete: false, cancellationToken: cancellationToken);
+                    _logger.LogInformation("Auto-created exchange (topic): {Topic}", topic);
+                }
+                else
+                {
+                    _logger.LogError("Cannot subscribe to non-existent topic: {Topic}", topic);
+                    throw new InvalidOperationException($"Topic '{topic}' does not exist.");
+                }
+            }
+
             await _channel.ExchangeDeclareAsync(topic, ExchangeType.Fanout, durable: true, autoDelete: false, cancellationToken: cancellationToken);
             var queueName = $"{consumerGroup}.{topic}";
             await _channel.QueueDeclareAsync(queueName, durable: true, exclusive: false, autoDelete: false, cancellationToken: cancellationToken);
@@ -87,7 +107,21 @@ public async Task PublishAsync(string topic, string message, CancellationToken c
 
             if (_channel is null) throw new InvalidOperationException("RabbitMQ channel is not available.");
 
-            await _channel.ExchangeDeclareAsync(topic, ExchangeType.Fanout, durable: true, autoDelete: false, cancellationToken: cancellationToken);
+            bool exchangeExists = await ExchangeExistsAsync(topic, cancellationToken);
+
+            if (!exchangeExists)
+            {
+                if (_autoCreateTopics)
+                {
+                    await _channel.ExchangeDeclareAsync(topic, ExchangeType.Fanout, durable: true, autoDelete: false, cancellationToken: cancellationToken);
+                    _logger.LogInformation("Auto-created exchange (topic): {Topic}", topic);
+                }
+                else
+                {
+                    _logger.LogError("Cannot publish to non-existent topic: {Topic}", topic);
+                    throw new InvalidOperationException($"Topic '{topic}' does not exist.");
+                }
+            }
 
             var body = Encoding.UTF8.GetBytes(message);
             await _channel.BasicPublishAsync(topic, "", body, cancellationToken: cancellationToken);
@@ -102,6 +136,29 @@ private async Task EnsureConnectedAsync(CancellationToken cancellationToken)
             _connection = await _connectionFactory.CreateConnectionAsync(cancellationToken);
             _channel = await _connection.CreateChannelAsync(null, cancellationToken);
         }
+        
+        private async Task<bool> ExchangeExistsAsync(string exchange, CancellationToken cancellationToken)
+        {
+            try
+            {
+                await _channel!.ExchangeDeclarePassiveAsync(exchange, cancellationToken);
+                return true;
+            }
+            catch (OperationInterruptedException ex) when (ex.ShutdownReason?.ReplyCode == 404)
+            {
+                _logger.LogDebug("Exchange '{Exchange}' does not exist.", exchange);
+
+                // Since the channel was closed, reopen it before returning
+                await EnsureConnectedAsync(cancellationToken);
+        
+                return false;
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError(ex, "Unexpected error while checking if exchange '{Exchange}' exists.", exchange);
+                throw;
+            }
+        }
 
         public async ValueTask DisposeAsync()
         {

Original file line number	Diff line number	Diff line change
`@@ -26,6 +26,7 @@ public RabbitMqMessagingProviderTests()`
`26`	`26`	`_provider = new RabbitMqMessagingProvider(`
`27`	`27`	`_mockConnectionFactory.Object,`
`28`	`28`	`_mockConsumerFactory.Object,`
	`29`	`+ autoCreateTopics: true,`
`29`	`30`	`_mockLogger.Object`
`30`	`31`	`);`
`31`	`32`	`}`
`@@ -42,7 +43,7 @@ public void Constructor_ShouldThrowException_WhenDependenciesAreNull(`
`42`	`43`	`var mockLogger = logger is null ? null! : _mockLogger.Object;`
`43`	`44`
`44`	`45`	`Assert.Throws<ArgumentNullException>(() =>`
`45`		`- new RabbitMqMessagingProvider(mockConnectionFactory, mockConsumerFactory, mockLogger));`
	`46`	`+ new RabbitMqMessagingProvider(mockConnectionFactory, mockConsumerFactory, autoCreateTopics: true, mockLogger));`
`46`	`47`	`}`
`47`	`48`
`48`	`49`	`[Theory]`
Original file line number	Diff line number	Diff line change
`@@ -7,5 +7,6 @@ public class OpenDddRabbitMqOptions`
`7`	`7`	`public string Username { get; set; } = "guest";`
`8`	`8`	`public string Password { get; set; } = "guest";`
`9`	`9`	`public string VirtualHost { get; set; } = "/";`
	`10`	`+ public bool AutoCreateTopics { get; set; } = true;`
`10`	`11`	`}`
`11`	`12`	`}`