comment test

Signed-off-by: Gabriele Santomaggio <[email protected]>

Author: Gsantomaggio

.ci/ubuntu/cluster/rmq/Dockerfile

@@ -12,5 +12,6 @@ COPY --chown=rabbitmq:rabbitmq --chmod=0644 enabled_plugins /etc/rabbitmq/enable
 COPY --chown=rabbitmq:rabbitmq rabbitmq-env.conf /etc/rabbitmq/rabbitmq-env.conf
 COPY --chown=rabbitmq:rabbitmq rabbitmq.conf /etc/rabbitmq/rabbitmq.conf
 COPY --chown=rabbitmq:rabbitmq advanced.config /etc/rabbitmq/advanced.config
+COPY --chown=rabbitmq:rabbitmq 21-enable-management-collector.conf /etc/rabbitmq/conf.d/21-enable-management-collector.conf
 
 EXPOSE 4369 5671 5672 15672 15692 25672 35672-35682

.ci/ubuntu/one-node/gha-setup.sh

@@ -90,6 +90,7 @@ function start_rabbitmq
         --volume "$GITHUB_WORKSPACE/.ci/ubuntu/one-node/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf:ro" \
         --volume "$GITHUB_WORKSPACE/.ci/certs:/etc/rabbitmq/certs:ro" \
         --volume "$GITHUB_WORKSPACE/.ci/ubuntu/one-node/advanced.config:/etc/rabbitmq/advanced.config:ro" \
+        --volume "$GITHUB_WORKSPACE/.ci/ubuntu/one-node/21-enable-management-collector.conf:/etc/rabbitmq/conf.d/21-enable-management-collector.conf:ro" \
         --volume "$GITHUB_WORKSPACE/.ci/ubuntu/log:/var/log/rabbitmq" \
         "$rabbitmq_image"
 }

RabbitMQ.AMQP.Client/Consts.cs

@@ -28,6 +28,7 @@ public static class Consts
         private const string AmqpSqlFilter = "amqp:sql-filter";
         internal static readonly Symbol s_streamSqlFilterSymbol = new(AmqpSqlFilter);
         internal const string SqlFilter = "sql-filter";
+        internal static readonly Symbol s_sqlFilterSymbol = new(SqlFilter);
 
         internal const string AmqpPropertiesFilter = "amqp:properties-filter";
         internal const string AmqpApplicationPropertiesFilter = "amqp:application-properties-filter";

RabbitMQ.AMQP.Client/Impl/AmqpConsumer.cs

@@ -88,6 +88,12 @@ void OnAttached(ILink argLink, Attach argAttach)
 
                 // TODO configurable timeout
                 var waitSpan = TimeSpan.FromSeconds(5);
+
+                // TODO
+                // Even 10ms is enough to allow the links to establish,
+                // which tells me it allows the .NET runtime to process
+                await Task.Delay(10).ConfigureAwait(false);
+
                 _receiverLink = await attachCompletedTcs.Task.WaitAsync(waitSpan)
                     .ConfigureAwait(false);
 

RabbitMQ.AMQP.Client/Impl/AmqpConsumerBuilder.cs

@@ -86,7 +86,7 @@ public async Task<IConsumer> BuildAndStartAsync(CancellationToken cancellationTo
                 throw new ConsumerException("Message handler is not set");
             }
 
