diff --git a/spring-web/src/main/java/org/springframework/http/client/reactive/ReactorClientHttpConnector.java b/spring-web/src/main/java/org/springframework/http/client/reactive/ReactorClientHttpConnector.java index 6c523e9fb7..e3a5fa0ee0 100644 --- a/spring-web/src/main/java/org/springframework/http/client/reactive/ReactorClientHttpConnector.java +++ b/spring-web/src/main/java/org/springframework/http/client/reactive/ReactorClientHttpConnector.java @@ -20,6 +20,8 @@ import java.net.URI; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Function; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import reactor.core.publisher.Mono; import reactor.netty.NettyOutbound; import reactor.netty.http.client.HttpClient; @@ -27,23 +29,42 @@ import reactor.netty.http.client.HttpClientRequest; import reactor.netty.resources.ConnectionProvider; import reactor.netty.resources.LoopResources; +import org.springframework.context.Lifecycle; +import org.springframework.context.SmartLifecycle; import org.springframework.http.HttpMethod; +import org.springframework.lang.Nullable; import org.springframework.util.Assert; /** * Reactor-Netty implementation of {@link ClientHttpConnector}. * + *

This class implements {@link Lifecycle} and can be optionally declared + * as a Spring-managed bean. + * * @author Brian Clozel * @author Rossen Stoyanchev + * @author Sebastien Deleuze * @since 5.0 * @see reactor.netty.http.client.HttpClient */ -public class ReactorClientHttpConnector implements ClientHttpConnector { +public class ReactorClientHttpConnector implements ClientHttpConnector, SmartLifecycle { + + private static final Log logger = LogFactory.getLog(ReactorClientHttpConnector.class); private final static Function defaultInitializer = client -> client.compress(true); - private final HttpClient httpClient; + private HttpClient httpClient; + + @Nullable + private final ReactorResourceFactory resourceFactory; + + @Nullable + private final Function mapper; + + private volatile boolean running = true; + + private final Object lifecycleMonitor = new Object(); /** @@ -54,6 +75,8 @@ public class ReactorClientHttpConnector implements ClientHttpConnector { */ public ReactorClientHttpConnector() { this.httpClient = defaultInitializer.apply(HttpClient.create()); + this.resourceFactory = null; + this.mapper = null; } /** @@ -68,14 +91,20 @@ public class ReactorClientHttpConnector implements ClientHttpConnector { * 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 closed. - * @param factory the resource factory to obtain the resources from + * @param resourceFactory the resource factory to obtain the resources from * @param mapper a mapper for further initialization of the created client * @since 5.1 */ - public ReactorClientHttpConnector(ReactorResourceFactory factory, Function mapper) { - ConnectionProvider provider = factory.getConnectionProvider(); + public ReactorClientHttpConnector(ReactorResourceFactory resourceFactory, Function mapper) { + this.httpClient = createHttpClient(resourceFactory, mapper); + this.resourceFactory = resourceFactory; + this.mapper = mapper; + } + + private static HttpClient createHttpClient(ReactorResourceFactory resourceFactory, Function mapper) { + ConnectionProvider provider = resourceFactory.getConnectionProvider(); Assert.notNull(provider, "No ConnectionProvider: is ReactorResourceFactory not initialized yet?"); - this.httpClient = defaultInitializer.andThen(mapper).andThen(applyLoopResources(factory)) + return defaultInitializer.andThen(mapper).andThen(applyLoopResources(resourceFactory)) .apply(HttpClient.create(provider)); } @@ -96,6 +125,8 @@ public class ReactorClientHttpConnector implements ClientHttpConnector { public ReactorClientHttpConnector(HttpClient httpClient) { Assert.notNull(httpClient, "HttpClient is required"); this.httpClient = httpClient; + this.resourceFactory = null; + this.mapper = null; } @@ -131,4 +162,52 @@ public class ReactorClientHttpConnector implements ClientHttpConnector { return new ReactorClientHttpRequest(method, uri, request, nettyOutbound); } + @Override + public void start() { + synchronized (this.lifecycleMonitor) { + if (!isRunning()) { + if (this.resourceFactory != null && this.mapper != null) { + this.httpClient = createHttpClient(this.resourceFactory, this.mapper); + } + else { + logger.warn("Restarting a ReactorClientHttpConnector bean is only supported with externally managed Reactor Netty resources"); + } + this.running = true; + } + } + } + + @Override + public void stop() { + synchronized (this.lifecycleMonitor) { + if (isRunning()) { + this.running = false; + } + } + } + + @Override + public final void stop(Runnable callback) { + synchronized (this.lifecycleMonitor) { + stop(); + callback.run(); + } + } + + @Override + public boolean isRunning() { + return this.running; + } + + @Override + public boolean isAutoStartup() { + return false; + } + + @Override + public int getPhase() { + // Start after ReactorResourceFactory + return 1; + } + } diff --git a/spring-web/src/test/java/org/springframework/http/client/reactive/ReactorClientHttpConnectorTests.java b/spring-web/src/test/java/org/springframework/http/client/reactive/ReactorClientHttpConnectorTests.java new file mode 100644 index 0000000000..e096d44117 --- /dev/null +++ b/spring-web/src/test/java/org/springframework/http/client/reactive/ReactorClientHttpConnectorTests.java @@ -0,0 +1,71 @@ +/* + * Copyright 2002-2023 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.util.function.Function; + +import org.junit.jupiter.api.Test; +import reactor.netty.http.client.HttpClient; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * @author Sebastien Deleuze + */ +public class ReactorClientHttpConnectorTests { + + @Test + void restartWithDefaultConstructor() { + ReactorClientHttpConnector connector = new ReactorClientHttpConnector(); + assertThat(connector.isRunning()).isTrue(); + connector.start(); + assertThat(connector.isRunning()).isTrue(); + connector.stop(); + assertThat(connector.isRunning()).isFalse(); + connector.start(); + assertThat(connector.isRunning()).isTrue(); + } + + @Test + void restartWithExternalResourceFactory() { + ReactorResourceFactory resourceFactory = new ReactorResourceFactory(); + resourceFactory.afterPropertiesSet(); + Function mapper = Function.identity(); + ReactorClientHttpConnector connector = new ReactorClientHttpConnector(resourceFactory, mapper); + assertThat(connector.isRunning()).isTrue(); + connector.start(); + assertThat(connector.isRunning()).isTrue(); + connector.stop(); + assertThat(connector.isRunning()).isFalse(); + connector.start(); + assertThat(connector.isRunning()).isTrue(); + } + + @Test + void restartWithHttpClient() { + HttpClient httpClient = HttpClient.create(); + ReactorClientHttpConnector connector = new ReactorClientHttpConnector(httpClient); + assertThat(connector.isRunning()).isTrue(); + connector.start(); + assertThat(connector.isRunning()).isTrue(); + connector.stop(); + assertThat(connector.isRunning()).isFalse(); + connector.start(); + assertThat(connector.isRunning()).isTrue(); + } + +}