Add non-destructive queues support

Author: Havret

src/ArtemisNetClient.Extensions.DependencyInjection/ActiveMqArtemisClientDependencyInjectionExtensions.cs

@@ -236,7 +236,8 @@ public static IActiveMqBuilder AddConsumer(this IActiveMqBuilder builder, string
                     MaxConsumers = queueOptions.MaxConsumers,
                     AutoCreateAddress = queueOptions.AutoCreateAddress,
                     PurgeOnNoConsumers = queueOptions.PurgeOnNoConsumers,
-                    LastValueKey = queueOptions.LastValueKey
+                    LastValueKey = queueOptions.LastValueKey,
+                    NonDestructive = queueOptions.NonDestructive
                 });
                 if (options.AddressConfigurations.TryGetValue(address, out var routingTypes))
                 {

src/ArtemisNetClient.Extensions.DependencyInjection/Options/QueueOptions.cs

@@ -23,5 +23,10 @@ public class QueueOptions
         /// Configures Last-Value key for a queue 
         /// </summary>
         public string LastValueKey { get; set; }
+        
+        /// <summary>
+        /// True if the queue is non-destructive 
+        /// </summary>
+        public bool NonDestructive { get; set; }
     }
 }

src/ArtemisNetClient/Configuration/QueueConfiguration.cs

@@ -36,5 +36,10 @@ public class QueueConfiguration
         /// Configures Last-Value key for a queue 
         /// </summary>
         public string LastValueKey { get; set; }
+
+        /// <summary>
+        /// True if the queue is non-destructive 
+        /// </summary>
+        public bool NonDestructive { get; set; }
     }
 }

src/ArtemisNetClient/Management/RequestSerializer.cs

@@ -39,6 +39,7 @@ public static async ValueTask<string> QueueConfigurationToJson(QueueConfiguratio
                 writer.WriteString("filter-string", configuration.FilterExpression ?? string.Empty);
                 writer.WriteBoolean("auto-delete", configuration.AutoDelete);
                 writer.WriteNumber("auto-delete-message-count", configuration.AutoDeleteMessageCount);
+                writer.WriteBoolean("non-destructive", configuration.NonDestructive);
                 if (configuration.AutoDeleteDelay.HasValue)
                 {
                     writer.WriteNumber("auto-delete-delay", Convert.ToInt64(configuration.AutoDeleteDelay.Value.TotalMilliseconds));    

test/ArtemisNetClient.IntegrationTests/TopologyManagement/CreateQueueSpec.cs

@@ -188,5 +188,41 @@ await topologyManager.CreateQueueAsync(new QueueConfiguration
             Assert.Equal("foo3", (await consumer.ReceiveAsync(CancellationToken)).GetBody<string>());
             Assert.Equal("foo6", (await consumer.ReceiveAsync(CancellationToken)).GetBody<string>());
         }
+
+        [Fact]
+        public async Task Should_create_non_destructive_queue()
+        {
+            var connection = await CreateConnection();
+            var address = Guid.NewGuid().ToString();
+            var queue = Guid.NewGuid().ToString();
+
+            var topologyManager = await connection.CreateTopologyManagerAsync();
+            await topologyManager.CreateAddressAsync(address, RoutingType.Multicast);
+            await topologyManager.CreateQueueAsync(new QueueConfiguration
+            {
+                Address = address,
+                Name = queue,
+                RoutingType = RoutingType.Multicast,
+                NonDestructive = true
+            });
+            
+            var producer = await connection.CreateProducerAsync(address, RoutingType.Multicast);
+
+            await producer.SendAsync(new Message("foo"));
+
+            var consumer = await connection.CreateConsumerAsync(address, queue, CancellationToken);
+            var msg = await consumer.ReceiveAsync(CancellationToken);
+            await consumer.AcceptAsync(msg);
+            Assert.Equal("foo", msg.GetBody<string>());
+
+            // close consumer
+            await consumer.DisposeAsync();
+            
+            // recreate consumer, and the message should be redelivered as the queue was created as non-destructive
+            consumer = await connection.CreateConsumerAsync(address, queue, CancellationToken);
+            msg = await consumer.ReceiveAsync(CancellationToken);
+            await consumer.AcceptAsync(msg);
+            Assert.Equal("foo", msg.GetBody<string>());
+        }
     }
 }