diff --git a/spring-web/src/main/java/org/springframework/http/client/reactive/JdkClientHttpConnector.java b/spring-web/src/main/java/org/springframework/http/client/reactive/JdkClientHttpConnector.java index 1cc8e13be9..3fcf281948 100644 --- a/spring-web/src/main/java/org/springframework/http/client/reactive/JdkClientHttpConnector.java +++ b/spring-web/src/main/java/org/springframework/http/client/reactive/JdkClientHttpConnector.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2021 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. @@ -27,37 +27,40 @@ import org.springframework.core.io.buffer.DefaultDataBufferFactory; import org.springframework.http.HttpMethod; /** - * {@link ClientHttpConnector} for the Java 11 HTTP client. + * {@link ClientHttpConnector} for Java's {@link HttpClient}. * * @author Julien Eyraud - * @since 5.2 - * @see Java HttpClient + * @since 6.0 + * @see HttpClient */ public class JdkClientHttpConnector implements ClientHttpConnector { private final HttpClient httpClient; - private final DataBufferFactory dataBufferFactory; + private final DataBufferFactory bufferFactory; /** - * Default constructor that creates a new instance of {@link HttpClient} and a {@link DataBufferFactory}. + * Default constructor that uses {@link HttpClient#newHttpClient()}. */ public JdkClientHttpConnector() { this(HttpClient.newHttpClient(), new DefaultDataBufferFactory()); } /** - * Constructor with an initialized {@link HttpClient} and a initialized {@link DataBufferFactory}. + * Constructor with an initialized {@link HttpClient} and a {@link DataBufferFactory}. */ - public JdkClientHttpConnector(final HttpClient httpClient, final DataBufferFactory dataBufferFactory) { + public JdkClientHttpConnector(HttpClient httpClient, DataBufferFactory bufferFactory) { this.httpClient = httpClient; - this.dataBufferFactory = dataBufferFactory; + this.bufferFactory = bufferFactory; } + @Override - public Mono connect(final HttpMethod method, final URI uri, final Function> requestCallback) { - final JdkClientHttpRequest request = new JdkClientHttpRequest(this.httpClient, method, uri, this.dataBufferFactory); + public Mono connect( + HttpMethod method, URI uri, Function> requestCallback) { + + JdkClientHttpRequest request = new JdkClientHttpRequest(this.httpClient, method, uri, this.bufferFactory); return requestCallback.apply(request).then(Mono.defer(request::getResponse)); } diff --git a/spring-web/src/main/java/org/springframework/http/client/reactive/JdkClientHttpRequest.java b/spring-web/src/main/java/org/springframework/http/client/reactive/JdkClientHttpRequest.java index 567e888277..f514915230 100644 --- a/spring-web/src/main/java/org/springframework/http/client/reactive/JdkClientHttpRequest.java +++ b/spring-web/src/main/java/org/springframework/http/client/reactive/JdkClientHttpRequest.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2021 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. @@ -37,18 +37,20 @@ 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.lang.Nullable; import org.springframework.util.Assert; /** - * {@link ClientHttpRequest} implementation for the Java 11 HTTP client. + * {@link ClientHttpRequest} implementation for Java's {@link HttpClient}. * * @author Julien Eyraud - * @since 5.2 - * @see Java HttpClient + * @since 6.0 */ class JdkClientHttpRequest extends AbstractClientHttpRequest { - private static final Set DISALLOWED_HEADERS = Set.of("connection", "content-length", "date", "expect", "from", "host", "upgrade", "via", "warning"); + private static final Set DISALLOWED_HEADERS = + Set.of("connection", "content-length", "date", "expect", "from", "host", "upgrade", "via", "warning"); + private final HttpClient httpClient; @@ -60,14 +62,18 @@ class JdkClientHttpRequest extends AbstractClientHttpRequest { private final DataBufferFactory bufferFactory; + @Nullable private Mono response; - public JdkClientHttpRequest(final HttpClient httpClient, final HttpMethod httpMethod, final URI uri, final DataBufferFactory bufferFactory) { + 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"); + this.httpClient = httpClient; this.method = httpMethod; this.uri = uri; @@ -75,26 +81,6 @@ class JdkClientHttpRequest extends AbstractClientHttpRequest { this.bufferFactory = bufferFactory; } - @Override - protected void applyHeaders() { - HttpHeaders headers = getHeaders(); - for (Map.Entry> 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() { @@ -117,16 +103,52 @@ class JdkClientHttpRequest extends AbstractClientHttpRequest { return (T) this.builder.build(); } + Mono getResponse() { + Assert.notNull(this.response, "Response is not set"); + return this.response; + } + + @Override - public Mono writeWith(final Publisher body) { + protected void applyHeaders() { + for (Map.Entry> header : getHeaders().entrySet()) { + if (DISALLOWED_HEADERS.contains(header.getKey().toLowerCase())) { + continue; + } + for (String value : header.getValue()) { + this.builder.header(header.getKey(), value); + } + } + if (!getHeaders().containsKey(HttpHeaders.ACCEPT)) { + this.builder.header(HttpHeaders.ACCEPT, "*/*"); + } + } + + @Override + protected void applyCookies() { + this.builder.header(HttpHeaders.COOKIE, + getCookies().values().stream() + .flatMap(List::stream) + .map(cookie -> cookie.getName() + "=" + cookie.getValue()) + .collect(Collectors.joining("; "))); + } + + @Override + public Mono writeWith(Publisher body) { return doCommit(() -> { - final Flow.Publisher 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)); + Flow.Publisher flow = + JdkFlowAdapter.publisherToFlowPublisher(Flux.from(body).map(DataBuffer::asByteBuffer)); + + HttpRequest.BodyPublisher bodyPublisher = (getHeaders().getContentLength() >= 0 ? + HttpRequest.BodyPublishers.fromPublisher(flow, getHeaders().getContentLength()) : + HttpRequest.BodyPublishers.fromPublisher(flow)); + + 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)); + return Mono.empty(); }); } @@ -141,17 +163,17 @@ class JdkClientHttpRequest extends AbstractClientHttpRequest { 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(); - }); - } + + 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)); + + return Mono.empty(); + }); } - public Mono getResponse() { - return this.response; - } } diff --git a/spring-web/src/main/java/org/springframework/http/client/reactive/JdkClientHttpResponse.java b/spring-web/src/main/java/org/springframework/http/client/reactive/JdkClientHttpResponse.java index a705be046b..c00308e4e4 100644 --- a/spring-web/src/main/java/org/springframework/http/client/reactive/JdkClientHttpResponse.java +++ b/spring-web/src/main/java/org/springframework/http/client/reactive/JdkClientHttpResponse.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2021 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. @@ -17,10 +17,12 @@ package org.springframework.http.client.reactive; import java.net.HttpCookie; +import java.net.http.HttpClient; import java.net.http.HttpResponse; import java.nio.ByteBuffer; import java.util.List; import java.util.concurrent.Flow; +import java.util.function.Function; import java.util.regex.Matcher; import java.util.regex.Pattern; @@ -38,34 +40,32 @@ import org.springframework.util.LinkedMultiValueMap; import org.springframework.util.MultiValueMap; /** - * {@link ClientHttpResponse} implementation for the Java 11 HTTP client. + * {@link ClientHttpResponse} implementation for Java's {@link HttpClient}. * * @author Julien Eyraud - * @since 5.2 - * @see Java HttpClient + * @since 6.0 */ class JdkClientHttpResponse implements ClientHttpResponse { - private static final Pattern SAMESITE_PATTERN = Pattern.compile("(?i).*SameSite=(Strict|Lax|None).*"); + + private static final Pattern SAME_SITE_PATTERN = Pattern.compile("(?i).*SameSite=(Strict|Lax|None).*"); + private final HttpResponse>> response; private final DataBufferFactory bufferFactory; - public JdkClientHttpResponse(final HttpResponse>> response, final DataBufferFactory bufferFactory) { + public JdkClientHttpResponse( + HttpResponse>> response, 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()); + return HttpStatus.valueOf(this.response.statusCode()); } @Override @@ -73,35 +73,44 @@ class JdkClientHttpResponse implements ClientHttpResponse { return this.response.statusCode(); } + @Override + public HttpHeaders getHeaders() { + return this.response.headers().map().entrySet().stream() + .collect(HttpHeaders::new, + (headers, entry) -> headers.addAll(entry.getKey(), entry.getValue()), + HttpHeaders::addAll); + } + @Override public MultiValueMap 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()) + 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() - ) - ).collect(LinkedMultiValueMap::new, (m, v) -> m.add(v.getName(), v), LinkedMultiValueMap::addAll); + .build())) + .collect(LinkedMultiValueMap::new, + (valueMap, cookie) -> valueMap.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); } @Override public Flux getBody() { - return JdkFlowAdapter - .flowPublisherToFlux(this.response.body()) - .flatMap(Flux::fromIterable) + return JdkFlowAdapter.flowPublisherToFlux(this.response.body()) + .flatMapIterable(Function.identity()) .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); - } } diff --git a/spring-webflux/src/test/java/org/springframework/web/reactive/function/client/WebClientIntegrationTests.java b/spring-webflux/src/test/java/org/springframework/web/reactive/function/client/WebClientIntegrationTests.java index 11cc814a2d..d08c7c56d9 100644 --- a/spring-webflux/src/test/java/org/springframework/web/reactive/function/client/WebClientIntegrationTests.java +++ b/spring-webflux/src/test/java/org/springframework/web/reactive/function/client/WebClientIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2020 the original author or authors. + * Copyright 2002-2021 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. diff --git a/spring-webflux/src/test/java/org/springframework/web/reactive/result/method/annotation/SseIntegrationTests.java b/spring-webflux/src/test/java/org/springframework/web/reactive/result/method/annotation/SseIntegrationTests.java index daf092d1e3..3d83767813 100644 --- a/spring-webflux/src/test/java/org/springframework/web/reactive/result/method/annotation/SseIntegrationTests.java +++ b/spring-webflux/src/test/java/org/springframework/web/reactive/result/method/annotation/SseIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2020 the original author or authors. + * Copyright 2002-2021 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. diff --git a/src/docs/asciidoc/web/webflux-webclient.adoc b/src/docs/asciidoc/web/webflux-webclient.adoc index c9c725a38e..b0b8e4a951 100644 --- a/src/docs/asciidoc/web/webflux-webclient.adoc +++ b/src/docs/asciidoc/web/webflux-webclient.adoc @@ -12,9 +12,9 @@ decode request and response content on the server side. support for the following: * https://github.com/reactor/reactor-netty[Reactor Netty] +* https://docs.oracle.com/en/java/javase/11/docs/api/java.net.http/java/net/http/HttpClient.html[JDK HttpClient] * https://github.com/jetty-project/jetty-reactive-httpclient[Jetty Reactive HttpClient] * https://hc.apache.org/index.html[Apache HttpComponents] -* https://docs.oracle.com/en/java/javase/11/docs/api/java.net.http/java/net/http/HttpClient.html[JDK HttpClient] * Others can be plugged via `ClientHttpConnector`. @@ -336,6 +336,40 @@ To configure a response timeout for a specific request: +[[webflux-client-builder-jdk-httpclient]] +=== JDK HttpClient + +The following example shows how to customize the JDK `HttpClient`: + +[source,java,indent=0,subs="verbatim,quotes",role="primary"] +.Java +---- + HttpClient httpClient = HttpClient.newBuilder() + .followRedirects(Redirect.NORMAL) + .connectTimeout(Duration.ofSeconds(20)) + .build(); + + ClientHttpConnector connector = + new JdkClientHttpConnector(httpClient, new DefaultDataBufferFactory()); + + WebClient webClient = WebClient.builder().clientConnector(connector).build(); +---- + +[source,kotlin,indent=0,subs="verbatim,quotes",role="secondary"] +.Kotlin +---- + val httpClient = HttpClient.newBuilder() + .followRedirects(Redirect.NORMAL) + .connectTimeout(Duration.ofSeconds(20)) + .build() + + val connector = JdkClientHttpConnector(httpClient, DefaultDataBufferFactory()) + + val webClient = WebClient.builder().clientConnector(connector).build() +---- + + + [[webflux-client-builder-jetty]] === Jetty @@ -426,6 +460,7 @@ The following example shows how to customize Apache HttpComponents `HttpClient` HttpAsyncClientBuilder clientBuilder = HttpAsyncClients.custom(); clientBuilder.setDefaultRequestConfig(...); CloseableHttpAsyncClient client = clientBuilder.build(); + ClientHttpConnector connector = new HttpComponentsClientHttpConnector(client); WebClient webClient = WebClient.builder().clientConnector(connector).build(); @@ -440,43 +475,6 @@ The following example shows how to customize Apache HttpComponents `HttpClient` val webClient = WebClient.builder().clientConnector(connector).build() ---- -[[webflux-client-builder-jdk-httpclient]] -=== JDK -The following example shows how to customize JDK `HttpClient` settings: - -[source,java,indent=0,subs="verbatim,quotes",role="primary"] -.Java ----- - @Bean - public WebClient webClient() { - HttpClient httpClient = HttpClient.newBuilder() - .executor(...) - .connectTimeout(...) - .build(); - ClientHttpConnector connector = - new JdkClientHttpConnector(httpClient, new DefaultDataBufferFactory()); // <1> - - return WebClient webClient = WebClient.builder().clientConnector(connector).build(); // <2> - } ----- -<1> Use the `JdkClientHttpConnector` constructor with customized `HttpClient` instance. -<2> Plug the connector into the `WebClient.Builder`. - -[source,kotlin,indent=0,subs="verbatim,quotes",role="secondary"] -.Kotlin ----- - @Bean - fun webClient() : WebClient { - val httpClient = HttpClient.newBuilder() - .executor(...) - .connectTimeout(...) - .build() - val connector = JdkClientHttpConnector(httpClient, DefaultDataBufferFactory()) // <1> - - return WebClient webClient = WebClient.builder() - .clientConnector(connector).build() // <2> - } ----- [[webflux-client-retrieve]] == `retrieve()`