Add JdkHttpClientResourceFactory

See gh-23432

Author: rstoyanchev

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

@@ -23,6 +23,7 @@
 import java.nio.ByteBuffer;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
 import java.util.concurrent.Flow;
 import java.util.function.Function;
 
@@ -31,6 +32,8 @@
 import org.springframework.core.io.buffer.DataBufferFactory;
 import org.springframework.core.io.buffer.DefaultDataBufferFactory;
 import org.springframework.http.HttpMethod;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
 
 /**
  * {@link ClientHttpConnector} for the Java {@link HttpClient}.
@@ -44,21 +47,47 @@ public class JdkClientHttpConnector implements ClientHttpConnector {
 
 	private final HttpClient httpClient;
 
-	private final DataBufferFactory bufferFactory;
+	private DataBufferFactory bufferFactory = DefaultDataBufferFactory.sharedInstance;
 
 
 	/**
 	 * Default constructor that uses {@link HttpClient#newHttpClient()}.
 	 */
 	public JdkClientHttpConnector() {
-		this(HttpClient.newHttpClient(), new DefaultDataBufferFactory());
+		this(HttpClient.newHttpClient());
 	}
 
 	/**
 	 * Constructor with an initialized {@link HttpClient} and a {@link DataBufferFactory}.
 	 */
