JDK 11 HttpClient integration with WebClient

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

@@ -0,0 +1,64 @@
+/*
+ * Copyright 2002-2019 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.http.HttpClient;
+import java.util.function.Function;
+
+import reactor.core.publisher.Mono;
+
+import org.springframework.core.io.buffer.DataBufferFactory;
+import org.springframework.core.io.buffer.DefaultDataBufferFactory;
+import org.springframework.http.HttpMethod;
+
+/**
+ * {@link ClientHttpConnector} for the Java 11 HTTP client.
+ *
+ * @author Julien Eyraud
+ * @since 5.2
+ * @see <a href="https://docs.oracle.com/en/java/javase/11/docs/api/java.net.http/java/net/http/HttpClient.html">Java HttpClient</a>
+ */
+public class JdkClientHttpConnector implements ClientHttpConnector {
+
+	private final HttpClient httpClient;
+
+	private final DataBufferFactory dataBufferFactory;
+
+
+	/**
+	 * Default constructor that creates a new instance of {@link HttpClient} and a {@link DataBufferFactory}.
+	 */
+	public JdkClientHttpConnector() {
+		this(HttpClient.newHttpClient(), new DefaultDataBufferFactory());
+	}
+
+	/**
+	 * Constructor with an initialized {@link HttpClient} and a initialized {@link DataBufferFactory}.
+	 */
+	public JdkClientHttpConnector(final HttpClient httpClient, final DataBufferFactory dataBufferFactory) {
+		this.httpClient = httpClient;
+		this.dataBufferFactory = dataBufferFactory;
+	}
+
+	@Override
+	public Mono<ClientHttpResponse> connect(final HttpMethod method, final URI uri, final Function<? super ClientHttpRequest, Mono<Void>> requestCallback) {
+		final JdkClientHttpRequest request = new JdkClientHttpRequest(this.httpClient, method, uri, this.dataBufferFactory);
+		return requestCallback.apply(request).then(Mono.defer(request::getResponse));
+	}
+
+}

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

