Use "filter."-like subscription properties

acogoluegnes · acogoluegnes · commit 4c908f3ba5bb · 2023-05-26T11:58:33.000+02:00
And deal with null filter values.

 Conflicts:
	pom.xml
diff --git a/pom.xml b/pom.xml
@@ -92,7 +92,7 @@
     <buildnumber.plugin.version>1.4</buildnumber.plugin.version>
     <jmh.version>1.36</jmh.version>
     <spotless.version>2.37.0</spotless.version>
-    <google-java-format.version>1.15.0</google-java-format.version>
+    <google-java-format.version>1.17.0</google-java-format.version>
     <jacoco.version>0.8.10</jacoco.version>
     <!-- for documentation -->
     <broker.version>3.11</broker.version>
diff --git a/src/main/java/com/rabbitmq/stream/ConsumerBuilder.java b/src/main/java/com/rabbitmq/stream/ConsumerBuilder.java
@@ -219,6 +219,10 @@ interface FilterConfiguration {
 
     FilterConfiguration filter(Predicate<Message> filter);
 
+    FilterConfiguration matchUnfiltered();
+
+    FilterConfiguration matchUnfiltered(boolean matchUnfiltered);
+
     ConsumerBuilder builder();
   }
 }
diff --git a/src/main/java/com/rabbitmq/stream/impl/Client.java b/src/main/java/com/rabbitmq/stream/impl/Client.java
@@ -13,34 +13,7 @@
 // info@rabbitmq.com.
 package com.rabbitmq.stream.impl;
 
-import static com.rabbitmq.stream.Constants.COMMAND_CLOSE;
-import static com.rabbitmq.stream.Constants.COMMAND_CONSUMER_UPDATE;
-import static com.rabbitmq.stream.Constants.COMMAND_CREATE_STREAM;
-import static com.rabbitmq.stream.Constants.COMMAND_CREDIT;
-import static com.rabbitmq.stream.Constants.COMMAND_DECLARE_PUBLISHER;
-import static com.rabbitmq.stream.Constants.COMMAND_DELETE_PUBLISHER;
-import static com.rabbitmq.stream.Constants.COMMAND_DELETE_STREAM;
-import static com.rabbitmq.stream.Constants.COMMAND_EXCHANGE_COMMAND_VERSIONS;
-import static com.rabbitmq.stream.Constants.COMMAND_HEARTBEAT;
-import static com.rabbitmq.stream.Constants.COMMAND_METADATA;
-import static com.rabbitmq.stream.Constants.COMMAND_OPEN;
-import static com.rabbitmq.stream.Constants.COMMAND_PARTITIONS;
-import static com.rabbitmq.stream.Constants.COMMAND_PEER_PROPERTIES;
-import static com.rabbitmq.stream.Constants.COMMAND_PUBLISH;
-import static com.rabbitmq.stream.Constants.COMMAND_QUERY_OFFSET;
-import static com.rabbitmq.stream.Constants.COMMAND_QUERY_PUBLISHER_SEQUENCE;
-import static com.rabbitmq.stream.Constants.COMMAND_ROUTE;
-import static com.rabbitmq.stream.Constants.COMMAND_SASL_AUTHENTICATE;
-import static com.rabbitmq.stream.Constants.COMMAND_SASL_HANDSHAKE;
-import static com.rabbitmq.stream.Constants.COMMAND_STORE_OFFSET;
-import static com.rabbitmq.stream.Constants.COMMAND_STREAM_STATS;
-import static com.rabbitmq.stream.Constants.COMMAND_SUBSCRIBE;
-import static com.rabbitmq.stream.Constants.COMMAND_UNSUBSCRIBE;
-import static com.rabbitmq.stream.Constants.RESPONSE_CODE_AUTHENTICATION_FAILURE;
-import static com.rabbitmq.stream.Constants.RESPONSE_CODE_AUTHENTICATION_FAILURE_LOOPBACK;
-import static com.rabbitmq.stream.Constants.RESPONSE_CODE_OK;
-import static com.rabbitmq.stream.Constants.RESPONSE_CODE_SASL_CHALLENGE;
-import static com.rabbitmq.stream.Constants.VERSION_1;
+import static com.rabbitmq.stream.Constants.*;
 import static com.rabbitmq.stream.impl.Utils.encodeRequestCode;
 import static com.rabbitmq.stream.impl.Utils.encodeResponseCode;
 import static com.rabbitmq.stream.impl.Utils.extractResponseCode;
