Add release call to websocket ByteBuff after using it

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>

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.
@@ -15,30 +15,18 @@
  */
 package io.helidon.webserver.websocket;
 
-import java.io.IOException;
-import java.net.URI;
-import java.nio.ByteBuffer;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.CompletableFuture;
-import java.util.logging.Level;
-import java.util.logging.Logger;
-import java.util.stream.Collectors;
-
 import io.helidon.common.http.Parameters;
 import io.helidon.common.http.UriComponent;
 import io.helidon.common.reactive.BufferedEmittingPublisher;
 import io.helidon.common.reactive.Multi;
-
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.SimpleChannelInboundHandler;
 import io.netty.handler.codec.http.FullHttpRequest;
 import io.netty.handler.codec.http.HttpHeaders;
+import io.netty.util.ReferenceCountUtil;
 import jakarta.websocket.CloseReason;
 import jakarta.websocket.DeploymentException;
 import jakarta.websocket.Extension;
@@ -52,6 +40,18 @@
 import org.glassfish.tyrus.spi.WebSocketEngine;
 import org.glassfish.tyrus.spi.Writer;
 
+import java.io.IOException;
+import java.net.URI;
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import java.util.stream.Collectors;
+
 import static jakarta.websocket.CloseReason.CloseCodes.UNEXPECTED_CONDITION;
 
 class WebSocketHandler extends SimpleChannelInboundHandler<Object> {
@@ -69,12 +69,12 @@ 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,
-                            HttpHeaders upgradeResponseHeaders,
-                            WebSocketRouting webSocketRouting) {
+                     FullHttpRequest upgradeRequest,
+                     HttpHeaders upgradeResponseHeaders,
+                     WebSocketRouting webSocketRouting) {
         int k = path.indexOf('?');
         if (k > 0) {
             this.path = path.substring(0, k);
@@ -154,23 +154,28 @@ 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) {
-        // Pass all data to Tyrus spi
-        int retries = MAX_RETRIES;
-        while (nioBuffer.remaining() > 0 && retries-- > 0) {
-            connection.getReadHandler().handle(nioBuffer);
-        }
+    private void sendBytesToTyrus(ChannelHandlerContext ctx, ByteBuf byteBuf) {
+        try {
+            ByteBuffer nioBuffer = byteBuf.nioBuffer();
+            // Pass all data to Tyrus spi
+            int retries = MAX_RETRIES;
+            while (nioBuffer.remaining() > 0 && retries-- > 0) {
+                connection.getReadHandler().handle(nioBuffer);
+            }
 
-        // If we can't push all data to Tyrus, cancel and report problem
-        if (retries == 0) {
-            ctx.close();
-            connection.close(
-                    new CloseReason(UNEXPECTED_CONDITION, "Tyrus did not consume all data after " + MAX_RETRIES + " retries")
-            );
+            // If we can't push all data to Tyrus, cancel and report problem
+            if (retries == 0) {
+                ctx.close();
+                connection.close(
+                        new CloseReason(UNEXPECTED_CONDITION, "Tyrus did not consume all data after " + MAX_RETRIES + " retries")
+                );
+            }
+        } finally {
+            ReferenceCountUtil.release(byteBuf);
         }
     }
 
@@ -223,20 +228,20 @@ 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);
         }
 
 
     }
 
-    private void logError(Throwable throwable){
+    private void logError(Throwable throwable) {
         LOGGER.log(Level.SEVERE, "WS handler ERROR ", throwable);
     }