@@ -0,0 +1,157 @@
+/*
+ * Copyright 2002-2019 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.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Flow;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import org.reactivestreams.Publisher;
+import reactor.adapter.JdkFlowAdapter;
+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 org.springframework.util.Assert;
+
+/**
+ * {@link ClientHttpRequest} implementation for the Java 11 HTTP client.
+ *
+ * @author Julien Eyraud
+ * @since 5.2
+ * @see <a href="https://docs.oracle.com/en/java/javase/11/docs/api/java.net.http/java/net/http/HttpClient.html">Java HttpClient</a>
+ */
+class JdkClientHttpRequest extends AbstractClientHttpRequest {
+
+	private static final Set<String> DISALLOWED_HEADERS = Set.of("connection", "content-length", "date", "expect", "from", "host", "upgrade", "via", "warning");
+
+	private final HttpClient httpClient;
+
+	private final HttpMethod method;
+
+	private final URI uri;
+
+	private final HttpRequest.Builder builder;
+
+	private final DataBufferFactory bufferFactory;
+
+	private Mono<ClientHttpResponse> response;
+
+
+	public JdkClientHttpRequest(final HttpClient httpClient, final HttpMethod httpMethod, final URI uri, final DataBufferFactory bufferFactory) {
+		Assert.notNull(httpClient, "HttpClient should not be null");
+		Assert.notNull(httpMethod, "HttpMethod should not be null");
+		Assert.notNull(uri, "URI should not be null");
+		Assert.notNull(bufferFactory, "DataBufferFactory should not be null");
+		this.httpClient = httpClient;
+		this.method = httpMethod;
+		this.uri = uri;
+		this.builder = HttpRequest.newBuilder(uri);
+		this.bufferFactory = bufferFactory;
+	}
+
+	@Override
+	protected void applyHeaders() {
+		HttpHeaders headers = getHeaders();
+		for (Map.Entry<String, List<String>> header : getHeaders().entrySet()) {
+			if (!DISALLOWED_HEADERS.contains(header.getKey().toLowerCase())) {
+				for (String value : header.getValue()) {
+					this.builder.header(header.getKey(), value);
+				}
+			}
+		}
+		if (!headers.containsKey(HttpHeaders.ACCEPT)) {
+			this.builder.header(HttpHeaders.ACCEPT, "*/*");
+		}
+	}
+
+	@Override
+	protected void applyCookies() {
+		final String cookies = getCookies().values().stream().flatMap(List::stream).map(c -> c.getName() + "=" + c.getValue()).collect(Collectors.joining("; "));
+		this.builder.header(HttpHeaders.COOKIE, cookies);
+	}
+
+	@Override
+	public HttpMethod getMethod() {
+		return this.method;
+	}
+
+	@Override
+	public URI getURI() {
+		return this.uri;
+	}
+
+	@Override
+	public DataBufferFactory bufferFactory() {
+		return this.bufferFactory;
+	}
+
+	@Override
+	@SuppressWarnings("unchecked")
+	public <T> T getNativeRequest() {
+		return (T) this.builder.build();
+	}
+
+	@Override
+	public Mono<Void> writeWith(final Publisher<? extends DataBuffer> body) {
+		return doCommit(() -> {
+			final Flow.Publisher<ByteBuffer> flowAdapter = JdkFlowAdapter.publisherToFlowPublisher(Flux.from(body).map(DataBuffer::asByteBuffer));
+			final long contentLength = getHeaders().getContentLength();
+			final HttpRequest.BodyPublisher bodyPublisher = contentLength >= 0 ? HttpRequest.BodyPublishers.fromPublisher(flowAdapter, contentLength)
+					: HttpRequest.BodyPublishers.fromPublisher(flowAdapter);
+			this.response = Mono
+					.fromCompletionStage(() -> this.httpClient.sendAsync(this.builder.method(this.method.name(), bodyPublisher).build(), HttpResponse.BodyHandlers.ofPublisher()))
+					.map(r -> new JdkClientHttpResponse(r, this.bufferFactory));
+			return Mono.empty();
+		});
+	}
+
+	@Override
+	public Mono<Void> writeAndFlushWith(final Publisher<? extends Publisher<? extends DataBuffer>> body) {
+		return writeWith(Flux.from(body).flatMap(Function.identity()));
+	}
+
+	@Override
+	public Mono<Void> setComplete() {
+		if (isCommitted()) {
+			return Mono.empty();
+		}
+		else {
+			return doCommit(() -> {
+				this.response = Mono
+						.fromCompletionStage(() -> this.httpClient.sendAsync(this.builder.method(this.method.name(), HttpRequest.BodyPublishers.noBody()).build(), HttpResponse.BodyHandlers.ofPublisher()))
+						.map(r -> new JdkClientHttpResponse(r, this.bufferFactory));
+				return Mono.empty();
+			});
+		}
+	}
+
+	public Mono<ClientHttpResponse> getResponse() {
+		return this.response;
+	}
+}

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

@@ -0,0 +1,107 @@
+/*
+ * Copyright 2002-2019 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.HttpCookie;
+import java.net.http.HttpResponse;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.concurrent.Flow;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import reactor.adapter.JdkFlowAdapter;
+import reactor.core.publisher.Flux;
+
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferFactory;
+import org.springframework.core.io.buffer.DataBufferUtils;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseCookie;
+import org.springframework.lang.Nullable;
+import org.springframework.util.LinkedMultiValueMap;
+import org.springframework.util.MultiValueMap;
+
+/**
+ * {@link ClientHttpResponse} implementation for the Java 11 HTTP client.
+ *
+ * @author Julien Eyraud
+ * @since 5.2
+ * @see <a href="https://docs.oracle.com/en/java/javase/11/docs/api/java.net.http/java/net/http/HttpClient.html">Java HttpClient</a>
+ */
+class JdkClientHttpResponse implements ClientHttpResponse {
+	private static final Pattern SAMESITE_PATTERN = Pattern.compile("(?i).*SameSite=(Strict|Lax|None).*");
+
+	private final HttpResponse<Flow.Publisher<List<ByteBuffer>>> response;
+
+	private final DataBufferFactory bufferFactory;
+
+
+	public JdkClientHttpResponse(final HttpResponse<Flow.Publisher<List<ByteBuffer>>> response, final DataBufferFactory bufferFactory) {
+		this.response = response;
+		this.bufferFactory = bufferFactory;
+	}
+
+	@Nullable
+	private static String parseSameSite(String headerValue) {
+		Matcher matcher = SAMESITE_PATTERN.matcher(headerValue);
+		return (matcher.matches() ? matcher.group(1) : null);
+	}
+
+	@Override
+	public HttpStatus getStatusCode() {
+		return HttpStatus.resolve(this.response.statusCode());
+	}
+
+	@Override
+	public int getRawStatusCode() {
+		return this.response.statusCode();
+	}
+
+	@Override
+	public MultiValueMap<String, ResponseCookie> getCookies() {
+		return this.response.headers().allValues(HttpHeaders.SET_COOKIE).stream()
+				.flatMap(header ->
+						HttpCookie.parse(header).stream().map(httpCookie ->
+								ResponseCookie
+										.from(httpCookie.getName(), httpCookie.getValue())
+										.domain(httpCookie.getDomain())
+										.httpOnly(httpCookie.isHttpOnly())
+										.maxAge(httpCookie.getMaxAge())
+										.path(httpCookie.getPath())
+										.secure(httpCookie.getSecure())
+										.sameSite(parseSameSite(header))
+										.build()
+						)
+				).collect(LinkedMultiValueMap::new, (m, v) -> m.add(v.getName(), v), LinkedMultiValueMap::addAll);
+	}
+
+	@Override
+	public Flux<DataBuffer> getBody() {
+		return JdkFlowAdapter
+				.flowPublisherToFlux(this.response.body())
+				.flatMap(Flux::fromIterable)
+				.map(this.bufferFactory::wrap)
+				.doOnDiscard(DataBuffer.class, DataBufferUtils::release);
+	}
+
+	@Override
+	public HttpHeaders getHeaders() {
+		return this.response.headers().map().entrySet().stream().collect(HttpHeaders::new, (headers, entry) -> headers.addAll(entry.getKey(), entry.getValue()), HttpHeaders::addAll);
+	}
+}

