Stage Netty 3 provider so that we may use it to work on the Netty 4 implementation.

Author: md-5

providers/netty-4/pom.xml

@@ -0,0 +1,24 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <groupId>com.ning</groupId>
+        <artifactId>async-http-client-providers-parent</artifactId>
+        <version>1.8.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+    <artifactId>async-http-client-netty-provider</artifactId>
+    <name>Asynchronous Http Client Netty Provider</name>
+    <description>
+        The Async Http Client Netty Provider.
+    </description>
+
+    <dependencies>
+        <dependency>
+            <groupId>io.netty</groupId>
+            <artifactId>netty</artifactId>
+            <version>3.6.3.Final</version>
+        </dependency>
+    </dependencies>
+
+</project>

providers/netty-4/src/main/java/com/ning/http/client/providers/netty_4/BodyChunkedInput.java

@@ -0,0 +1,93 @@
+/*
+ * Copyright (c) 2010-2012 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.netty_4;
+
+import com.ning.http.client.Body;
+import org.jboss.netty.buffer.ChannelBuffers;
+import org.jboss.netty.handler.stream.ChunkedInput;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+/**
+ * Adapts a {@link Body} to Netty's {@link ChunkedInput}.
+ */
+class BodyChunkedInput
+        implements ChunkedInput {
+
+    private final Body body;
+
+    private final int chunkSize = 1024 * 8;
+
+    private ByteBuffer nextChunk;
+
+    private static final ByteBuffer EOF = ByteBuffer.allocate(0);
+
+    private boolean endOfInput = false;
+
+    public BodyChunkedInput(Body body) {
+        if (body == null) {
+            throw new IllegalArgumentException("no body specified");
+        }
+        this.body = body;
+    }
+
+    private ByteBuffer peekNextChunk()
+            throws IOException {
+
+        if (nextChunk == null) {
+            ByteBuffer buffer = ByteBuffer.allocate(chunkSize);
+            long length = body.read(buffer);
+            if (length < 0) {
+                // Negative means this is finished
+                buffer.flip();
+                nextChunk = buffer;
+                endOfInput = true;
+            } else if (length == 0) {
+                // Zero means we didn't get anything this time, but may get next time
+                buffer.flip();
+                nextChunk = null;
+            } else {
+                buffer.flip();
+                nextChunk = buffer;
+            }
+        }
+        return nextChunk;
+    }
+
+    /**
+     * Having no next chunk does not necessarily means end of input, other chunks may arrive later
+     */
+    public boolean hasNextChunk() throws Exception {
+        return peekNextChunk() != null;
+    }
+
+    public Object nextChunk() throws Exception {
+        ByteBuffer buffer = peekNextChunk();
+        if (buffer == null || buffer == EOF) {
+            return null;
+        }
+        nextChunk = null;
+
+        return ChannelBuffers.wrappedBuffer(buffer);
+    }
+
+    public boolean isEndOfInput() throws Exception {
+        return endOfInput;
+    }
+
+    public void close() throws Exception {
+        body.close();
+    }
+
+}

providers/netty-4/src/main/java/com/ning/http/client/providers/netty_4/BodyFileRegion.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright (c) 2010-2012 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.netty_4;
+
+import com.ning.http.client.RandomAccessBody;
+import org.jboss.netty.channel.FileRegion;
+
+import java.io.IOException;
+import java.nio.channels.WritableByteChannel;
+
+/**
+ * Adapts a {@link RandomAccessBody} to Netty's {@link FileRegion}.
+ */
+class BodyFileRegion
+        implements FileRegion {
+
+    private final RandomAccessBody body;
+
+    public BodyFileRegion(RandomAccessBody body) {
+        if (body == null) {
+            throw new IllegalArgumentException("no body specified");
+        }
+        this.body = body;
+    }
+
+    public long getPosition() {
+        return 0;
+    }
+
+    public long getCount() {
+        return body.getContentLength();
+    }
+
+    public long transferTo(WritableByteChannel target, long position)
+            throws IOException {
+        return body.transferTo(position, Long.MAX_VALUE, target);
+    }
+
+    public void releaseExternalResources() {
+        try {
+            body.close();
+        } catch (IOException e) {
+            // we tried
+        }
+    }
+
+}

providers/netty-4/src/main/java/com/ning/http/client/providers/netty_4/FeedableBodyGenerator.java

@@ -0,0 +1,119 @@
+/*
+ * Copyright (c) 2012 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.netty_4;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import com.ning.http.client.Body;
+import com.ning.http.client.BodyGenerator;
+
+/**
+ * {@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.
+ */
+public class FeedableBodyGenerator implements BodyGenerator {
+    private final static byte[] END_PADDING = "\r\n".getBytes();
+    private final static byte[] ZERO = "0".getBytes();
+    private final Queue<BodyPart> queue = new ConcurrentLinkedQueue<BodyPart>();
+    private final AtomicInteger queueSize = new AtomicInteger();
+    private FeedListener listener;
+
+    @Override
+    public Body createBody() throws IOException {
+        return new PushBody();
+    }
+
+    public void feed(final ByteBuffer buffer, final boolean isLast) throws IOException {
+        queue.offer(new BodyPart(buffer, isLast));
+        queueSize.incrementAndGet();
+        if (listener != null) {
+            listener.onContentAdded();
+        }
+    }
+
+    public static interface FeedListener {
+        public void onContentAdded();
+    }
+
+    public void setListener(FeedListener listener) {
+        this.listener = listener;
+    }
+
+    private final class PushBody implements Body {
+        private final int ONGOING = 0;
+        private final int CLOSING = 1;
+        private final int FINISHED = 2;
+
+        private int finishState = 0;
+
+        @Override
+        public long getContentLength() {
+            return -1;
+        }
+
+        @Override
+        public long read(final ByteBuffer buffer) throws IOException {
+            BodyPart nextPart = queue.peek();
+            if (nextPart == null) {
+                // Nothing in the queue
+                switch (finishState) {
+                case ONGOING:
+                    return 0;
+                case CLOSING:
+                    buffer.put(ZERO);
+                    buffer.put(END_PADDING);
+                    finishState = FINISHED;
+                    return buffer.position();
+                case FINISHED:
+                    buffer.put(END_PADDING);
+                    return -1;
+                }
+            }
+            int capacity = buffer.remaining() - 10; // be safe (we'll have to add size, ending, etc.)
+            int size = Math.min(nextPart.buffer.remaining(), capacity);
+            buffer.put(Integer.toHexString(size).getBytes());
+            buffer.put(END_PADDING);
+            for (int i=0; i < size; i++) {
+              buffer.put(nextPart.buffer.get());
+            }
+            buffer.put(END_PADDING);
+            if (!nextPart.buffer.hasRemaining()) {
+                if (nextPart.isLast) {
+                    finishState = CLOSING;
+                }
+                queue.remove();
+            }
+            return size;
+        }
+
+        @Override
+        public void close() throws IOException {
+        }
+
+    }
+
+    private final static class BodyPart {
+        private final boolean isLast;
+        private final ByteBuffer buffer;
+
+        public BodyPart(final ByteBuffer buffer, final boolean isLast) {
+            this.buffer = buffer;
+            this.isLast = isLast;
+        }
+    }
+}