Netty support for (Async)RestTemplate

This commit introduces an AsyncClientHttpRequestFactory based on Netty
4, for use with the (Async)RestTemplate.

build.gradle

@@ -672,6 +672,7 @@ project("spring-web") {
 		optional("commons-fileupload:commons-fileupload:1.3.1")
 		optional("org.apache.httpcomponents:httpclient:4.3.5")
 		optional("org.apache.httpcomponents:httpasyncclient:4.0.2")
+		optional("io.netty:netty-all:4.0.23.Final")
 		optional("com.fasterxml.jackson.core:jackson-databind:${jackson2Version}")
 		optional("com.fasterxml.jackson.dataformat:jackson-dataformat-xml:${jackson2Version}")
 		optional("com.google.code.gson:gson:${gsonVersion}")

spring-web/src/main/java/org/springframework/http/client/Netty4ClientHttpRequest.java

@@ -0,0 +1,186 @@
+/*
+ * Copyright 2002-2014 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.http.client;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.URI;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.ByteBufOutputStream;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.handler.codec.http.DefaultFullHttpRequest;
+import io.netty.handler.codec.http.FullHttpRequest;
+import io.netty.handler.codec.http.FullHttpResponse;
+import io.netty.handler.codec.http.HttpVersion;
+
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpMethod;
+import org.springframework.util.concurrent.ListenableFuture;
+import org.springframework.util.concurrent.SettableListenableFuture;
+
+/**
+ * {@link org.springframework.http.client.ClientHttpRequest} implementation that uses
+ * Netty 4 to execute requests.
+ *
+ * <p>Created via the {@link Netty4ClientHttpRequestFactory}.
+ *
+ * @author Arjen Poutsma
+ * @since 4.2
+ */
+class Netty4ClientHttpRequest extends AbstractAsyncClientHttpRequest implements ClientHttpRequest {
+
+	private final Bootstrap bootstrap;
+
+	private final URI uri;
+
+	private final HttpMethod method;
+
+	private final ByteBufOutputStream body;
+
+	Netty4ClientHttpRequest(Bootstrap bootstrap, URI uri, HttpMethod method, int maxRequestSize) {
+		this.bootstrap = bootstrap;
+		this.uri = uri;
+		this.method = method;
+		this.body = new ByteBufOutputStream(Unpooled.buffer(maxRequestSize));
+	}
+
+	@Override
+	public HttpMethod getMethod() {
+		return this.method;
+	}
+
+	@Override
+	public URI getURI() {
+		return this.uri;
+	}
+
+	@Override
+	protected OutputStream getBodyInternal(HttpHeaders headers) throws IOException {
+		return body;
+	}
+
+	@Override
+	protected ListenableFuture<ClientHttpResponse> executeInternal(final HttpHeaders headers)
+			throws IOException {
+		final SettableListenableFuture<ClientHttpResponse> responseFuture =
+				new SettableListenableFuture<ClientHttpResponse>();
+
+		ChannelFutureListener connectionListener = new ChannelFutureListener() {
+			@Override
+			public void operationComplete(ChannelFuture future) throws Exception {
+				if (future.isSuccess()) {
+					Channel channel = future.channel();
+					channel.pipeline()
+							.addLast(new SimpleChannelInboundHandler<FullHttpResponse>() {
+
+										@Override
+										protected void channelRead0(
+												ChannelHandlerContext ctx,
+												FullHttpResponse msg) throws Exception {
+											responseFuture
+													.set(new Netty4ClientHttpResponse(ctx,
+															msg));
+										}
+
+										@Override
+										public void exceptionCaught(
+												ChannelHandlerContext ctx,
+												Throwable cause) throws Exception {
+											responseFuture.setException(cause);
+										}
+									});
+
+					FullHttpRequest nettyRequest =
+							createFullHttpRequest(headers);
+
+					channel.writeAndFlush(nettyRequest);
+				}
+				else {
+					responseFuture.setException(future.cause());
+				}
+
+			}
+		};
+
+		bootstrap.connect(uri.getHost(), getPort(uri)).addListener(connectionListener);
+
+		return responseFuture;
+
+	}
+
+	@Override
+	public ClientHttpResponse execute() throws IOException {
+		try {
+			return executeAsync().get();
+		}
+		catch (InterruptedException ex) {
+			throw new IOException(ex.getMessage(), ex);
+		}
+		catch (ExecutionException ex) {
+			if (ex.getCause() instanceof IOException) {
+				throw (IOException) ex.getCause();
+			} else {
+				throw new IOException(ex.getMessage(), ex);
+			}
+		}
+	}
+
+	private static int getPort(URI uri) {
+		int port = uri.getPort();
+		if (port == -1) {
+			if ("http".equalsIgnoreCase(uri.getScheme())) {
+				port = 80;
+			}
+			else if ("https".equalsIgnoreCase(uri.getScheme())) {
+				port = 443;
+			}
+		}
+		return port;
+	}
+
+	private FullHttpRequest createFullHttpRequest(HttpHeaders headers) {
+		io.netty.handler.codec.http.HttpMethod nettyMethod =
+				io.netty.handler.codec.http.HttpMethod.valueOf(method.name());
+
+		FullHttpRequest nettyRequest = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1,
+				nettyMethod, this.uri.getRawPath(),
+				this.body.buffer());
+
+		nettyRequest.headers()
+				.set(io.netty.handler.codec.http.HttpHeaders.Names.HOST, uri.getHost());
+		nettyRequest.headers()
+				.set(io.netty.handler.codec.http.HttpHeaders.Names.CONNECTION,
+						io.netty.handler.codec.http.HttpHeaders.Values.CLOSE);
+
+		for (Map.Entry<String, List<String>> entry : headers.entrySet()) {
+			nettyRequest.headers().add(entry.getKey(), entry.getValue());
+		}
+
+		return nettyRequest;
+	}
+
+
+}

