Support filter size on stream creation

Author: acogoluegnes

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

@@ -14,6 +14,7 @@
 package com.rabbitmq.stream;
 
 import java.time.Duration;
+import java.util.function.Function;
 
 /** API to configure and create a stream. */
 public interface StreamCreator {
@@ -63,6 +64,16 @@ public interface StreamCreator {
    */
   StreamCreator leaderLocator(LeaderLocator leaderLocator);
 
+  /**
+   * Set the size of the stream chunk filters.
+   *
+   * @param size
+   * @return this creator instance
+   * @see ProducerBuilder#filterValue( Function)
+   * @see ConsumerBuilder#filter()
+   */
+  StreamCreator filterSize(int size);
+
   /**
    * Create the stream.
    *

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

@@ -2582,6 +2582,14 @@ public StreamParametersBuilder leaderLocator(LeaderLocator leaderLocator) {
       return this;
     }
 
+    public StreamParametersBuilder filterSize(int size) {
+      if (size < 16 || size > 255) {
+        throw new IllegalArgumentException("Stream filter size must be between 16 and 255");
+      }
+      this.parameters.put("stream-filter-size-bytes", String.valueOf(size));
+      return this;
+    }
+
     public StreamParametersBuilder put(String key, String value) {
       parameters.put(key, value);
       return this;

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

@@ -67,6 +67,12 @@ public StreamCreator leaderLocator(LeaderLocator leaderLocator) {
     return this;
   }
 
+  @Override
+  public StreamCreator filterSize(int size) {
+    streamParametersBuilder.filterSize(size);
+    return this;
+  }
+
   @Override
   public void create() {
     if (stream == null) {

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

@@ -15,6 +15,7 @@
 
 import static com.rabbitmq.stream.impl.TestUtils.*;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import com.rabbitmq.stream.*;
 import io.netty.channel.EventLoopGroup;
@@ -28,6 +29,8 @@
 import java.util.stream.IntStream;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInfo;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.CsvSource;
@@ -183,6 +186,17 @@ void matchUnfilteredShouldReturnNullFilteredValueAndFilteredValues(
         });
   }
 
+  @Test
+  void setFilterSizeOnCreation(TestInfo info) {
+    String s = streamName(info);
+    this.environment.streamCreator().stream(s).filterSize(128).create();
+    this.environment.deleteStream(s);
+    assertThatThrownBy(() -> this.environment.streamCreator().filterSize(15))
+        .isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(() -> this.environment.streamCreator().filterSize(256))
+        .isInstanceOf(IllegalArgumentException.class);
+  }
+
   private ProducerBuilder producerBuilder() {
     return this.environment.producerBuilder().stream(stream);
   }