DATAES-719 - Add customization hook for reactive WebClient.

Original PR: #363

(cherry picked from commit f7a14c1135189a62675a3345d70ccb1a6ec16af4)

src/main/java/org/springframework/data/elasticsearch/client/ClientConfiguration.java

@@ -20,11 +20,13 @@ import java.net.SocketAddress;
 import java.time.Duration;
 import java.util.List;
 import java.util.Optional;
+import java.util.function.Function;
 
 import javax.net.ssl.HostnameVerifier;
 import javax.net.ssl.SSLContext;
 
 import org.springframework.http.HttpHeaders;
+import org.springframework.web.reactive.function.client.WebClient;
 
 /**
  * Configuration interface exposing common client configuration properties for Elasticsearch clients.
@@ -161,6 +163,11 @@ public interface ClientConfiguration {
 	 */
 	Optional<String> getProxy();
 
+	/**
+	 * @return the function for configuring a WebClient.
+	 */
+	Function<WebClient, WebClient> getWebClientConfigurer();
+
 	/**
 	 * @author Christoph Strobl
 	 */
@@ -314,9 +321,17 @@ public interface ClientConfiguration {
 
 		/**
 		 * @param proxy a proxy formatted as String {@literal host:port}.
-		 * @return the {@link MaybeSecureClientConfigurationBuilder}.
+		 * @return the {@link TerminalClientConfigurationBuilder}.
 		 */
-		MaybeSecureClientConfigurationBuilder withProxy(String proxy);
+		TerminalClientConfigurationBuilder withProxy(String proxy);
+
+		/**
+		 * set customization hook in case of a reactive configuration
+		 * 
+		 * @param webClientConfigurer function to configure the WebClient
+		 * @return the {@link TerminalClientConfigurationBuilder}.
+		 */
+		TerminalClientConfigurationBuilder withWebClientConfigurer(Function<WebClient, WebClient> webClientConfigurer);
 
 		/**
 		 * Build the {@link ClientConfiguration} object.

src/main/java/org/springframework/data/elasticsearch/client/ClientConfigurationBuilder.java

@@ -20,6 +20,7 @@ import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 
 import javax.net.ssl.HostnameVerifier;
@@ -31,6 +32,7 @@ import org.springframework.data.elasticsearch.client.ClientConfiguration.Termina
 import org.springframework.http.HttpHeaders;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
+import org.springframework.web.reactive.function.client.WebClient;
 
 /**
  * Default builder implementation for {@link ClientConfiguration}.
@@ -56,6 +58,7 @@ class ClientConfigurationBuilder
 	private String password;
 	private String pathPrefix;
 	private String proxy;
+	private Function<WebClient, WebClient> webClientConfigurer;
 
 	/*
 	 * (non-Javadoc)
@@ -187,12 +190,20 @@ class ClientConfigurationBuilder
 
 	@Override
 	public TerminalClientConfigurationBuilder withPathPrefix(String pathPrefix) {
-
 		this.pathPrefix = pathPrefix;
 
 		return this;
 	}
 
+	@Override
+	public TerminalClientConfigurationBuilder withWebClientConfigurer(Function<WebClient, WebClient> webClientConfigurer) {
+
+		Assert.notNull(webClientConfigurer, "webClientConfigurer must not be null");
+
+		this.webClientConfigurer = webClientConfigurer;
+		return this;
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.elasticsearch.client.ClientConfiguration.ClientConfigurationBuilderWithOptionalDefaultHeaders#build()
@@ -208,7 +219,7 @@ class ClientConfigurationBuilder
 		}
 
 		return new DefaultClientConfiguration(hosts, headers, useSsl, sslContext, soTimeout, connectTimeout, pathPrefix,
-				hostnameVerifier, proxy);
+				hostnameVerifier, proxy, webClientConfigurer);
 	}
 
 	private static InetSocketAddress parse(String hostAndPort) {

src/main/java/org/springframework/data/elasticsearch/client/DefaultClientConfiguration.java

@@ -21,12 +21,14 @@ import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
+import java.util.function.Function;
 
 import javax.net.ssl.HostnameVerifier;
 import javax.net.ssl.SSLContext;
 
 import org.springframework.http.HttpHeaders;
 import org.springframework.lang.Nullable;
+import org.springframework.web.reactive.function.client.WebClient;
 
 /**
  * Default {@link ClientConfiguration} implementation.
@@ -48,10 +50,11 @@ class DefaultClientConfiguration implements ClientConfiguration {
 	private final String pathPrefix;
 	private final @Nullable HostnameVerifier hostnameVerifier;
 	private final String proxy;
+	private final Function<WebClient, WebClient> webClientConfigurer;
 
 	DefaultClientConfiguration(List<InetSocketAddress> hosts, HttpHeaders headers, boolean useSsl,
 			@Nullable SSLContext sslContext, Duration soTimeout, Duration connectTimeout, @Nullable String pathPrefix,
-			@Nullable HostnameVerifier hostnameVerifier, String proxy) {
+			@Nullable HostnameVerifier hostnameVerifier, String proxy, Function<WebClient, WebClient> webClientConfigurer) {
 
 		this.hosts = Collections.unmodifiableList(new ArrayList<>(hosts));
 		this.headers = new HttpHeaders(headers);
@@ -62,86 +65,56 @@ class DefaultClientConfiguration implements ClientConfiguration {
 		this.pathPrefix = pathPrefix;
 		this.hostnameVerifier = hostnameVerifier;
 		this.proxy = proxy;
+		this.webClientConfigurer = webClientConfigurer;
 	}
 
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.elasticsearch.client.ClientConfiguration#getEndpoints()
-	 */
 	@Override
 	public List<InetSocketAddress> getEndpoints() {
 		return this.hosts;
 	}
 
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.elasticsearch.client.ClientConfiguration#getDefaultHeaders()
-	 */
 	@Override
 	public HttpHeaders getDefaultHeaders() {
 		return this.headers;
 	}
 
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.elasticsearch.client.ClientConfiguration#useSsl()
-	 */
 	@Override
 	public boolean useSsl() {
 		return this.useSsl;
 	}
 
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.elasticsearch.client.ClientConfiguration#getSslContext()
-	 */
 	@Override
 	public Optional<SSLContext> getSslContext() {
 		return Optional.ofNullable(this.sslContext);
 	}
 
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.elasticsearch.client.ClientConfiguration#getHostNameVerifier()
-	 */
 	@Override
 	public Optional<HostnameVerifier> getHostNameVerifier() {
 		return Optional.ofNullable(this.hostnameVerifier);
 	}
 
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.elasticsearch.client.ClientConfiguration#getConnectTimeout()
-	 */
 	@Override
 	public Duration getConnectTimeout() {
 		return this.connectTimeout;
 	}
 
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.elasticsearch.client.ClientConfiguration#getSocketTimeout()
-	 */
 	@Override
 	public Duration getSocketTimeout() {
 		return this.soTimeout;
 	}
 
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.elasticsearch.client.ClientConfiguration#getPathPrefix()
-	 */
 	@Override
 	public String getPathPrefix() {
 		return this.pathPrefix;
 	}
 
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.elasticsearch.client.ClientConfiguration#getProxy()
-	 */
 	@Override
 	public Optional<String> getProxy() {
 		return Optional.ofNullable(proxy);
 	}
+
+	@Override
+	public Function<WebClient, WebClient> getWebClientConfigurer() {
+		return webClientConfigurer != null ? webClientConfigurer : Function.identity();
+	}
 }

