diff --git a/spring-web/src/main/java/org/springframework/http/client/ReactorClientHttpRequest.java b/spring-web/src/main/java/org/springframework/http/client/ReactorClientHttpRequest.java index 3d38e18aa3..090b8b783f 100644 --- a/spring-web/src/main/java/org/springframework/http/client/ReactorClientHttpRequest.java +++ b/spring-web/src/main/java/org/springframework/http/client/ReactorClientHttpRequest.java @@ -58,8 +58,8 @@ final class ReactorClientHttpRequest extends AbstractStreamingClientHttpRequest private final Duration readTimeout; - public ReactorClientHttpRequest(HttpClient httpClient, URI uri, HttpMethod method, - Duration exchangeTimeout, Duration readTimeout) { + public ReactorClientHttpRequest( + HttpClient httpClient, URI uri, HttpMethod method, Duration exchangeTimeout, Duration readTimeout) { this.httpClient = httpClient; this.method = method; @@ -82,55 +82,51 @@ final class ReactorClientHttpRequest extends AbstractStreamingClientHttpRequest @Override protected ClientHttpResponse executeInternal(HttpHeaders headers, @Nullable Body body) throws IOException { - HttpClient.RequestSender requestSender = this.httpClient + + HttpClient.RequestSender sender = this.httpClient .request(io.netty.handler.codec.http.HttpMethod.valueOf(this.method.name())); - requestSender = (this.uri.isAbsolute() ? requestSender.uri(this.uri) : requestSender.uri(this.uri.toString())); + sender = (this.uri.isAbsolute() ? sender.uri(this.uri) : sender.uri(this.uri.toString())); try { - ReactorClientHttpResponse result = requestSender.send((reactorRequest, nettyOutbound) -> - send(headers, body, reactorRequest, nettyOutbound)) - .responseConnection((reactorResponse, connection) -> - Mono.just(new ReactorClientHttpResponse(reactorResponse, connection, this.readTimeout))) - .next() - .block(this.exchangeTimeout); + ReactorClientHttpResponse result = + sender.send((request, outbound) -> send(headers, body, request, outbound)) + .responseConnection((response, conn) -> + Mono.just(new ReactorClientHttpResponse(response, conn, this.readTimeout))) + .next() + .block(this.exchangeTimeout); if (result == null) { throw new IOException("HTTP exchange resulted in no result"); } - else { - return result; - } + + return result; } catch (RuntimeException ex) { throw convertException(ex); } } - private Publisher send(HttpHeaders headers, @Nullable Body body, - HttpClientRequest reactorRequest, NettyOutbound nettyOutbound) { + private Publisher send( + HttpHeaders headers, @Nullable Body body, HttpClientRequest request, NettyOutbound outbound) { - headers.forEach((key, value) -> reactorRequest.requestHeaders().set(key, value)); + headers.forEach((key, value) -> request.requestHeaders().set(key, value)); - if (body != null) { - ByteBufMapper byteMapper = new ByteBufMapper(nettyOutbound.alloc()); - AtomicReference executor = new AtomicReference<>(); - - return nettyOutbound - .withConnection(connection -> executor.set(connection.channel().eventLoop())) - .send(FlowAdapters.toPublisher(new OutputStreamPublisher<>( - os -> body.writeTo(StreamUtils.nonClosing(os)), byteMapper, - executor.getAndSet(null), null))); - } - else { - return nettyOutbound; + if (body == null) { + return outbound; } + + AtomicReference executorRef = new AtomicReference<>(); + + return outbound + .withConnection(connection -> executorRef.set(connection.channel().eventLoop())) + .send(FlowAdapters.toPublisher(new OutputStreamPublisher<>( + os -> body.writeTo(StreamUtils.nonClosing(os)), new ByteBufMapper(outbound), + executorRef.getAndSet(null), null))); } static IOException convertException(RuntimeException ex) { - // Exceptions.ReactiveException is package private - Throwable cause = ex.getCause(); - + Throwable cause = ex.getCause(); // Exceptions.ReactiveException is private if (cause instanceof IOException ioEx) { return ioEx; } @@ -148,22 +144,22 @@ final class ReactorClientHttpRequest extends AbstractStreamingClientHttpRequest private final ByteBufAllocator allocator; - public ByteBufMapper(ByteBufAllocator allocator) { - this.allocator = allocator; + public ByteBufMapper(NettyOutbound outbound) { + this.allocator = outbound.alloc(); } @Override public ByteBuf map(int b) { - ByteBuf byteBuf = this.allocator.buffer(1); - byteBuf.writeByte(b); - return byteBuf; + ByteBuf buf = this.allocator.buffer(1); + buf.writeByte(b); + return buf; } @Override public ByteBuf map(byte[] b, int off, int len) { - ByteBuf byteBuf = this.allocator.buffer(len); - byteBuf.writeBytes(b, off, len); - return byteBuf; + ByteBuf buf = this.allocator.buffer(len); + buf.writeBytes(b, off, len); + return buf; } } diff --git a/spring-web/src/main/java/org/springframework/http/client/ReactorClientHttpRequestFactory.java b/spring-web/src/main/java/org/springframework/http/client/ReactorClientHttpRequestFactory.java index 1b4ea28cc8..199027383e 100644 --- a/spring-web/src/main/java/org/springframework/http/client/ReactorClientHttpRequestFactory.java +++ b/spring-web/src/main/java/org/springframework/http/client/ReactorClientHttpRequestFactory.java @@ -71,45 +71,41 @@ public class ReactorClientHttpRequestFactory implements ClientHttpRequestFactory /** - * Create a new instance of the {@code ReactorClientHttpRequestFactory} - * with a default {@link HttpClient} that has compression enabled. + * Constructor with default client, created via {@link HttpClient#create()}, + * and with {@link HttpClient#compress compression} enabled. */ public ReactorClientHttpRequestFactory() { - this.httpClient = defaultInitializer.apply(HttpClient.create()); - this.resourceFactory = null; - this.mapper = null; + this(defaultInitializer.apply(HttpClient.create())); } /** - * Create a new instance of the {@code ReactorClientHttpRequestFactory} - * based on the given {@link HttpClient}. - * @param httpClient the client to base on + * Constructor with a given {@link HttpClient} instance. + * @param client the client to use */ - public ReactorClientHttpRequestFactory(HttpClient httpClient) { - Assert.notNull(httpClient, "HttpClient must not be null"); - this.httpClient = httpClient; + public ReactorClientHttpRequestFactory(HttpClient client) { + Assert.notNull(client, "HttpClient must not be null"); this.resourceFactory = null; this.mapper = null; + this.httpClient = client; } /** * Constructor with externally managed Reactor Netty resources, including * {@link LoopResources} for event loop threads, and {@link ConnectionProvider} - * for the connection pool. - *

This constructor should be used only when you don't want the client - * to participate in the Reactor Netty global resources. By default the - * client participates in the Reactor Netty global resources held in - * {@link reactor.netty.http.HttpResources}, which is recommended since - * fixed, shared resources are favored for event loop concurrency. However, - * consider declaring a {@link ReactorResourceFactory} bean with - * {@code globalResources=true} in order to ensure the Reactor Netty global - * resources are shut down when the Spring ApplicationContext is stopped or closed - * and restarted properly when the Spring ApplicationContext is - * (with JVM Checkpoint Restore for example). - * @param resourceFactory the resource factory to obtain the resources from - * @param mapper a mapper for further initialization of the created client + * for connection pooling. + *

Generally, it is recommended to share resources for event loop + * concurrency. This can be achieved either by participating in the JVM-wide, + * global resources held in {@link reactor.netty.http.HttpResources}, or by + * using a specific, shared set of resources through a + * {@link ReactorResourceFactory} bean. The latter can ensure that resources + * are shut down when the Spring ApplicationContext is stopped/closed and + * restarted again (e.g. JVM checkpoint restore). + * @param resourceFactory the resource factory to get resources from + * @param mapper for further initialization of the client */ - public ReactorClientHttpRequestFactory(ReactorResourceFactory resourceFactory, Function mapper) { + public ReactorClientHttpRequestFactory( + ReactorResourceFactory resourceFactory, Function mapper) { + this.resourceFactory = resourceFactory; this.mapper = mapper; if (resourceFactory.isRunning()) { @@ -117,13 +113,26 @@ public class ReactorClientHttpRequestFactory implements ClientHttpRequestFactory } } + private HttpClient createHttpClient(ReactorResourceFactory factory, Function mapper) { + HttpClient client = HttpClient.create(factory.getConnectionProvider()); + client = defaultInitializer.andThen(mapper).apply(client); + client = client.runOn(factory.getLoopResources()); + if (this.connectTimeout != null) { + client = client.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, this.connectTimeout); + } + return client; + } + /** - * Set the underlying connect timeout in milliseconds. - * A value of 0 specifies an infinite timeout. - *

Default is 30 seconds. + * Set the underlying connect timeout value on the underlying client. + * Effectively, a shortcut for + * {@code httpClient.option(CONNECT_TIMEOUT_MILLIS, timeout)}. + *

By default, set to 30 seconds. + * @param connectTimeout the timeout value in millis; use 0 to never time out. * @see HttpClient#option(ChannelOption, Object) * @see ChannelOption#CONNECT_TIMEOUT_MILLIS + * @see Connection Timeout */ public void setConnectTimeout(int connectTimeout) { Assert.isTrue(connectTimeout >= 0, "Timeout must be a non-negative value"); @@ -135,11 +144,7 @@ public class ReactorClientHttpRequestFactory implements ClientHttpRequestFactory } /** - * Set the underlying connect timeout in milliseconds. - * A value of 0 specifies an infinite timeout. - *

Default is 30 seconds. - * @see HttpClient#option(ChannelOption, Object) - * @see ChannelOption#CONNECT_TIMEOUT_MILLIS + * Variant of {@link #setConnectTimeout(int)} with a {@link Duration} value. */ public void setConnectTimeout(Duration connectTimeout) { Assert.notNull(connectTimeout, "ConnectTimeout must not be null"); @@ -156,13 +161,11 @@ public class ReactorClientHttpRequestFactory implements ClientHttpRequestFactory } /** - * Set the underlying read timeout as {@code Duration}. - *

Default is 10 seconds. + * Variant of {@link #setConnectTimeout(int)} with a {@link Duration} value. */ public void setReadTimeout(Duration readTimeout) { Assert.notNull(readTimeout, "ReadTimeout must not be null"); - Assert.isTrue(!readTimeout.isNegative(), "Timeout must be a non-negative value"); - this.readTimeout = readTimeout; + setReadTimeout((int) readTimeout.toMillis()); } /** @@ -180,29 +183,20 @@ public class ReactorClientHttpRequestFactory implements ClientHttpRequestFactory */ public void setExchangeTimeout(Duration exchangeTimeout) { Assert.notNull(exchangeTimeout, "ExchangeTimeout must not be null"); - Assert.isTrue(!exchangeTimeout.isNegative(), "Timeout must be a non-negative value"); - this.exchangeTimeout = exchangeTimeout; - } - - private HttpClient createHttpClient(ReactorResourceFactory factory, Function mapper) { - HttpClient httpClient = defaultInitializer.andThen(mapper) - .apply(HttpClient.create(factory.getConnectionProvider())); - httpClient = httpClient.runOn(factory.getLoopResources()); - if (this.connectTimeout != null) { - httpClient = httpClient.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, this.connectTimeout); - } - return httpClient; + setExchangeTimeout((int) exchangeTimeout.toMillis()); } @Override public ClientHttpRequest createRequest(URI uri, HttpMethod httpMethod) throws IOException { - HttpClient httpClient = this.httpClient; - if (httpClient == null) { - Assert.state(this.resourceFactory != null && this.mapper != null, "Illegal configuration"); - httpClient = createHttpClient(this.resourceFactory, this.mapper); + HttpClient client = this.httpClient; + if (client == null) { + Assert.state(this.resourceFactory != null && this.mapper != null, + "Expected HttpClient or ResourceFactory and mapper"); + client = createHttpClient(this.resourceFactory, this.mapper); } - return new ReactorClientHttpRequest(httpClient, uri, httpMethod, this.exchangeTimeout, this.readTimeout); + return new ReactorClientHttpRequest( + client, uri, httpMethod, this.exchangeTimeout, this.readTimeout); } @@ -237,8 +231,7 @@ public class ReactorClientHttpRequestFactory implements ClientHttpRequestFactory @Override public int getPhase() { - // Start after ReactorResourceFactory - return 1; + return 1; // start after ReactorResourceFactory (0) } } diff --git a/spring-web/src/main/java/org/springframework/http/client/ReactorClientHttpResponse.java b/spring-web/src/main/java/org/springframework/http/client/ReactorClientHttpResponse.java index 593ace979c..9d1a1cf486 100644 --- a/spring-web/src/main/java/org/springframework/http/client/ReactorClientHttpResponse.java +++ b/spring-web/src/main/java/org/springframework/http/client/ReactorClientHttpResponse.java @@ -52,11 +52,14 @@ final class ReactorClientHttpResponse implements ClientHttpResponse { private volatile InputStream body; - public ReactorClientHttpResponse(HttpClientResponse response, Connection connection, Duration readTimeout) { + public ReactorClientHttpResponse( + HttpClientResponse response, Connection connection, Duration readTimeout) { + this.response = response; this.connection = connection; this.readTimeout = readTimeout; - this.headers = HttpHeaders.readOnlyHttpHeaders(new Netty4HeadersAdapter(response.responseHeaders())); + this.headers = HttpHeaders.readOnlyHttpHeaders( + new Netty4HeadersAdapter(response.responseHeaders())); } @@ -106,7 +109,8 @@ final class ReactorClientHttpResponse implements ClientHttpResponse { StreamUtils.drain(body); body.close(); } - catch (IOException ignored) { + catch (IOException ex) { + // ignore } }