Add --credits option to grant credits every n chunks

References rabbitmq/rabbitmq-stream-java-client#843

Author: acogoluegnes

src/main/java/com/rabbitmq/stream/perf/Converters.java

@@ -142,7 +142,7 @@ static class NotNegativeIntegerTypeConverter implements CommandLine.ITypeConvert
     @Override
     public Integer convert(String input) {
       try {
-        Integer value = Integer.valueOf(input);
+        int value = Integer.parseInt(input);
         if (value < 0) {
           throw new IllegalArgumentException();
         }
@@ -153,6 +153,35 @@ public Integer convert(String input) {
     }
   }
 
+  static class CreditsTypeConverter implements CommandLine.ITypeConverter<Credits> {
+
+    @Override
+    public Credits convert(String input) {
+      try {
+        int[] credits;
+        if (input.contains("-")) {
+          String[] creditsStr = input.split("-");
+          int initialCredits = Integer.parseInt(creditsStr[0]);
+          int n = Integer.parseInt(creditsStr[1]);
+          if (initialCredits < 0 || n < 0) {
+            throw new IllegalArgumentException();
+          }
+          credits = new int[] {initialCredits, n};
+        } else {
+          int initialCredits = Integer.parseInt(input);
+          if (initialCredits < 0) {
+            throw new IllegalArgumentException();
+          }
+          credits = new int[] {initialCredits, initialCredits / 2};
+        }
+        return new Credits(credits[0], credits[1]);
+      } catch (Exception e) {
+        throw new CommandLine.TypeConversionException(
+            "'" + input + "' is not valid, valid example values: 10-5, 5-2");
+      }
+    }
+  }
+
   static class ByteCapacityTypeConverter implements CommandLine.ITypeConverter<ByteCapacity> {
 
     @Override
@@ -451,4 +480,23 @@ public Boolean convert(String value) {
   private static void throwConversionException(String format, String... arguments) {
     throw new CommandLine.TypeConversionException(String.format(format, (Object[]) arguments));
   }
+
+  static class Credits {
+
+    private final int initialCredits;
+    private final int n;
+
+    Credits(int initialCredits, int n) {
+      this.initialCredits = initialCredits;
+      this.n = n;
+    }
+
+    int initialCredits() {
+      return this.initialCredits;
+    }
+
+    int n() {
+      return this.n;
+    }
+  }
 }

src/main/java/com/rabbitmq/stream/perf/StreamPerfTest.java

@@ -31,6 +31,7 @@
 import com.rabbitmq.stream.impl.Client;
 import com.rabbitmq.stream.impl.ClientProperties;
 import com.rabbitmq.stream.metrics.MetricsCollector;
+import com.rabbitmq.stream.perf.Converters.Credits;
 import com.rabbitmq.stream.perf.ShutdownService.CloseCallback;
 import com.rabbitmq.stream.perf.Utils.NamedThreadFactory;
 import com.rabbitmq.stream.perf.Utils.PerformanceMicrometerMetricsCollector;
@@ -648,6 +649,13 @@ static class InstanceSyncOptions {
       converter = Converters.NotNegativeIntegerTypeConverter.class)
   private int initialCredits;
 
+  @CommandLine.Option(
+      names = {"--credits"},
+      description = "initial credits and credits to grant every n chunks (e.g. 10-5)",
+      defaultValue = "0",
+      converter = Converters.CreditsTypeConverter.class)
+  private Credits credits;
+
   @CommandLine.Option(
       names = {"--heartbeat", "-b"},
       description = "requested heartbeat in seconds",
@@ -1309,12 +1317,17 @@ public Integer call() throws Exception {
 
                         String stream = stream(streams, i);
                         ConsumerBuilder consumerBuilder =
-                            environment
-                                .consumerBuilder()
-                                .offset(this.offset)
-                                .flow()
-                                .initialCredits(this.initialCredits)
-                                .builder();
+                            environment.consumerBuilder().offset(this.offset);
+
+                        if (this.credits.initialCredits() > 0) {
+                          consumerBuilder
+                              .flow()
+                              .strategy(
+                                  ConsumerFlowStrategy.creditEveryNthChunk(
+                                      this.credits.initialCredits(), this.credits.n()));
+                        } else {
+                          consumerBuilder.flow().initialCredits(this.initialCredits);
+                        }
 
                         if (this.superStreams) {
                           consumerBuilder.superStream(stream);

src/test/java/com/rabbitmq/stream/perf/ConvertersTest.java

@@ -225,6 +225,23 @@ void durationTypeConverterKo(String value) {
         .hasMessageContaining("valid");
   }
 
+  @ParameterizedTest
+  @CsvSource({"10-5,10,5", "5-2,5,2", "10,10,5", "5,5,2", "0,0,0"})
+  void creditsTypeConverterOk(String value, int initialCredits, int n) {
+    Converters.CreditsTypeConverter converter = new Converters.CreditsTypeConverter();
+    Converters.Credits credits = converter.convert(value);
+    assertThat(credits.initialCredits()).isEqualTo(initialCredits);
+    assertThat(credits.n()).isEqualTo(n);
+  }
+
+  @ParameterizedTest
+  @ValueSource(strings = {"a", "1-", ""})
+  void creditsTypeConverterKo(String value) {
+    Converters.CreditsTypeConverter converter = new Converters.CreditsTypeConverter();
+    assertThatThrownBy(() -> converter.convert(value))
+        .isInstanceOfAny(IllegalArgumentException.class, CommandLine.TypeConversionException.class);
+  }
+
   private static Tag tag(String key, String value) {
     return Tag.of(key, value);
   }