src/main/java/org/springframework/data/elasticsearch/client/reactive/DefaultReactiveElasticsearchClient.java

@@ -226,7 +226,9 @@ public class DefaultReactiveElasticsearchClient implements ReactiveElasticsearch
 			provider = provider.withPathPrefix(clientConfiguration.getPathPrefix());
 		}
 
-		return provider.withDefaultHeaders(clientConfiguration.getDefaultHeaders());
+		provider = provider.withDefaultHeaders(clientConfiguration.getDefaultHeaders()) //
+				.withWebClientConfigurer(clientConfiguration.getWebClientConfigurer());
+		return provider;
 	}
 
 	/*

src/main/java/org/springframework/data/elasticsearch/client/reactive/WebClientProvider.java

@@ -17,6 +17,7 @@ package org.springframework.data.elasticsearch.client.reactive;
 
 import java.net.InetSocketAddress;
 import java.util.function.Consumer;
+import java.util.function.Function;
 
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.client.reactive.ClientHttpConnector;
@@ -130,4 +131,12 @@ public interface WebClientProvider {
 	 * @since 3.2.4
 	 */
 	WebClientProvider withPathPrefix(String pathPrefix);
+
+	/**
+	 * Create a new instance of {@link WebClientProvider} calling the given {@link Function} to configure the {@link WebClient}.
+	 * @param webClientConfigurer configuration function
+	 * @return new instance of {@link WebClientProvider}
+	 * @since 3.2.4
+	 */
+	WebClientProvider withWebClientConfigurer(Function<WebClient, WebClient> webClientConfigurer);
 }

src/test/java/org/springframework/data/elasticsearch/client/reactive/ReactiveMockClientTestsUtils.java

@@ -277,6 +277,11 @@ public class ReactiveMockClientTestsUtils {
 			throw new UnsupportedOperationException();
 		}
 
+		@Override
+		public WebClientProvider withWebClientConfigurer(Function<WebClient, WebClient> webClientConfigurer) {
+			throw new UnsupportedOperationException("not implemented");
+		}
+
 		public Send when(String host) {
 			InetSocketAddress inetSocketAddress = getInetSocketAddress(host);
 			return new CallbackImpl(get(host), headersUriSpecMap.get(inetSocketAddress),