Have Netty bypass Body API in case of a File or a Stream

Author: Stephane Landelle

api/src/main/java/org/asynchttpclient/BodyGenerator.java

@@ -28,7 +28,5 @@ public interface BodyGenerator {
      * @return The request body, never {@code null}.
      * @throws IOException If the body could not be created.
      */
-    Body createBody()
-            throws IOException;
-
+    Body createBody() throws IOException;
 }

api/src/main/java/org/asynchttpclient/generators/FileBodyGenerator.java

@@ -25,63 +25,59 @@
 /**
  * Creates a request body from the contents of a file.
  */
-public class FileBodyGenerator
-        implements BodyGenerator {
+public class FileBodyGenerator implements BodyGenerator {
 
     private final File file;
     private final long regionSeek;
     private final long regionLength;
 
     public FileBodyGenerator(File file) {
-        if (file == null) {
-            throw new IllegalArgumentException("no file specified");
-        }
-        this.file = file;
-        this.regionLength = file.length();
-        this.regionSeek = 0;
+        this(file, 0L, file.length());
     }
 
     public FileBodyGenerator(File file, long regionSeek, long regionLength) {
         if (file == null) {
-            throw new IllegalArgumentException("no file specified");
+            throw new NullPointerException("file");
         }
         this.file = file;
         this.regionLength = regionLength;
         this.regionSeek = regionSeek;
     }
 
+    public File getFile() {
+        return file;
+    }
+
+    public long getRegionLength() {
+        return regionLength;
+    }
+
+    public long getRegionSeek() {
+        return regionSeek;
+    }
+
     /**
      * {@inheritDoc}
      */
     @Override
-    public RandomAccessBody createBody()
-            throws IOException {
+    public RandomAccessBody createBody() throws IOException {
         return new FileBody(file, regionSeek, regionLength);
     }
 
-    protected static class FileBody
-            implements RandomAccessBody {
+    private static class FileBody implements RandomAccessBody {
 
-        private final RandomAccessFile file;
+        private final RandomAccessFile raf;
 
         private final FileChannel channel;
 
         private final long length;
 
-        public FileBody(File file)
-                throws IOException {
-            this.file = new RandomAccessFile(file, "r");
-            channel = this.file.getChannel();
-            length = file.length();
-        }
-
-        public FileBody(File file, long regionSeek, long regionLength)
-                throws IOException {
-            this.file = new RandomAccessFile(file, "r");
-            channel = this.file.getChannel();
+        private FileBody(File file, long regionSeek, long regionLength) throws IOException {
+            raf = new RandomAccessFile(file, "r");
+            channel = raf.getChannel();
             length = regionLength;
             if (regionSeek > 0) {
-                this.file.seek(regionSeek);
+                raf.seek(regionSeek);
             }
         }
 
@@ -94,19 +90,15 @@ public long read(ByteBuffer buffer)
             return channel.read(buffer);
         }
 
-        public long transferTo(long position, long count, WritableByteChannel target)
-                throws IOException {
+        public long transferTo(long position, long count, WritableByteChannel target) throws IOException {
             if (count > length) {
                 count = length;
             }
             return channel.transferTo(position, count, target);
         }
 
-        public void close()
-                throws IOException {
-            file.close();
+        public void close() throws IOException {
+            raf.close();
         }
-
     }
-
 }

api/src/main/java/org/asynchttpclient/generators/InputStreamBodyGenerator.java

@@ -23,25 +23,17 @@
 import org.slf4j.LoggerFactory;
 
 /**
- * A {@link BodyGenerator} which use an {@link InputStream} for reading bytes, without having to read the entire
- * stream in memory.
+ * A {@link BodyGenerator} which use an {@link InputStream} for reading bytes, without having to read the entire stream in memory.
  * <p/>
- * NOTE: The {@link InputStream} must support the {@link InputStream#mark} and {@link java.io.InputStream#reset()} operation.
- * If not, mechanisms like authentication, redirect, or resumable download will not works.
+ * NOTE: The {@link InputStream} must support the {@link InputStream#mark} and {@link java.io.InputStream#reset()} operation. If not, mechanisms like authentication, redirect, or
+ * resumable download will not works.
  */
 public class InputStreamBodyGenerator implements BodyGenerator {
 
     private final InputStream inputStream;
-    private final static Logger logger = LoggerFactory.getLogger(InputStreamBodyGenerator.class);
 
     public InputStreamBodyGenerator(InputStream inputStream) {
         this.inputStream = inputStream;
-
-        if (inputStream.markSupported()) {
-            inputStream.mark(0);
-        } else {
-            logger.info("inputStream.markSupported() not supported. Some features will not work.");
-        }
     }
 
     public InputStream getInputStream() {
@@ -53,27 +45,39 @@ public InputStream getInputStream() {
      */
     @Override
     public Body createBody() throws IOException {
-        return new InputStreamBody();
+        return new InputStreamBody(inputStream);
     }
 
-    protected class InputStreamBody implements Body {
+    private static class InputStreamBody implements Body {
+
+        private static final Logger LOGGER = LoggerFactory.getLogger(InputStreamBody.class);
+
+        private final InputStream inputStream;
         private byte[] chunk;
 
+        private InputStreamBody(InputStream inputStream) {
+            this.inputStream = inputStream;
+            if (inputStream.markSupported()) {
+                inputStream.mark(0);
+            } else {
+                LOGGER.info("inputStream.markSupported() not supported. Some features will not work.");
+            }
+        }
+
         public long getContentLength() {
-            return -1;
+            return -1L;
         }
 
         public long read(ByteBuffer buffer) throws IOException {
 
             // To be safe.
             chunk = new byte[buffer.remaining() - 10];
 
-
             int read = -1;
             try {
                 read = inputStream.read(chunk);
             } catch (IOException ex) {
-                logger.warn("Unable to read", ex);
+                LOGGER.warn("Unable to read", ex);
             }
 
             if (read > 0) {

providers/netty/src/main/java/org/asynchttpclient/providers/netty/request/NettyRequestSender.java

@@ -37,6 +37,7 @@
 import java.io.RandomAccessFile;
 import java.net.InetSocketAddress;
 import java.net.URI;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.Future;
 import java.util.concurrent.RejectedExecutionException;
@@ -57,9 +58,11 @@
 import org.asynchttpclient.filter.FilterContext;
 import org.asynchttpclient.filter.FilterException;
 import org.asynchttpclient.filter.IOExceptionFilter;
+import org.asynchttpclient.generators.FileBodyGenerator;
 import org.asynchttpclient.generators.InputStreamBodyGenerator;
 import org.asynchttpclient.listener.TransferCompletionHandler;
 import org.asynchttpclient.multipart.MultipartBody;
+import org.asynchttpclient.multipart.Part;
 import org.asynchttpclient.providers.netty.Constants;
 import org.asynchttpclient.providers.netty.channel.Channels;
 import org.asynchttpclient.providers.netty.future.FutureReaper;
@@ -255,8 +258,7 @@ private <T> ListenableFuture<T> sendRequestWithNewChannel(Request request, URI u
         return cl.future();
     }
 
-    public <T> ListenableFuture<T> sendRequest(final Request request, final AsyncHandler<T> asyncHandler, NettyResponseFuture<T> future, boolean reclaimCache)
-            throws IOException {
+    public <T> ListenableFuture<T> sendRequest(final Request request, final AsyncHandler<T> asyncHandler, NettyResponseFuture<T> future, boolean reclaimCache) throws IOException {
 
         if (closed.get()) {
             throw new IOException("Closed");
@@ -278,17 +280,15 @@ public <T> ListenableFuture<T> sendRequest(final Request request, final AsyncHan
         }
     }
 
-    private void sendFileBody(Channel channel, File file, NettyResponseFuture<?> future) throws IOException {
+    private void sendFileBody(Channel channel, File file, long offset, long fileLength, NettyResponseFuture<?> future) throws IOException {
         final RandomAccessFile raf = new RandomAccessFile(file, "r");
 
         try {
-            long fileLength = raf.length();
-
             ChannelFuture writeFuture;
             if (Channels.getSslHandler(channel) != null) {
-                writeFuture = channel.write(new ChunkedFile(raf, 0, fileLength, Constants.MAX_BUFFERED_BYTES), channel.newProgressivePromise());
+                writeFuture = channel.write(new ChunkedFile(raf, offset, fileLength, Constants.MAX_BUFFERED_BYTES), channel.newProgressivePromise());
             } else {
-                FileRegion region = new DefaultFileRegion(raf.getChannel(), 0, fileLength);
+                FileRegion region = new DefaultFileRegion(raf.getChannel(), offset, fileLength);
                 writeFuture = channel.write(region, channel.newProgressivePromise());
             }
             // FIXME probably useless in Netty 4
@@ -345,10 +345,13 @@ public void operationComplete(ChannelProgressiveFuture cf) {
     public void sendBody(final Channel channel, final Body body, NettyResponseFuture<?> future) {
         Object msg;
         if (Channels.getSslHandler(channel) == null && body instanceof RandomAccessBody) {
+            // FIXME also do something for multipart and use a ChunkedInput
             msg = new BodyFileRegion((RandomAccessBody) body);
+
         } else {
-            BodyGenerator bg = future.getRequest().getBodyGenerator();
             msg = new BodyChunkedInput(body);
+
+            BodyGenerator bg = future.getRequest().getBodyGenerator();
             if (bg instanceof FeedableBodyGenerator) {
                 FeedableBodyGenerator.class.cast(bg).setListener(new FeedListener() {
                     @Override
@@ -374,38 +377,48 @@ public void operationComplete(ChannelProgressiveFuture cf) {
         channel.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
     }
 
-    private Body computeBody(HttpRequest nettyRequest, NettyResponseFuture<?> future) {
-
-        if (nettyRequest.getMethod().equals(HttpMethod.CONNECT)) {
-            return null;
-        }
+    private Body computeBodyAndSetContentLengthOrTransferEncodingHeader(HttpRequest nettyRequest, NettyResponseFuture<?> future) {
 
-        HttpHeaders headers = nettyRequest.headers();
         BodyGenerator bg = future.getRequest().getBodyGenerator();
+        List<Part> parts = future.getRequest().getParts();
         Body body = null;
-        if (bg != null) {
-            try {
-                body = bg.createBody();
-            } catch (IOException ex) {
-                throw new IllegalStateException(ex);
+
+        if (bg != null || parts != null) {
+            HttpHeaders headers = nettyRequest.headers();
+            long length = -1L;
+
+            if (bg instanceof FileBodyGenerator) {
+                // don't even compute body, file be be directly passed
+                length = FileBodyGenerator.class.cast(bg).getRegionLength();
+
+            } else if (bg instanceof InputStreamBodyGenerator) {
+                // don't even compute body, stream be be directly passed
+                length = -1L;
+
+            } else if (bg != null) {
+                try {
+                    body = bg.createBody();
+                    length = body.getContentLength();
+
+                } catch (IOException ex) {
+                    throw new IllegalStateException(ex);
+                }
+
+            } else {
+                String contentType = headers.get(HttpHeaders.Names.CONTENT_TYPE);
+                String contentLength = nettyRequest.headers().get(HttpHeaders.Names.CONTENT_LENGTH);
+
+                if (contentLength != null) {
+                    length = Long.parseLong(contentLength);
+                }
+                body = new MultipartBody(parts, contentType, length);
             }
-            long length = body.getContentLength();
+
             if (length >= 0) {
                 headers.set(HttpHeaders.Names.CONTENT_LENGTH, length);
             } else {
                 headers.set(HttpHeaders.Names.TRANSFER_ENCODING, HttpHeaders.Values.CHUNKED);
             }
-        } else if (future.getRequest().getParts() != null) {
-            String contentType = headers.get(HttpHeaders.Names.CONTENT_TYPE);
-            String contentLength = nettyRequest.headers().get(HttpHeaders.Names.CONTENT_LENGTH);
-
-            long length = -1;
-            if (contentLength != null) {
-                length = Long.parseLong(contentLength);
-            } else {
-                nettyRequest.headers().add(HttpHeaders.Names.TRANSFER_ENCODING, HttpHeaders.Values.CHUNKED);
-            }
-            body = new MultipartBody(future.getRequest().getParts(), contentType, length);
         }
 
         return body;
@@ -449,7 +462,9 @@ public final <T> void writeRequest(final Channel channel, final AsyncHttpClientC
 
             HttpRequest nettyRequest = future.getNettyRequest();
             AsyncHandler<T> handler = future.getAsyncHandler();
-            Body body = computeBody(nettyRequest, future);
+
+            // beware: compute early because actually also write headers
+            Body body = computeBodyAndSetContentLengthOrTransferEncodingHeader(nettyRequest, future);
 
             if (handler instanceof TransferCompletionHandler) {
                 configureTransferAdapter(handler, nettyRequest);
@@ -477,14 +492,22 @@ public final <T> void writeRequest(final Channel channel, final AsyncHttpClientC
 
             // FIXME OK, why? and what's the point of not having a is/get?
             if (future.getAndSetWriteBody(true)) {
-                if (!future.getNettyRequest().getMethod().equals(HttpMethod.CONNECT)) {
+                if (!nettyRequest.getMethod().equals(HttpMethod.CONNECT)) {
+
                     if (future.getRequest().getFile() != null) {
-                        sendFileBody(channel, future.getRequest().getFile(), future);
+                        File file = future.getRequest().getFile();
+                        sendFileBody(channel, file, 0, file.length(), future);
+
+                    } else if (future.getRequest().getBodyGenerator() instanceof FileBodyGenerator) {
+                        // 2 different ways of passing a file, wouhou!
+                        FileBodyGenerator fileBodyGenerator = (FileBodyGenerator) future.getRequest().getBodyGenerator();
+                        sendFileBody(channel, fileBodyGenerator.getFile(), fileBodyGenerator.getRegionSeek(), fileBodyGenerator.getRegionLength(), future);
 
                     } else if (future.getRequest().getStreamData() != null) {
                         if (sendStreamAndExit(channel, future.getRequest().getStreamData(), future))
                             return;
                     } else if (future.getRequest().getBodyGenerator() instanceof InputStreamBodyGenerator) {
+                        // 2 different ways of passing a stream, wouhou!
                         if (sendStreamAndExit(channel, InputStreamBodyGenerator.class.cast(future.getRequest().getBodyGenerator()).getInputStream(), future))
                             return;