Integrate Apache Http client with WebClient

build.gradle

@@ -153,6 +153,8 @@ configure(allprojects) { project ->
 				exclude group: "commons-logging", name: "commons-logging"
 			}
 			dependency "org.eclipse.jetty:jetty-reactive-httpclient:1.1.2"
+			dependency 'org.apache.httpcomponents.client5:httpclient5:5.0'
+			dependency 'org.apache.httpcomponents.core5:httpcore5-reactive:5.0'
 
 			dependency "org.jruby:jruby:9.2.11.0"
 			dependency "org.python:jython-standalone:2.7.1"

spring-web/spring-web.gradle

@@ -35,6 +35,8 @@ dependencies {
 		exclude group: "javax.servlet", module: "javax.servlet-api"
 	}
 	optional("org.eclipse.jetty:jetty-reactive-httpclient")
+	optional('org.apache.httpcomponents.client5:httpclient5:5.0')
+	optional('org.apache.httpcomponents.core5:httpcore5-reactive:5.0')
 	optional("com.squareup.okhttp3:okhttp")
 	optional("org.apache.httpcomponents:httpclient")
 	optional("org.apache.httpcomponents:httpasyncclient")

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

@@ -48,7 +48,8 @@
  * @author Arjen Poutsma
  * @since 4.0
  * @see HttpComponentsClientHttpRequestFactory#createRequest
