Document filtering

Author: acogoluegnes

src/docs/asciidoc/advanced-topics.adoc

@@ -2,6 +2,76 @@
 
 === Advanced Topics
 
+==== Filtering
+
+RabbitMQ Stream provides a server-side filtering feature that avoids reading all the messages of a stream and filtering only on the client side.
+This helps to save network bandwidth when a consuming application needs only a subset of messages, e.g. the messages from a given geographical region.
+
+The filtering feature works as follows:
+
+* each message is published with an associated _filter value_
+* a consumer that wants to enable filtering must:
+** define one or several filter values
+** define some client-side filtering logic
+
+Why does the consumer need to define some client-side filtering logic?
+Because the server-side filtering is probabilistic: messages that do not match the filter value(s) can still be sent to the consumer.
+The server uses a https://en.wikipedia.org/wiki/Bloom_filter[bloom filter], _a space-efficient probabilistic data structure_, where false positives are possible.
+Despite this, the filtering saves some bandwidth, which is its primary goal.
+
+===== Filtering on the Publishing Side
+
+Filtering on the publishing side consists in defining some logic to extract the filter value from a message.
+The following snippet shows how to extract the filter value from an application property:
+
+.Declaring a producer with logic to extract a filter value from each message
+[source,java,indent=0]
+--------
+include::{test-examples}/FilteringUsage.java[tag=producer-simple]
+--------
+<1> Get filter value from `state` application property
+
+Note the filter value can be null: the message is then published in a regular way.
+It is called in this context an _unfiltered_ message.
+
+===== Filtering on the Consuming Side
+
+A consumer needs to set up one or several filter values and some filtering logic to enable filtering.
+The filtering logic must be consistent with the filter values.
+In the next snippet, the consumer wants to process only messages from the state of California.
+It sets a filter value to `california` and a predicate that accepts a message only if the `state` application properties is `california`:
+
+.Declaring a consumer with a filter value and filtering logic
+[source,java,indent=0]
+--------
+include::{test-examples}/FilteringUsage.java[tag=consumer-simple]
+--------
+<1> Set filter value
+<2> Set filtering logic
+
+The filter logic is a `Predicate<Message>`.
+It must return `true` if a message is accepted, following the same semantics as `java.util.stream.Stream#filter(Predicate)`.
+
+As stated above, not all messages must have an associated filter value.
+Many applications may not need some filtering, so they can publish messages the regular way.
+And a stream can contain messages with and without an associated filter value.
+
+By default, messages without a filter value (a.k.a _unfiltered_ messages) are not sent to a consumer that enabled filtering.
+
+But what if a consumer wants to process messages with a filter value and messages without any filter value as well?
+It must use the `matchUnfiltered()` method in its declaration and also make sure to keep the filtering logic consistent:
+
+.Declaring a consumer with a filter value and filtering logic
+[source,java,indent=0]
+--------
+include::{test-examples}/FilteringUsage.java[tag=consumer-match-unfiltered]
+--------
+<1> Request messages from California
+<2> Request messages without a filter value as well
+<3> Let both types of messages pass
+
+In the example above, the filtering logic has been adapted to let pass `california` messages _and_ messages without a state set as well.
+
 ==== Using Native `epoll`
 
 The stream Java client uses the https://netty.io/[Netty] network framework and its Java NIO transport implementation by default.

src/main/java/com/rabbitmq/stream/ConsumerBuilder.java

