Fix sharing of processors for Azure transports that messes up deadlet…

…ters (#529)
tinglesoftware · Jul 6, 2023 · 025d144 · 025d144
1 parent 6872499
commit 025d144
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 7 deletions.
diff --git a/src/Tingle.EventBus.Transports.Azure.EventHubs/AzureEventHubsTransport.cs b/src/Tingle.EventBus.Transports.Azure.EventHubs/AzureEventHubsTransport.cs
@@ -300,7 +300,8 @@ Task<EventHubProducerClient> creator((Type, bool) _, CancellationToken ct)
     private Task<EventProcessorClient> GetProcessorAsync(EventRegistration reg, EventConsumerRegistration ecr, CancellationToken cancellationToken)
     {
         var name = reg.EventName;
-        if (ecr.Deadletter) name += Options.DeadLetterSuffix;
+        var deadletter = ecr.Deadletter;
+        if (deadletter) name += Options.DeadLetterSuffix;
 
         // For events configured as sourced from IoT Hub,
         // 1. The event hub name is in the metadata
@@ -360,7 +361,7 @@ async Task<EventProcessorClient> creator(string key, CancellationToken ct)
             return processor;
         }
 
-        var key = $"{eventHubName}/{consumerGroup}";
+        var key = $"{eventHubName}/{consumerGroup}/{deadletter}";
         return processorsCache.GetOrAddAsync(key, creator, cancellationToken);
     }
 

diff --git a/src/Tingle.EventBus.Transports.Azure.ServiceBus/AzureServiceBusTransport.cs b/src/Tingle.EventBus.Transports.Azure.ServiceBus/AzureServiceBusTransport.cs
@@ -322,6 +322,7 @@ private Task<ServiceBusProcessor> GetProcessorAsync(EventRegistration reg, Event
     {
         var topicName = reg.EventName!;
         var subscriptionName = ecr.ConsumerName!;
+        var deadletter = ecr.Deadletter;
 
         async Task<ServiceBusProcessor> creator(string key, CancellationToken ct)
         {
@@ -343,7 +344,7 @@ async Task<ServiceBusProcessor> creator(string key, CancellationToken ct)
                 PrefetchCount = Options.DefaultPrefetchCount,
 
                 // Set the sub-queue to be used
-                SubQueue = ecr.Deadletter ? SubQueue.DeadLetter : SubQueue.None,
+                SubQueue = deadletter ? SubQueue.DeadLetter : SubQueue.None,
             };
 
             // Allow for the defaults to be overridden
@@ -376,7 +377,7 @@ await CreateSubscriptionIfNotExistsAsync(ecr: ecr,
             }
         }
 
-        var key = $"{topicName}/{subscriptionName}";
+        var key = $"{topicName}/{subscriptionName}/{deadletter}";
         return processorsCache.GetOrAddAsync(key, creator, cancellationToken);
     }
 

diff --git a/src/Tingle.EventBus.Transports.InMemory/InMemoryTransport.cs b/src/Tingle.EventBus.Transports.InMemory/InMemoryTransport.cs
@@ -294,8 +294,9 @@ private Task<InMemoryProcessor> GetProcessorAsync(EventRegistration reg, EventCo
     {
         var topicName = reg.EventName!;
         var subscriptionName = ecr.ConsumerName!;
+        var deadletter = ecr.Deadletter;
 
-        var key = $"{topicName}/{subscriptionName}";
+        var key = $"{topicName}/{subscriptionName}/{deadletter}";
         Task<InMemoryProcessor> creator(string _, CancellationToken ct)
         {
             // Create the processor options
@@ -307,13 +308,13 @@ Task<InMemoryProcessor> creator(string _, CancellationToken ct)
             {
                 // Create the processor for the Queue
                 Logger.CreatingQueueProcessor(queueName: topicName);
-                processor = inMemoryClient.CreateProcessor(queueName: topicName, options: inpo);
+                processor = inMemoryClient.CreateProcessor(queueName: topicName, options: inpo); // TODO: support deadletter
             }
             else
             {
                 // Create the processor for the Subscription
                 Logger.CreatingSubscriptionProcessor(topicName: topicName, subscriptionName: subscriptionName);
-                processor = inMemoryClient.CreateProcessor(topicName: topicName, subscriptionName: subscriptionName, options: inpo);
+                processor = inMemoryClient.CreateProcessor(topicName: topicName, subscriptionName: subscriptionName, options: inpo); // TODO: support deadletter
             }
 
             return Task.FromResult(processor);