Cache topic checks for ten minutes in messaging providers.

Author: runemalm

src/OpenDDD.Tests/Integration/Infrastructure/Events/Kafka/KafkaMessagingProviderTests.cs

@@ -365,7 +365,7 @@ public async Task CompetingConsumers_ShouldDistributeMessages_WhenMultipleConsum
             var consumerGroup = "test-consumer-group";
             var totalMessages = 100;
             var numConsumers = 2;
-            var variancePercentage = 0.2;
+            var variancePercentage = 0.3;
             var perConsumerMessageCount = new ConcurrentDictionary<Guid, int>(); // Track messages per consumer
             var allMessagesProcessed = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
 

src/OpenDDD/Infrastructure/Events/Azure/AzureServiceBusMessagingProvider.cs

@@ -13,6 +13,8 @@ public class AzureServiceBusMessagingProvider : IMessagingProvider, IAsyncDispos
         private readonly bool _autoCreateTopics;
         private readonly ILogger<AzureServiceBusMessagingProvider> _logger;
         private readonly ConcurrentDictionary<string, AzureServiceBusSubscription> _subscriptions = new();
+        private readonly ConcurrentDictionary<string, DateTime> _topicCache = new();
+        private readonly TimeSpan _cacheExpiration = TimeSpan.FromSeconds(600);
         private bool _disposed;
 
         public AzureServiceBusMessagingProvider(
@@ -105,18 +107,29 @@ public async Task PublishAsync(string topic, string message, CancellationToken c
 
         private async Task EnsureTopicExistsAsync(string topic, CancellationToken cancellationToken)
         {
-            if (!await _adminClient.TopicExistsAsync(topic, cancellationToken))
+            if (_topicCache.TryGetValue(topic, out var lastChecked) && DateTime.UtcNow - lastChecked < _cacheExpiration)
             {
-                if (_autoCreateTopics)
-                {
-                    await _adminClient.CreateTopicAsync(topic, cancellationToken);
-                    _logger.LogInformation("Created topic: {Topic}", topic);
-                }
-                else
-                {
-                    throw new InvalidOperationException($"Topic '{topic}' does not exist. Enable 'autoCreateTopics' to create topics automatically.");
-                }
+                _logger.LogDebug("Skipping topic check for '{Topic}' (cached result).", topic);
+                return;
+            }
+
+            var topicExists = await _adminClient.TopicExistsAsync(topic, cancellationToken);
+    
+            if (topicExists)
+            {
+                _topicCache[topic] = DateTime.UtcNow;
+                return;
+            }
+
+            if (_autoCreateTopics)
+            {
+                await _adminClient.CreateTopicAsync(topic, cancellationToken);
+                _logger.LogInformation("Created topic: {Topic}", topic);
+                _topicCache[topic] = DateTime.UtcNow;
+                return;
             }
+            
+            throw new InvalidOperationException($"Topic '{topic}' does not exist. Enable 'autoCreateTopics' to create topics automatically.");
         }
 
         private async Task CreateSubscriptionIfNotExistsAsync(string topic, string subscriptionName, CancellationToken cancellationToken)

src/OpenDDD/Infrastructure/Events/Kafka/KafkaMessagingProvider.cs

@@ -15,6 +15,8 @@ public class KafkaMessagingProvider : IMessagingProvider, IAsyncDisposable
         private readonly IKafkaConsumerFactory _consumerFactory;
         private readonly ILogger<KafkaMessagingProvider> _logger;
         private readonly ConcurrentDictionary<string, KafkaSubscription> _subscriptions = new();
+        private readonly ConcurrentDictionary<string, DateTime> _topicCache = new();
+        private readonly TimeSpan _cacheExpiration = TimeSpan.FromSeconds(600);
         private readonly CancellationTokenSource _cts = new();
         private bool _disposed;
 
@@ -93,11 +95,18 @@ public async Task PublishAsync(string topic, string message, CancellationToken c
 
         private async Task EnsureTopicExistsAsync(string topic, CancellationToken cancellationToken)
         {
+            if (_topicCache.TryGetValue(topic, out var lastChecked) && DateTime.UtcNow - lastChecked < _cacheExpiration)
+            {
+                _logger.LogDebug("Skipping topic check for '{Topic}' (cached result).", topic);
+                return;
+            }
+
             var metadata = _adminClient.GetMetadata(TimeSpan.FromSeconds(5));
 
             if (metadata.Topics.Any(t => t.Topic == topic))
             {
                 _logger.LogDebug("Topic '{Topic}' already exists.", topic);
+                _topicCache[topic] = DateTime.UtcNow;
                 return;
             }
 
@@ -109,6 +118,8 @@ await _adminClient.CreateTopicsAsync(new[]
                     new TopicSpecification { Name = topic, NumPartitions = 2, ReplicationFactor = 1 }
                 }, null);
 
+                _topicCache[topic] = DateTime.UtcNow;
+                
                 // Wait for the topic to be available
                 for (int i = 0; i < 30; i++)
                 {

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

@@ -17,6 +17,9 @@ public class RabbitMqMessagingProvider : IMessagingProvider, IAsyncDisposable
         private IConnection? _connection;
         private IChannel? _channel;
         private readonly ConcurrentDictionary<string, RabbitMqSubscription> _subscriptions = new();
+        private readonly ConcurrentDictionary<string, DateTime> _topicCache = new();
+        private readonly TimeSpan _cacheExpiration = TimeSpan.FromSeconds(600);
+        private bool _disposed;
 
         public RabbitMqMessagingProvider(
             IConnectionFactory factory,
@@ -104,23 +107,29 @@ private async Task EnsureConnectedAsync(CancellationToken cancellationToken)
 
         private async Task EnsureTopicExistsAsync(string topic, CancellationToken cancellationToken)
         {
-            bool exchangeExists = await ExchangeExistsAsync(topic, cancellationToken);
+            if (_topicCache.TryGetValue(topic, out var lastChecked) && DateTime.UtcNow - lastChecked < _cacheExpiration)
+            {
+                _logger.LogDebug("Skipping exchange check for '{Topic}' (cached result).", topic);
+                return;
+            }
 
+            bool exchangeExists = await ExchangeExistsAsync(topic, cancellationToken);
+    
             if (exchangeExists)
             {
-                _logger.LogDebug("Exchange '{Topic}' already exists.", topic);
+                _topicCache[topic] = DateTime.UtcNow;
                 return;
             }
 
             if (_autoCreateTopics)
             {
                 await _channel.ExchangeDeclareAsync(topic, ExchangeType.Fanout, durable: true, autoDelete: false, cancellationToken: cancellationToken);
                 _logger.LogInformation("Auto-created exchange (topic): {Topic}", topic);
+                _topicCache[topic] = DateTime.UtcNow;
+                return;
             }
-            else
-            {
-                throw new InvalidOperationException($"Topic '{topic}' does not exist. Enable 'autoCreateTopics' to create topics automatically.");
-            }
+            
+            throw new InvalidOperationException($"Topic '{topic}' does not exist. Enable 'autoCreateTopics' to create topics automatically.");
         }
 
         private async Task<bool> ExchangeExistsAsync(string exchange, CancellationToken cancellationToken)
@@ -147,6 +156,9 @@ private async Task<bool> ExchangeExistsAsync(string exchange, CancellationToken
 
         public async ValueTask DisposeAsync()
         {
+            if (_disposed) return;
+            _disposed = true;
+
             _logger.LogDebug("Disposing RabbitMqMessagingProvider...");
 
             foreach (var subscription in _subscriptions.Values)