Fix for #670

tejksat · tejksat · commit 27bffd66ef5f · 2016-08-28T14:18:51.000+03:00
LinkedTransferQueue replaced with SynchronousQueue to fix OutOfDirectMemoryError. InputStream.read() bytes method implemented to improve performance.
diff --git a/src/main/java/com/github/dockerjava/netty/handler/HttpResponseStreamHandler.java b/src/main/java/com/github/dockerjava/netty/handler/HttpResponseStreamHandler.java
@@ -1,16 +1,15 @@
 package com.github.dockerjava.netty.handler;
 
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.concurrent.SynchronousQueue;
+
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.SimpleChannelInboundHandler;
 
-import java.io.IOException;
-import java.io.InputStream;
-import java.util.concurrent.LinkedTransferQueue;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-
 import com.github.dockerjava.api.async.ResultCallback;
+import com.google.common.base.Optional;
 
 /**
  * Handler that converts an incoming byte stream to an {@link InputStream}.
@@ -20,46 +19,37 @@
 public class HttpResponseStreamHandler extends SimpleChannelInboundHandler<ByteBuf> {
 
     private final HttpResponseInputStream stream = new HttpResponseInputStream();
-    private ResultCallback<InputStream> resultCallback;
 
     public HttpResponseStreamHandler(ResultCallback<InputStream> resultCallback) {
-        this.resultCallback = resultCallback;
+        resultCallback.onNext(stream);
     }
 
     @Override
     protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
-        invokeCallbackOnFirstRead();
-
         stream.write(msg.copy());
     }
 
-    private void invokeCallbackOnFirstRead() {
-        if (resultCallback != null) {
-            resultCallback.onNext(stream);
-            resultCallback = null;
-        }
-    }
-
     @Override
-    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
+    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
         stream.writeComplete();
-        super.channelReadComplete(ctx);
+
+        super.channelInactive(ctx);
     }
 
     public static class HttpResponseInputStream extends InputStream {
 
-        private AtomicBoolean writeCompleted = new AtomicBoolean(false);
+        private boolean writeCompleted = false;
 
-        private LinkedTransferQueue<ByteBuf> queue = new LinkedTransferQueue<>();
+        private SynchronousQueue<Optional<ByteBuf>> queue = new SynchronousQueue<>();
 
         private ByteBuf current = null;
 
-        public void write(ByteBuf byteBuf) {
-            queue.put(byteBuf);
+        public void write(ByteBuf byteBuf) throws InterruptedException {
+            queue.put(Optional.of(byteBuf));
         }
 
-        public void writeComplete() {
-            writeCompleted.set(true);
+        public void writeComplete() throws InterruptedException {
+            queue.put(Optional.<ByteBuf>absent());
         }
 
         @Override
@@ -70,16 +60,16 @@ public void close() throws IOException {
         }
 
         private void releaseQueued() {
-            ByteBuf byteBuf = queue.poll();
-            while (byteBuf != null) {
-                byteBuf.release();
+            Optional<ByteBuf> byteBuf = queue.poll();
+            while (byteBuf != null && byteBuf.isPresent()) {
+                byteBuf.get().release();
                 byteBuf = queue.poll();
             }
         }
 
         @Override
         public int available() throws IOException {
-            poll();
+            poll(0);
             return readableBytes();
         }
 
@@ -94,38 +84,52 @@ private int readableBytes() {
 
         @Override
         public int read() throws IOException {
+            byte[] b = new byte[1];
+            int n = read(b, 0, 1);
+            return n != -1 ? b[0] : -1;
+        }
 
-            poll();
-
-            if (readableBytes() == 0) {
-                if (writeCompleted.get()) {
-                    return -1;
-                }
-            }
-
-            if (current != null && current.readableBytes() > 0) {
-                return current.readByte() & 0xff;
+        @Override
+        public int read(byte[] b, int off, int len) throws IOException {
+            off = poll(off);
+            if (current == null) {
+                return -1;
             } else {
-                return read();
+                int availableBytes = Math.min(len, current.readableBytes() - off);
+                current.readBytes(b, off, availableBytes);
+                return availableBytes;
             }
         }
 
-        private void poll() {
-            if (readableBytes() == 0) {
+        private int poll(int off) {
+            if (writeCompleted) {
+                return off;
+            }
+            while (readableBytes() <= off) {
                 try {
-                    releaseCurrent();
-                    current = queue.poll(500, TimeUnit.MILLISECONDS);
+                    off -= releaseCurrent();
+                    Optional<ByteBuf> optional = queue.take();
+                    if (optional.isPresent()) {
+                        current = optional.get();
+                    } else {
+                        writeCompleted = true;
+                        return off;
+                    }
                 } catch (InterruptedException e) {
                     throw new RuntimeException(e);
                 }
             }
+            return off;
         }
 
-        private void releaseCurrent() {
+        private int releaseCurrent() {
             if (current != null) {
+                int n = current.readableBytes();
                 current.release();
                 current = null;
+                return n;
             }
+            return 0;
         }
     }
 }
diff --git a/src/test/java/com/github/dockerjava/netty/exec/CopyArchiveFromContainerCmdExecTest.java b/src/test/java/com/github/dockerjava/netty/exec/CopyArchiveFromContainerCmdExecTest.java
@@ -93,8 +93,6 @@ public void copyFromContainerBinaryFile() throws Exception {
         }
 
         InputStream response = dockerClient.copyArchiveFromContainerCmd(container.getId(), "/binary.dat").exec();
-        Boolean bytesAvailable = response.available() > 0;
-        assertTrue(bytesAvailable, "The file was not copied from the container.");
 
         try (TarArchiveInputStream tarInputStream = new TarArchiveInputStream(response)) {
             TarArchiveEntry nextTarEntry = tarInputStream.getNextTarEntry();

Original file line number	Diff line number	Diff line change
`@@ -93,8 +93,6 @@ public void copyFromContainerBinaryFile() throws Exception {`
`93`	`93`	`}`
`94`	`94`
`95`	`95`	`InputStream response = dockerClient.copyArchiveFromContainerCmd(container.getId(), "/binary.dat").exec();`
`96`		`- Boolean bytesAvailable = response.available() > 0;`
`97`		`- assertTrue(bytesAvailable, "The file was not copied from the container.");`
`98`	`96`
`99`	`97`	`try (TarArchiveInputStream tarInputStream = new TarArchiveInputStream(response)) {`
`100`	`98`	`TarArchiveEntry nextTarEntry = tarInputStream.getNextTarEntry();`