+ initial implementation of partial body generator

Author: oleksiys

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

@@ -1960,9 +1960,14 @@ public boolean doHandle(final FilterChainContext ctx,
                         last = true;
                         buffer = Buffers.EMPTY_BUFFER;
                     } else {
-                        // @TODO pass the context to bodyLocal to be able to
+                        // pass the context to bodyLocal to be able to
                         // continue body transferring once more data is available
-                        return false;
+                        if (generator instanceof PartialBodyGenerator) {
+                            ((PartialBodyGenerator) generator).initializeAsynchronousTransfer(ctx, requestPacket);
+                            return false;
+                        } else {
+                            throw new IllegalStateException("BodyGenerator unexpectedly returned 0 bytes available");
+                        }
                     }
                 }
 

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

@@ -0,0 +1,113 @@
+/*
+ * Copyright (c) 2011 Sonatype, Inc. All rights reserved.
+ *
+ * This program is licensed to you under the Apache License Version 2.0,
+ * and you may not use this file except in compliance with the Apache License Version 2.0.
+ * You may obtain a copy of the Apache License Version 2.0 at http://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the Apache License Version 2.0 is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the Apache License Version 2.0 for the specific language governing permissions and limitations there under.
+ */
+package com.ning.http.client.providers.grizzly;
+
+import com.ning.http.client.Body;
+import com.ning.http.client.BodyGenerator;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.glassfish.grizzly.Buffer;
+import org.glassfish.grizzly.filterchain.FilterChainContext;
+import org.glassfish.grizzly.http.HttpContent;
+import org.glassfish.grizzly.http.HttpRequestPacket;
+
+/**
+ * {@link BodyGenerator} which may return just part of the payload at the time
+ * handler is requesting it. If it happens - PartialBodyGenerator becomes responsible
+ * for finishing payload transferring asynchronously.
+ *
+ * @author The Grizzly Team
+ * @since 1.7.0
+ */
+public class PartialBodyGenerator implements BodyGenerator {
+    private final Queue<BodyPart> queue = new ConcurrentLinkedQueue<BodyPart>();
+    private final AtomicInteger queueSize = new AtomicInteger();
+    
+    private volatile HttpRequestPacket requestPacket;
+    private volatile FilterChainContext context;
+    
+    @Override
+    public Body createBody() throws IOException {
+        return new PartialBody();
+    }
+    
+    public void append(final Buffer buffer, final boolean isLast)
+            throws IOException {
+        queue.offer(new BodyPart(buffer, isLast));
+        queueSize.incrementAndGet();
+        
+        if (context != null) {
+            flushQueue();            
+        }
+    }
+    
+    void initializeAsynchronousTransfer(final FilterChainContext context, 
+            final HttpRequestPacket requestPacket) throws IOException {
+        this.context = context;
+        this.requestPacket = requestPacket;
+        flushQueue();
+    }
+
+    private void flushQueue() throws IOException {
+        if (queueSize.get() > 0) {
+            synchronized(this) {
+                while(queueSize.get() > 0) {
+                    final BodyPart bodyPart = queue.poll();
+                    queueSize.decrementAndGet();
+                    final HttpContent content =
+                            requestPacket.httpContentBuilder()
+                            .content(bodyPart.buffer)
+                            .last(bodyPart.isLast)
+                            .build();
+                    context.write(content, ((!requestPacket.isCommitted()) ?
+                            context.getTransportContext().getCompletionHandler() :
+                            null));
+                    
+                }
+            }
+        }
+    }
+    
+    private final class PartialBody implements Body {
+
+        @Override
+        public long getContentLength() {
+            return -1;
+        }
+
+        @Override
+        public long read(final ByteBuffer buffer) throws IOException {
+            return 0;
+        }
+
+        @Override
+        public void close() throws IOException {
+            context.completeAndRecycle();
+            context = null;
+            requestPacket = null;
+        }
+    }
+    
+    private final static class BodyPart {
+        private final boolean isLast;
+        private final Buffer buffer;
+
+        public BodyPart(final Buffer buffer, final boolean isLast) {
+            this.buffer = buffer;
+            this.isLast = isLast;
+        }
+    }
+}