Merge pull request #635 from rabbitmq/initial-member-count-in-stream-creator

Add initial member count to stream creator

Author: acogoluegnes

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

@@ -92,6 +92,16 @@ public interface StreamCreator {
    */
   StreamCreator filterSize(int size);
 
+  /**
+   * Set the number of initial members the stream should have.
+   *
+   * @param initialMemberCount initial number of nodes
+   * @return this creator instance
+   * @see <a href="https://www.rabbitmq.com/docs/streams#replication-factor">Initial Replication
+   *     Factor</a>
+   */
+  StreamCreator initialMemberCount(int initialMemberCount);
+
   /**
    * Configure the super stream to create.
    *

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

@@ -2744,6 +2744,14 @@ public StreamParametersBuilder filterSize(int size) {
       return this;
     }
 
+    public StreamParametersBuilder initialMemberCount(int initialMemberCount) {
+      if (initialMemberCount <= 0) {
+        throw new IllegalArgumentException("The initial member count must be greater than 0");
+      }
+      this.parameters.put("initial-cluster-size", String.valueOf(initialMemberCount));
+      return this;
+    }
+
     public StreamParametersBuilder put(String key, String value) {
       parameters.put(key, value);
       return this;

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

@@ -86,6 +86,12 @@ public StreamCreator filterSize(int size) {
     return this;
   }
 
+  @Override
+  public StreamCreator initialMemberCount(int initialMemberCount) {
+    streamParametersBuilder.initialMemberCount(initialMemberCount);
+    return this;
+  }
+
   @Override
   public SuperStreamConfiguration superStream() {
     if (this.superStreamConfiguration == null) {

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

@@ -813,4 +813,17 @@ void enforceEntityPerConnectionLimits() {
       executor.shutdownNow();
     }
   }
+
+  @Test
+  void brokerShouldAcceptInitialMemberCountArgument(TestInfo info) {
+    String s = streamName(info);
+    Environment env = environmentBuilder.build();
+    try {
+      env.streamCreator().name(s).initialMemberCount(1).create();
+      assertThat(env.streamExists(s)).isTrue();
+    } finally {
+      env.deleteStream(s);
+      env.close();
+    }
+  }
 }