Move event loop group setup to Netty configuration

References #269

Author: acogoluegnes

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

@@ -92,17 +92,6 @@ public interface EnvironmentBuilder {
    */
   EnvironmentBuilder codec(Codec codec);
 
-  /**
-   * The Netty {@link EventLoopGroup} instance to use.
-   *
-   * <p>The environment uses its own instance by default. It is the developer's responsibility to
-   * close the {@link EventLoopGroup} they provide.
-   *
-   * @param eventLoopGroup
-   * @return this builder instance
-   */
-  EnvironmentBuilder eventLoopGroup(EventLoopGroup eventLoopGroup);
-
   /**
    * Informational ID for this environment instance.
    *
@@ -198,6 +187,19 @@ public interface EnvironmentBuilder {
    */
   EnvironmentBuilder requestedMaxFrameSize(int requestedMaxFrameSize);
 
+  /**
+   * The Netty {@link EventLoopGroup} instance to use.
+   *
+   * <p>The environment uses its own instance by default. It is the developer's responsibility to
+   * close the {@link EventLoopGroup} they provide.
+   *
+   * @param eventLoopGroup
+   * @return this builder instance
+   * @deprecated use {@link NettyConfiguration#eventLoopGroup(EventLoopGroup)} from {@link #netty()}
+   *     instead
+   */
+  EnvironmentBuilder eventLoopGroup(EventLoopGroup eventLoopGroup);
+
   /**
    * Netty's {@link io.netty.buffer.ByteBuf} allocator.
    *
@@ -408,6 +410,17 @@ interface TlsConfiguration {
   /** Helper to configure Netty */
   interface NettyConfiguration {
 
+    /**
+     * The {@link EventLoopGroup} instance to use.
+     *
+     * <p>The environment uses its own instance by default. It is the developer's responsibility to
+     * close the {@link EventLoopGroup} they provide.
+     *
+     * @param eventLoopGroup
+     * @return the Netty configuration helper
+     */
+    NettyConfiguration eventLoopGroup(EventLoopGroup eventLoopGroup);
+
     /**
      * Netty's {@link io.netty.buffer.ByteBuf} allocator.
      *

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

@@ -134,8 +134,10 @@ public EnvironmentBuilder compressionCodecFactory(
     return this;
   }
 
+  @SuppressWarnings("deprecation")
+  @Override
   public EnvironmentBuilder eventLoopGroup(EventLoopGroup eventLoopGroup) {
-    this.clientParameters.eventLoopGroup(eventLoopGroup);
+    this.netty().eventLoopGroup(eventLoopGroup);
     return this;
   }
 
@@ -309,6 +311,7 @@ public Environment build() {
     }
     this.id = this.id == null ? "rabbitmq-stream" : this.id;
     this.connectionNamingStrategy = Utils.defaultConnectionNamingStrategy(this.id + "-");
+    this.clientParameters.eventLoopGroup(this.netty.eventLoopGroup);
     this.clientParameters.byteBufAllocator(this.netty.byteBufAllocator);
     this.clientParameters.channelCustomizer(this.netty.channelCustomizer);
     this.clientParameters.bootstrapCustomizer(this.netty.bootstrapCustomizer);
@@ -400,14 +403,20 @@ public SslContext sslContext() {
   static class DefaultNettyConfiguration implements NettyConfiguration {
 
     private final EnvironmentBuilder environmentBuilder;
+    private EventLoopGroup eventLoopGroup;
+    private ByteBufAllocator byteBufAllocator = ByteBufAllocator.DEFAULT;
+    private Consumer<Channel> channelCustomizer = noOpConsumer();
+    private Consumer<Bootstrap> bootstrapCustomizer = noOpConsumer();
 
     private DefaultNettyConfiguration(EnvironmentBuilder environmentBuilder) {
       this.environmentBuilder = environmentBuilder;
     }
 
-    private ByteBufAllocator byteBufAllocator = ByteBufAllocator.DEFAULT;
-    private Consumer<Channel> channelCustomizer = noOpConsumer();
-    private Consumer<Bootstrap> bootstrapCustomizer = noOpConsumer();
+    @Override
+    public NettyConfiguration eventLoopGroup(EventLoopGroup eventLoopGroup) {
+      this.eventLoopGroup = eventLoopGroup;
+      return this;
+    }
 
     @Override
     public NettyConfiguration byteBufAllocator(ByteBufAllocator byteBufAllocator) {

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

@@ -673,9 +673,9 @@ public Integer call() throws Exception {
               .addressResolver(addrResolver)
               .scheduledExecutorService(envExecutor)
               .metricsCollector(metricsCollector)
-              .eventLoopGroup(eventLoopGroup)
               .netty()
               .byteBufAllocator(byteBufAllocator)
+              .eventLoopGroup(eventLoopGroup)
               .environmentBuilder()
               .codec(codec)
               .maxProducersByConnection(this.producersByConnection)

src/test/java/com/rabbitmq/stream/DefaultEnvironmentTest.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2020-2021 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2020-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
 // Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").
@@ -44,7 +44,12 @@ void defaultEnvironmentCanBeInstantiatedAndCanConnect() {
     try (Client client = new Client(new Client.ClientParameters().eventLoopGroup(eventLoopGroup))) {
       Client.Response response = client.create(stream);
       assertThat(response.isOk()).isTrue();
-      try (Environment environment = Environment.builder().eventLoopGroup(eventLoopGroup).build()) {
+      try (Environment environment =
+          Environment.builder()
+              .netty()
+              .eventLoopGroup(eventLoopGroup)
+              .environmentBuilder()
+              .build()) {
         environment.producerBuilder().stream(stream);
       } finally {
         response = client.delete(stream);

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

@@ -1,4 +1,4 @@
-// Copyright (c) 2021-2022 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2021-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
 // Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").
@@ -68,7 +68,7 @@ static void afterAll() throws Exception {
   void init() {
     environmentBuilder = Environment.builder();
     environmentBuilder.addressResolver(add -> localhost());
-    env = environmentBuilder.eventLoopGroup(eventLoopGroup).build();
+    env = environmentBuilder.netty().eventLoopGroup(eventLoopGroup).environmentBuilder().build();
   }
 
   @AfterEach

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

@@ -1,4 +1,4 @@
-// Copyright (c) 2021 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2021-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
 // Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").
@@ -69,7 +69,7 @@ static void afterAll() throws Exception {
   void init() {
     environmentBuilder = Environment.builder();
     environmentBuilder.addressResolver(add -> localhost());
-    env = environmentBuilder.eventLoopGroup(eventLoopGroup).build();
+    env = environmentBuilder.netty().eventLoopGroup(eventLoopGroup).environmentBuilder().build();
   }
 
   @AfterEach

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

@@ -1,4 +1,4 @@
-// Copyright (c) 2021-2022 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2021-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
 // Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").
@@ -48,7 +48,11 @@ public class SacStreamConsumerTest {
   @BeforeEach
   void init() {
     EnvironmentBuilder environmentBuilder =
-        Environment.builder().eventLoopGroup(eventLoopGroup).maxConsumersByConnection(1);
+        Environment.builder()
+            .netty()
+            .eventLoopGroup(eventLoopGroup)
+            .environmentBuilder()
+            .maxConsumersByConnection(1);
     environmentBuilder.addressResolver(add -> localhost());
     environment = environmentBuilder.build();
   }

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

@@ -1,4 +1,4 @@
-// Copyright (c) 2021-2022 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2021-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
 // Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").
@@ -64,7 +64,8 @@ public class SacSuperStreamConsumerTest {
 
   @BeforeEach
   void init(TestInfo info) throws Exception {
-    EnvironmentBuilder environmentBuilder = Environment.builder().eventLoopGroup(eventLoopGroup);
+    EnvironmentBuilder environmentBuilder =
+        Environment.builder().netty().eventLoopGroup(eventLoopGroup).environmentBuilder();
     environmentBuilder.addressResolver(add -> localhost());
     environment = environmentBuilder.build();
     superStream = TestUtils.streamName(info);

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

@@ -1,4 +1,4 @@
-// Copyright (c) 2021 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2021-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
 // Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").
@@ -110,7 +110,7 @@ static long offset(String line) {
   void init() throws Exception {
     environmentBuilder = Environment.builder();
     environmentBuilder.addressResolver(add -> localhost());
-    env = environmentBuilder.eventLoopGroup(eventLoopGroup).build();
+    env = environmentBuilder.netty().eventLoopGroup(eventLoopGroup).environmentBuilder().build();
     socket = new Socket("localhost", 61613);
     out = socket.getOutputStream();
     in = new BufferedReader(new InputStreamReader(socket.getInputStream()));

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

@@ -1,4 +1,4 @@
-// Copyright (c) 2020-2022 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2020-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
 // Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").
@@ -113,7 +113,9 @@ void init() {
     }
     EnvironmentBuilder environmentBuilder =
         Environment.builder()
+            .netty()
             .eventLoopGroup(eventLoopGroup)
+            .environmentBuilder()
             .recoveryBackOffDelayPolicy(
                 BackOffDelayPolicy.fixedWithInitialDelay(recoveryInitialDelay, RECOVERY_DELAY))
             .topologyUpdateBackOffDelayPolicy(

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

@@ -108,7 +108,7 @@ void init() {
     environmentBuilder = Environment.builder();
     environmentBuilder.addressResolver(
         add -> add.port() == Client.DEFAULT_PORT ? localhost() : localhostTls());
-    environmentBuilder.eventLoopGroup(eventLoopGroup);
+    environmentBuilder.netty().eventLoopGroup(eventLoopGroup);
   }
 
   @Test

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

@@ -1,4 +1,4 @@
-// Copyright (c) 2020-2022 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2020-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
 // Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").
@@ -79,7 +79,9 @@ public class StreamProducerTest {
   void init() {
     EnvironmentBuilder environmentBuilder =
         Environment.builder()
+            .netty()
             .eventLoopGroup(eventLoopGroup)
+            .environmentBuilder()
             .recoveryBackOffDelayPolicy(BackOffDelayPolicy.fixed(Duration.ofSeconds(2)))
             .topologyUpdateBackOffDelayPolicy(BackOffDelayPolicy.fixed(Duration.ofSeconds(2)));
     environmentBuilder.addressResolver(add -> localhost());

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

@@ -1,4 +1,4 @@
-// Copyright (c) 2021-2022 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2021-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
 // Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").
@@ -60,7 +60,8 @@ public class SuperStreamConsumerTest {
 
   @BeforeEach
   void init(TestInfo info) throws Exception {
-    EnvironmentBuilder environmentBuilder = Environment.builder().eventLoopGroup(eventLoopGroup);
+    EnvironmentBuilder environmentBuilder =
+        Environment.builder().netty().eventLoopGroup(eventLoopGroup).environmentBuilder();
     environmentBuilder.addressResolver(add -> localhost());
     environment = environmentBuilder.build();
     superStream = TestUtils.streamName(info);

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

@@ -1,4 +1,4 @@
-// Copyright (c) 2021 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2021-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
 // Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").
@@ -59,7 +59,8 @@ public class SuperStreamProducerTest {
 
   @BeforeEach
   void init(TestInfo info) throws Exception {
-    EnvironmentBuilder environmentBuilder = Environment.builder().eventLoopGroup(eventLoopGroup);
+    EnvironmentBuilder environmentBuilder =
+        Environment.builder().netty().eventLoopGroup(eventLoopGroup).environmentBuilder();
     environmentBuilder.addressResolver(add -> localhost());
     environment = environmentBuilder.build();
     connection = new ConnectionFactory().newConnection();

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

@@ -1,4 +1,4 @@
-// Copyright (c) 2021 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2021-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
 // Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").
@@ -61,7 +61,8 @@ public class SuperStreamTest {
 
   @BeforeEach
   void init(TestInfo info) throws Exception {
-    EnvironmentBuilder environmentBuilder = Environment.builder().eventLoopGroup(eventLoopGroup);
+    EnvironmentBuilder environmentBuilder =
+        Environment.builder().netty().eventLoopGroup(eventLoopGroup).environmentBuilder();
     environmentBuilder.addressResolver(add -> localhost());
     environment = environmentBuilder.build();
     connection = new ConnectionFactory().newConnection();