-	public JdkClientHttpConnector(HttpClient httpClient, DataBufferFactory bufferFactory) {
+	public JdkClientHttpConnector(HttpClient httpClient) {
 		this.httpClient = httpClient;
+	}
+
+	/**
+	 * Constructor with a {@link JdkHttpClientResourceFactory} that provides
+	 * shared resources.
+	 * @param clientBuilder a pre-initialized builder for the client that will
+	 * be further initialized with the shared resources to use
+	 * @param resourceFactory the {@link JdkHttpClientResourceFactory} to use
+	 */
+	public JdkClientHttpConnector(
+			HttpClient.Builder clientBuilder, @Nullable JdkHttpClientResourceFactory resourceFactory) {
+
+		if (resourceFactory != null) {
+			Executor executor = resourceFactory.getExecutor();
+			clientBuilder.executor(executor);
+		}
+		this.httpClient = clientBuilder.build();
+	}
+
+
+	/**
+	 * Set the buffer factory to use.
+	 * <p>By default, this is {@link DefaultDataBufferFactory#sharedInstance}.
+	 */
+	public void setBufferFactory(DataBufferFactory bufferFactory) {
+		Assert.notNull(bufferFactory, "DataBufferFactory is required");
 		this.bufferFactory = bufferFactory;
 	}
 

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

@@ -0,0 +1,108 @@
+/*
+ * 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.
+ * 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.http.HttpClient;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
+
+import org.eclipse.jetty.util.component.LifeCycle;
+import org.eclipse.jetty.util.thread.QueuedThreadPool;
+
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.lang.Nullable;
+import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
+import org.springframework.util.Assert;
+
+/**
+ * Factory to manage JDK HttpClient resources such as a shared {@link Executor}
+ * within the lifecycle of a Spring {@code ApplicationContext}.
+ *
+ * <p>This factory implements {@link InitializingBean} and {@link DisposableBean}
+ * and is expected typically to be declared as a Spring-managed bean.
+ *
+ * @author Rossen Stoyanchev
+ * @since 6.0
+ * @see JdkClientHttpConnector#JdkClientHttpConnector(HttpClient.Builder, JdkHttpClientResourceFactory)
+ */
+public class JdkHttpClientResourceFactory implements InitializingBean, DisposableBean {
+
+	@Nullable
+	private Executor executor;
+
+	private String threadPrefix = "jdk-http";
+
+
+	/**
+	 * Configure the {@link Executor} to use for {@link HttpClient} exchanges.
+	 * The given executor is started and stopped via {@link InitializingBean}
+	 * and {@link DisposableBean}.
+	 * <p>By default, this is set to {@link Executors#newCachedThreadPool(ThreadFactory)},
+	 * which mirrors {@link HttpClient.Builder#executor(Executor)}.
+	 * @param executor the executor to use
+	 */
+	public void setExecutor(@Nullable Executor executor) {
+		this.executor = executor;
+	}
+
+	/**
+	 * Return the configured {@link Executor}.
+	 */
+	@Nullable
+	public Executor getExecutor() {
+		return this.executor;
+	}
+
+	/**
+	 * Configure the thread prefix to initialize {@link QueuedThreadPool} executor with. This
+	 * is used only when a {@link Executor} instance isn't
+	 * {@link #setExecutor(Executor) provided}.
+	 * <p>By default set to "jetty-http".
+	 * @param threadPrefix the thread prefix to use
+	 */
+	public void setThreadPrefix(String threadPrefix) {
+		Assert.notNull(threadPrefix, "Thread prefix is required");
+		this.threadPrefix = threadPrefix;
+	}
+
+
+	@Override
+	public void afterPropertiesSet() throws Exception {
+		if (this.executor == null) {
+			String name = this.threadPrefix + "@" + Integer.toHexString(hashCode());
+			this.executor = Executors.newCachedThreadPool(new CustomizableThreadFactory(name));
+		}
+		if (this.executor instanceof LifeCycle) {
+			((LifeCycle)this.executor).start();
+		}
+	}
+
+	@Override
+	public void destroy() throws Exception {
+		try {
+			if (this.executor instanceof LifeCycle) {
+				((LifeCycle)this.executor).stop();
+			}
+		}
+		catch (Throwable ex) {
+			// ignore
+		}
+	}
+
+}

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

@@ -99,9 +99,9 @@ class WebClientIntegrationTests {
 	static Stream<ClientHttpConnector> arguments() {
 		return Stream.of(
 				new ReactorClientHttpConnector(),
+				new JdkClientHttpConnector(),
 				new JettyClientHttpConnector(),
-				new HttpComponentsClientHttpConnector(),
-				new JdkClientHttpConnector()
+				new HttpComponentsClientHttpConnector()
 		);
 	}
 

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

@@ -75,21 +75,21 @@ class SseIntegrationTests extends AbstractHttpHandlerIntegrationTests {
 	static Object[][] arguments() {
 		return new Object[][] {
 			{new JettyHttpServer(), new ReactorClientHttpConnector()},
+			{new JettyHttpServer(), new JdkClientHttpConnector()},
 			{new JettyHttpServer(), new JettyClientHttpConnector()},
 			{new JettyHttpServer(), new HttpComponentsClientHttpConnector()},
-			{new JettyHttpServer(), new JdkClientHttpConnector()},
 			{new ReactorHttpServer(), new ReactorClientHttpConnector()},
+			{new ReactorHttpServer(), new JdkClientHttpConnector()},
 			{new ReactorHttpServer(), new JettyClientHttpConnector()},
 			{new ReactorHttpServer(), new HttpComponentsClientHttpConnector()},
-			{new ReactorHttpServer(), new JdkClientHttpConnector()},
 			{new TomcatHttpServer(), new ReactorClientHttpConnector()},
+			{new TomcatHttpServer(), new JdkClientHttpConnector()},
 			{new TomcatHttpServer(), new JettyClientHttpConnector()},
 			{new TomcatHttpServer(), new HttpComponentsClientHttpConnector()},
-			{new TomcatHttpServer(), new JdkClientHttpConnector()},
 			{new UndertowHttpServer(), new ReactorClientHttpConnector()},
+			{new UndertowHttpServer(), new JdkClientHttpConnector()},
 			{new UndertowHttpServer(), new JettyClientHttpConnector()},
 			{new UndertowHttpServer(), new HttpComponentsClientHttpConnector()},
-			{new UndertowHttpServer(), new JdkClientHttpConnector()},
 		};
 	}