fixup

Author: lukebakken

RabbitMQ.AMQP.Client/Impl/AmqpConsumerBuilder.cs

@@ -139,8 +139,9 @@ public IConsumerBuilder.IStreamOptions Offset(string interval)
 
         public IConsumerBuilder.IStreamOptions FilterValue(string value)
         {
+            var values = new List<string> { value };
             _filters[s_streamFilterSymbol] =
-                new DescribedValue(s_streamFilterSymbol, value);
+                new DescribedValue(s_streamFilterSymbol, values);
             return this;
         }
 

RabbitMQ.AMQP.Client/Impl/AmqpMessage.cs

@@ -228,8 +228,14 @@ public IDictionary<object, object> Properties()
             return NativeMessage.ApplicationProperties.Map;
         }
 
+#if TODO
+        static void validateMessageAnnotationKey(String key) {
+    if (!key.startsWith("x-")) {
+      throw new IllegalArgumentException("Message annotation keys must start with 'x-': " + key);
+    }
+  }
+#endif
         // Annotations
-
         public IMessage Annotation(string key, object value)
         {
             EnsureAnnotations();

Tests/Consumer/StreamConsumerTests.cs

@@ -231,13 +231,13 @@ void ml(ulong idx, IMessage msg)
         await WhenAllComplete(publishTasks);
         publishTasks.Clear();
 
-        var messages = new ConcurrentBag<IMessage>();
+        long receivedCount = 0;
         Exception? messageHandlerEx = null;
         async Task messageHandler(IContext cxt, IMessage msg)
         {
             try
             {
-                messages.Add(msg);
+                Interlocked.Increment(ref receivedCount);
                 await cxt.AcceptAsync();
             }
             catch (Exception ex)
@@ -257,14 +257,14 @@ async Task messageHandler(IContext cxt, IMessage msg)
 
         using (IConsumer consumer = await consumerBuilder0.BuildAndStartAsync())
         {
-            await WaitUntilStable(() => messages.Count);
+            await WaitUntilStable(() => (int)Interlocked.Read(ref receivedCount));
         }
 
         Assert.Null(messageHandlerEx);
-        Assert.True(messages.Count >= messageWaveCount);
-        Assert.True(messages.Count < allMessagesCount,
-            $"0: messages.Count {messages.Count}, waveCount * messageWaveCount {allMessagesCount}");
-        messages.Clear();
+        Assert.True(receivedCount >= messageWaveCount);
+        Assert.True(receivedCount < allMessagesCount,
+            $"0: receivedCount {receivedCount}, waveCount * messageWaveCount {allMessagesCount}");
+        receivedCount = 0;
 
         IConsumerBuilder consumerBuilder1 = _connection.ConsumerBuilder()
             .Queue(_queueName)
@@ -277,15 +277,13 @@ async Task messageHandler(IContext cxt, IMessage msg)
 
         using (IConsumer consumer = await consumerBuilder1.BuildAndStartAsync())
         {
-            await WaitUntilStable(() => messages.Count);
+            await WaitUntilStable(() => (int)Interlocked.Read(ref receivedCount));
         }
 
         Assert.Null(messageHandlerEx);
-        Assert.True(messages.Count >= 2 * messageWaveCount);
-        //TODO note the Java test still asserts that messages.Count is less than allMessagesCount
-        Assert.True(messages.Count <= allMessagesCount,
-            $"1: messages.Count {messages.Count}, waveCount * messageWaveCount {allMessagesCount}");
-        messages.Clear();
+        Assert.True(receivedCount >= 2 * messageWaveCount);
+        Assert.True(receivedCount < allMessagesCount,
+            $"1: messages.Count {receivedCount}, waveCount * messageWaveCount {allMessagesCount}");
     }
 
     /// <summary>