* Finish porting last two tests from SourceFiltersTest.java

Author: lukebakken

Tests/Consumer/StreamConsumerTests.cs

@@ -901,4 +901,60 @@ public async Task FilterExpressionPropertiesAndApplicationProperties()
             Assert.Equal(body3, mb);
         }
     }
+
+    /// <summary>
+    /// Port of filterExpressionFilterFewMessagesFromManyToTestFlowControl
+    /// </summary>
+    [SkippableFact]
+    public async Task FilterExpressionFilterFewMessagesFromManyToTestFlowControl()
+    {
+        Assert.NotNull(_connection);
+        Assert.NotNull(_management);
+        SkipIfNotRabbitMQ_4_1_0();
+
+        string groupId = RandomString();
+
+        IQueueSpecification q = _management.Queue(_queueName).Stream().Queue();
+        await q.DeclareAsync();
+
+        await PublishAsync(q, 1, (_, msg) => msg.GroupId(groupId));
+        await PublishAsync(q, 1000);
+        await PublishAsync(q, 1, (_, msg) => msg.GroupId(groupId));
+
+        IEnumerable<IMessage> msgs = await ConsumeAsync(2, options => options.GroupId(groupId));
+        foreach (IMessage m in msgs)
+        {
+            Assert.Equal(groupId, m.GroupId());
+        }
+    }
+
+    /// <summary>
+    /// Port of filterExpressionStringModifier
+    /// </summary>
+    [SkippableFact]
+    public async Task FilterExpressionStringModifier()
+    {
+        Assert.NotNull(_connection);
+        Assert.NotNull(_management);
+        SkipIfNotRabbitMQ_4_1_0();
+
+        IQueueSpecification q = _management.Queue(_queueName).Stream().Queue();
+        await q.DeclareAsync();
+
+        await PublishAsync(q, 1, (_, msg) => msg.Subject("abc 123"));
+        await PublishAsync(q, 1, (_, msg) => msg.Subject("foo bar"));
+        await PublishAsync(q, 1, (_, msg) => msg.Subject("ab 12"));
+
+        IEnumerable<IMessage> msgs = await ConsumeAsync(2, options => options.Subject("$p:ab"));
+        foreach (IMessage m in msgs)
+        {
+            Assert.StartsWith("ab", m.Subject());
+        }
+
+        msgs = await ConsumeAsync(1, options => options.Subject("$s:bar"));
+        foreach (IMessage m in msgs)
+        {
+            Assert.Equal("foo bar", m.Subject());
+        }
+    }
 }