Support filter expressions

References rabbitmq/rabbitmq-server#12415
rabbitmq · Oct 7, 2024 · fbb3d4e · fbb3d4e
1 parent 63a91d0
commit fbb3d4e
Show file tree

Hide file tree

Showing 6 changed files with 232 additions and 37 deletions.
diff --git a/.github/workflows/test-pr.yml b/.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:amqp-filtex'
       - name: Start toxiproxy
         run: ci/start-toxiproxy.sh
       - name: Display Java version

diff --git a/src/main/java/com/rabbitmq/client/amqp/ConsumerBuilder.java b/src/main/java/com/rabbitmq/client/amqp/ConsumerBuilder.java
@@ -209,6 +209,22 @@ interface FilterOptions<T> {
 
     T subject(String subject);
 
+    T replyTo(String replyTo);
+
+    T contentType(String contentType);
+
+    T contentEncoding(String contentEncoding);
+
+    T absoluteExpiryTime(long absoluteExpiryTime);
+
+    T creationTime(long creationTime);
+
+    T groupId(String groupId);
+
+    T groupSequence(int groupSequence);
+
+    T replyToGroupId(String groupId);
+
     T property(String key, boolean value);
 
     T property(String key, byte value);

diff --git a/src/main/java/com/rabbitmq/client/amqp/impl/AmqpConsumerBuilder.java b/src/main/java/com/rabbitmq/client/amqp/impl/AmqpConsumerBuilder.java
@@ -296,6 +296,46 @@ public StreamFilterOptions subject(String subject) {
       return propertyFilter("subject", subject);
     }
 
+    @Override
+    public StreamFilterOptions replyTo(String replyTo) {
+      return propertyFilter("reply-to", replyTo);
+    }
+
+    @Override
+    public StreamFilterOptions contentType(String contentType) {
+      return propertyFilter("content-type", Symbol.valueOf(contentType));
+    }
+
+    @Override
+    public StreamFilterOptions contentEncoding(String contentEncoding) {
+      return propertyFilter("content-encoding", Symbol.valueOf(contentEncoding));
+    }
+
+    @Override
+    public StreamFilterOptions absoluteExpiryTime(long absoluteExpiryTime) {
+      return propertyFilter("absolute-expiry-time", new Date(absoluteExpiryTime));
+    }
+
+    @Override
+    public StreamFilterOptions creationTime(long creationTime) {
+      return propertyFilter("creation-time", new Date(creationTime));
+    }
+
+    @Override
+    public StreamFilterOptions groupId(String groupId) {
+      return propertyFilter("group-id", groupId);
+    }
+
+    @Override
+    public StreamFilterOptions groupSequence(int groupSequence) {
+      return propertyFilter("group-sequence", UnsignedInteger.valueOf(groupSequence));
+    }
+
+    @Override
+    public StreamFilterOptions replyToGroupId(String groupId) {
+      return propertyFilter("reply-to-group-id", groupId);
+    }
+
     @Override
     public StreamFilterOptions property(String key, boolean value) {
       return this.applicationPropertyFilter(key, value);

diff --git a/src/main/java/com/rabbitmq/client/amqp/impl/AmqpMessage.java b/src/main/java/com/rabbitmq/client/amqp/impl/AmqpMessage.java
@@ -241,7 +241,7 @@ public Message groupSequence(int groupSequence) {
 
   @Override
   public Message replyToGroupId(String groupId) {
-    callOnDelegate(m -> replyToGroupId(groupId));
+    callOnDelegate(m -> m.replyToGroupId(groupId));
     return this;
   }
 

diff --git a/src/test/java/com/rabbitmq/client/amqp/impl/Assertions.java b/src/test/java/com/rabbitmq/client/amqp/impl/Assertions.java
@@ -23,6 +23,7 @@
 import com.rabbitmq.client.amqp.Message;
 import java.time.Duration;
 import java.util.List;
+import java.util.Objects;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
@@ -232,23 +233,15 @@ MessageAssert hasId(long id) {
     }
 
     MessageAssert hasId(Object id) {
-      isNotNull();
-      if (!actual.messageId().equals(id)) {
-        fail("Message ID should be '%s' but is '%s'", id, actual.messageId());
-      }
-      return this;
+      return hasField("id", actual.messageId(), id);
     }
 
     MessageAssert hasCorrelationId(long correlationId) {
       return this.hasCorrelationId(new UnsignedLong(correlationId));
     }
 
     MessageAssert hasCorrelationId(Object id) {
-      isNotNull();
-      if (!actual.correlationId().equals(id)) {
-        fail("Correlation ID should be '%s' but is '%s'", id, actual.correlationId());
-      }
-      return this;
+      return hasField("correlation-id", actual.correlationId(), id);
     }
 
     MessageAssert hasUserId(byte[] userId) {
@@ -258,18 +251,55 @@ MessageAssert hasUserId(byte[] userId) {
     }
 
     MessageAssert hasTo(String to) {
+      return hasField("to", actual.to(), to);
+    }
+
+    MessageAssert hasSubject(String subject) {
+      return hasField("subject", actual.subject(), subject);
+    }
+
+    MessageAssert hasReplyTo(String replyTo) {
+      return hasField("reply-to", actual.replyTo(), replyTo);
+    }
+
+    MessageAssert hasContentType(String contentType) {
+      return hasField("content-type", actual.contentType(), contentType);
+    }
+
+    MessageAssert hasContentEncoding(String contentEncoding) {
+      return hasField("content-encoding", actual.contentEncoding(), contentEncoding);
+    }
+
+    MessageAssert hasAbsoluteExpiryTime(long absoluteExpiryTime) {
       isNotNull();
-      if (!actual.to().equals(to)) {
-        fail("To field should be '%s' but is '%s'", to, actual.to());
-      }
+      org.assertj.core.api.Assertions.assertThat(actual.absoluteExpiryTime())
+          .isEqualTo(absoluteExpiryTime);
       return this;
     }
 
-    MessageAssert hasSubject(String subject) {
+    MessageAssert hasCreationTime(long creationTime) {
       isNotNull();
-      if (!actual.subject().equals(subject)) {
-        fail("Subject should be '%s' but is '%s'", subject, actual.subject());
-      }
+      org.assertj.core.api.Assertions.assertThat(actual.creationTime()).isEqualTo(creationTime);
+      return this;
+    }
+
+    MessageAssert hasGroupId(String groupId) {
+      return hasField("group-id", actual.groupId(), groupId);
+    }
+
+    MessageAssert hasGroupSequence(long groupSequence) {
+      isNotNull();
+      org.assertj.core.api.Assertions.assertThat(actual.groupSequence()).isEqualTo(groupSequence);
+      return this;
+    }
+
+    MessageAssert hasReplyToGroupId(String groupId) {
+      return hasField("reply-to-group-id", actual.replyToGroupId(), groupId);
+    }
+
+    MessageAssert hasBody(byte[] body) {
+      isNotNull();
+      org.assertj.core.api.Assertions.assertThat(actual.body()).isEqualTo(body);
       return this;
     }
 
@@ -324,6 +354,14 @@ MessageAssert doesNotHaveAnnotation(String key) {
       }
       return this;
     }
+
+    private MessageAssert hasField(String fieldLabel, Object value, Object expected) {
+      isNotNull();
+      if (!Objects.equals(value, expected)) {
+        fail("Field '%s' should be '%s' but is '%s'", fieldLabel, expected, value);
+      }
+      return this;
+    }
   }
 
   static class ConnectionAssert extends AbstractObjectAssert<ConnectionAssert, AmqpConnection> {