Fix topic support in DLE+TTL Scheduler (#1124)

Pliner · web-flow · commit 26bc9b1a2661 · 2020-10-05T14:46:10.000+05:00
* Do not set dead letter routing key to support custom routing keys

* Delete test
diff --git a/Source/EasyNetQ.IntegrationTests/Scheduler/When_future_publish_using_dead_letter_exchange.cs b/Source/EasyNetQ.IntegrationTests/Scheduler/When_future_publish_using_dead_letter_exchange.cs
@@ -38,7 +38,7 @@ public async Task Test()
 
             using (bus.Subscribe<Message>(subscriptionId, messagesSink.Receive))
             {
-                await bus.FuturePublishBatchAsync(messages, TimeSpan.FromSeconds(5), cts.Token)
+                await bus.FuturePublishBatchAsync(messages, TimeSpan.FromSeconds(5), "#", cts.Token)
                     .ConfigureAwait(false);
 
                 await messagesSink.WaitAllReceivedAsync(cts.Token).ConfigureAwait(false);
diff --git a/Source/EasyNetQ.IntegrationTests/Scheduler/When_future_publish_using_dead_letter_exchange_with_publish_confirms.cs b/Source/EasyNetQ.IntegrationTests/Scheduler/When_future_publish_using_dead_letter_exchange_with_publish_confirms.cs
@@ -39,7 +39,7 @@ public async Task Test()
 
             using (bus.Subscribe<Message>(subscriptionId, messagesSink.Receive))
             {
-                await bus.FuturePublishBatchAsync(messages, TimeSpan.FromSeconds(5), cts.Token)
+                await bus.FuturePublishBatchAsync(messages, TimeSpan.FromSeconds(5), "#", cts.Token)
                     .ConfigureAwait(false);
 
                 await messagesSink.WaitAllReceivedAsync(cts.Token).ConfigureAwait(false);
diff --git a/Source/EasyNetQ.IntegrationTests/Scheduler/When_future_publish_using_dead_letter_exchange_with_topic.cs b/Source/EasyNetQ.IntegrationTests/Scheduler/When_future_publish_using_dead_letter_exchange_with_topic.cs
@@ -0,0 +1,58 @@
+﻿using System;
+using System.Threading;
+using System.Threading.Tasks;
+using EasyNetQ.IntegrationTests.Utils;
+using EasyNetQ.Scheduling;
+using FluentAssertions;
+using Xunit;
+
+namespace EasyNetQ.IntegrationTests.Scheduler
+{
+    [Collection("RabbitMQ")]
+    public class When_publish_and_subscribe_with_delay_using_dead_letter_exchange_with_topic : IDisposable
+    {
+        public When_publish_and_subscribe_with_delay_using_dead_letter_exchange_with_topic(RabbitMQFixture fixture)
+        {
+            bus = RabbitHutch.CreateBus(
+                $"host={fixture.Host};prefetchCount=1;timeout=5", c => c.EnableDeadLetterExchangeAndMessageTtlScheduler()
+            );
+        }
+
+        public void Dispose()
+        {
+            bus.Dispose();
+        }
+
+        private const int MessagesCount = 10;
+
+        private readonly IBus bus;
+
+        [Fact]
+        public async Task Test()
+        {
+            using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
+
+            var subscriptionId = Guid.NewGuid().ToString();
+            var firstTopicMessagesSink = new MessagesSink(MessagesCount);
+            var firstTopicMessages = MessagesFactories.Create(MessagesCount);
+            var secondTopicMessagesSink = new MessagesSink(MessagesCount);
+            var secondTopicMessages = MessagesFactories.Create(MessagesCount);
+            using (bus.Subscribe<Message>(subscriptionId, firstTopicMessagesSink.Receive, x => x.WithTopic("first")))
+            using (bus.Subscribe<Message>(subscriptionId, secondTopicMessagesSink.Receive, x => x.WithTopic("second")))
+            {
+                await Task.WhenAll(
+                    bus.FuturePublishBatchAsync(firstTopicMessages, TimeSpan.FromSeconds(5), "first", cts.Token),
+                    bus.FuturePublishBatchAsync(secondTopicMessages, TimeSpan.FromSeconds(5), "second", cts.Token)
+                ).ConfigureAwait(false);
+
+                await Task.WhenAll(
+                    firstTopicMessagesSink.WaitAllReceivedAsync(cts.Token),
+                    secondTopicMessagesSink.WaitAllReceivedAsync(cts.Token)
+                ).ConfigureAwait(false);
+
+                firstTopicMessagesSink.ReceivedMessages.Should().Equal(firstTopicMessages);
+                secondTopicMessagesSink.ReceivedMessages.Should().Equal(secondTopicMessages);
+            }
+        }
+    }
+}
diff --git a/Source/EasyNetQ.IntegrationTests/Scheduler/When_future_publish_using_delayed_exchange.cs b/Source/EasyNetQ.IntegrationTests/Scheduler/When_future_publish_using_delayed_exchange.cs
@@ -38,7 +38,7 @@ public async Task Test()
 
             using (bus.Subscribe<Message>(subscriptionId, messagesSink.Receive))
             {
-                await bus.FuturePublishBatchAsync(messages, TimeSpan.FromSeconds(5), cts.Token)
+                await bus.FuturePublishBatchAsync(messages, TimeSpan.FromSeconds(5), "#", cts.Token)
                     .ConfigureAwait(false);
 
                 await messagesSink.WaitAllReceivedAsync(cts.Token).ConfigureAwait(false);
diff --git a/Source/EasyNetQ.IntegrationTests/Scheduler/When_future_publish_using_delayed_exchange_with_publish_confirms.cs b/Source/EasyNetQ.IntegrationTests/Scheduler/When_future_publish_using_delayed_exchange_with_publish_confirms.cs
@@ -39,7 +39,7 @@ public async Task Test()
 
             using (bus.Subscribe<Message>(subscriptionId, messagesSink.Receive))
             {
-                await bus.FuturePublishBatchAsync(messages, TimeSpan.FromSeconds(5), cts.Token)
+                await bus.FuturePublishBatchAsync(messages, TimeSpan.FromSeconds(5), "#", cts.Token)
                     .ConfigureAwait(false);
 
                 await messagesSink.WaitAllReceivedAsync(cts.Token).ConfigureAwait(false);
diff --git a/Source/EasyNetQ.IntegrationTests/Utils/BusExtensions.cs b/Source/EasyNetQ.IntegrationTests/Utils/BusExtensions.cs
@@ -35,14 +35,15 @@ public static async Task FuturePublishBatchAsync<T>(
             this IBus bus,
             IEnumerable<T> messages,
             TimeSpan delay,
+            string topic = "#",
             CancellationToken cancellationToken = default
         ) where T : class
         {
             foreach (var message in messages)
             {
                 cancellationToken.ThrowIfCancellationRequested();
 
-                await bus.FuturePublishAsync(delay, message).ConfigureAwait(false);
+                await bus.FuturePublishAsync(delay, message, topic).ConfigureAwait(false);
             }
         }
 
diff --git a/Source/EasyNetQ.Tests/Scheduling/SchedulingExtensionsTests.cs b/Source/EasyNetQ.Tests/Scheduling/SchedulingExtensionsTests.cs
diff --git a/Source/EasyNetQ/Scheduling/DeadLetterExchangeAndMessageTtlScheduler.cs b/Source/EasyNetQ/Scheduling/DeadLetterExchangeAndMessageTtlScheduler.cs
@@ -11,11 +11,14 @@ public class DeadLetterExchangeAndMessageTtlScheduler : IScheduler
         private readonly IAdvancedBus advancedBus;
         private readonly IConventions conventions;
         private readonly IMessageDeliveryModeStrategy messageDeliveryModeStrategy;
+        private readonly bool setDeadLetterRoutingKey;
 
         public DeadLetterExchangeAndMessageTtlScheduler(
             IAdvancedBus advancedBus,
             IConventions conventions,
-            IMessageDeliveryModeStrategy messageDeliveryModeStrategy)
+            IMessageDeliveryModeStrategy messageDeliveryModeStrategy,
+            bool setDeadLetterRoutingKey = false
+        )
         {
             Preconditions.CheckNotNull(advancedBus, "advancedBus");
             Preconditions.CheckNotNull(conventions, "conventions");
@@ -24,6 +27,7 @@ public DeadLetterExchangeAndMessageTtlScheduler(
             this.advancedBus = advancedBus;
             this.conventions = conventions;
             this.messageDeliveryModeStrategy = messageDeliveryModeStrategy;
+            this.setDeadLetterRoutingKey = setDeadLetterRoutingKey;
         }
 
         public Task FuturePublishAsync<T>(DateTime futurePublishDate, T message, string cancellationKey = null) where T : class
@@ -77,13 +81,19 @@ private async Task FuturePublishInternalAsync<T>(TimeSpan messageDelay, T messag
             Preconditions.CheckNotNull(message, "message");
             Preconditions.CheckLess(messageDelay, MaxMessageDelay, "messageDelay");
             Preconditions.CheckNull(cancellationKey, "cancellationKey");
+
             var delay = Round(messageDelay);
             var delayString = delay.ToString(@"dd\_hh\_mm\_ss");
             var exchangeName = conventions.ExchangeNamingConvention(typeof (T));
             var futureExchangeName = exchangeName + "_" + delayString;
             var futureQueueName = conventions.QueueNamingConvention(typeof (T), delayString);
             var futureExchange = await advancedBus.ExchangeDeclareAsync(futureExchangeName, ExchangeType.Topic).ConfigureAwait(false);
-            var futureQueue = await advancedBus.QueueDeclareAsync(futureQueueName, perQueueMessageTtl: (int) delay.TotalMilliseconds, deadLetterExchange: exchangeName, deadLetterRoutingKey: topic).ConfigureAwait(false);
+            var futureQueue = await advancedBus.QueueDeclareAsync(
+                futureQueueName,
+                perQueueMessageTtl: (int) delay.TotalMilliseconds,
+                deadLetterExchange: exchangeName,
+                deadLetterRoutingKey: setDeadLetterRoutingKey ? topic : null
+            ).ConfigureAwait(false);
             await advancedBus.BindAsync(futureExchange, futureQueue, topic).ConfigureAwait(false);
             var easyNetQMessage = new Message<T>(message)
             {
@@ -100,13 +110,19 @@ private void FuturePublishInternal<T>(TimeSpan messageDelay, T message, string t
             Preconditions.CheckNotNull(message, "message");
             Preconditions.CheckLess(messageDelay, MaxMessageDelay, "messageDelay");
             Preconditions.CheckNull(cancellationKey, "cancellationKey");
+
             var delay = Round(messageDelay);
             var delayString = delay.ToString(@"dd\_hh\_mm\_ss");
             var exchangeName = conventions.ExchangeNamingConvention(typeof (T));
             var futureExchangeName = exchangeName + "_" + delayString;
             var futureQueueName = conventions.QueueNamingConvention(typeof (T), delayString);
             var futureExchange = advancedBus.ExchangeDeclare(futureExchangeName, ExchangeType.Topic);
-            var futureQueue = advancedBus.QueueDeclare(futureQueueName, perQueueMessageTtl: (int) delay.TotalMilliseconds, deadLetterExchange: exchangeName, deadLetterRoutingKey: topic);
+            var futureQueue = advancedBus.QueueDeclare(
+                futureQueueName,
+                perQueueMessageTtl: (int) delay.TotalMilliseconds,
+                deadLetterExchange: exchangeName,
+                deadLetterRoutingKey: setDeadLetterRoutingKey ? topic : null
+            );
             advancedBus.Bind(futureExchange, futureQueue, topic);
             var easyNetQMessage = new Message<T>(message)
             {
@@ -123,4 +139,4 @@ private static TimeSpan Round(TimeSpan timeSpan)
             return new TimeSpan(timeSpan.Days, timeSpan.Hours, timeSpan.Minutes, timeSpan.Seconds, 0);
         }
     }
-}
+}
diff --git a/Source/EasyNetQ/Scheduling/SchedulingExtensions.cs b/Source/EasyNetQ/Scheduling/SchedulingExtensions.cs
@@ -11,7 +11,25 @@ public static IServiceRegister EnableDelayedExchangeScheduler(this IServiceRegis
 
         public static IServiceRegister EnableDeadLetterExchangeAndMessageTtlScheduler(this IServiceRegister serviceRegister)
         {
-            return serviceRegister.Register<IScheduler, DeadLetterExchangeAndMessageTtlScheduler>();
+            return serviceRegister.Register<IScheduler>(
+                x => new DeadLetterExchangeAndMessageTtlScheduler(
+                    x.Resolve<IAdvancedBus>(),
+                    x.Resolve<IConventions>(),
+                    x.Resolve<IMessageDeliveryModeStrategy>()
+                )
+            );
+        }
+
+        public static IServiceRegister EnableLegacyDeadLetterExchangeAndMessageTtlScheduler(this IServiceRegister serviceRegister)
+        {
+            return serviceRegister.Register<IScheduler>(
+                x => new DeadLetterExchangeAndMessageTtlScheduler(
+                    x.Resolve<IAdvancedBus>(),
+                    x.Resolve<IConventions>(),
+                    x.Resolve<IMessageDeliveryModeStrategy>(),
+                    true
+                )
+            );
         }
     }
-}
+}

Original file line number	Diff line number	Diff line change
`@@ -38,7 +38,7 @@ public async Task Test()`
`38`	`38`
`39`	`39`	`using (bus.Subscribe<Message>(subscriptionId, messagesSink.Receive))`
`40`	`40`	`{`
`41`		`- await bus.FuturePublishBatchAsync(messages, TimeSpan.FromSeconds(5), cts.Token)`
	`41`	`+ await bus.FuturePublishBatchAsync(messages, TimeSpan.FromSeconds(5), "#", cts.Token)`
`42`	`42`	`.ConfigureAwait(false);`
`43`	`43`
`44`	`44`	`await messagesSink.WaitAllReceivedAsync(cts.Token).ConfigureAwait(false);`
Original file line number	Diff line number	Diff line change
`@@ -39,7 +39,7 @@ public async Task Test()`
`39`	`39`
`40`	`40`	`using (bus.Subscribe<Message>(subscriptionId, messagesSink.Receive))`
`41`	`41`	`{`
`42`		`- await bus.FuturePublishBatchAsync(messages, TimeSpan.FromSeconds(5), cts.Token)`
	`42`	`+ await bus.FuturePublishBatchAsync(messages, TimeSpan.FromSeconds(5), "#", cts.Token)`
`43`	`43`	`.ConfigureAwait(false);`
`44`	`44`
`45`	`45`	`await messagesSink.WaitAllReceivedAsync(cts.Token).ConfigureAwait(false);`
Original file line number	Diff line number	Diff line change
`@@ -35,14 +35,15 @@ public static async Task FuturePublishBatchAsync<T>(`
`35`	`35`	`this IBus bus,`
`36`	`36`	`IEnumerable<T> messages,`
`37`	`37`	`TimeSpan delay,`
	`38`	`+ string topic = "#",`
`38`	`39`	`CancellationToken cancellationToken = default`
`39`	`40`	`) where T : class`
`40`	`41`	`{`
`41`	`42`	`foreach (var message in messages)`
`42`	`43`	`{`
`43`	`44`	`cancellationToken.ThrowIfCancellationRequested();`
`44`	`45`
`45`		`- await bus.FuturePublishAsync(delay, message).ConfigureAwait(false);`
	`46`	`+ await bus.FuturePublishAsync(delay, message, topic).ConfigureAwait(false);`
`46`	`47`	`}`
`47`	`48`	`}`
`48`	`49`