Be more careful about release netty buffers

Author: justinsb

cloudata-files/src/main/java/com/cloudata/files/fs/SliceByteInput.java

@@ -38,43 +38,53 @@ public void close() throws Exception {
     @Override
     public ByteBuf readChunk(ChannelHandlerContext ctx) throws Exception {
         while (true) {
-            ByteBuf buffer = inner.readChunk(ctx);
+            ByteBuf buffer = null;
+            boolean release = true;
 
-            if (buffer == null) {
-                return null;
-            }
+            try {
+                buffer = inner.readChunk(ctx);
+
+                if (buffer == null) {
+                    return null;
+                }
 
-            // log.info("SliceByteInput position=" + position);
+                // log.info("SliceByteInput position=" + position);
 
-            if (from != null) {
-                if (position < from) {
-                    int skip = (int) Math.min(from - position, buffer.readableBytes());
-                    if (skip > 0) {
-                        buffer.skipBytes(skip);
-                        position += skip;
+                if (from != null) {
+                    if (position < from) {
+                        int skip = (int) Math.min(from - position, buffer.readableBytes());
+                        if (skip > 0) {
+                            buffer.skipBytes(skip);
+                            position += skip;
+                        }
                     }
                 }
-            }
 
-            int consumed = buffer.readableBytes();
+                int consumed = buffer.readableBytes();
 
-            if (to != null) {
-                long n = to - position;
-                long limitWriterIndex = buffer.readerIndex() + n;
+                if (to != null) {
+                    long n = to - position;
+                    long limitWriterIndex = buffer.readerIndex() + n;
 
-                if (limitWriterIndex < buffer.writerIndex()) {
-                    buffer.writerIndex((int) limitWriterIndex);
-                }
+                    if (limitWriterIndex < buffer.writerIndex()) {
+                        buffer.writerIndex((int) limitWriterIndex);
+                    }
 
-                if (position > to && !buffer.isReadable()) {
-                    return null;
+                    if (position > to && !buffer.isReadable()) {
+                        return null;
+                    }
                 }
-            }
 
-            position += consumed;
+                position += consumed;
 
-            if (buffer.isReadable()) {
-                return buffer;
+                if (buffer.isReadable()) {
+                    release = false;
+                    return buffer;
+                }
+            } finally {
+                if (release && buffer != null) {
+                    buffer.release();
+                }
             }
         }
     }