Add Last-Value Queues support

Author: Havret

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

@@ -1,7 +1,7 @@
 namespace ActiveMQ.Artemis.Client.Extensions.DependencyInjection
 {
     /// <summary>
-    /// Provides programmatic configuration for a queue to be declared. 
+    /// Provides programmatic configuration for a queue to be declared.
     /// </summary>
     public class QueueOptions
     {
@@ -18,5 +18,10 @@ public class QueueOptions
         public bool AutoCreateAddress { get; set; }
 
         public string FilterExpression { get; set; }
+
+        /// <summary>
+        /// Configures Last-Value key for a queue 
+        /// </summary>
+        public string LastValueKey { get; set; }
     }
 }

src/ArtemisNetClient/Configuration/QueueConfiguration.cs

@@ -31,5 +31,10 @@ public class QueueConfiguration
         /// Delay for auto-deleting queues that do not have any consumers attached.
         /// </summary>
         public TimeSpan? AutoDeleteDelay { get; set; }
+
+        /// <summary>
+        /// Configures Last-Value key for a queue 
+        /// </summary>
+        public string LastValueKey { get; set; }
     }
 }

src/ArtemisNetClient/Management/RequestSerializer.cs

@@ -47,6 +47,10 @@ public static async ValueTask<string> QueueConfigurationToJson(QueueConfiguratio
                 {
                     writer.WriteNumber("auto-delete-delay", -1);
                 }
+                if (string.IsNullOrEmpty(configuration.LastValueKey) == false)
+                {
+                    writer.WriteString("last-value-key", configuration.LastValueKey);
+                }
                 writer.WriteEndObject();
             }
 

test/ArtemisNetClient.IntegrationTests/TopologyManagement/CreateQueueSpec.cs

@@ -151,5 +151,42 @@ await topologyManager.CreateQueueAsync(new QueueConfiguration
 
             Assert.Equal("foo2", (await newQueue1Consumer.ReceiveAsync(CancellationToken)).GetBody<string>());
         }
+
+        [Fact]
+        public async Task Should_create_last_value_queue()
+        {
+            var connection = await CreateConnection();
+            var address = Guid.NewGuid().ToString();
+            var queue = Guid.NewGuid().ToString();
+            var lastValueKey = "my-last-value-key";
+
+            var topologyManager = await connection.CreateTopologyManagerAsync();
+            await topologyManager.CreateAddressAsync(address, RoutingType.Multicast);
+            await topologyManager.CreateQueueAsync(new QueueConfiguration
+            {
+                Address = address,
+                Name = queue,
+                RoutingType = RoutingType.Multicast,
+                LastValueKey = lastValueKey
+            });
+            
+            var producer = await connection.CreateProducerAsync(address, RoutingType.Multicast);
+            
+            // Send series of messages with the same last value key
+            await producer.SendAsync(new Message("foo1") { ApplicationProperties = { [lastValueKey] = "1" } });
+            await producer.SendAsync(new Message("foo2") { ApplicationProperties = { [lastValueKey] = "1" } });
+            await producer.SendAsync(new Message("foo3") { ApplicationProperties = { [lastValueKey] = "1" } });
+            
+            // Send series of messages with another last value key
+            await producer.SendAsync(new Message("foo4") { ApplicationProperties = { [lastValueKey] = "2" } });
+            await producer.SendAsync(new Message("foo5") { ApplicationProperties = { [lastValueKey] = "2" } });
+            await producer.SendAsync(new Message("foo6") { ApplicationProperties = { [lastValueKey] = "2" } });
+            
+            var consumer = await connection.CreateConsumerAsync(address, queue, CancellationToken);
+            
+            // Only messages foo3 and foo6 should survive
+            Assert.Equal("foo3", (await consumer.ReceiveAsync(CancellationToken)).GetBody<string>());
+            Assert.Equal("foo6", (await consumer.ReceiveAsync(CancellationToken)).GetBody<string>());
+        }
     }
 }