spring-web/src/main/java/org/springframework/http/client/Netty4ClientHttpRequestFactory.java

@@ -0,0 +1,159 @@
+/*
+ * Copyright 2002-2014 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.http.client;
+
+import java.io.IOException;
+import java.net.URI;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.codec.http.HttpClientCodec;
+import io.netty.handler.codec.http.HttpObjectAggregator;
+import io.netty.handler.ssl.SslContext;
+
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.http.HttpMethod;
+import org.springframework.util.Assert;
+
+/**
+ * {@link org.springframework.http.client.ClientHttpRequestFactory} implementation that
+ * uses <a href="http://netty.io/">Netty 4</a> to create requests.
+ *
+ * <p>Allows to use a pre-configured {@link EventLoopGroup} instance - useful for sharing
+ * across multiple clients.
+ *
+ * @author Arjen Poutsma
+ * @since 4.2
+ */
+public class Netty4ClientHttpRequestFactory
+		implements ClientHttpRequestFactory, AsyncClientHttpRequestFactory,
+		InitializingBean, DisposableBean {
+
+	/**
+	 * The default maximum request size.
+	 * @see #setMaxRequestSize(int)
+	 */
+	public static final int DEFAULT_MAX_REQUEST_SIZE = 1024 * 1024 * 10;
+
+	private final EventLoopGroup eventLoopGroup;
+
+	private final boolean defaultEventLoopGroup;
+
+	private SslContext sslContext;
+
+	private int maxRequestSize = DEFAULT_MAX_REQUEST_SIZE;
+
+	private Bootstrap bootstrap;
+
+	/**
+	 * Creates a new {@code Netty4ClientHttpRequestFactory} with a default
+	 * {@link NioEventLoopGroup}.
+	 */
+	public Netty4ClientHttpRequestFactory() {
+		int ioWorkerCount = Runtime.getRuntime().availableProcessors() * 2;
+		eventLoopGroup = new NioEventLoopGroup(ioWorkerCount);
+		defaultEventLoopGroup = true;
+	}
+
+	/**
+	 * Creates a new {@code Netty4ClientHttpRequestFactory} with the given
+	 * {@link EventLoopGroup}.
+	 *
+	 * <p><b>NOTE:</b> the given group will <strong>not</strong> be
+	 * {@linkplain EventLoopGroup#shutdownGracefully() shutdown} by this factory; doing
+	 * so becomes the responsibility of the caller.
+	 */
+	public Netty4ClientHttpRequestFactory(EventLoopGroup eventLoopGroup) {
+		Assert.notNull(eventLoopGroup, "'eventLoopGroup' must not be null");
+		this.eventLoopGroup = eventLoopGroup;
+		this.defaultEventLoopGroup = false;
+	}
+
+	/**
+	 * Sets the default maximum request size. The default is
+	 * {@link #DEFAULT_MAX_REQUEST_SIZE}.
+	 * @see HttpObjectAggregator#HttpObjectAggregator(int)
+	 */
+	public void setMaxRequestSize(int maxRequestSize) {
+		this.maxRequestSize = maxRequestSize;
+	}
+
+	/**
+	 * Sets the SSL context.
+	 */
+	public void setSslContext(SslContext sslContext) {
+		this.sslContext = sslContext;
+	}
+
+	private Bootstrap getBootstrap() {
+		if (this.bootstrap == null) {
+			Bootstrap bootstrap = new Bootstrap();
+			bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class)
+					.handler(new ChannelInitializer<SocketChannel>() {
+						@Override
+						protected void initChannel(SocketChannel ch) throws Exception {
+							ChannelPipeline pipeline = ch.pipeline();
+
+							if (sslContext != null) {
+								pipeline.addLast(sslContext.newHandler(ch.alloc()));
+							}
+							pipeline.addLast(new HttpClientCodec());
+							pipeline.addLast(new HttpObjectAggregator(maxRequestSize));
+						}
+					});
+			this.bootstrap = bootstrap;
+		}
+		return this.bootstrap;
+	}
+
+	@Override
+	public void afterPropertiesSet() throws Exception {
+		getBootstrap();
+	}
+
+	private Netty4ClientHttpRequest createRequestInternal(URI uri, HttpMethod httpMethod) {
+		return new Netty4ClientHttpRequest(getBootstrap(), uri, httpMethod, maxRequestSize);
+	}
+
+	@Override
+	public ClientHttpRequest createRequest(URI uri, HttpMethod httpMethod)
+			throws IOException {
+		return createRequestInternal(uri, httpMethod);
+	}
+
+	@Override
+	public AsyncClientHttpRequest createAsyncRequest(URI uri, HttpMethod httpMethod)
+			throws IOException {
+		return createRequestInternal(uri, httpMethod);
+	}
+
+	@Override
+	public void destroy() throws InterruptedException {
+		if (defaultEventLoopGroup) {
+			// clean up the EventLoopGroup if we created it in the constructor
+			eventLoopGroup.shutdownGracefully().sync();
+		}
+	}
+
+
+}