Refactoring in the JDK HttpClient support

See gh-23432

Author: rstoyanchev

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

@@ -18,6 +18,12 @@
 
 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.concurrent.CompletableFuture;
+import java.util.concurrent.Flow;
 import java.util.function.Function;
 
 import reactor.core.publisher.Mono;
@@ -27,9 +33,10 @@
 import org.springframework.http.HttpMethod;
 
 /**
- * {@link ClientHttpConnector} for Java's {@link HttpClient}.
+ * {@link ClientHttpConnector} for the Java {@link HttpClient}.
  *
  * @author Julien Eyraud
+ * @author Rossen Stoyanchev
  * @since 6.0
  * @see <a href="https://docs.oracle.com/en/java/javase/11/docs/api/java.net.http/java/net/http/HttpClient.html">HttpClient</a>
  */
@@ -60,8 +67,17 @@ public JdkClientHttpConnector(HttpClient httpClient, DataBufferFactory bufferFac
 	public Mono<ClientHttpResponse> connect(
 			HttpMethod method, URI uri, Function<? super ClientHttpRequest, Mono<Void>> requestCallback) {
 
-		JdkClientHttpRequest request = new JdkClientHttpRequest(this.httpClient, method, uri, this.bufferFactory);
-		return requestCallback.apply(request).then(Mono.defer(request::getResponse));
+		JdkClientHttpRequest jdkClientHttpRequest = new JdkClientHttpRequest(method, uri, this.bufferFactory);
+
+		return requestCallback.apply(jdkClientHttpRequest).then(Mono.defer(() -> {
+			HttpRequest httpRequest = jdkClientHttpRequest.getNativeRequest();
+
+			CompletableFuture<HttpResponse<Flow.Publisher<List<ByteBuffer>>>> future =
+					this.httpClient.sendAsync(httpRequest, HttpResponse.BodyHandlers.ofPublisher());
+
+			return Mono.fromCompletionStage(future)
+					.map(response -> new JdkClientHttpResponse(response, this.bufferFactory));
+		}));
 	}
 
 }

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

@@ -19,11 +19,9 @@
 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;
@@ -35,50 +33,38 @@
 
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferFactory;
+import org.springframework.http.HttpCookie;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpMethod;
-import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
 /**
- * {@link ClientHttpRequest} implementation for Java's {@link HttpClient}.
+ * {@link ClientHttpRequest} for the Java {@link HttpClient}.
  *
  * @author Julien Eyraud
+ * @author Rossen Stoyanchev
  * @since 6.0
  */
 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;
 
-	@Nullable
-	private Mono<ClientHttpResponse> response;
-
+	private final HttpRequest.Builder builder;
 
