Add code snippets for stream filtering documentation

Author: acogoluegnes

src/test/java/com/rabbitmq/client/amqp/docs/WebsiteDocumentation.java

@@ -104,6 +104,16 @@ void publishAddressFormatInMessages() {
         .message();
   }
 
+  void publishStreamFiltering() {
+    Publisher publisher = null;
+    byte[] body = null;
+    Message message = publisher.message(body)
+        .annotation("x-stream-filter-value", "invoices");
+    publisher.publish(message, context -> {
+      // confirm callback
+    });
+  }
+
   void consuming() {
     Connection connection = null;
     Consumer consumer = connection.consumerBuilder()
@@ -147,8 +157,13 @@ void consumingStreamFiltering() {
         .filterValues("invoices", "orders")
         .filterMatchUnfiltered(true)
         .builder()
-        .messageHandler((context, message) -> {
-          // message processing
+        .messageHandler((ctx, msg) -> {
+          String filterValue = (String) msg.annotation("x-stream-filter-value");
+          // there must be some client-side filter logic
+          if ("invoices".equals(filterValue) || "orders".equals(filterValue)) {
+            // message processing
+          }
+          ctx.accept();
         })
         .build();
 

src/test/java/com/rabbitmq/client/amqp/impl/SourceFiltersTest.java

@@ -326,7 +326,8 @@ void streamFilteringWithClientSideFiltering() {
             .messageHandler(
                 (ctx, msg) -> {
                   receivedCount.incrementAndGet();
-                  if (selection.equals(msg.subject())) {
+                  if (selection.equals(msg.subject())
+                      && selection.equals(msg.annotation("x-stream-filter-value"))) {
                     selectedMessageCount.incrementAndGet();
                   }
                   ctx.accept();