Test kit improve message filters

src/ArtemisNetClient.Testing/SharedMessageSource.cs

@@ -1,4 +1,5 @@
 using System.Data;
+using System.Text.RegularExpressions;
 using ActiveMQ.Artemis.Client.InternalUtilities;
 using ActiveMQ.Artemis.Client.Testing.Utils;
 
@@ -94,7 +95,25 @@ private bool FilterExpressionMatches(Message message)
         }
 
         dataTable.Rows.Add(dataRow);
-
-        return dataTable.Select(Info.FilterExpression).Any();
+
+        while (true)
+        {
+            try
+            {
+                return dataTable.Select(Info.FilterExpression).Any();
+            }
+            catch (EvaluateException e) when(e.Message.StartsWith("Cannot find column"))
+            {
+                var match = Regex.Match(e.Message, @"Cannot find column \[(\w+)\]\.");
+                if (match.Success)
+                {
+                    dataTable.Columns.Add(match.Groups[1].Value);
+                }
+                else
+                {
+                    throw;
+                }
+            }
+        }
     }
 }

test/ArtemisNetClient.Testing.UnitTests/FilterExpressionsSpec.cs

@@ -197,6 +197,30 @@ public async Task Should_filter_messages_based_on_application_property()
         Assert.All(blueMessages, x => Assert.Equal("blue", x.GetBody<string>()));
     }
 
+    [Fact]
+    public async Task Should_filter_messages_when_property_used_in_filter_is_missing()
+    {
+        var endpoint = EndpointUtil.GetUniqueEndpoint();
+        using var testKit = new TestKit(endpoint);
+
+        var connectionFactory = new ConnectionFactory();
+        await using var connection = await connectionFactory.CreateAsync(endpoint);
+
+        var address = "test_address";
+        await using var messageConsumer = await connection.CreateConsumerAsync(new ConsumerConfiguration
+        {
+            Address = address,
+            RoutingType = RoutingType.Multicast,
+            FilterExpression = "color IS NULL"
+        });
+
+        await testKit.SendMessageAsync(address, new Message("msg"));
+
+        var message = await messageConsumer.ReceiveAsync();
+
+        Assert.Equal("msg", message.GetBody<string>());
+    }
+
     static async Task<IReadOnlyList<Message>> ReceiveMessages(IConsumer consumer, int count)
     {
         var messages = new List<Message>();