Fix NettyInputStream ByteBuf leak (#4222)

* Fix NettyInputStream ByteBuf leak in jersey/netty HTTP2 Handler Signed-off-by: Franck Séhédic <[email protected]>
eclipse-ee4j · Aug 13, 2019 · 14e8f18 · 14e8f18
1 parent 0a867e9
commit 14e8f18
Show file tree

Hide file tree

Showing 3 changed files with 15 additions and 27 deletions.
diff --git a/...nnector/src/main/java/org/glassfish/jersey/netty/connector/internal/NettyInputStream.java b/...nnector/src/main/java/org/glassfish/jersey/netty/connector/internal/NettyInputStream.java
@@ -68,9 +68,11 @@ public NettyInputStream(LinkedBlockingDeque<InputStream> isList) {
         this.isList = isList;
     }
 
-    @Override
-    public int read(byte[] b, int off, int len) throws IOException {
+    private interface ISReader {
+        int readFrom(InputStream take) throws IOException;
+    }
 
+    private int readInternal(ISReader isReader) throws IOException {
         if (end) {
             return -1;
         }
@@ -83,10 +85,12 @@ public int read(byte[] b, int off, int len) throws IOException {
                 return -1;
             }
 
-            int read = take.read(b, off, len);
+            int read = isReader.readFrom(take);
 
             if (take.available() > 0) {
                 isList.addFirst(take);
+            } else {
+                take.close();
             }
 
             return read;
@@ -96,29 +100,13 @@ public int read(byte[] b, int off, int len) throws IOException {
     }
 
     @Override
-    public int read() throws IOException {
-
-        if (end) {
-            return -1;
-        }
-
-        try {
-            InputStream take = isList.take();
-
-            if (checkEndOfInput(take)) {
-                return -1;
-            }
-
-            int read = take.read();
-
-            if (take.available() > 0) {
-                isList.addFirst(take);
-            }
+    public int read(byte[] b, int off, int len) throws IOException {
+        return readInternal(take -> take.read(b, off, len));
+    }
 
-            return read;
-        } catch (InterruptedException e) {
-            throw new IOException("Interrupted.", e);
-        }
+    @Override
+    public int read() throws IOException {
+        return readInternal(InputStream::read);
     }
 
     @Override

diff --git a/...ty-http/src/main/java/org/glassfish/jersey/netty/httpserver/JerseyHttp2ServerHandler.java b/...ty-http/src/main/java/org/glassfish/jersey/netty/httpserver/JerseyHttp2ServerHandler.java
@@ -88,7 +88,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
      * Process incoming data.
      */
     private void onDataRead(ChannelHandlerContext ctx, Http2DataFrame data) throws Exception {
-        isList.add(new ByteBufInputStream(data.content()));
+        isList.add(new ByteBufInputStream(data.content(), true));
         if (data.isEndStream()) {
             isList.add(NettyInputStream.END_OF_INPUT);
         }

diff --git a/...s/netty-http/src/main/java/org/glassfish/jersey/netty/httpserver/JerseyServerHandler.java b/...s/netty-http/src/main/java/org/glassfish/jersey/netty/httpserver/JerseyServerHandler.java
@@ -100,7 +100,7 @@ public void run() {
             ByteBuf content = httpContent.content();
 
             if (content.isReadable()) {
-                isList.add(new ByteBufInputStream(content));
+                isList.add(new ByteBufInputStream(content, true));
             }
 
             if (msg instanceof LastHttpContent) {