More clean up

Author: slandelle

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

@@ -13,25 +13,18 @@
  */
 package org.asynchttpclient.request.body.generator;
 
-import java.util.Queue;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
 
-public final class BlockingFeedableBodyGenerator extends QueueBasedFeedableBodyGenerator<BlockingQueue<BodyChunk>> {
-    private final ArrayBlockingQueue<BodyChunk> queue;
 
-    public BlockingFeedableBodyGenerator(int capacity) {
-        queue = new ArrayBlockingQueue<>(capacity, true);
     }
 
     @Override
     protected boolean offer(BodyChunk chunk) throws InterruptedException {
         queue.put(chunk);
         return true;
     }
-
-    @Override
-    protected Queue<org.asynchttpclient.request.body.generator.BodyChunk> queue() {
-        return queue;
-    }
 }

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

@@ -16,11 +16,11 @@
 import java.nio.ByteBuffer;
 
 public final class BodyChunk {
-    final boolean isLast;
-    final ByteBuffer buffer;
 
-    public BodyChunk(final ByteBuffer buffer, final boolean isLast) {
         this.buffer = buffer;
-        this.isLast = isLast;
     }
 }

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

@@ -54,7 +54,7 @@ private BodyState readNextChunk(ByteBuf target) throws IOException {
             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();
             } else {
@@ -69,7 +69,7 @@ private void readChunk(ByteBuf target, BodyChunk part) {
         move(target, part.buffer);
 
         if (!part.buffer.hasRemaining()) {
-            if (part.isLast) {
                 state = BodyState.STOP;
             }
             queue.remove();

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

@@ -20,17 +20,20 @@
 
 public abstract class QueueBasedFeedableBodyGenerator<T extends Queue<BodyChunk>> implements FeedableBodyGenerator {
 
+    protected final T queue;
     private FeedListener listener;
 
+    public QueueBasedFeedableBodyGenerator(T queue) {
+        this.queue = queue;
+    }
+
     @Override
     public Body createBody() {
-        return new PushBody(queue());
     }
 
     protected abstract boolean offer(BodyChunk chunk) throws Exception;
 
-    protected abstract Queue<BodyChunk> queue();
-
     @Override
     public boolean feed(final ByteBuffer buffer, final boolean isLast) throws Exception {
         boolean offered = offer(new BodyChunk(buffer, isLast));

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

@@ -35,7 +35,7 @@ public class ReactiveStreamsBodyGenerator implements FeedableBodyGenerator {
 
     public ReactiveStreamsBodyGenerator(Publisher<ByteBuffer> publisher) {
         this.publisher = publisher;
-        this.feedableBodyGenerator = new UnboundedFeedableBodyGenerator();
     }
 
     public Publisher<ByteBuffer> getPublisher() {

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

@@ -13,19 +13,16 @@
  */
 package org.asynchttpclient.request.body.generator;
 
-import java.util.Queue;
 import java.util.concurrent.ConcurrentLinkedQueue;
 
-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.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.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 UnboundedFeedableBodyGenerator();
             Request r = post(getTargetUrl()).setBody(feedableBodyGenerator).build();
 
             ListenableFuture<Response> responseFuture = c.executeRequest(r);

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

@@ -28,12 +28,12 @@
 
 public class FeedableBodyGeneratorTest {
 
-    private UnboundedFeedableBodyGenerator feedableBodyGenerator;
     private TestFeedListener listener;
 
     @BeforeMethod
     public void setUp() throws Exception {
-        feedableBodyGenerator = new UnboundedFeedableBodyGenerator();
         listener = new TestFeedListener();
         feedableBodyGenerator.setListener(listener);
     }