TestKit - add support for filter expressions

src/ArtemisNetClient.Testing/MessageSourceInfo.cs

@@ -0,0 +1,3 @@
+namespace ActiveMQ.Artemis.Client.Testing;
+
+internal record MessageSourceInfo(string Address, string Queue, string? FilterExpression);

src/ArtemisNetClient.Testing/SharedMessageSource.cs

@@ -1,3 +1,5 @@
+using System.Data;
+using ActiveMQ.Artemis.Client.InternalUtilities;
 using ActiveMQ.Artemis.Client.Testing.Utils;
 
 namespace ActiveMQ.Artemis.Client.Testing;
@@ -7,13 +9,13 @@ internal class SharedMessageSource
     private readonly List<MessageSource> _messageSources = new();
     private int _cursor;
 
-    public SharedMessageSource(MessageSource messageSource, string queue)
+    public SharedMessageSource(MessageSourceInfo info, MessageSource messageSource)
     {
         AddMessageSource(messageSource);
-        Queue = queue;
+        Info = info;
     }
 
-    public string Queue { get; }
+    public MessageSourceInfo Info { get; }
 
     public void AddMessageSource(MessageSource messageSource)
     {
@@ -22,6 +24,11 @@ public void AddMessageSource(MessageSource messageSource)
 
     public void Enqueue(Message message)
     {
+        if (Info.FilterExpression != null && FilterExpressionMatches(message) == false)
+        {
+            return;
+        }
+
         if (message.GroupId is { Length: > 0 } groupId)
         {
             var cursor = ArtemisBucketHelper.GetBucket(groupId, _messageSources.Count);
@@ -41,4 +48,53 @@ public void Enqueue(Message message)
             }
         }
     }
+
+    private bool FilterExpressionMatches(Message message)
+    {
+        var dataTable = new DataTable();
+        var dataRow = dataTable.NewRow();
+        foreach (var key in message.ApplicationProperties.Keys)
+        {
+            var value = message.ApplicationProperties[key];
+            if (value != null)
+            {
+                dataTable.Columns.Add(key, value.GetType());
+                dataRow[key] = value;
+            }
+        }
+
+        if (message.Priority is { } priority)
+        {
+            var columnName = "AMQPriority";
+            dataTable.Columns.Add(columnName, priority.GetType());
+            dataRow[columnName] = priority;
+        }
+
+        if (message.TimeToLive is { } timeToLive)
+        {
+            var columnName = "AMQExpiration";
+            var value = DateTimeOffset.UtcNow.Add(timeToLive).ToUnixTimeMilliseconds();
+            dataTable.Columns.Add(columnName, value.GetType());
+            dataRow[columnName] = value;
+        }
+
+        if (message.DurabilityMode is { } durabilityMode)
+        {
+            var columnName = "AMQDurable";
+            dataTable.Columns.Add(columnName, typeof(string));
+            dataRow[columnName] = durabilityMode == DurabilityMode.Durable ? "DURABLE" : "NON_DURABLE";
+        }
+
+        if (message.CreationTime is { } creationTime)
+        {
+            var columnName = "AMQTimestamp";
+            var value = DateTime.SpecifyKind(creationTime, DateTimeKind.Utc).ToUnixTimeMilliseconds();
+            dataTable.Columns.Add(columnName, value.GetType());
+            dataRow[columnName] = value;
+        }
+
+        dataTable.Rows.Add(dataRow);
+
+        return dataTable.Select(Info.FilterExpression).Any();
+    }
 }

src/ArtemisNetClient.Testing/TestKit.cs

@@ -69,24 +69,24 @@ private void OnMessage(Message message)
         }
     }
 