-	public JdkClientHttpRequest(
-			HttpClient httpClient, HttpMethod httpMethod, URI uri, 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");
+	public JdkClientHttpRequest(HttpMethod httpMethod, URI uri, DataBufferFactory bufferFactory) {
+		Assert.notNull(httpMethod, "HttpMethod is required");
+		Assert.notNull(uri, "URI is required");
+		Assert.notNull(bufferFactory, "DataBufferFactory is required");
 
-		this.httpClient = httpClient;
 		this.method = httpMethod;
 		this.uri = uri;
-		this.builder = HttpRequest.newBuilder(uri);
 		this.bufferFactory = bufferFactory;
+		this.builder = HttpRequest.newBuilder(uri);
 	}
 
 
@@ -103,20 +89,16 @@ public <T> T getNativeRequest() {
 		return (T) this.builder.build();
 	}
 
-	Mono<ClientHttpResponse> getResponse() {
-		Assert.notNull(this.response, "Response is not set");
-		return this.response;
-	}
-
 
 	@Override
 	protected void applyHeaders() {
-		for (Map.Entry<String, List<String>> header : getHeaders().entrySet()) {
-			if (DISALLOWED_HEADERS.contains(header.getKey().toLowerCase())) {
+		for (Map.Entry<String, List<String>> entry : getHeaders().entrySet()) {
+			if (entry.getKey().equalsIgnoreCase(HttpHeaders.CONTENT_LENGTH)) {
+				// content-length is specified when writing
 				continue;
 			}
-			for (String value : header.getValue()) {
-				this.builder.header(header.getKey(), value);
+			for (String value : entry.getValue()) {
+				this.builder.header(entry.getKey(), value);
 			}
 		}
 		if (!getHeaders().containsKey(HttpHeaders.ACCEPT)) {
@@ -126,31 +108,28 @@ protected void applyHeaders() {
 
 	@Override
 	protected void applyCookies() {
-		this.builder.header(HttpHeaders.COOKIE,
-				getCookies().values().stream()
-						.flatMap(List::stream)
-						.map(cookie -> cookie.getName() + "=" + cookie.getValue())
-						.collect(Collectors.joining("; ")));
+		this.builder.header(HttpHeaders.COOKIE, getCookies().values().stream()
+				.flatMap(List::stream).map(HttpCookie::toString).collect(Collectors.joining(";")));
 	}
 
 	@Override
 	public Mono<Void> writeWith(Publisher<? extends DataBuffer> body) {
 		return doCommit(() -> {
-			Flow.Publisher<ByteBuffer> flow =
-					JdkFlowAdapter.publisherToFlowPublisher(Flux.from(body).map(DataBuffer::asByteBuffer));
+			this.builder.method(this.method.name(), toBodyPublisher(body));
+			return Mono.empty();
+		});
+	}
 
-			HttpRequest.BodyPublisher bodyPublisher = (getHeaders().getContentLength() >= 0 ?
-					HttpRequest.BodyPublishers.fromPublisher(flow, getHeaders().getContentLength()) :
-					HttpRequest.BodyPublishers.fromPublisher(flow));
+	private HttpRequest.BodyPublisher toBodyPublisher(Publisher<? extends DataBuffer> body) {
+		Publisher<ByteBuffer> byteBufferBody = (body instanceof Mono ?
+				Mono.from(body).map(DataBuffer::asByteBuffer) :
+				Flux.from(body).map(DataBuffer::asByteBuffer));
 
-			this.response = Mono.fromCompletionStage(() -> {
-						HttpRequest request = this.builder.method(this.method.name(), bodyPublisher).build();
-						return this.httpClient.sendAsync(request, HttpResponse.BodyHandlers.ofPublisher());
-					})
-					.map(response -> new JdkClientHttpResponse(response, this.bufferFactory));
+		Flow.Publisher<ByteBuffer> bodyFlow = JdkFlowAdapter.publisherToFlowPublisher(byteBufferBody);
 
-			return Mono.empty();
-		});
+		return (getHeaders().getContentLength() > 0 ?
+				HttpRequest.BodyPublishers.fromPublisher(bodyFlow, getHeaders().getContentLength()) :
+				HttpRequest.BodyPublishers.fromPublisher(bodyFlow));
 	}
 
 	@Override
@@ -160,18 +139,8 @@ public Mono<Void> writeAndFlushWith(final Publisher<? extends Publisher<? extend
 
 	@Override
 	public Mono<Void> setComplete() {
-		if (isCommitted()) {
-			return Mono.empty();
-		}
-
 		return doCommit(() -> {
-			this.response = Mono.fromCompletionStage(() -> {
-						HttpRequest.BodyPublisher bodyPublisher = HttpRequest.BodyPublishers.noBody();
-						HttpRequest request = this.builder.method(this.method.name(), bodyPublisher).build();
-						return this.httpClient.sendAsync(request, HttpResponse.BodyHandlers.ofPublisher());
-					})
-					.map(response -> new JdkClientHttpResponse(response, this.bufferFactory));
-
+			this.builder.method(this.method.name(), HttpRequest.BodyPublishers.noBody());
 			return Mono.empty();
 		});
 	}

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

@@ -21,6 +21,8 @@
 import java.net.http.HttpResponse;
 import java.nio.ByteBuffer;
 import java.util.List;
+import java.util.Locale;
+import java.util.Map;
 import java.util.concurrent.Flow;
 import java.util.function.Function;
 import java.util.regex.Matcher;
@@ -36,13 +38,16 @@
 import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseCookie;
 import org.springframework.lang.Nullable;
+import org.springframework.util.CollectionUtils;
+import org.springframework.util.LinkedCaseInsensitiveMap;
 import org.springframework.util.LinkedMultiValueMap;
 import org.springframework.util.MultiValueMap;
 
 /**
- * {@link ClientHttpResponse} implementation for Java's {@link HttpClient}.
+ * {@link ClientHttpResponse} for the Java {@link HttpClient}.
  *
  * @author Julien Eyraud
+ * @author Rossen Stoyanchev
  * @since 6.0
  */
 class JdkClientHttpResponse implements ClientHttpResponse {
@@ -54,12 +59,23 @@ class JdkClientHttpResponse implements ClientHttpResponse {
 
 	private final DataBufferFactory bufferFactory;
 
+	private final HttpHeaders headers;
+
 
 	public JdkClientHttpResponse(
 			HttpResponse<Flow.Publisher<List<ByteBuffer>>> response, DataBufferFactory bufferFactory) {
 
 		this.response = response;
 		this.bufferFactory = bufferFactory;
+		this.headers = adaptHeaders(response);
+	}
+
+	private static HttpHeaders adaptHeaders(HttpResponse<Flow.Publisher<List<ByteBuffer>>> response) {
+		Map<String, List<String>> rawHeaders = response.headers().map();
+		Map<String, List<String>> map = new LinkedCaseInsensitiveMap<>(rawHeaders.size(), Locale.ENGLISH);
+		MultiValueMap<String, String> multiValueMap = CollectionUtils.toMultiValueMap(map);
+		multiValueMap.putAll(rawHeaders);
+		return HttpHeaders.readOnlyHttpHeaders(multiValueMap);
 	}
 
 
@@ -75,34 +91,31 @@ public int getRawStatusCode() {
 
 	@Override
 	public HttpHeaders getHeaders() {
-		return this.response.headers().map().entrySet().stream()
-				.collect(HttpHeaders::new,
-						(headers, entry) -> headers.addAll(entry.getKey(), entry.getValue()),
-						HttpHeaders::addAll);
+		return this.headers;
 	}
 
 	@Override
 	public MultiValueMap<String, ResponseCookie> getCookies() {
 		return this.response.headers().allValues(HttpHeaders.SET_COOKIE).stream()
-				.flatMap(header ->
-						HttpCookie.parse(header).stream().map(cookie ->
-								ResponseCookie.from(cookie.getName(), cookie.getValue())
-										.domain(cookie.getDomain())
-										.httpOnly(cookie.isHttpOnly())
-										.maxAge(cookie.getMaxAge())
-										.path(cookie.getPath())
-										.secure(cookie.getSecure())
-										.sameSite(parseSameSite(header))
-										.build()))
+				.flatMap(header -> {
+					Matcher matcher = SAME_SITE_PATTERN.matcher(header);
+					String sameSite = (matcher.matches() ? matcher.group(1) : null);
+					return HttpCookie.parse(header).stream().map(cookie -> toResponseCookie(cookie, sameSite));
+				})
 				.collect(LinkedMultiValueMap::new,
-						(valueMap, cookie) -> valueMap.add(cookie.getName(), cookie),
+						(cookies, cookie) -> cookies.add(cookie.getName(), cookie),
 						LinkedMultiValueMap::addAll);
 	}
 
-	@Nullable
-	private static String parseSameSite(String headerValue) {
-		Matcher matcher = SAME_SITE_PATTERN.matcher(headerValue);
-		return (matcher.matches() ? matcher.group(1) : null);
+	private ResponseCookie toResponseCookie(HttpCookie cookie, @Nullable String sameSite) {
+		return ResponseCookie.from(cookie.getName(), cookie.getValue())
+				.domain(cookie.getDomain())
+				.httpOnly(cookie.isHttpOnly())
+				.maxAge(cookie.getMaxAge())
+				.path(cookie.getPath())
+				.secure(cookie.getSecure())
+				.sameSite(sameSite)
+				.build();
 	}
 
 	@Override