Split writeRequest

Author: Stephane Landelle

api/src/main/java/org/asynchttpclient/listener/TransferCompletionHandler.java

@@ -219,7 +219,7 @@ private void fireOnThrowable(Throwable t) {
     public static class TransferAdapter {
         private final FluentCaseInsensitiveStringsMap headers;
 
-        public TransferAdapter(FluentCaseInsensitiveStringsMap headers) throws IOException {
+        public TransferAdapter(FluentCaseInsensitiveStringsMap headers) {
             this.headers = headers;
         }
 

providers/netty4/src/main/java/org/asynchttpclient/providers/netty4/NettyRequestSender.java

@@ -182,7 +182,7 @@ public <T> ListenableFuture<T> doConnect(final Request request, final AsyncHandl
         boolean acquiredConnection = !reclaimCache && channels.acquireConnection(asyncHandler);
 
         NettyConnectListener<T> cl = new NettyConnectListener.Builder<T>(config, this, request, asyncHandler, future).build(uri);
-        
+
         boolean avoidProxy = ProxyUtils.avoidProxy(proxyServer, uri.getHost());
 
         if (useSSl) {
@@ -261,7 +261,158 @@ public <T> ListenableFuture<T> doConnect(final Request request, final AsyncHandl
         }
         return cl.future();
     }
-    
+
+    private void sendFileBody(Channel channel, File file, 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());
+            } else {
+                // FIXME why not use io.netty.channel.DefaultFileRegion?
+                FileRegion region = new OptimizedFileRegion(raf, 0, fileLength);
+                writeFuture = channel.write(region, channel.newProgressivePromise());
+            }
+            writeFuture.addListener(new ProgressListener(config, false, future.getAsyncHandler(), future) {
+                public void operationComplete(ChannelProgressiveFuture cf) {
+                    try {
+                        raf.close();
+                    } catch (IOException e) {
+                        LOGGER.warn("Failed to close request body: {}", e.getMessage(), e);
+                    }
+                    super.operationComplete(cf);
+                }
+            });
+            channel.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
+        } catch (IOException ex) {
+            if (raf != null) {
+                try {
+                    raf.close();
+                } catch (IOException e) {
+                }
+            }
+            throw ex;
+        }
+    }
+
+    private boolean sendStreamAndExit(Channel channel, final InputStream is, NettyResponseFuture<?> future) throws IOException {
+
+        if (future.getAndSetStreamWasAlreadyConsumed()) {
+            if (is.markSupported())
+                is.reset();
+            else {
+                LOGGER.warn("Stream has already been consumed and cannot be reset");
+                return true;
+            }
+        }
+
+        channel.write(new ChunkedStream(is), channel.newProgressivePromise()).addListener(new ProgressListener(config, false, future.getAsyncHandler(), future) {
+            public void operationComplete(ChannelProgressiveFuture cf) {
+                try {
+                    is.close();
+                } catch (IOException e) {
+                    LOGGER.warn("Failed to close request body: {}", e.getMessage(), e);
+                }
+                super.operationComplete(cf);
+            }
+        });
+        channel.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
+
+        return false;
+    }
+
+    public void sendBody(final Channel channel, final Body body, NettyResponseFuture<?> future) {
+        Object msg;
+        if (Channels.getSslHandler(channel) == null && body instanceof RandomAccessBody) {
+            msg = new BodyFileRegion((RandomAccessBody) body);
+        } else {
+            BodyGenerator bg = future.getRequest().getBodyGenerator();
+            msg = new BodyChunkedInput(body);
+            if (bg instanceof FeedableBodyGenerator) {
+                FeedableBodyGenerator.class.cast(bg).setListener(new FeedListener() {
+                    @Override
+                    public void onContentAdded() {
+                        channel.pipeline().get(ChunkedWriteHandler.class).resumeTransfer();
+                    }
+                });
+            }
+        }
+        ChannelFuture writeFuture = channel.write(msg, channel.newProgressivePromise());
+
+        final Body b = body;
+        writeFuture.addListener(new ProgressListener(config, false, future.getAsyncHandler(), future) {
+            public void operationComplete(ChannelProgressiveFuture cf) {
+                try {
+                    b.close();
+                } catch (IOException e) {
+                    LOGGER.warn("Failed to close request body: {}", e.getMessage(), e);
+                }
+                super.operationComplete(cf);
+            }
+        });
+        channel.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
+    }
+
+    private Body computeBody(HttpRequest nettyRequest, NettyResponseFuture<?> future) {
+
+        if (nettyRequest.getMethod().equals(HttpMethod.CONNECT)) {
+            return null;
+        }
+
+        HttpHeaders headers = nettyRequest.headers();
+        BodyGenerator bg = future.getRequest().getBodyGenerator();
+        Body body = null;
+        if (bg != null) {
+            try {
+                body = bg.createBody();
+            } catch (IOException ex) {
+                throw new IllegalStateException(ex);
+            }
+            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 length = headers.get(HttpHeaders.Names.CONTENT_LENGTH);
+            body = new MultipartBody(future.getRequest().getParts(), contentType, length);
+        }
+
+        return body;
+    }
+
+    private void configureTransferAdapter(AsyncHandler<?> handler, HttpRequest nettyRequest) {
+        FluentCaseInsensitiveStringsMap h = new FluentCaseInsensitiveStringsMap();
+        for (Map.Entry<String, String> entries : nettyRequest.headers()) {
+            h.add(entries.getKey(), entries.getValue());
+        }
+
+        TransferCompletionHandler.class.cast(handler).transferAdapter(new TransferAdapter(h));
+    }
+
+    private void scheduleReaper(NettyResponseFuture<?> future) {
+        try {
+            future.touch();
+            int requestTimeout = AsyncHttpProviderUtils.requestTimeout(config, future.getRequest());
+            int schedulePeriod = requestTimeout != -1 ? (config.getIdleConnectionTimeoutInMs() != -1 ? Math.min(requestTimeout, config.getIdleConnectionTimeoutInMs())
+                    : requestTimeout) : config.getIdleConnectionTimeoutInMs();
+
+            if (schedulePeriod != -1 && !future.isDone() && !future.isCancelled()) {
+                ReaperFuture reaperFuture = new ReaperFuture(future, config, isClose, channels);
+                Future<?> scheduledFuture = config.reaper().scheduleAtFixedRate(reaperFuture, 0, schedulePeriod, TimeUnit.MILLISECONDS);
+                reaperFuture.setScheduledFuture(scheduledFuture);
+                future.setReaperFuture(reaperFuture);
+            }
+        } catch (RejectedExecutionException ex) {
+            channels.abort(future, ex);
+        }
+    }
+
     protected final <T> void writeRequest(final Channel channel, final AsyncHttpClientConfig config, final NettyResponseFuture<T> future) {
         try {
             // If the channel is dead because it was pooled and the remote
@@ -272,41 +423,15 @@ protected final <T> void writeRequest(final Channel channel, final AsyncHttpClie
             }
 
             HttpRequest nettyRequest = future.getNettyRequest();
-            Body body = null;
-            if (!nettyRequest.getMethod().equals(HttpMethod.CONNECT)) {
-                BodyGenerator bg = future.getRequest().getBodyGenerator();
-                if (bg != null) {
-                    try {
-                        body = bg.createBody();
-                    } catch (IOException ex) {
-                        throw new IllegalStateException(ex);
-                    }
-                    long length = body.getContentLength();
-                    if (length >= 0) {
-                        nettyRequest.headers().set(HttpHeaders.Names.CONTENT_LENGTH, length);
-                    } else {
-                        nettyRequest.headers().set(HttpHeaders.Names.TRANSFER_ENCODING, HttpHeaders.Values.CHUNKED);
-                    }
-                } else if (future.getRequest().getParts() != null) {
-                    String contentType = nettyRequest.headers().get(HttpHeaders.Names.CONTENT_TYPE);
-                    String length = nettyRequest.headers().get(HttpHeaders.Names.CONTENT_LENGTH);
-                    body = new MultipartBody(future.getRequest().getParts(), contentType, length);
-                }
-            }
-
-            if (future.getAsyncHandler() instanceof TransferCompletionHandler) {
-
-                FluentCaseInsensitiveStringsMap h = new FluentCaseInsensitiveStringsMap();
-                for (Map.Entry<String, String> entries : future.getNettyRequest().headers()) {
-                    h.add(entries.getKey(), entries.getValue());
-                }
+            AsyncHandler<T> handler = future.getAsyncHandler();
+            Body body = computeBody(nettyRequest, future);
 
-                TransferCompletionHandler.class.cast(future.getAsyncHandler()).transferAdapter(new TransferAdapter(h));
+            if (handler instanceof TransferCompletionHandler) {
+                configureTransferAdapter(handler, nettyRequest);
             }
 
             // Leave it to true.
-            // FIXME Yeah... explain why instead of saying the same thing as the
-            // code
+            // FIXME That doesn't just leave to true, the set is always done? and what's the point of not having a is/get?
             if (future.getAndSetWriteHeaders(true)) {
                 try {
                     channel.writeAndFlush(nettyRequest, channel.newProgressivePromise()).addListener(new ProgressListener(config, true, future.getAsyncHandler(), future));
@@ -322,104 +447,25 @@ protected final <T> void writeRequest(final Channel channel, final AsyncHttpClie
                 }
             }
 
+            // 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 (future.getRequest().getFile() != null) {
-                        final File file = future.getRequest().getFile();
-                        long fileLength = 0;
-                        final RandomAccessFile raf = new RandomAccessFile(file, "r");
-
-                        try {
-                            fileLength = raf.length();
-
-                            ChannelFuture writeFuture;
-                            if (Channels.getSslHandler(channel) != null) {
-                                writeFuture = channel.write(new ChunkedFile(raf, 0, fileLength, Constants.MAX_BUFFERED_BYTES), channel.newProgressivePromise());
-                            } else {
-                                // FIXME why not use io.netty.channel.DefaultFileRegion?
-                                FileRegion region = new OptimizedFileRegion(raf, 0, fileLength);
-                                writeFuture = channel.write(region, channel.newProgressivePromise());
-                            }
-                            writeFuture.addListener(new ProgressListener(config, false, future.getAsyncHandler(), future) {
-                                public void operationComplete(ChannelProgressiveFuture cf) {
-                                    try {
-                                        raf.close();
-                                    } catch (IOException e) {
-                                        LOGGER.warn("Failed to close request body: {}", e.getMessage(), e);
-                                    }
-                                    super.operationComplete(cf);
-                                }
-                            });
-                            channel.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
-                        } catch (IOException ex) {
-                            if (raf != null) {
-                                try {
-                                    raf.close();
-                                } catch (IOException e) {
-                                }
-                            }
-                            throw ex;
-                        }
-                    } else if (future.getRequest().getStreamData() != null || future.getRequest().getBodyGenerator() instanceof InputStreamBodyGenerator) {
-                        final InputStream is = future.getRequest().getStreamData() != null ? future.getRequest().getStreamData() : InputStreamBodyGenerator.class.cast(
-                                future.getRequest().getBodyGenerator()).getInputStream();
-
-                        if (future.getAndSetStreamWasAlreadyConsumed()) {
-                            if (is.markSupported())
-                                is.reset();
-                            else {
-                                LOGGER.warn("Stream has already been consumed and cannot be reset");
-                                return;
-                            }
-                        }
-
-                        channel.write(new ChunkedStream(is), channel.newProgressivePromise()).addListener(new ProgressListener(config, false, future.getAsyncHandler(), future) {
-                            public void operationComplete(ChannelProgressiveFuture cf) {
-                                try {
-                                    is.close();
-                                } catch (IOException e) {
-                                    LOGGER.warn("Failed to close request body: {}", e.getMessage(), e);
-                                }
-                                super.operationComplete(cf);
-                            }
-                        });
-                        channel.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
+                        sendFileBody(channel, future.getRequest().getFile(), future);
 
-                    } else if (body != null) {
+                    } else if (future.getRequest().getStreamData() != null) {
+                        if (sendStreamAndExit(channel, future.getRequest().getStreamData(), future))
+                            return;
+                    } else if (future.getRequest().getBodyGenerator() instanceof InputStreamBodyGenerator) {
+                        if (sendStreamAndExit(channel, InputStreamBodyGenerator.class.cast(future.getRequest().getBodyGenerator()).getInputStream(), future))
+                            return;
 
-                        Object msg;
-                        if (Channels.getSslHandler(channel) == null && body instanceof RandomAccessBody) {
-                            msg = new BodyFileRegion((RandomAccessBody) body);
-                        } else {
-                            BodyGenerator bg = future.getRequest().getBodyGenerator();
-                            msg = new BodyChunkedInput(body);
-                            if (bg instanceof FeedableBodyGenerator) {
-                                FeedableBodyGenerator.class.cast(bg).setListener(new FeedListener() {
-                                    @Override
-                                    public void onContentAdded() {
-                                        channel.pipeline().get(ChunkedWriteHandler.class).resumeTransfer();
-                                    }
-                                });
-                            }
-                        }
-                        ChannelFuture writeFuture = channel.write(msg, channel.newProgressivePromise());
-
-                        final Body b = body;
-                        writeFuture.addListener(new ProgressListener(config, false, future.getAsyncHandler(), future) {
-                            public void operationComplete(ChannelProgressiveFuture cf) {
-                                try {
-                                    b.close();
-                                } catch (IOException e) {
-                                    LOGGER.warn("Failed to close request body: {}", e.getMessage(), e);
-                                }
-                                super.operationComplete(cf);
-                            }
-                        });
-                        channel.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
+                    } else if (body != null) {
+                        sendBody(channel, body, future);
                     }
                 }
             }
+
         } catch (Throwable ioe) {
             try {
                 channel.close();
@@ -428,23 +474,9 @@ public void operationComplete(ChannelProgressiveFuture cf) {
             }
         }
 
-        try {
-            future.touch();
-            int requestTimeout = AsyncHttpProviderUtils.requestTimeout(config, future.getRequest());
-            int schedulePeriod = requestTimeout != -1 ? (config.getIdleConnectionTimeoutInMs() != -1 ? Math.min(requestTimeout, config.getIdleConnectionTimeoutInMs())
-                    : requestTimeout) : config.getIdleConnectionTimeoutInMs();
-
-            if (schedulePeriod != -1 && !future.isDone() && !future.isCancelled()) {
-                ReaperFuture reaperFuture = new ReaperFuture(future, config, isClose, channels);
-                Future<?> scheduledFuture = config.reaper().scheduleAtFixedRate(reaperFuture, 0, schedulePeriod, TimeUnit.MILLISECONDS);
-                reaperFuture.setScheduledFuture(scheduledFuture);
-                future.setReaperFuture(reaperFuture);
-            }
-        } catch (RejectedExecutionException ex) {
-            channels.abort(future, ex);
-        }
+        scheduleReaper(future);
     }
-    
+
     // FIXME Clean up Netty 3: replayRequest's response parameter is unused +
     // WTF return???
     public void replayRequest(final NettyResponseFuture<?> future, FilterContext fc, ChannelHandlerContext ctx) throws IOException {