@@ -152,6 +152,11 @@ public interface ConsumerBuilder {
    */
   ConsumerBuilder noTrackingStrategy();
 
+  /**
+   * Configure the filtering.
+   *
+   * @return the filtering configuration
+   */
   FilterConfiguration filter();
 
   /**
@@ -213,16 +218,53 @@ interface AutoTrackingStrategy {
     ConsumerBuilder builder();
   }
 
+  /** Filter configuration. */
   interface FilterConfiguration {
 
+    /**
+     * Set the filter values.
+     *
+     * @param filterValues
+     * @return this filter configuration instance
+     */
     FilterConfiguration values(String... filterValues);
 
+    /**
+     * Client-side filtering logic.
+     *
+     * <p>It must be consistent with the requested filter {@link #values( String...)} and the {@link
+     * #matchUnfiltered()} flag.
+     *
+     * @param filter a predicate that returns <code>true</code> if a message should go to the {@link
+     *     MessageHandler}
+     * @return this filter configuration instance
+     */
     FilterConfiguration filter(Predicate<Message> filter);
 
+    /**
+     * Whether messages without a filter value should be sent as well.
+     *
+     * <p>Default is false.
+     *
+     * @return this filter configuration instance
+     */
     FilterConfiguration matchUnfiltered();
 
+    /**
+     * Whether messages without a filter value should be sent as well.
+     *
+     * <p>Default is false.
+     *
+     * @param matchUnfiltered
+     * @return this filter configuration instance
+     */
     FilterConfiguration matchUnfiltered(boolean matchUnfiltered);
 
+    /**
+     * Go back to the builder.
+     *
+     * @return the consumer builder
+     */
     ConsumerBuilder builder();
   }
 }

src/main/java/com/rabbitmq/stream/ProducerBuilder.java

@@ -132,6 +132,12 @@ public interface ProducerBuilder {
    */
   ProducerBuilder enqueueTimeout(Duration timeout);
 
+  /**
+   * Logic to extract a filter value from a message.
+   *
+   * @param filterValueExtractor
+   * @return
+   */
   ProducerBuilder filter(Function<Message, String> filterValueExtractor);
 
   /**

src/main/java/com/rabbitmq/stream/impl/StreamConsumerBuilder.java

@@ -389,6 +389,9 @@ private DefaultFilterConfiguration(StreamConsumerBuilder builder) {
 
     @Override
     public FilterConfiguration values(String... filterValues) {
+      if (filterValues == null || filterValues.length == 0) {
+        throw new IllegalArgumentException("At least one filter value must be specified");
+      }
       this.filterValues = Arrays.asList(filterValues);
       return this;
     }

src/test/java/com/rabbitmq/stream/docs/FilteringUsage.java

@@ -0,0 +1,69 @@
+// Copyright (c) 2023 VMware, Inc. or its affiliates.  All rights reserved.
+//
+// This software, the RabbitMQ Stream Java client library, is dual-licensed under the
+// Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").
+// For the MPL, please see LICENSE-MPL-RabbitMQ. For the ASL,
+// please see LICENSE-APACHE2.
+//
+// This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND,
+// either express or implied. See the LICENSE file for specific language governing
+// rights and limitations of this software.
+//
+// If you have any questions regarding licensing, please contact us at
+// [email protected].
+
+package com.rabbitmq.stream.docs;
+
+import com.rabbitmq.stream.Consumer;
+import com.rabbitmq.stream.Environment;
+import com.rabbitmq.stream.Producer;
+
+public class FilteringUsage {
+
+  void producerSimple() {
+    Environment environment = Environment.builder().build();
+    // tag::producer-simple[]
+    Producer producer = environment.producerBuilder()
+      .stream("invoices")
+      .filter(msg ->
+        msg.getApplicationProperties().get("state").toString())  // <1>
+      .build();
+    // end::producer-simple[]
+  }
+
+  void consumerSimple() {
+    Environment environment = Environment.builder().build();
+    // tag::consumer-simple[]
+    String filterValue = "california";
+    Consumer consumer = environment.consumerBuilder()
+      .stream("invoices")
+      .filter()
+        .values(filterValue)  // <1>
+        .filter(msg ->
+          filterValue.equals(msg.getApplicationProperties().get("state")))  // <2>
+      .builder()
+      .messageHandler((ctx, msg) -> { })
+      .build();
+    // end::consumer-simple[]
+  }
+
+  void consumerMatchUnfiltered() {
+    Environment environment = Environment.builder().build();
+    // tag::consumer-match-unfiltered[]
+    String filterValue = "california";
+    Consumer consumer = environment.consumerBuilder()
+      .stream("invoices")
+      .filter()
+        .values(filterValue)  // <1>
+        .matchUnfiltered()  // <2>
+        .filter(msg ->
+            filterValue.equals(msg.getApplicationProperties().get("state"))
+            || !msg.getApplicationProperties().containsKey("state")  // <3>
+        )
+      .builder()
+      .messageHandler((ctx, msg) -> { })
+      .build();
+    // end::consumer-match-unfiltered[]
+  }
+
+}