Merge pull request AsyncHttpClient#262 from md-5/master

[AsyncHttpClient#261] Update to use Netty 4 as a transport provider

Author: jfarcand

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-4-provider</artifactId>
+ <name>Asynchronous Http Client Netty 4 Provider</name>
+ <description>
+ The Async Http Client Netty 4 Provider.
+ </description>
+
+ <dependencies>
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-all</artifactId>
+ <version>4.0.0.Beta3</version>
+ </dependency>
+ </dependencies>
+
+</project>

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

@@ -0,0 +1,96 @@
+/*
+ * 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 io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.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<ByteBuf> {
+
+ 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;
+ }
+
+ @Override
+ public boolean readChunk(ByteBuf b) throws Exception {
+ ByteBuffer buffer = peekNextChunk();
+ if (buffer == null || buffer == EOF) {
+ return false;
+ }
+ nextChunk = null;
+
+ b.writeBytes(buffer);
+ return true;
+ }
+
+ 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,59 @@
+/*
+ * 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 io.netty.buffer.AbstractReferenceCounted;
+import io.netty.buffer.ReferenceCounted;
+import io.netty.channel.FileRegion;
+
+import java.io.IOException;
+import java.nio.channels.WritableByteChannel;
+
+/**
+ * Adapts a {@link RandomAccessBody} to Netty's {@link FileRegion}.
+ */
+class BodyFileRegion
+ extends AbstractReferenceCounted
+ implements FileRegion {
+
+ private final RandomAccessBody body;
+
+ public BodyFileRegion(RandomAccessBody body) {
+ if (body == null) {
+ throw new IllegalArgumentException("no body specified");
+ }
+ this.body = body;
+ }
+
+ public long position() {
+ return 0;
+ }
+
+ public long count() {
+ return body.getContentLength();
+ }
+
+ public long transferTo(WritableByteChannel target, long position)
+ throws IOException {
+ return body.transferTo(position, Long.MAX_VALUE, target);
+ }
+
+ public void deallocate() {
+ 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;
+ }
+ }
+}