@@ -216,6 +189,7 @@ public long applyAsLong(Object value) {
   private final Duration rpcTimeout;
   private volatile ShutdownReason shutdownReason = null;
   private final Runnable exchangeCommandVersionsCheck;
+  private final boolean filteringSupported;
 
   public Client() {
     this(new ClientParameters());
@@ -398,18 +372,25 @@ public void initChannel(SocketChannel ch) {
           tuneState.getHeartbeat());
       this.connectionProperties = open(parameters.virtualHost);
       Set<FrameHandlerInfo> supportedCommands = maybeExchangeCommandVersions();
-      if (supportedCommands.stream()
-          .filter(i -> i.getKey() == COMMAND_STREAM_STATS)
-          .findAny()
-          .isPresent()) {
-        this.exchangeCommandVersionsCheck = () -> {};
-      } else {
-        this.exchangeCommandVersionsCheck =
-            () -> {
-              throw new UnsupportedOperationException(
-                  "QueryStreamInfo is available only on RabbitMQ 3.11 or more.");
-            };
-      }
+      AtomicReference<Runnable> exchangeCommandVersionsCheckReference = new AtomicReference<>();
+      AtomicBoolean filteringSupportedReference = new AtomicBoolean(false);
+      supportedCommands.forEach(
+          c -> {
+            if (c.getKey() == COMMAND_STREAM_STATS) {
+              exchangeCommandVersionsCheckReference.set(() -> {});
+            }
+            if (c.getKey() == COMMAND_PUBLISH && c.getMaxVersion() >= VERSION_2) {
+              filteringSupportedReference.set(true);
+            }
+          });
+      this.exchangeCommandVersionsCheck =
+          exchangeCommandVersionsCheckReference.get() == null
+              ? () -> {
+                throw new UnsupportedOperationException(
+                    "QueryStreamInfo is available only on RabbitMQ 3.11 or more.");
+              }
+              : exchangeCommandVersionsCheckReference.get();
+      this.filteringSupported = filteringSupportedReference.get();
       started.set(true);
       this.metricsCollector.openConnection();
     } catch (RuntimeException e) {
@@ -1414,6 +1395,10 @@ private String serverAddress() {
     }
   }
 
+  boolean filteringSupported() {
+    return this.filteringSupported;
+  }
+
   public List<String> route(String routingKey, String superStream) {
     if (routingKey == null || superStream == null) {
       throw new IllegalArgumentException("routing key and stream must not be null");
@@ -1615,11 +1600,7 @@ private Set<FrameHandlerInfo> maybeExchangeCommandVersions() {
     Set<FrameHandlerInfo> supported = new HashSet<>();
     try {
       if (Utils.is3_11_OrMore(brokerVersion())) {
-        for (FrameHandlerInfo info : exchangeCommandVersions()) {
-          if (info.getKey() == COMMAND_STREAM_STATS) {
-            supported.add(info);
-          }
-        }
+        supported.addAll(exchangeCommandVersions());
       }
     } catch (Exception e) {
       LOGGER.info("Error while exchanging command versions: {}", e.getMessage());
diff --git a/src/main/java/com/rabbitmq/stream/impl/StreamConsumer.java b/src/main/java/com/rabbitmq/stream/impl/StreamConsumer.java
@@ -76,7 +76,9 @@ class StreamConsumer implements Consumer {
       ConsumerUpdateListener consumerUpdateListener,
       int initialCredits,
       int additionalCredits) {
-
+    if (Utils.filteringEnabled(subscriptionProperties) && !environment.filteringSupported()) {
+      throw new IllegalArgumentException("Filtering is not supported by the broker");
+    }
     this.id = ID_SEQUENCE.getAndIncrement();
     Runnable trackingClosingCallback;
     try {
diff --git a/src/main/java/com/rabbitmq/stream/impl/StreamConsumerBuilder.java b/src/main/java/com/rabbitmq/stream/impl/StreamConsumerBuilder.java
@@ -13,6 +13,9 @@
 // info@rabbitmq.com.
 package com.rabbitmq.stream.impl;
 
+import static com.rabbitmq.stream.impl.Utils.SUBSCRIPTION_PROPERTY_FILTER_PREFIX;
+import static com.rabbitmq.stream.impl.Utils.SUBSCRIPTION_PROPERTY_MATCH_UNFILTERED;
+
 import com.rabbitmq.stream.*;
 import java.lang.reflect.Field;
 import java.lang.reflect.Modifier;
@@ -21,6 +24,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Predicate;
 
 class StreamConsumerBuilder implements ConsumerBuilder {
@@ -88,7 +92,7 @@ public ConsumerBuilder name(String name) {
 
   @Override
   public ConsumerBuilder singleActiveConsumer() {
-    this.subscriptionProperties.put("single-active-consumer", "true");
+    this.subscriptionProperties.put(Utils.SUBSCRIPTION_PROPERTY_SAC, "true");
     return this;
   }
 
@@ -200,8 +204,14 @@ public Consumer build() {
       handler = this.messageHandler;
     } else {
       this.filterConfiguration.validate();
+      AtomicInteger i = new AtomicInteger(0);
+      this.filterConfiguration.filterValues.forEach(
+          v ->
+              this.subscriptionProperties.put(
+                  SUBSCRIPTION_PROPERTY_FILTER_PREFIX + i.getAndIncrement(), v));
       this.subscriptionProperties.put(
-          "filters", String.join(",", this.filterConfiguration.filterValues));
+          SUBSCRIPTION_PROPERTY_MATCH_UNFILTERED,
+          this.filterConfiguration.matchUnfiltered ? "true" : "false");
       final Predicate<Message> filter = this.filterConfiguration.filter;
       final MessageHandler delegate = this.messageHandler;
       handler =
@@ -231,7 +241,7 @@ public Consumer build() {
       environment.addConsumer((StreamConsumer) consumer);
     } else {
       if (Utils.isSac(this.subscriptionProperties)) {
-        this.subscriptionProperties.put("super-stream", this.superStream);
+        this.subscriptionProperties.put(Utils.SUBSCRIPTION_PROPERTY_SUPER_STREAM, this.superStream);
       }
       consumer =
           new SuperStreamConsumer(this, this.superStream, this.environment, trackingConfiguration);
@@ -371,14 +381,14 @@ private static final class DefaultFilterConfiguration implements FilterConfigura
     private final StreamConsumerBuilder builder;
     private List<String> filterValues;
     private Predicate<Message> filter;
+    private boolean matchUnfiltered = false;
 
     private DefaultFilterConfiguration(StreamConsumerBuilder builder) {
       this.builder = builder;
     }
 
     @Override
     public FilterConfiguration values(String... filterValues) {
-      // FIXME: check for ',' in values
       this.filterValues = Arrays.asList(filterValues);
       return this;
     }
@@ -389,6 +399,18 @@ public FilterConfiguration filter(Predicate<Message> filter) {
       return this;
     }
 
+    @Override
+    public FilterConfiguration matchUnfiltered() {
+      this.matchUnfiltered = true;
+      return this;
+    }
+
+    @Override
+    public FilterConfiguration matchUnfiltered(boolean matchUnfiltered) {
+      this.matchUnfiltered = matchUnfiltered;
+      return this;
+    }
+
     @Override
     public ConsumerBuilder builder() {
       return this.builder;
diff --git a/src/main/java/com/rabbitmq/stream/impl/StreamEnvironment.java b/src/main/java/com/rabbitmq/stream/impl/StreamEnvironment.java
@@ -744,6 +744,10 @@ static <T> T locatorOperation(
     return result;
   }
 
+  boolean filteringSupported() {
+    return this.locatorOperation(Client::filteringSupported);
+  }
+
   Clock clock() {
     return this.clock;
   }
diff --git a/src/main/java/com/rabbitmq/stream/impl/StreamProducer.java b/src/main/java/com/rabbitmq/stream/impl/StreamProducer.java
@@ -94,6 +94,9 @@ class StreamProducer implements Producer {
       Duration enqueueTimeout,
       Function<Message, String> filterValueExtractor,
       StreamEnvironment environment) {
+    if (filterValueExtractor != null && !environment.filteringSupported()) {
+      throw new IllegalArgumentException("Filtering is not supported by the broker");
+    }
     this.id = ID_SEQUENCE.getAndIncrement();
     this.environment = environment;
     this.name = name;
@@ -606,8 +609,12 @@ private static final class OutboundMessageFilterValueWriterCallback
     public int write(ByteBuf bb, Object entity, long publishingId) {
       AccumulatedEntity accumulatedEntity = (AccumulatedEntity) entity;
       String filterValue = accumulatedEntity.filterValue();
-      bb.writeShort(filterValue.length());
-      bb.writeBytes(filterValue.getBytes(StandardCharsets.UTF_8));
+      if (filterValue == null) {
+        bb.writeShort(-1);
+      } else {
+        bb.writeShort(filterValue.length());
+        bb.writeBytes(filterValue.getBytes(StandardCharsets.UTF_8));
+      }
       Codec.EncodedMessage messageToPublish =
           (Codec.EncodedMessage) accumulatedEntity.encodedEntity();
       bb.writeInt(messageToPublish.getSize());
@@ -619,7 +626,12 @@ public int write(ByteBuf bb, Object entity, long publishingId) {
     public int fragmentLength(Object entity) {
       AccumulatedEntity accumulatedEntity = (AccumulatedEntity) entity;
       Codec.EncodedMessage message = (Codec.EncodedMessage) accumulatedEntity.encodedEntity();
-      return 8 + 2 + accumulatedEntity.filterValue().length() + 4 + message.getSize();
+      String filterValue = accumulatedEntity.filterValue();
+      if (filterValue == null) {
+        return 8 + 2 + 4 + message.getSize();
+      } else {
+        return 8 + 2 + accumulatedEntity.filterValue().length() + 4 + message.getSize();
+      }
     }
   }
 }
diff --git a/src/main/java/com/rabbitmq/stream/impl/Utils.java b/src/main/java/com/rabbitmq/stream/impl/Utils.java
@@ -65,6 +65,11 @@ final class Utils {
   private static final Logger LOGGER = LoggerFactory.getLogger(Utils.class);
   private static final Map<Short, String> CONSTANT_LABELS;
 
+  static final String SUBSCRIPTION_PROPERTY_SAC = "single-active-consumer";
+  static final String SUBSCRIPTION_PROPERTY_SUPER_STREAM = "super-stream";
+  static final String SUBSCRIPTION_PROPERTY_FILTER_PREFIX = "filter.";
+  static final String SUBSCRIPTION_PROPERTY_MATCH_UNFILTERED = "match-unfiltered";
+
   static {
     Map<Short, String> labels = new HashMap<>();
     Arrays.stream(Constants.class.getDeclaredFields())
@@ -119,6 +124,15 @@ static boolean isSac(Map<String, String> properties) {
     }
   }
 
+  static boolean filteringEnabled(Map<String, String> properties) {
+    if (properties == null || properties.isEmpty()) {
+      return false;
+    } else {
+      return properties.keySet().stream()
+          .anyMatch(k -> k.startsWith(SUBSCRIPTION_PROPERTY_FILTER_PREFIX));
+    }
+  }
+
   static short encodeRequestCode(Short code) {
     return code;
   }
diff --git a/src/main/java/com/rabbitmq/stream/perf/StreamPerfTest.java b/src/main/java/com/rabbitmq/stream/perf/StreamPerfTest.java
@@ -123,6 +123,7 @@ public class StreamPerfTest implements Callable<Integer> {
 
   @CommandLine.Mixin
   private final CommandLine.HelpCommand helpCommand = new CommandLine.HelpCommand();
+
   // for testing
   private final AddressResolver addressResolver;
   private final PrintWriter err, out;
@@ -477,6 +478,7 @@ static class InstanceSyncOptions {
   private List<Monitoring> monitorings;
   private volatile Environment environment;
   private volatile EventLoopGroup eventLoopGroup;
+
   // constructor for completion script generation
   public StreamPerfTest() {
     this(null, null, null, null);
diff --git a/src/test/java/com/rabbitmq/stream/impl/ClientTest.java b/src/test/java/com/rabbitmq/stream/impl/ClientTest.java
@@ -1093,7 +1093,7 @@ public int fragmentLength(Object obj) {
         stream,
         OffsetSpecification.first(),
         1,
-        Collections.singletonMap("filters", newFilterValue));
+        Collections.singletonMap("filter.1", newFilterValue));
 
     int expectedCount = filterValueCount.get(newFilterValue).get();
     waitAtMost(() -> filteredConsumedMessageCount.get() == expectedCount);
diff --git a/src/test/java/com/rabbitmq/stream/impl/FilteringTest.java b/src/test/java/com/rabbitmq/stream/impl/FilteringTest.java
diff --git a/src/test/java/com/rabbitmq/stream/impl/MicroStreamTest.java b/src/test/java/com/rabbitmq/stream/impl/MicroStreamTest.java
diff --git a/src/test/java/com/rabbitmq/stream/impl/TestUtils.java b/src/test/java/com/rabbitmq/stream/impl/TestUtils.java

Original file line number	Diff line number	Diff line change
`@@ -219,6 +219,10 @@ interface FilterConfiguration {`
`219`	`219`
`220`	`220`	`FilterConfiguration filter(Predicate<Message> filter);`
`221`	`221`
	`222`	`+ FilterConfiguration matchUnfiltered();`
	`223`	`+`
	`224`	`+ FilterConfiguration matchUnfiltered(boolean matchUnfiltered);`
	`225`	`+`
`222`	`226`	`ConsumerBuilder builder();`
`223`	`227`	`}`
`224`	`228`	`}`
Original file line number	Diff line number	Diff line change
`@@ -744,6 +744,10 @@ static <T> T locatorOperation(`
`744`	`744`	`return result;`
`745`	`745`	`}`
`746`	`746`
	`747`	`+ boolean filteringSupported() {`
	`748`	`+ return this.locatorOperation(Client::filteringSupported);`
	`749`	`+ }`
	`750`	`+`
`747`	`751`	`Clock clock() {`
`748`	`752`	`return this.clock;`
`749`	`753`	`}`