Port changes for AsyncHttpClient#411 from 1.7 to master.

Author: rlubke

providers/grizzly/src/main/java/org/asynchttpclient/providers/grizzly/bodyhandler/PartsBodyHandler.java

@@ -13,17 +13,22 @@
 
 package org.asynchttpclient.providers.grizzly.bodyhandler;
 
+import org.asynchttpclient.Body;
+import org.asynchttpclient.Part;
 import org.asynchttpclient.Request;
+import org.asynchttpclient.multipart.MultipartBody;
 import org.asynchttpclient.multipart.MultipartRequestEntity;
+import org.asynchttpclient.providers.grizzly.FeedableBodyGenerator;
+import org.asynchttpclient.providers.grizzly.GrizzlyAsyncHttpProvider;
 import org.asynchttpclient.util.AsyncHttpProviderUtils;
 import org.glassfish.grizzly.Buffer;
 import org.glassfish.grizzly.filterchain.FilterChainContext;
-import org.glassfish.grizzly.http.HttpContent;
 import org.glassfish.grizzly.http.HttpRequestPacket;
+import org.glassfish.grizzly.memory.Buffers;
 import org.glassfish.grizzly.memory.MemoryManager;
-import org.glassfish.grizzly.utils.BufferOutputStream;
 
 import java.io.IOException;
+import java.util.List;
 
 import static org.asynchttpclient.util.MiscUtil.isNonEmpty;
 
@@ -42,25 +47,70 @@ public boolean doHandle(final FilterChainContext ctx,
                          final HttpRequestPacket requestPacket)
     throws IOException {
 
-        MultipartRequestEntity mre =
+        final List<Part> parts = request.getParts();
+        final MultipartRequestEntity mre =
                 AsyncHttpProviderUtils.createMultipartRequestEntity(
-                        request.getParts(),
-                        request.getHeaders());
-        requestPacket.setContentLengthLong(mre.getContentLength());
-        requestPacket.setContentType(mre.getContentType());
-        final MemoryManager mm = ctx.getMemoryManager();
-        Buffer b = mm.allocate(512);
-        BufferOutputStream o = new BufferOutputStream(mm, b, true);
-        mre.writeRequest(o);
-        b = o.getBuffer();
-        b.trim();
-        if (b.hasRemaining()) {
-            final HttpContent content = requestPacket.httpContentBuilder().content(b).build();
-            content.setLast(true);
-            ctx.write(content, ((!requestPacket.isCommitted()) ? ctx.getTransportContext().getCompletionHandler() : null));
+                        parts, request.getHeaders());
+        final long contentLength = mre.getContentLength();
+        final String contentType = mre.getContentType();
+        requestPacket.setContentLengthLong(contentLength);
+        requestPacket.setContentType(contentType);
+        if (GrizzlyAsyncHttpProvider.LOGGER.isDebugEnabled()) {
+            GrizzlyAsyncHttpProvider.LOGGER.debug(
+                    "REQUEST(modified): contentLength={}, contentType={}",
+                         new Object[]{
+                                 requestPacket.getContentLength(),
+                                 requestPacket.getContentType()
+                         });
         }
 
-        return true;
+        final FeedableBodyGenerator generator = new FeedableBodyGenerator() {
+            @Override
+            public Body createBody() throws IOException {
+                return new MultipartBody(parts, contentType,
+                                         String.valueOf(contentLength));
+            }
+        };
+        generator.setFeeder(new FeedableBodyGenerator.BaseFeeder(generator) {
+            @Override
+            public void flush() throws IOException {
+                final Body bodyLocal = feedableBodyGenerator.createBody();
+                try {
+                    final MemoryManager mm = ctx.getMemoryManager();
+                    boolean last = false;
+                    while (!last) {
+                        Buffer buffer = mm.allocate(BodyHandler.MAX_CHUNK_SIZE);
+                        buffer.allowBufferDispose(true);
+                        final long readBytes =
+                                bodyLocal.read(buffer.toByteBuffer());
+                        if (readBytes > 0) {
+                            buffer.position((int) readBytes);
+                            buffer.trim();
+                        } else {
+                            buffer.dispose();
+                            if (readBytes < 0) {
+                                last = true;
+                                buffer = Buffers.EMPTY_BUFFER;
+                            } else {
+                                throw new IllegalStateException(
+                                        "MultipartBody unexpectedly returned 0 bytes available");
+                            }
+                        }
+                        feed(buffer, last);
+                    }
+                } finally {
+                    if (bodyLocal != null) {
+                        try {
+                            bodyLocal.close();
+                        } catch (IOException ignore) {
+                        }
+                    }
+                }
+            }
+        });
+        generator.initializeAsynchronousTransfer(ctx, requestPacket);
+        return false;
+
     }
 
 } // END PartsBodyHandler