-    private void OnMessageSource(string address, string queue, MessageSource messageSource)
+    private void OnMessageSource(MessageSourceInfo info, MessageSource messageSource)
     {
-        if (!_messageSources.TryGetValue(address, out var messageSources))
+        if (!_messageSources.TryGetValue(info.Address, out var messageSources))
         {
             messageSources = new List<SharedMessageSource>();
-            _messageSources[address] = messageSources;
+            _messageSources[info.Address] = messageSources;
         }
 
         lock (messageSources)
         {
-            var existingMessageSource = messageSources.FirstOrDefault(x => x.Queue == queue);
+            var existingMessageSource = messageSources.FirstOrDefault(x => x.Info == info);
             if (existingMessageSource != null)
             {
                 existingMessageSource.AddMessageSource(messageSource);
             }
             else
             {
-                messageSources.Add(new SharedMessageSource(messageSource, queue));
+                messageSources.Add(new SharedMessageSource(info, messageSource));
             }
         }
     }

src/ArtemisNetClient.Testing/TestLinkProcessor.cs

@@ -1,4 +1,5 @@
 using System.Text.RegularExpressions;
+using Amqp;
 using Amqp.Framing;
 using Amqp.Listener;
 using Amqp.Transactions;
@@ -9,9 +10,9 @@ namespace ActiveMQ.Artemis.Client.Testing;
 internal class TestLinkProcessor : ILinkProcessor
 {
     private readonly Action<Message> _onMessage;
-    private readonly Action<string, string, MessageSource> _onMessageSource;
+    private readonly Action<MessageSourceInfo, MessageSource> _onMessageSource;
 
-    public TestLinkProcessor(Action<Message> onMessage, Action<string, string, MessageSource> onMessageSource)
+    public TestLinkProcessor(Action<Message> onMessage, Action<MessageSourceInfo, MessageSource> onMessageSource)
     {
         _onMessage = onMessage;
         _onMessageSource = onMessageSource;
@@ -26,10 +27,9 @@ public void Process(AttachContext attachContext)
                 attachContext.Attach.OfferedCapabilities = new[] { new Symbol("SHARED-SUBS") };
             }
 
-            var (address, queue) = ParseAddressAndQueue(source, attachContext.Link);
-
+            var messageSourceInfo = GetMessageSourceInfo(source, attachContext.Link);
             var messageSource = new MessageSource();
-            _onMessageSource(address, queue, messageSource);
+            _onMessageSource(messageSourceInfo, messageSource);
             attachContext.Complete(new SourceLinkEndpoint(messageSource, attachContext.Link), 0);
         }
         else if (attachContext.Attach.Target is Target)
@@ -44,14 +44,33 @@ public void Process(AttachContext attachContext)
         }
     }
 
-    private (string address, string queue) ParseAddressAndQueue(Source input, ListenerLink listenerLink)
+    private static MessageSourceInfo GetMessageSourceInfo(Source source, ILink link)
     {
+        var filterExpression = GetFilterExpression(source);
+
         // FQQN match
-        if (Regex.Match(input.Address, "(.+)::(.+)") is { Success: true, Groups: { Count: 3 } groups })
+        if (Regex.Match(source.Address, "(.+)::(.+)") is { Success: true, Groups: { Count: 3 } groups })
+        {
+            return new MessageSourceInfo( groups[1].Value,  groups[2].Value, FilterExpression: filterExpression);
+        }
+
+        return new MessageSourceInfo(source.Address, link.Name, FilterExpression: filterExpression);
+    }
+
+    private static readonly Symbol _selectorFilterSymbol = new("apache.org:selector-filter:string");
+
+    private static string? GetFilterExpression(Source source)
+    {
+        if (!source.FilterSet.TryGetValue(_selectorFilterSymbol, out var filterExpressionObj))
+        {
+            return null;
+        }
+
+        if (filterExpressionObj is DescribedValue { Value: string filterExpression })
         {
-            return (address: groups[1].Value, queue: groups[2].Value);
+            return filterExpression;
         }
 
-        return (address: input.Address, queue: listenerLink.Name);
+        return null;
     }
 }

src/ArtemisNetClient.Testing/Utils/IsExternalInitFix.cs

@@ -0,0 +1,5 @@
+// ReSharper disable once CheckNamespace
+namespace System.Runtime.CompilerServices
+{
+    internal static class IsExternalInit {}
+}