-            if (_configuration.Filters[Consts.SqlFilter] is not null &&
+            if (_configuration.Filters[Consts.s_sqlFilterSymbol] is not null &&
                 _amqpConnection._featureFlags.IsSqlFeatureEnabled == false)
             {
                 throw new ConsumerException("SQL filter is not supported by the connection." +
@@ -254,7 +254,7 @@ public IConsumerBuilder.IStreamFilterOptions Sql(string sql)
                 throw new ArgumentNullException(nameof(sql));
             }
 
-            _filters[Consts.SqlFilter] =
+            _filters[Consts.s_sqlFilterSymbol] =
                 new DescribedValue(Consts.s_streamSqlFilterSymbol, sql);
             return this;
         }

RabbitMQ.AMQP.Client/Impl/AmqpPublisher.cs

@@ -67,6 +67,12 @@ void OnAttached(ILink argLink, Attach argAttach)
 
                 // TODO configurable timeout
                 var waitSpan = TimeSpan.FromSeconds(5);
+                // TODO
+                // Even 10ms is enough to allow the links to establish,
+                // which tells me it allows the .NET runtime to process
+                await Task.Delay(10)
+                    .ConfigureAwait(false);
+
                 _senderLink = await attachCompletedTcs.Task.WaitAsync(waitSpan)
                     .ConfigureAwait(false);
 

Tests/Consumer/ConsumerSqlFilterTests.cs

@@ -21,7 +21,7 @@ public class ConsumerSqlFilterTests(ITestOutputHelper testOutputHelper) : Integr
         // Example test method (to be implemented):
         [SkippableFact]
         [Trait("Category", "SqlFilter")]
-        public async Task TestSqlFilterFunctionality()
+        public async Task TestSqlFilterFunctionalityAsync()
         {
             Assert.NotNull(_connection);
             Assert.NotNull(_management);
@@ -37,7 +37,7 @@ public async Task TestSqlFilterFunctionality()
                 new TaskCompletionSource<IMessage>(TaskCreationOptions.RunContinuationsAsynchronously);
             IConsumer consumer = await _connection.ConsumerBuilder()
                 .Queue(_queueName)
-                .Stream().Filter().Sql("properties.user_id = 'John'").Stream().Offset(StreamOffsetSpecification.First)
+                .Stream().Filter().Sql("properties.subject LIKE '%John%'").Stream().Offset(StreamOffsetSpecification.First)
                 .Builder().MessageHandler((IContext ctx, IMessage msg) =>
                 {
                     tcs.SetResult(msg);
@@ -49,20 +49,22 @@ public async Task TestSqlFilterFunctionality()
 
             IPublisher publisher = await _connection.PublisherBuilder().Queue(_queueName).BuildAsync();
 
-            var msgNotInTheFilter = new AmqpMessage("Test message for SQL filter")
-                .Property("user_id", "Gas"); // This property should not match the SQL filter
+            // var msgNotInTheFilter = new AmqpMessage("Test message for SQL filter")
+            //     .Property("user_id", "Gas"); // This property should not match the SQL filter
+            var msgNotInTheFilter = new AmqpMessage("Test message for SQL filter, should not match")
+                .Subject("Gas"); // This property should not match the SQL filter
             await publisher.PublishAsync(msgNotInTheFilter);
-            var msgInTheFilter = new AmqpMessage("Test message for NOT SQL filter")
-                .Property("user_id", "John"); // This property should match the SQL filter
+            var msgInTheFilter = new AmqpMessage("Test message for SQL filter")
+                .Subject("John"); // This property should match the SQL filter
             await publisher.PublishAsync(msgInTheFilter);
-            await tcs.Task.WaitAsync(TimeSpan.FromSeconds(10)).ConfigureAwait(false);
+            await tcs.Task.WaitAsync(TimeSpan.FromSeconds(10));
 
             Assert.Equal("Test message for SQL filter", tcs.Task.Result.BodyAsString());
-            Assert.Equal("John", tcs.Task.Result.Property("user_id"));
-            await consumer.CloseAsync().ConfigureAwait(false);
-            await publisher.CloseAsync().ConfigureAwait(false);
-            await q.DeleteAsync().ConfigureAwait(false);
-            await _connection.CloseAsync().ConfigureAwait(false);
+            Assert.Equal("John", tcs.Task.Result.Subject());
+            await consumer.CloseAsync();
+            await publisher.CloseAsync();
+            await q.DeleteAsync();
+            await _connection.CloseAsync();
         }
     }
 }

Tests/Management/ManagementTests.cs

@@ -454,6 +454,6 @@ public async Task PurgeQueueShouldReturnErrorForStream()
         await queueSpec.DeclareAsync();
         await PublishAsync(queueSpec, 19);
         await WaitUntilQueueMessageCount(_queueName, 19);
-        await Assert.ThrowsAsync<BadRequestException>(() => queueSpec.PurgeAsync());
+        await Assert.ThrowsAsync<BadRequestException>(queueSpec.PurgeAsync);
     }
 }