Test multi-threaded Netty server support

coap-mbedtls/build.gradle.kts

@@ -14,5 +14,6 @@ dependencies {
     testImplementation(testFixtures(project(":coap-core")))
     testImplementation("org.junit.jupiter:junit-jupiter-api:5.10.3")
     testRuntimeOnly("org.junit.jupiter:junit-jupiter-engine:5.10.3")
+    testImplementation("io.netty:netty-transport-native-epoll:4.1.112.Final:linux-x86_64")
     testImplementation("ch.qos.logback:logback-classic:1.3.14")
 }

coap-mbedtls/src/test/java/org/opencoap/transport/mbedtls/MbedtlsNettyTest.java

@@ -35,6 +35,7 @@
 import com.mbed.coap.packet.CoapResponse;
 import com.mbed.coap.packet.Code;
 import com.mbed.coap.server.CoapServer;
+import com.mbed.coap.server.CoapServerBuilder;
 import com.mbed.coap.server.RouterService;
 import io.netty.bootstrap.Bootstrap;
 import io.netty.buffer.ByteBuf;
@@ -44,6 +45,9 @@
 import io.netty.channel.ChannelOutboundHandlerAdapter;
 import io.netty.channel.ChannelPromise;
 import io.netty.channel.EventLoopGroup;
+import io.netty.channel.epoll.EpollChannelOption;
+import io.netty.channel.epoll.EpollDatagramChannel;
+import io.netty.channel.epoll.EpollEventLoopGroup;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.DatagramChannel;
 import io.netty.channel.socket.nio.NioDatagramChannel;
@@ -52,14 +56,20 @@
 import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.nio.ByteBuffer;
+import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.TestInstance;
+import org.junit.jupiter.api.condition.EnabledOnOs;
+import org.junit.jupiter.api.condition.OS;
 import org.opencoap.coap.netty.NettyCoapTransport;
 import org.opencoap.ssl.PskAuth;
 import org.opencoap.ssl.SslConfig;
@@ -121,7 +131,7 @@ void beforeAll() throws IOException {
     }
 
     @AfterAll
-    void afterAll() {
+    void afterAll() throws ExecutionException, InterruptedException {
         server.stop();
         eventLoopGroup.shutdownGracefully(0, 0, TimeUnit.SECONDS);
     }
@@ -221,6 +231,60 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
         serverTransport.getChannel().pipeline().remove("test-handler");
     }
 
+    @Test
+    @EnabledOnOs(OS.LINUX)
+    void multi_thread_server() throws Exception {
+        int threads = 4;
+        EventLoopGroup epollGroup = new EpollEventLoopGroup(threads, new DefaultThreadFactory("srv", true));
+        Bootstrap bootstrap = new Bootstrap()
+                .group(epollGroup)
+                .localAddress(new Random().nextInt(32000, 64000))
+                .channel(EpollDatagramChannel.class)
+                .option(EpollChannelOption.SO_REUSEPORT, true)
+                .handler(new ChannelInitializer<DatagramChannel>() {
+                    @Override
+                    protected void initChannel(DatagramChannel ch) {
+                        ch.pipeline().addFirst("DTLS", new DtlsChannelHandler(serverConf));
+                    }
+                });
+        CoapServerBuilder  serverBuilder = CoapServer.builder()
+                .executor(eventLoopGroup)
+                .route(RouterService.builder()
+                        .get("/test", __ -> ok("OK").toFuture())
+                        .post("/echo", req -> CompletableFuture.supplyAsync(() -> ok(req.getPayload()).build()))
+                );
+        ArrayList<CoapServer> servers = new ArrayList<>();
+        for (int i = 0; i < threads; i++) {
+            servers.add(
+                    serverBuilder
+                            .transport(new NettyCoapTransport(bootstrap, EMPTY_RESOLVER))
+                            .build()
+                            .start()
+            );
+        }
+
+        List<CoapClient> clients = new ArrayList<>();
+        InetSocketAddress srvAddr = localhost(servers.get(0).getLocalSocketAddress().getPort());
+        CoapServerBuilder clientBuilder = CoapServer.builder().executor(eventLoopGroup);
+        for (int i = 0; i < 100; i++) {
+            clients.add(
+                    clientBuilder
+                            .transport(new NettyCoapTransport(createClientBootstrap(srvAddr), EMPTY_RESOLVER))
+                            .buildClient(srvAddr)
+            );
+        }
+
+        for (CoapClient client : clients) {
+            assertEquals(ok("paska"), client.sendSync(post("/echo").payload("paska")));
+            client.close();
+        }
+
+        for (CoapServer srv : servers) {
+            srv.stop();
+        }
+        epollGroup.shutdown();
+    }
+
     private Bootstrap createBootstrap(int port) {
         return new Bootstrap()
                 .group(eventLoopGroup)

coap-mbedtls/src/test/resources/logback-test.xml

@@ -2,7 +2,7 @@
 
     <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
         <encoder>
-            <pattern>%d{HH:mm:ss} %.-1level %m %n</pattern>
+            <pattern>%d{HH:mm:ss} [tid:%10.10thread] %-5level %-80msg [%logger]%n</pattern>
         </encoder>
     </appender>