- * @deprecated as of Spring 5.0, with no direct replacement
+ * @deprecated as of Spring 5.0, in favor of
+ * {@link org.springframework.http.client.reactive.HttpComponentsClientHttpConnector}
  */
 @Deprecated
 final class HttpComponentsAsyncClientHttpRequest extends AbstractBufferingAsyncClientHttpRequest {

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

@@ -44,7 +44,8 @@
  * @author Stephane Nicoll
  * @since 4.0
  * @see HttpAsyncClient
- * @deprecated as of Spring 5.0, with no direct replacement
+ * @deprecated as of Spring 5.0, in favor of
+ * {@link org.springframework.http.client.reactive.HttpComponentsClientHttpConnector}
  */
 @Deprecated
 public class HttpComponentsAsyncClientHttpRequestFactory extends HttpComponentsClientHttpRequestFactory

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

@@ -37,7 +37,8 @@
  * @author Arjen Poutsma
  * @since 4.0
  * @see HttpComponentsAsyncClientHttpRequest#executeAsync()
- * @deprecated as of Spring 5.0, with no direct replacement
+ * @deprecated as of Spring 5.0, in favor of
+ * {@link org.springframework.http.client.reactive.HttpComponentsClientHttpConnector}
  */
 @Deprecated
 final class HttpComponentsAsyncClientHttpResponse extends AbstractClientHttpResponse {

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

@@ -0,0 +1,169 @@
+/*
+ * Copyright 2002-2020 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
+ *
+ *      https://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.reactive;
+
+import java.net.URI;
+import java.nio.ByteBuffer;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+import org.apache.hc.client5.http.cookie.BasicCookieStore;
+import org.apache.hc.client5.http.impl.async.CloseableHttpAsyncClient;
+import org.apache.hc.client5.http.impl.async.HttpAsyncClients;
+import org.apache.hc.client5.http.protocol.HttpClientContext;
+import org.apache.hc.core5.concurrent.FutureCallback;
+import org.apache.hc.core5.http.ContentType;
+import org.apache.hc.core5.http.HttpResponse;
+import org.apache.hc.core5.http.Message;
+import org.apache.hc.core5.http.nio.support.BasicRequestProducer;
+import org.apache.hc.core5.reactive.ReactiveEntityProducer;
+import org.apache.hc.core5.reactive.ReactiveResponseConsumer;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.MonoSink;
+
+import org.springframework.core.io.buffer.DataBufferFactory;
+import org.springframework.core.io.buffer.DefaultDataBufferFactory;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpMethod;
+import org.springframework.lang.Nullable;
+
+/**
+ * {@link ClientHttpConnector} implementation for the Apache HttpComponents HttpClient 5.x.
+ *
+ * @author Martin Tarjányi
+ * @since 5.3
+ * @see <a href="https://hc.apache.org/index.html">Apache HttpComponents</a>
+ */
+public class HttpComponentsClientHttpConnector implements ClientHttpConnector {
+
+	private final CloseableHttpAsyncClient client;
+
+	private final DataBufferFactory dataBufferFactory;
+
+	private final Supplier<? extends HttpClientContext> contextSupplier;
+
+
+	/**
+	 * Default constructor that creates and starts a new instance of {@link CloseableHttpAsyncClient}.
+	 */
+	public HttpComponentsClientHttpConnector() {
+		this(HttpAsyncClients.createDefault());
+	}
+
+	/**
+	 * Constructor with a pre-configured {@link CloseableHttpAsyncClient} instance.
+	 * @param client the client to use
+	 */
+	public HttpComponentsClientHttpConnector(CloseableHttpAsyncClient client) {
+		this(client, HttpClientContext::create);
+	}
+
+	/**
+	 * Constructor with a pre-configured {@link CloseableHttpAsyncClient} instance
+	 * and a {@link HttpClientContext} supplier lambda which is called before each request
+	 * and passed to the client.
+	 * @param client the client to use
+	 * @param contextSupplier a {@link HttpClientContext} supplier
+	 */
+	public HttpComponentsClientHttpConnector(CloseableHttpAsyncClient client,
+			Supplier<? extends HttpClientContext> contextSupplier) {
+
+		this.dataBufferFactory = new DefaultDataBufferFactory();
+		this.contextSupplier = contextSupplier;
+		this.client = client;
+		this.client.start();
+	}
+
+
+	@Override
+	public Mono<ClientHttpResponse> connect(HttpMethod method, URI uri,
+			Function<? super ClientHttpRequest, Mono<Void>> requestCallback) {
+
+		HttpClientContext context = this.contextSupplier.get();
+
+		if (context.getCookieStore() == null) {
+			context.setCookieStore(new BasicCookieStore());
+		}
+
+		HttpComponentsClientHttpRequest request = new HttpComponentsClientHttpRequest(method, uri,
+				context, this.dataBufferFactory);
+
+		return requestCallback.apply(request).then(Mono.defer(() -> execute(request, context)));
+	}
+
+	private Mono<ClientHttpResponse> execute(HttpComponentsClientHttpRequest request, HttpClientContext context) {
+		Flux<ByteBuffer> byteBufferFlux = request.getByteBufferFlux();
+
+		ReactiveEntityProducer reactiveEntityProducer = createReactiveEntityProducer(request, byteBufferFlux);
+
+		BasicRequestProducer basicRequestProducer = new BasicRequestProducer(request.getHttpRequest(),
+				reactiveEntityProducer);
+
+		return Mono.<Message<HttpResponse, Publisher<ByteBuffer>>>create(sink -> {
+			ReactiveResponseConsumer reactiveResponseConsumer =
+					new ReactiveResponseConsumer(new MonoFutureCallbackAdapter<>(sink));
+
+			this.client.execute(basicRequestProducer, reactiveResponseConsumer, context, null);
+		}).map(message -> new HttpComponentsClientHttpResponse(this.dataBufferFactory, message, context));
+	}
+
+	@Nullable
+	private ReactiveEntityProducer createReactiveEntityProducer(HttpComponentsClientHttpRequest request,
+			@Nullable Flux<ByteBuffer> byteBufferFlux) {
+
+		if (byteBufferFlux == null) {
+			return null;
+		}
+
+		String contentEncoding = request.getHeaders().getFirst(HttpHeaders.CONTENT_ENCODING);
+
+		ContentType contentType = null;
+
+		if (request.getHeaders().getContentType() != null) {
+			contentType = ContentType.parse(request.getHeaders().getContentType().toString());
+		}
+
+		return new ReactiveEntityProducer(byteBufferFlux, request.getContentLength(), contentType, contentEncoding);
+	}
+
+
+	private static class MonoFutureCallbackAdapter<T> implements FutureCallback<T> {
+
+		private final MonoSink<T> sink;
+
+		public MonoFutureCallbackAdapter(MonoSink<T> sink) {
+			this.sink = sink;
+		}
+
+		@Override
+		public void completed(T result) {
+			this.sink.success(result);
+		}
+
+		@Override
+		public void failed(Exception ex) {
+			this.sink.error(ex);
+		}
+
+		@Override
+		public void cancelled() {
+		}
+	}
+
+}

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

@@ -0,0 +1,152 @@
+/*
+ * Copyright 2002-2020 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
+ *
+ *      https://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.reactive;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.nio.ByteBuffer;
+import java.util.Collection;
+
+import org.apache.hc.client5.http.cookie.CookieStore;
+import org.apache.hc.client5.http.impl.cookie.BasicClientCookie;
+import org.apache.hc.client5.http.protocol.HttpClientContext;
+import org.apache.hc.core5.http.HttpRequest;
+import org.apache.hc.core5.http.message.BasicHttpRequest;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferFactory;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpMethod;
+
+import static org.springframework.http.MediaType.ALL_VALUE;
+
+/**
+ * {@link ClientHttpRequest} implementation for the Apache HttpComponents HttpClient 5.x.
+ *
+ * @author Martin Tarjányi
+ * @since 5.3
+ * @see <a href="https://hc.apache.org/index.html">Apache HttpComponents</a>
+ */
+class HttpComponentsClientHttpRequest extends AbstractClientHttpRequest {
+
+	private final HttpRequest httpRequest;
+
+	private final DataBufferFactory dataBufferFactory;
+
+	private final HttpClientContext context;
+
+	private Flux<ByteBuffer> byteBufferFlux;
+
+	private long contentLength = -1;
+
+
+	public HttpComponentsClientHttpRequest(HttpMethod method, URI uri, HttpClientContext context,
+			DataBufferFactory dataBufferFactory) {
+
+		this.context = context;
+		this.httpRequest = new BasicHttpRequest(method.name(), uri);
+		this.dataBufferFactory = dataBufferFactory;
+	}
+
+
+	@Override
+	public HttpMethod getMethod() {
+		return HttpMethod.resolve(this.httpRequest.getMethod());
+	}
+
+	@Override
+	public URI getURI() {
+		try {
+			return this.httpRequest.getUri();
+		}
+		catch (URISyntaxException ex) {
+			throw new IllegalArgumentException("Invalid URI syntax.", ex);
+		}
+	}
+
+	@Override
+	public DataBufferFactory bufferFactory() {
+		return this.dataBufferFactory;
+	}
+
+	@Override
+	public Mono<Void> writeWith(Publisher<? extends DataBuffer> body) {
+		return doCommit(() -> {
+			this.byteBufferFlux = Flux.from(body).map(DataBuffer::asByteBuffer);
+			return Mono.empty();
+		});
+	}
+
+	@Override
+	public Mono<Void> writeAndFlushWith(Publisher<? extends Publisher<? extends DataBuffer>> body) {
+		return writeWith(Flux.from(body).flatMap(p -> p));
+	}
+
+	@Override
+	public Mono<Void> setComplete() {
+		return doCommit();
+	}
+
+	@Override
+	protected void applyHeaders() {
+		HttpHeaders headers = getHeaders();
+		this.contentLength = headers.getContentLength();
+
+		headers.entrySet()
+				.stream()
+				.filter(entry -> !HttpHeaders.CONTENT_LENGTH.equals(entry.getKey()))
+				.forEach(entry -> entry.getValue().forEach(v -> this.httpRequest.addHeader(entry.getKey(), v)));
+
+		if (!this.httpRequest.containsHeader(HttpHeaders.ACCEPT)) {
+			this.httpRequest.addHeader(HttpHeaders.ACCEPT, ALL_VALUE);
+		}
+	}
+
+	@Override
+	protected void applyCookies() {
+		if (getCookies().isEmpty()) {
+			return;
+		}
+
+		CookieStore cookieStore = this.context.getCookieStore();
+
+		getCookies().values()
+				.stream()
+				.flatMap(Collection::stream)
+				.forEach(cookie -> {
+					BasicClientCookie clientCookie = new BasicClientCookie(cookie.getName(), cookie.getValue());
+					clientCookie.setDomain(getURI().getHost());
+					clientCookie.setPath(getURI().getPath());
+					cookieStore.addCookie(clientCookie);
+				});
+	}
+
+	public HttpRequest getHttpRequest() {
+		return this.httpRequest;
+	}
+
+	public Flux<ByteBuffer> getByteBufferFlux() {
+		return this.byteBufferFlux;
+	}
+
+	public long getContentLength() {
+		return this.contentLength;
+	}
+}