Add test with native epoll

References #269

Author: acogoluegnes

pom.xml

@@ -286,6 +286,14 @@
       <scope>test</scope>
     </dependency>
 
+    <dependency>
+      <groupId>io.netty</groupId>
+      <artifactId>netty-transport-native-epoll</artifactId>
+      <version>${netty.version}</version>
+      <classifier>linux-x86_64</classifier>
+      <scope>test</scope>
+    </dependency>
+
     <dependency>
       <groupId>org.openjdk.jmh</groupId>
       <artifactId>jmh-core</artifactId>

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

@@ -248,24 +248,38 @@ public Client(ClientParameters parameters) {
               }
             });
 
-    EventLoopGroup eventLoopGroup;
-    if (parameters.eventLoopGroup == null) {
-      this.eventLoopGroup = new NioEventLoopGroup();
-      eventLoopGroup = this.eventLoopGroup;
+    Consumer<Bootstrap> bootstrapCustomizer =
+        parameters.bootstrapCustomizer == null ? noOpConsumer() : parameters.bootstrapCustomizer;
+
+    Bootstrap b = new Bootstrap();
+    bootstrapCustomizer.accept(b);
+    if (b.config().group() == null) {
+      EventLoopGroup eventLoopGroup;
+      if (parameters.eventLoopGroup == null) {
+        this.eventLoopGroup = new NioEventLoopGroup();
+        eventLoopGroup = this.eventLoopGroup;
+      } else {
+        this.eventLoopGroup = null;
+        eventLoopGroup = parameters.eventLoopGroup;
+      }
+      b.group(eventLoopGroup);
     } else {
       this.eventLoopGroup = null;
-      eventLoopGroup = parameters.eventLoopGroup;
+    }
+    if (b.config().channelFactory() == null) {
+      b.channel(NioSocketChannel.class);
+    }
+    if (!b.config().options().containsKey(ChannelOption.SO_KEEPALIVE)) {
+      b.option(ChannelOption.SO_KEEPALIVE, true);
+    }
+    if (!b.config().options().containsKey(ChannelOption.ALLOCATOR)) {
+      b.option(
+          ChannelOption.ALLOCATOR,
+          parameters.byteBufAllocator == null
+              ? ByteBufAllocator.DEFAULT
+              : parameters.byteBufAllocator);
     }
 
-    Bootstrap b = new Bootstrap();
-    b.group(eventLoopGroup);
-    b.channel(NioSocketChannel.class);
-    b.option(ChannelOption.SO_KEEPALIVE, true);
-    b.option(
-        ChannelOption.ALLOCATOR,
-        parameters.byteBufAllocator == null
-            ? ByteBufAllocator.DEFAULT
-            : parameters.byteBufAllocator);
     Consumer<Channel> channelCustomizer =
         parameters.channelCustomizer == null ? noOpConsumer() : parameters.channelCustomizer;
     b.handler(
@@ -310,9 +324,6 @@ public void write(
             channelCustomizer.accept(ch);
           }
         });
-    Consumer<Bootstrap> bootstrapCustomizer =
-        parameters.bootstrapCustomizer == null ? noOpConsumer() : parameters.bootstrapCustomizer;
-    bootstrapCustomizer.accept(b);
 
     ChannelFuture f;
     String clientConnectionName =

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

@@ -53,6 +53,8 @@
 import com.rabbitmq.stream.impl.TestUtils.DisabledIfTlsNotEnabled;
 import io.netty.channel.Channel;
 import io.netty.channel.EventLoopGroup;
+import io.netty.channel.epoll.EpollEventLoopGroup;
+import io.netty.channel.epoll.EpollSocketChannel;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.handler.ssl.SslHandler;
 import java.net.ConnectException;
@@ -64,13 +66,16 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Random;
+import java.util.Set;
 import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 import javax.net.ssl.SNIHostName;
 import javax.net.ssl.SSLParameters;
 import org.assertj.core.api.ThrowableAssert.ThrowingCallable;
@@ -79,6 +84,9 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.TestInfo;
+import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
+import org.junit.jupiter.api.condition.EnabledOnOs;
+import org.junit.jupiter.api.condition.OS;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
@@ -615,4 +623,44 @@ void nettyInitializersAreCalled() {
     assertThat(bootstrapCalled).isTrue();
     assertThat(channelCalled).isTrue();
   }
+
+  @Test
+  @EnabledOnOs(OS.LINUX)
+  @EnabledIfSystemProperty(named = "os.arch", matches = "amd64")
+  void nativeEpollWorksOnLinux() {
+    int messageCount = 10_000;
+    EventLoopGroup epollEventLoopGroup = new EpollEventLoopGroup();
+    try {
+      Set<Channel> channels = ConcurrentHashMap.newKeySet();
+      try (Environment env =
+          environmentBuilder
+              .netty()
+              .eventLoopGroup(epollEventLoopGroup)
+              .bootstrapCustomizer(b -> b.channel(EpollSocketChannel.class))
+              .channelCustomizer(ch -> channels.add(ch))
+              .environmentBuilder()
+              .build()) {
+        Producer producer = env.producerBuilder().stream(this.stream).build();
+        ConfirmationHandler handler = confirmationStatus -> {};
+        IntStream.range(0, messageCount)
+            .forEach(
+                i ->
+                    producer.send(
+                        producer
+                            .messageBuilder()
+                            .addData("hello".getBytes(StandardCharsets.UTF_8))
+                            .build(),
+                        handler));
+        CountDownLatch latch = new CountDownLatch(messageCount);
+        env.consumerBuilder().stream(this.stream)
+            .offset(OffsetSpecification.first())
+            .messageHandler((context, message) -> latch.countDown())
+            .build();
+        assertThat(latchAssert(latch)).completes();
+      }
+      assertThat(channels).isNotEmpty().allMatch(ch -> ch instanceof EpollSocketChannel);
+    } finally {
+      epollEventLoopGroup.shutdownGracefully(0, 0, SECONDS);
+    }
+  }
 }