Start support for filter expressions

WIP

Author: acogoluegnes

pom.xml

@@ -50,6 +50,7 @@
         <junit.jupiter.version>5.11.2</junit.jupiter.version>
         <assertj.version>3.26.3</assertj.version>
         <mockito.version>5.14.1</mockito.version>
+        <jqwik.version>1.9.1</jqwik.version>
         <amqp-client.version>5.20.0</amqp-client.version>
         <micrometer-tracing-test.version>1.3.4</micrometer-tracing-test.version>
         <micrometer-docs-generator.version>1.0.4</micrometer-docs-generator.version>
@@ -184,6 +185,13 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>net.jqwik</groupId>
+            <artifactId>jqwik</artifactId>
+            <version>${jqwik.version}</version>
+            <scope>test</scope>
+        </dependency>
+
         <dependency>
             <groupId>eu.rekawek.toxiproxy</groupId>
             <artifactId>toxiproxy-java</artifactId>

src/main/java/com/rabbitmq/client/amqp/ConsumerBuilder.java

@@ -17,7 +17,10 @@
 // [email protected].
 package com.rabbitmq.client.amqp;
 
+import java.math.BigDecimal;
 import java.time.Instant;
+import java.util.UUID;
+import org.apache.qpid.protonj2.types.*;
 
 /** API to configure and create a {@link Consumer}. */
 public interface ConsumerBuilder {
@@ -158,6 +161,8 @@ interface StreamOptions {
      */
     StreamOptions filterMatchUnfiltered(boolean matchUnfiltered);
 
+    StreamFilterOptions filter();
+
     /**
      * Return the consumer builder.
      *
@@ -176,6 +181,80 @@ enum StreamOffsetSpecification {
     NEXT
   }
 
+  interface FilterOptions<T> {
+
+    T messageId(Object id);
+
+    T messageId(String id);
+
+    T messageId(long id);
+
+    T messageId(byte[] id);
+
+    T messageId(UUID id);
+
+    T correlationId(Object correlationId);
+
+    T correlationId(String correlationId);
+
+    T correlationId(long correlationId);
+
+    T correlationId(byte[] correlationId);
+
+    T correlationId(UUID correlationId);
+
+    T userId(byte[] userId);
+
+    T to(String to);
+
+    T subject(String subject);
+
+    T property(String key, boolean value);
+
+    T property(String key, byte value);
+
+    T property(String key, short value);
+
+    T property(String key, int value);
+
+    T property(String key, long value);
+
+    T propertyUnsigned(String key, byte value);
+
+    T propertyUnsigned(String key, short value);
+
+    T propertyUnsigned(String key, int value);
+
+    T propertyUnsigned(String key, long value);
+
+    T property(String key, float value);
+
+    T property(String key, double value);
+
+    T propertyDecimal32(String key, BigDecimal value);
+
+    T propertyDecimal64(String key, BigDecimal value);
+
+    T propertyDecimal128(String key, BigDecimal value);
+
+    T property(String key, char value);
+
+    T propertyTimestamp(String key, long value);
+
+    T property(String key, UUID value);
+
+    T property(String key, byte[] value);
+
+    T property(String key, String value);
+
+    T propertySymbol(String key, String value);
+  }
+
+  interface StreamFilterOptions extends FilterOptions<StreamFilterOptions> {
+
+    StreamOptions stream();
+  }
+
   /**
    * Callback to modify a consumer subscription before the link creation.
    *

src/main/java/com/rabbitmq/client/amqp/impl/AmqpConsumer.java

@@ -41,6 +41,7 @@
 import org.apache.qpid.protonj2.engine.impl.ProtonLinkCreditState;
 import org.apache.qpid.protonj2.engine.impl.ProtonReceiver;
 import org.apache.qpid.protonj2.engine.impl.ProtonSessionIncomingWindow;
+import org.apache.qpid.protonj2.types.DescribedType;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -58,7 +59,7 @@ final class AmqpConsumer extends ResourceBase implements Consumer {
   private final Long id;
   private final String address;
   private final String queue;
-  private final Map<String, Object> filters;
+  private final Map<String, DescribedType> filters;
   private final Map<String, Object> linkProperties;
   private final ConsumerBuilder.SubscriptionListener subscriptionListener;
   private final AmqpConnection connection;
@@ -161,7 +162,7 @@ private ClientReceiver createNativeReceiver(
       Session nativeSession,
       String address,
       Map<String, Object> properties,
-      Map<String, Object> filters,
+      Map<String, DescribedType> filters,
       SubscriptionListener subscriptionListener) {
     try {
       filters = new LinkedHashMap<>(filters);
@@ -175,7 +176,7 @@ private ClientReceiver createNativeReceiver(
               .creditWindow(0)
               .properties(properties);
       if (!filters.isEmpty()) {
-        receiverOptions.sourceOptions().filters(filters);
+        receiverOptions.sourceOptions().filters(Map.copyOf(filters));
       }
       return (ClientReceiver)
           ExceptionUtils.wrapGet(nativeSession.openReceiver(address, receiverOptions).openFuture());