Add options for TCP nodelay, so_sndbuf, so_rcvbuf

Author: acogoluegnes

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

@@ -46,6 +46,7 @@
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.ByteBufAllocatorMetric;
 import io.netty.buffer.ByteBufAllocatorMetricProvider;
+import io.netty.channel.ChannelOption;
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.epoll.EpollEventLoopGroup;
 import io.netty.channel.epoll.EpollSocketChannel;
@@ -513,6 +514,27 @@ static class InstanceSyncOptions {
       converter = Utils.BackOffDelayPolicyTypeConverter.class)
   private BackOffDelayPolicy topologyBackOffDelayPolicy;
 
+  @CommandLine.Option(
+      names = {"--tcp-send-buffer-size", "-tsbs"},
+      description = "TCP SO_SNDBUF option, default is platform value.",
+      defaultValue = "0",
+      converter = Utils.ByteCapacityTypeConverter.class)
+  private ByteCapacity tcpSndBuff;
+
+  @CommandLine.Option(
+      names = {"--tcp-receive-buffer-size", "-trbs"},
+      description = "TCP SO_RCVBUF option, default is platform value.",
+      defaultValue = "0",
+      converter = Utils.ByteCapacityTypeConverter.class)
+  private ByteCapacity tcpRcvBuff;
+
+  @CommandLine.Option(
+      names = {"--tcp-no-delay", "-tnd"},
+      description = "TCP NODELAY",
+      arity = "1",
+      defaultValue = "true")
+  private boolean tcpNoDelay;
+
   private MetricsCollector metricsCollector;
   private PerformanceMetrics performanceMetrics;
   private List<Monitoring> monitorings;
@@ -764,6 +786,18 @@ public Integer call() throws Exception {
         bootstrapCustomizer = b -> {};
       }
 
+      bootstrapCustomizer =
+          bootstrapCustomizer.andThen(
+              b -> {
+                if (this.tcpSndBuff.toBytes() > 0) {
+                  b.option(ChannelOption.SO_SNDBUF, (int) this.tcpSndBuff.toBytes());
+                }
+                if (this.tcpRcvBuff.toBytes() > 0) {
+                  b.option(ChannelOption.SO_RCVBUF, (int) this.tcpRcvBuff.toBytes());
+                }
+                b.option(ChannelOption.TCP_NODELAY, this.tcpNoDelay);
+              });
+
       EnvironmentBuilder environmentBuilder =
           Environment.builder()
               .id("stream-perf-test")