Release Netty ByteBuf after it is consumed by Tyrus. Issue #7002. (#7042

) Signed-off-by: Santiago Pericasgeertsen <[email protected]>
helidon-io · Jun 22, 2023 · 15a560f · 15a560f
1 parent b2f265b
commit 15a560f
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 6 deletions.
diff --git a/webserver/websocket/pom.xml b/webserver/websocket/pom.xml
@@ -29,6 +29,10 @@
     <artifactId>helidon-webserver-websocket</artifactId>
     <name>Helidon WebServer WebSocket</name>
 
+    <properties>
+        <surefire.argLine>-Dio.netty.leakDetectionLevel=paranoid</surefire.argLine>
+    </properties>
+
     <dependencies>
         <dependency>
             <groupId>io.helidon.webserver</groupId>

diff --git a/webserver/websocket/src/main/java/io/helidon/webserver/websocket/WebSocketHandler.java b/webserver/websocket/src/main/java/io/helidon/webserver/websocket/WebSocketHandler.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2022 Oracle and/or its affiliates.
+ * Copyright (c) 2022, 2023 Oracle and/or its affiliates.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -69,7 +69,7 @@ class WebSocketHandler extends SimpleChannelInboundHandler<Object> {
     private final TyrusServerContainer tyrusServerContainer;
     private volatile Connection connection;
     private final WebSocketEngine.UpgradeInfo upgradeInfo;
-    private final BufferedEmittingPublisher<ByteBuffer> emitter;
+    private final BufferedEmittingPublisher<ByteBuf> emitter;
 
     WebSocketHandler(ChannelHandlerContext ctx, String path,
                             FullHttpRequest upgradeRequest,
@@ -154,16 +154,18 @@ public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
     @Override
     protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
         if (msg instanceof ByteBuf byteBuf) {
-            emitter.emit(byteBuf.copy().nioBuffer());
+            emitter.emit(byteBuf.copy());
         }
     }
 
-    private void sendBytesToTyrus(ChannelHandlerContext ctx, ByteBuffer nioBuffer) {
+    private void sendBytesToTyrus(ChannelHandlerContext ctx, ByteBuf byteBuf) {
         // Pass all data to Tyrus spi
+        ByteBuffer nioBuffer = byteBuf.nioBuffer();
         int retries = MAX_RETRIES;
         while (nioBuffer.remaining() > 0 && retries-- > 0) {
             connection.getReadHandler().handle(nioBuffer);
         }
+        byteBuf.release();
 
         // If we can't push all data to Tyrus, cancel and report problem
         if (retries == 0) {
@@ -223,13 +225,13 @@ public void write(ByteBuffer byteBuffer, CompletionHandler<ByteBuffer> completio
                 return ctx;
             }, webSocketRouting.getExecutorService()).thenAccept(c -> Multi.create(emitter)
                     .observeOn(webSocketRouting.getExecutorService())
-                    .forEach(byteBuffer -> sendBytesToTyrus(c, byteBuffer))
+                    .forEach(byteBuf -> sendBytesToTyrus(c, byteBuf))
                     .onError(this::logError)
             );
         } else {
             this.connection = upgradeInfo.createConnection(writer, WebSocketHandler::close);
             Multi.create(emitter)
-                    .forEach(byteBuffer -> sendBytesToTyrus(ctx, byteBuffer))
+                    .forEach(byteBuf -> sendBytesToTyrus(ctx, byteBuf))
                     .onError(this::logError);
         }