Introduce BlockingFeedableBodyGenerator, let feed throw Exception

Author: slandelle

client/src/main/java/org/asynchttpclient/request/body/generator/BlockingFeedableBodyGenerator.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright (c) 2015 AsyncHttpClient Project. 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 org.asynchttpclient.request.body.generator;
+
+import java.util.Queue;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+
+import org.asynchttpclient.request.body.generator.QueueBasedFeedableBodyGenerator.BodyChunk;
+
+public final class BlockingFeedableBodyGenerator extends QueueBasedFeedableBodyGenerator<BlockingQueue<BodyChunk>> {
+    private final BlockingQueue<BodyChunk> queue;
+
+    public BlockingFeedableBodyGenerator(int capacity) {
+        queue = new ArrayBlockingQueue<>(capacity);
+    }
+
+    @Override
+    protected boolean offer(BodyChunk chunk) throws InterruptedException {
+        queue.put(chunk);
+        return true;
+    }
+
+    @Override
+    protected Queue<org.asynchttpclient.request.body.generator.QueueBasedFeedableBodyGenerator.BodyChunk> queue() {
+        return queue;
+    }
+}

client/src/main/java/org/asynchttpclient/request/body/generator/FeedableBodyGenerator.java

@@ -21,7 +21,7 @@
  */
 public interface FeedableBodyGenerator extends BodyGenerator {
 
-    boolean feed(ByteBuffer buffer, boolean isLast);
+    boolean feed(ByteBuffer buffer, boolean isLast) throws Exception;
 
     void setListener(FeedListener listener);
 

client/src/main/java/org/asynchttpclient/request/body/generator/SimpleFeedableBodyGenerator.java renamed to client/src/main/java/org/asynchttpclient/request/body/generator/QueueBasedFeedableBodyGenerator.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2014 AsyncHttpClient Project. All rights reserved.
+ * Copyright (c) 2015 AsyncHttpClient Project. 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.
@@ -18,22 +18,25 @@
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.Queue;
-import java.util.concurrent.ConcurrentLinkedQueue;
 
 import org.asynchttpclient.request.body.Body;
+import org.asynchttpclient.request.body.generator.QueueBasedFeedableBodyGenerator.BodyChunk;
+
+public abstract class QueueBasedFeedableBodyGenerator<T extends Queue<BodyChunk>> implements FeedableBodyGenerator, BodyGenerator {
 
-public final class SimpleFeedableBodyGenerator implements FeedableBodyGenerator, BodyGenerator {
-    private final Queue<BodyChunk> queue = new ConcurrentLinkedQueue<>();
     private FeedListener listener;
 
     @Override
     public Body createBody() {
         return new PushBody();
     }
 
+    protected abstract boolean offer(BodyChunk chunk) throws Exception;
+    protected abstract Queue<BodyChunk> queue();
+    
     @Override
-    public boolean feed(final ByteBuffer buffer, final boolean isLast) {
-        boolean offered = queue.offer(new BodyChunk(buffer, isLast));
+    public boolean feed(final ByteBuffer buffer, final boolean isLast) throws Exception {
+        boolean offered = offer(new BodyChunk(buffer, isLast));
         if (offered && listener != null) {
             listener.onContentAdded();
         }
@@ -69,13 +72,13 @@ public BodyState transferTo(final ByteBuf target) throws IOException {
         private BodyState readNextChunk(ByteBuf target) throws IOException {
             BodyState res = BodyState.SUSPEND;
             while (target.isWritable() && state != BodyState.STOP) {
-                BodyChunk nextChunk = queue.peek();
+                BodyChunk nextChunk = queue().peek();
                 if (nextChunk == null) {
                     // Nothing in the queue. suspend stream if nothing was read. (reads == 0)
                     return res;
                 } else if (!nextChunk.buffer.hasRemaining() && !nextChunk.isLast) {
                     // skip empty buffers
-                    queue.remove();
+                    queue().remove();
                 } else {
                     res = BodyState.CONTINUE;
                     readChunk(target, nextChunk);
@@ -91,7 +94,7 @@ private void readChunk(ByteBuf target, BodyChunk part) {
                 if (part.isLast) {
                     state = BodyState.STOP;
                 }
-                queue.remove();
+                queue().remove();
             }
         }
 
@@ -110,7 +113,7 @@ private void move(ByteBuf target, ByteBuffer source) {
         }
     }
 
-    private final class BodyChunk {
+    public static final class BodyChunk {
         private final boolean isLast;
         private final ByteBuffer buffer;
 

client/src/main/java/org/asynchttpclient/request/body/generator/ReactiveStreamsBodyGenerator.java

@@ -35,15 +35,15 @@ public class ReactiveStreamsBodyGenerator implements FeedableBodyGenerator {
 
     public ReactiveStreamsBodyGenerator(Publisher<ByteBuffer> publisher) {
         this.publisher = publisher;
-        this.feedableBodyGenerator = new SimpleFeedableBodyGenerator();
+        this.feedableBodyGenerator = new UnboundedFeedableBodyGenerator();
     }
 
     public Publisher<ByteBuffer> getPublisher() {
         return this.publisher;
     }
 
     @Override
-    public boolean feed(ByteBuffer buffer, boolean isLast) {
+    public boolean feed(ByteBuffer buffer, boolean isLast) throws Exception {
         return feedableBodyGenerator.feed(buffer, isLast);
     }
 

client/src/main/java/org/asynchttpclient/request/body/generator/UnboundedFeedableBodyGenerator.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright (c) 2014 AsyncHttpClient Project. 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 org.asynchttpclient.request.body.generator;
+
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+import org.asynchttpclient.request.body.generator.QueueBasedFeedableBodyGenerator.BodyChunk;
+
+public final class UnboundedFeedableBodyGenerator extends QueueBasedFeedableBodyGenerator<ConcurrentLinkedQueue<BodyChunk>> {
+    private final Queue<BodyChunk> queue = new ConcurrentLinkedQueue<>();
+
+    @Override
+    protected boolean offer(BodyChunk chunk) throws Exception {
+        return queue.offer(chunk);
+    }
+
+    @Override
+    protected Queue<org.asynchttpclient.request.body.generator.QueueBasedFeedableBodyGenerator.BodyChunk> queue() {
+        return queue;
+    }
+}

client/src/test/java/org/asynchttpclient/request/body/ChunkingTest.java

@@ -32,7 +32,7 @@
 import org.asynchttpclient.Response;
 import org.asynchttpclient.request.body.generator.FeedableBodyGenerator;
 import org.asynchttpclient.request.body.generator.InputStreamBodyGenerator;
-import org.asynchttpclient.request.body.generator.SimpleFeedableBodyGenerator;
+import org.asynchttpclient.request.body.generator.UnboundedFeedableBodyGenerator;
 import org.testng.annotations.Test;
 
 public class ChunkingTest extends AbstractBasicTest {
@@ -74,7 +74,7 @@ public void doTestWithInputStreamBodyGenerator(InputStream is) throws Throwable
     public void doTestWithFeedableBodyGenerator(InputStream is) throws Throwable {
         try (AsyncHttpClient c = asyncHttpClient(httpClientBuilder())) {
 
-            final FeedableBodyGenerator feedableBodyGenerator = new SimpleFeedableBodyGenerator();
+            final FeedableBodyGenerator feedableBodyGenerator = new UnboundedFeedableBodyGenerator();
             Request r = post(getTargetUrl()).setBody(feedableBodyGenerator).build();
 
             ListenableFuture<Response> responseFuture = c.executeRequest(r);
@@ -85,7 +85,7 @@ public void doTestWithFeedableBodyGenerator(InputStream is) throws Throwable {
         }
     }
 
-    private void feed(FeedableBodyGenerator feedableBodyGenerator, InputStream is) throws IOException {
+    private void feed(FeedableBodyGenerator feedableBodyGenerator, InputStream is) throws Exception {
         try (InputStream inputStream = is) {
             byte[] buffer = new byte[512];
             for (int i = 0; (i = inputStream.read(buffer)) > -1;) {

client/src/test/java/org/asynchttpclient/request/body/generator/FeedableBodyGeneratorTest.java

@@ -28,12 +28,12 @@
 
 public class FeedableBodyGeneratorTest {
 
-    private SimpleFeedableBodyGenerator feedableBodyGenerator;
+    private UnboundedFeedableBodyGenerator feedableBodyGenerator;
     private TestFeedListener listener;
 
     @BeforeMethod
     public void setUp() throws Exception {
-        feedableBodyGenerator = new SimpleFeedableBodyGenerator();
+        feedableBodyGenerator = new UnboundedFeedableBodyGenerator();
         listener = new TestFeedListener();
         feedableBodyGenerator.setListener(listener);
     }
@@ -73,7 +73,7 @@ private byte[] readFromBody(Body body) throws IOException {
         return readBytes;
     }
 
-    private static class TestFeedListener implements SimpleFeedableBodyGenerator.FeedListener {
+    private static class TestFeedListener implements UnboundedFeedableBodyGenerator.FeedListener {
 
         private int calls;