* Fix the StreamFilter test.

Author: lukebakken

RabbitMQ.AMQP.Client/Impl/AmqpConsumer.cs

@@ -28,7 +28,7 @@ private enum PauseStatus
         private readonly UnsettledMessageCounter _unsettledMessageCounter = new();
         private readonly ConsumerConfiguration _configuration;
 
-        public AmqpConsumer(ConsumerConfiguration configuration)
+        internal AmqpConsumer(ConsumerConfiguration configuration)
         {
             _configuration = configuration;
             _configuration.Connection.AddConsumer(_id, this);
@@ -108,6 +108,7 @@ await receiverLinkTask.WaitAsync(waitSpan)
                     // TODO save / cancel task
                     _ = Task.Run(ProcessMessages);
 
+                    // TODO cancellation token
                     await base.OpenAsync()
                         .ConfigureAwait(false);
                 }

RabbitMQ.AMQP.Client/Impl/AmqpConsumerBuilder.cs

@@ -14,11 +14,11 @@ namespace RabbitMQ.AMQP.Client.Impl
     /// <summary>
     /// ConsumerConfiguration is a helper class that holds the configuration for the consumer
     /// </summary>
-    public class ConsumerConfiguration
+    internal sealed class ConsumerConfiguration
     {
         public AmqpConnection Connection { get; set; } = null!;
         public string Address { get; set; } = "";
-        public int InitialCredits { get; set; } = 10;
+        public int InitialCredits { get; set; } = 100; // TODO use constant, check with Java lib
         public Map Filters { get; set; } = new();
         public MessageHandler? Handler { get; set; }
         // TODO re-name to ListenerContextAction? Callback?

RabbitMQ.AMQP.Client/PublicAPI.Unshipped.txt

@@ -287,7 +287,6 @@ RabbitMQ.AMQP.Client.Impl.AmqpConnection.Publishers.get -> System.Collections.Ge
 RabbitMQ.AMQP.Client.Impl.AmqpConnection.RpcClientBuilder() -> RabbitMQ.AMQP.Client.IRpcClientBuilder!
 RabbitMQ.AMQP.Client.Impl.AmqpConnection.RpcServerBuilder() -> RabbitMQ.AMQP.Client.IRpcServerBuilder!
 RabbitMQ.AMQP.Client.Impl.AmqpConsumer
-RabbitMQ.AMQP.Client.Impl.AmqpConsumer.AmqpConsumer(RabbitMQ.AMQP.Client.Impl.ConsumerConfiguration! configuration) -> void
 RabbitMQ.AMQP.Client.Impl.AmqpConsumer.Pause() -> void
 RabbitMQ.AMQP.Client.Impl.AmqpConsumer.Unpause() -> void
 RabbitMQ.AMQP.Client.Impl.AmqpConsumer.UnsettledMessageCount.get -> long
@@ -488,19 +487,6 @@ RabbitMQ.AMQP.Client.Impl.Consts
 RabbitMQ.AMQP.Client.Impl.Consts.Consts() -> void
 RabbitMQ.AMQP.Client.Impl.ConsumerBuilderStreamOptions
 RabbitMQ.AMQP.Client.Impl.ConsumerBuilderStreamOptions.ConsumerBuilderStreamOptions(RabbitMQ.AMQP.Client.IConsumerBuilder! consumerBuilder, Amqp.Types.Map! filters) -> void
-RabbitMQ.AMQP.Client.Impl.ConsumerConfiguration
-RabbitMQ.AMQP.Client.Impl.ConsumerConfiguration.Address.get -> string!
-RabbitMQ.AMQP.Client.Impl.ConsumerConfiguration.Address.set -> void
-RabbitMQ.AMQP.Client.Impl.ConsumerConfiguration.Connection.get -> RabbitMQ.AMQP.Client.Impl.AmqpConnection!
-RabbitMQ.AMQP.Client.Impl.ConsumerConfiguration.Connection.set -> void
-RabbitMQ.AMQP.Client.Impl.ConsumerConfiguration.ConsumerConfiguration() -> void
-RabbitMQ.AMQP.Client.Impl.ConsumerConfiguration.Filters.get -> Amqp.Types.Map!
-RabbitMQ.AMQP.Client.Impl.ConsumerConfiguration.Filters.set -> void
-RabbitMQ.AMQP.Client.Impl.ConsumerConfiguration.Handler.get -> RabbitMQ.AMQP.Client.MessageHandler?
-RabbitMQ.AMQP.Client.Impl.ConsumerConfiguration.Handler.set -> void
-RabbitMQ.AMQP.Client.Impl.ConsumerConfiguration.InitialCredits.get -> int
-RabbitMQ.AMQP.Client.Impl.ConsumerConfiguration.InitialCredits.set -> void
-RabbitMQ.AMQP.Client.Impl.ConsumerConfiguration.ListenerContext -> System.Action<RabbitMQ.AMQP.Client.IConsumerBuilder.ListenerContext!>?
 RabbitMQ.AMQP.Client.Impl.DefaultAddressBuilder<T>
 RabbitMQ.AMQP.Client.Impl.DefaultAddressBuilder<T>.Address() -> string!
 RabbitMQ.AMQP.Client.Impl.DefaultAddressBuilder<T>.DefaultAddressBuilder() -> void

Tests/Consumer/StreamConsumerTests.cs

@@ -217,19 +217,22 @@ public async Task StreamFiltering()
         IQueueSpecification queueSpec = _management.Queue().Name(_queueName).Type(QueueType.STREAM);
         await queueSpec.DeclareAsync();
 
-        var publishTasks = new List<Task>();
         foreach (string w in waves)
         {
             void ml(ulong idx, IMessage msg)
             {
-                msg.MessageId(idx);
-                msg.Annotation("x-stream-filter-value", w);
+                msg.MessageId(idx)
+                    .Annotation("x-stream-filter-value", w);
             }
-            publishTasks.Add(PublishAsync(queueSpec, messageWaveCount, ml));
 
+            /*
+             * Note:
+             * If publishing is done async, then messages must be filtered
+             * as they are received because you could get "wrong" values
+             * due to how the bloom filter works
+             */
+            await PublishAsync(queueSpec, messageWaveCount, ml);
         }
-        await WhenAllComplete(publishTasks);
-        publishTasks.Clear();
 
         long receivedCount = 0;
         Exception? messageHandlerEx = null;
@@ -238,6 +241,8 @@ async Task messageHandler(IContext cxt, IMessage msg)
             try
             {
                 Interlocked.Increment(ref receivedCount);
+                // _testOutputHelper.WriteLine($"id {0} x-stream-filter-value {1}",
+                //    msg.MessageId(), msg.Annotation("x-stream-filter-value"));
                 await cxt.AcceptAsync();
             }
             catch (Exception ex)
@@ -260,6 +265,7 @@ async Task messageHandler(IContext cxt, IMessage msg)
             await WaitUntilStable(() => (int)Interlocked.Read(ref receivedCount));
         }
 
+        // _testOutputHelper.WriteLine("0: receivedCount: {0}", receivedCount);
         Assert.Null(messageHandlerEx);
         Assert.True(receivedCount >= messageWaveCount);
         Assert.True(receivedCount < allMessagesCount,
@@ -280,6 +286,7 @@ async Task messageHandler(IContext cxt, IMessage msg)
             await WaitUntilStable(() => (int)Interlocked.Read(ref receivedCount));
         }
 
+        // _testOutputHelper.WriteLine("1: receivedCount: {0}", receivedCount);
         Assert.Null(messageHandlerEx);
         Assert.True(receivedCount >= 2 * messageWaveCount);
         Assert.True(receivedCount < allMessagesCount,