Merge pull request #14 from facundovs/BODY_GENERATOR_NOT_CLOSES_STREAM

Body generator is not closing the given source.

pom.xml

@@ -137,6 +137,12 @@
                 </exclusion>
             </exclusions>
         </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-all</artifactId>
+            <version>1.10.19</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.eclipse.jetty</groupId>
             <artifactId>jetty-server</artifactId>

src/main/java/com/ning/http/client/providers/grizzly/PayloadGenFactory.java

@@ -38,6 +38,7 @@
 import org.glassfish.grizzly.memory.Buffers;
 import org.glassfish.grizzly.memory.MemoryManager;
 import org.glassfish.grizzly.utils.Charsets;
+import sun.misc.IOUtils;
 
 import static com.ning.http.client.providers.grizzly.PayloadGenerator.MAX_CHUNK_SIZE;
 import static com.ning.http.util.MiscUtils.isNonEmpty;
@@ -480,52 +481,60 @@ public boolean generate(final FilterChainContext ctx,
                              final Request request,
                              final HttpRequestPacket requestPacket)
         throws IOException {
-
+            Body bodyLocal = null;
             final BodyGenerator generator = request.getBodyGenerator();
-            final Body bodyLocal = generator.createBody();
-            final long len = bodyLocal.getContentLength();
-            if (len >= 0) {
-                requestPacket.setContentLengthLong(len);
-            } else {
-                requestPacket.setChunked(true);
-            }
 
-            final MemoryManager mm = ctx.getMemoryManager();
-            boolean last = false;
-
-            while (!last) {
-                Buffer buffer = mm.allocate(MAX_CHUNK_SIZE);
-                buffer.allowBufferDispose(true);
-
-                final long readBytes = bodyLocal.read(buffer.toByteBuffer());
-                if (readBytes > 0) {
-                    buffer.position((int) readBytes);
-                    buffer.trim();
+            try {
+                bodyLocal = generator.createBody();
+                final long len = bodyLocal.getContentLength();
+                if (len >= 0) {
+                    requestPacket.setContentLengthLong(len);
                 } else {
-                    buffer.dispose();
-
-                    if (readBytes < 0) {
-                        last = true;
-                        buffer = Buffers.EMPTY_BUFFER;
+                    requestPacket.setChunked(true);
+                }
+
+                final MemoryManager mm = ctx.getMemoryManager();
+                boolean last = false;
+
+                while (!last) {
+                    Buffer buffer = mm.allocate(MAX_CHUNK_SIZE);
+                    buffer.allowBufferDispose(true);
+
+                    final long readBytes = bodyLocal.read(buffer.toByteBuffer());
+                    if (readBytes > 0) {
+                        buffer.position((int) readBytes);
+                        buffer.trim();
                     } else {
-                        // pass the context to bodyLocal to be able to
-                        // continue body transferring once more data is available
-                        if (generator instanceof FeedableBodyGenerator) {
-                            ((FeedableBodyGenerator) generator).initializeAsynchronousTransfer(ctx, requestPacket);
-                            return false;
+                        buffer.dispose();
+
+                        if (readBytes < 0) {
+                            last = true;
+                            buffer = Buffers.EMPTY_BUFFER;
                         } else {
-                            throw new IllegalStateException("BodyGenerator unexpectedly returned 0 bytes available");
+                            // pass the context to bodyLocal to be able to
+                            // continue body transferring once more data is available
+                            if (generator instanceof FeedableBodyGenerator) {
+                                ((FeedableBodyGenerator) generator).initializeAsynchronousTransfer(ctx, requestPacket);
+                                return false;
+                            } else {
+                                throw new IllegalStateException("BodyGenerator unexpectedly returned 0 bytes available");
+                            }
                         }
                     }
+
+                    final HttpContent content =
+                            requestPacket.httpContentBuilder().content(buffer).
+                                    last(last).build();
+                    ctx.write(content, ((!requestPacket.isCommitted()) ? ctx.getTransportContext().getCompletionHandler() : null));
                 }
 
-                final HttpContent content =
-                        requestPacket.httpContentBuilder().content(buffer).
-                                last(last).build();
-                ctx.write(content, ((!requestPacket.isCommitted()) ? ctx.getTransportContext().getCompletionHandler() : null));
+                return true;
+            }
+            finally {
+                if (bodyLocal != null && ! ( generator instanceof  FeedableBodyGenerator)){
+                    bodyLocal.close();
+                }
             }
-
-            return true;
         }
 
     } // END BodyGeneratorAdapter        

src/test/java/com/ning/http/client/providers/grizzly/BodyGeneratorTest.java

@@ -0,0 +1,66 @@
+/*
+ * Copyright 2010 Ning, Inc.
+ *
+ * Ning licenses this file to you under the Apache License, version 2.0
+ * (the "License"); you may not use this file except in compliance with the
+ * License.  You may obtain a copy of the License at:
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.ning.http.client.providers.grizzly;
+
+import com.ning.http.client.Body;
+import com.ning.http.client.BodyGenerator;
+import com.ning.http.client.Request;
+
+import org.glassfish.grizzly.Buffer;
+import org.glassfish.grizzly.filterchain.FilterChainContext;
+import org.glassfish.grizzly.http.HttpRequestPacket;
+import org.glassfish.grizzly.memory.MemoryManager;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import static com.ning.http.client.providers.grizzly.PayloadGenFactory.getPayloadGenerator;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyInt;
+import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.nio.ByteBuffer;
+
+public class BodyGeneratorTest {
+
+    private final Request request = mock(Request.class);
+    private final BodyGenerator bodyGenerator = mock(BodyGenerator.class);
+    private final Body body = mock(Body.class);
+    private final FilterChainContext context = mock(FilterChainContext.class, RETURNS_DEEP_STUBS);
+    private final HttpRequestPacket requestPacket = mock(HttpRequestPacket.class);
+    private final MemoryManager memoryManager = mock(MemoryManager.class);
+    private final Buffer buffer = mock(Buffer.class);
+
+    @BeforeMethod
+    public void setUp() throws Exception {
+        when(request.getBodyGenerator()).thenReturn(bodyGenerator);
+        when(bodyGenerator.createBody()).thenReturn(body);
+        when(context.getMemoryManager()).thenReturn(memoryManager);
+        when(requestPacket.isCommitted()).thenReturn(true);
+        when(memoryManager.allocate(anyInt())).thenReturn(buffer);
+        when(body.read(any(ByteBuffer.class))).thenReturn(-1L);
+    }
+
+    @Test
+    public void testBodyIsClosed() throws Exception {
+        getPayloadGenerator(request).generate(context, request, requestPacket);
+        verify(body).close();
+    }
+
+}