Refactor rabbit provider by breaking out helper method for reuse.

Author: runemalm

src/OpenDDD/Infrastructure/Events/RabbitMq/RabbitMqMessagingProvider.cs

@@ -45,22 +45,11 @@ 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)
             {
-                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 CreateTopicIfNotExistsAsync(topic, cancellationToken);
             }
-
+            
             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);
@@ -76,7 +65,7 @@ public async Task<ISubscription> SubscribeAsync(string topic, string consumerGro
 
             return subscription;
         }
-        
+
         public async Task UnsubscribeAsync(ISubscription subscription, CancellationToken cancellationToken = default)
         {
             if (subscription == null)
@@ -107,20 +96,9 @@ public async Task PublishAsync(string topic, string message, CancellationToken c
 
             if (_channel is null) throw new InvalidOperationException("RabbitMQ channel is not available.");
 
-            bool exchangeExists = await ExchangeExistsAsync(topic, cancellationToken);
-
-            if (!exchangeExists)
+            if (_autoCreateTopics)
             {
-                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.");
-                }
+                await CreateTopicIfNotExistsAsync(topic, cancellationToken);
             }
 
             var body = Encoding.UTF8.GetBytes(message);
@@ -137,6 +115,25 @@ private async Task EnsureConnectedAsync(CancellationToken cancellationToken)
             _channel = await _connection.CreateChannelAsync(null, cancellationToken);
         }
 
+        private async Task CreateTopicIfNotExistsAsync(string topic, 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 subscribe to non-existent topic: {Topic}", topic);
+                    throw new InvalidOperationException($"Topic '{topic}' does not exist.");
+                }
+            }
+        }
+        
         private async Task<bool> ExchangeExistsAsync(string exchange, CancellationToken cancellationToken)
         {
             try