spring-webflux/src/test/java/org/springframework/web/reactive/function/client/WebClientIntegrationTests.java

@@ -62,6 +62,7 @@
 import org.springframework.http.ResponseEntity;
 import org.springframework.http.client.reactive.ClientHttpConnector;
 import org.springframework.http.client.reactive.HttpComponentsClientHttpConnector;
+import org.springframework.http.client.reactive.JdkClientHttpConnector;
 import org.springframework.http.client.reactive.JettyClientHttpConnector;
 import org.springframework.http.client.reactive.ReactorClientHttpConnector;
 import org.springframework.web.reactive.function.BodyExtractors;
@@ -92,7 +93,8 @@ static Stream<ClientHttpConnector> arguments() {
 		return Stream.of(
 				new ReactorClientHttpConnector(),
 				new JettyClientHttpConnector(),
-				new HttpComponentsClientHttpConnector()
+				new HttpComponentsClientHttpConnector(),
+				new JdkClientHttpConnector()
 		);
 	}
 

spring-webflux/src/test/java/org/springframework/web/reactive/result/method/annotation/SseIntegrationTests.java

@@ -36,6 +36,7 @@
 import org.springframework.core.ParameterizedTypeReference;
 import org.springframework.http.client.reactive.ClientHttpConnector;
 import org.springframework.http.client.reactive.HttpComponentsClientHttpConnector;
+import org.springframework.http.client.reactive.JdkClientHttpConnector;
 import org.springframework.http.client.reactive.JettyClientHttpConnector;
 import org.springframework.http.client.reactive.ReactorClientHttpConnector;
 import org.springframework.http.codec.ServerSentEvent;
@@ -76,15 +77,19 @@ static Object[][] arguments() {
 			{new JettyHttpServer(), new ReactorClientHttpConnector()},
 			{new JettyHttpServer(), new JettyClientHttpConnector()},
 			{new JettyHttpServer(), new HttpComponentsClientHttpConnector()},
+			{new JettyHttpServer(), new JdkClientHttpConnector()},
 			{new ReactorHttpServer(), new ReactorClientHttpConnector()},
 			{new ReactorHttpServer(), new JettyClientHttpConnector()},
 			{new ReactorHttpServer(), new HttpComponentsClientHttpConnector()},
+			{new ReactorHttpServer(), new JdkClientHttpConnector()},
 			{new TomcatHttpServer(), new ReactorClientHttpConnector()},
 			{new TomcatHttpServer(), new JettyClientHttpConnector()},
 			{new TomcatHttpServer(), new HttpComponentsClientHttpConnector()},
+			{new TomcatHttpServer(), new JdkClientHttpConnector()},
 			{new UndertowHttpServer(), new ReactorClientHttpConnector()},
 			{new UndertowHttpServer(), new JettyClientHttpConnector()},
-			{new UndertowHttpServer(), new HttpComponentsClientHttpConnector()}
+			{new UndertowHttpServer(), new HttpComponentsClientHttpConnector()},
+			{new UndertowHttpServer(), new JdkClientHttpConnector()},
 		};
 	}