Refine filtering API

As-per discussion with @kjnilsson and @ggreen.

Author: acogoluegnes

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

@@ -230,7 +230,7 @@ interface FilterConfiguration {
     FilterConfiguration values(String... filterValues);
 
     /**
-     * Client-side filtering logic.
+     * Client-side filtering logic, occurring after the server-side filtering.
      *
      * <p>It must be consistent with the requested filter {@link #values( String...)} and the {@link
      * #matchUnfiltered()} flag.
@@ -239,7 +239,7 @@ interface FilterConfiguration {
      *     MessageHandler}
      * @return this filter configuration instance
      */
-    FilterConfiguration filter(Predicate<Message> filter);
+    FilterConfiguration postFilter(Predicate<Message> filter);
 
     /**
      * Whether messages without a filter value should be sent as well.

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

@@ -136,9 +136,9 @@ public interface ProducerBuilder {
    * Logic to extract a filter value from a message.
    *
    * @param filterValueExtractor
-   * @return
+   * @return this builder instance
    */
-  ProducerBuilder filter(Function<Message, String> filterValueExtractor);
+  ProducerBuilder filterValue(Function<Message, String> filterValueExtractor);
 
   /**
    * Create the {@link Producer} instance.

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

@@ -397,7 +397,7 @@ public FilterConfiguration values(String... filterValues) {
     }
 
     @Override
-    public FilterConfiguration filter(Predicate<Message> filter) {
+    public FilterConfiguration postFilter(Predicate<Message> filter) {
       this.filter = filter;
       return this;
     }

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

@@ -131,7 +131,7 @@ public ProducerBuilder enqueueTimeout(Duration timeout) {
   }
 
   @Override
-  public ProducerBuilder filter(Function<Message, String> filterValueExtractor) {
+  public ProducerBuilder filterValue(Function<Message, String> filterValueExtractor) {
     this.filterValueExtractor = filterValueExtractor;
     return this;
   }

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

@@ -25,7 +25,7 @@ void producerSimple() {
     // tag::producer-simple[]
     Producer producer = environment.producerBuilder()
       .stream("invoices")
-      .filter(msg ->
+      .filterValue(msg ->
         msg.getApplicationProperties().get("state").toString())  // <1>
       .build();
     // end::producer-simple[]
@@ -39,7 +39,7 @@ void consumerSimple() {
       .stream("invoices")
       .filter()
         .values(filterValue)  // <1>
-        .filter(msg ->
+        .postFilter(msg ->
           filterValue.equals(msg.getApplicationProperties().get("state")))  // <2>
       .builder()
       .messageHandler((ctx, msg) -> { })
@@ -56,7 +56,7 @@ void consumerMatchUnfiltered() {
       .filter()
         .values(filterValue)  // <1>
         .matchUnfiltered()  // <2>
-        .filter(msg ->
+        .postFilter(msg ->
             filterValue.equals(msg.getApplicationProperties().get("state"))
             || !msg.getApplicationProperties().containsKey("state")  // <3>
         )

src/test/java/com/rabbitmq/stream/impl/FilteringTest.java

@@ -91,7 +91,7 @@ void publishConsume(String producerName) throws Exception {
     consumerBuilder()
         .filter()
         .values(newFilterValue)
-        .filter(
+        .postFilter(
             m -> {
               receivedMessageCount.incrementAndGet();
               return newFilterValue.equals(m.getProperties().getGroupId());
@@ -142,7 +142,7 @@ void matchUnfilteredShouldReturnNullFilteredValueAndFilteredValues(
         .filter()
         .values(filterValues.get(0))
         .matchUnfiltered(matchUnfiltered)
-        .filter(m -> true)
+        .postFilter(m -> true)
         .builder()
         .messageHandler(
             (ctx, msg) -> {
@@ -176,7 +176,10 @@ private ConsumerBuilder consumerBuilder() {
   private void publish(
       int messageCount, String producerName, Supplier<String> filterValueSupplier) {
     Producer producer =
-        producerBuilder().name(producerName).filter(m -> m.getProperties().getGroupId()).build();
+        producerBuilder()
+            .name(producerName)
+            .filterValue(m -> m.getProperties().getGroupId())
+            .build();
     CountDownLatch latch = new CountDownLatch(messageCount);
     ConfirmationHandler confirmationHandler = ctx -> latch.countDown();
     IntStream.range(0, messageCount)