Support SQL filter expressions for stream consumers

This is a RabbitMQ 4.2 feature.

Author: acogoluegnes

.github/workflows/test-pr.yml

@@ -24,6 +24,8 @@ jobs:
           cache: 'maven'
       - name: Start broker
         run: ci/start-broker.sh
+        env:
+          RABBITMQ_IMAGE: pivotalrabbitmq/rabbitmq:pr-14110-otp27
       - name: Start toxiproxy
         run: ci/start-toxiproxy.sh
       - name: Display Java version

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

@@ -565,6 +565,27 @@ interface FilterOptions<T> {
      * @return type-parameter object
      */
     T propertySymbol(String key, String value);
+
+    /**
+     * Set an SQL filter expression.
+     *
+     * <p>Section 6 of the AMQP Filter Expressions specification defines the semantics of the
+     * feature, but the <b>SQL syntax follows the <a
+     * href="https://jakarta.ee/specifications/messaging/3.1/jakarta-messaging-spec-3.1#message-selector-syntax">JMS
+     * message selector syntax</a></b>.
+     *
+     * <p>Requires RabbitMQ 4.2 or more.
+     *
+     * @param sql SQL expression
+     * @return type-parameter object
+     * @see <a
+     *     href="https://jakarta.ee/specifications/messaging/3.1/jakarta-messaging-spec-3.1#message-selector-syntax">JMS
+     *     message selector syntax</a>
+     * @see <a
+     *     href="https://groups.oasis-open.org/higherlogic/ws/public/document?document_id=66227">AMQP
+     *     Filter Expressions</a>
+     */
+    T sql(String sql);
   }
 
   /**

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

@@ -26,6 +26,7 @@
 import static com.rabbitmq.client.amqp.impl.Tuples.pair;
 import static com.rabbitmq.client.amqp.impl.Utils.supportFilterExpressions;
 import static com.rabbitmq.client.amqp.impl.Utils.supportSetToken;
+import static com.rabbitmq.client.amqp.impl.Utils.supportSqlFilterExpressions;
 import static java.lang.System.nanoTime;
 import static java.time.Duration.ofNanos;
 
@@ -115,7 +116,9 @@ final class AmqpConnection extends ResourceBase implements Connection {
   private final ConnectionSettings.AffinityStrategy affinityStrategy;
   private final String name;
   private final Lock instanceLock = new ReentrantLock();
-  private final boolean filterExpressionsSupported, setTokenSupported;
+  private final boolean filterExpressionsSupported,
+      setTokenSupported,
+      sqlFilterExpressionsSupported;
   private volatile ConsumerWorkService consumerWorkService;
   private volatile Executor dispatchingExecutor;
   private final boolean privateDispatchingExecutor;
@@ -212,6 +215,7 @@ final class AmqpConnection extends ResourceBase implements Connection {
     String brokerVersion = brokerVersion(this.nativeConnection);
     this.filterExpressionsSupported = supportFilterExpressions(brokerVersion);
     this.setTokenSupported = supportSetToken(brokerVersion);
+    this.sqlFilterExpressionsSupported = supportSqlFilterExpressions(brokerVersion);
     LOGGER.debug("Opened connection '{}' on node '{}'.", this.name(), this.connectionNodename());
     this.state(OPEN);
     this.environment.metricsCollector().openConnection();
@@ -856,6 +860,10 @@ boolean filterExpressionsSupported() {
     return this.filterExpressionsSupported;
   }
 
+  boolean sqlFilterExpressionsSupported() {
+    return this.sqlFilterExpressionsSupported;
+  }
+
   boolean setTokenSupported() {
     return this.setTokenSupported;
   }

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

@@ -226,11 +226,11 @@ static StreamOptions streamOptions(Map<String, DescribedType> filters) {
 
   private static class DefaultStreamFilterOptions implements StreamFilterOptions {
 
-    private final StreamOptions streamOptions;
+    private final DefaultStreamOptions streamOptions;
     private final Map<String, DescribedType> filters;
 
     private DefaultStreamFilterOptions(
-        StreamOptions streamOptions, Map<String, DescribedType> filters) {
+        DefaultStreamOptions streamOptions, Map<String, DescribedType> filters) {
       this.streamOptions = streamOptions;
       this.filters = filters;
     }
@@ -440,6 +440,16 @@ public StreamFilterOptions propertySymbol(String key, String value) {
       return this.applicationPropertyFilter(key, Symbol.valueOf(value));
     }
 
+    @Override
+    public StreamFilterOptions sql(String sql) {
+      if (!this.streamOptions.builder.connection.filterExpressionsSupported()) {
+        throw new IllegalArgumentException(
+            "AMQP SQL filter expressions requires at least RabbitMQ 4.2.0");
+      }
+      this.filters.put("sql-filter", filterValue("apache.org:selector-filter:string", sql));
+      return this;
+    }
+
     @Override
     public StreamOptions stream() {
       return this.streamOptions;

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

@@ -236,6 +236,15 @@ static boolean is4_0_OrMore(String brokerVersion) {
     }
   }
 
+  static boolean is4_2_OrMore(String brokerVersion) {
+    try {
+      return versionCompare(currentVersion(brokerVersion), "4.2.0") >= 0;
+    } catch (Exception e) {
+      LOGGER.debug("Unable to parse broker version {}", brokerVersion, e);
+      return true;
+    }
+  }
+
   static boolean is4_1_OrMore(String brokerVersion) {
     try {
       return versionCompare(currentVersion(brokerVersion), "4.1.0") >= 0;
@@ -253,6 +262,11 @@ static boolean supportSetToken(String brokerVersion) {
     return is4_1_OrMore(brokerVersion);
   }
 
+  static boolean supportSqlFilterExpressions(String brokerVersion) {
+    // TODO should be 4.2
+    return is4_1_OrMore(brokerVersion);
+  }
+
   static final class ObservationConnectionInfo implements ObservationCollector.ConnectionInfo {
 
     private final String address;

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

@@ -469,6 +469,21 @@ void filterExpressionStringModifier() {
     msgs.forEach(m -> assertThat(m).hasSubject("foo bar"));
   }
 
+  @Test
+  // TODO should be 4.2
+  @BrokerVersionAtLeast(RABBITMQ_4_1_0)
+  void filterExpressionSql() {
+    publish(1, m -> m.subject("abc 123"));
+    publish(1, m -> m.subject("foo bar"));
+    publish(1, m -> m.subject("ab 12"));
+
+    List<Message> msgs = consume(2, m -> m.sql("properties.subject LIKE 'ab%'"));
+    msgs.forEach(m -> assertThat(m.subject()).startsWith("ab"));
+
+    msgs = consume(1, m -> m.sql("properties.subject like 'foo%'"));
+    msgs.forEach(m -> assertThat(m).hasSubject("foo bar"));
+  }
+
   void publish(int messageCount) {
     this.publish(messageCount, UnaryOperator.identity());
   }