DATAES-488 - Polishing & Documentation.

Rename VerificationMode -> Verification. Reorder methods in ReactiveElasticsearchClient, add test for DefaultWebClientProvider. Enforce assertions and fix some overall code style issues.
Add client reference documentation section.

Author: christophstrobl

src/main/asciidoc/index.adoc

@@ -1,10 +1,19 @@
-= Spring Data Elasticsearch
-BioMed Central Development Team
+= Spring Data Elasticsearch - Reference Documentation
+BioMed Central Development Team; Oliver Drotbohm; Greg Turnquist; Christoph Strobl;
 :revnumber: {version}
 :revdate: {localdate}
+:toc:
+:toc-placement!:
+:linkcss:
+:doctype: book
+:docinfo: shared
+:source-highlighter: prettify
+:icons: font
+:imagesdir: images
+ifdef::backend-epub3[:front-cover-image: image:epub-cover.png[Front Cover,1050,1600]]
 :spring-data-commons-docs: ../../../../spring-data-commons/src/main/asciidoc
 
-(C) 2013-2015 The original author(s).
+(C) 2013-2018 The original author(s).
 
 NOTE: Copies of this document may be made for your own use and for distribution to others, provided that you do not charge any fee for such copies and further provided that each copy contains this Copyright Notice, whether distributed in print or electronically.
 
@@ -19,6 +28,7 @@ include::{spring-data-commons-docs}/repositories.adoc[]
 = Reference Documentation
 
 :leveloffset: +1
+include::reference/elasticsearch-clients.adoc[]
 include::reference/data-elasticsearch.adoc[]
 include::reference/elasticsearch-misc.adoc[]
 :leveloffset: -1

src/main/asciidoc/reference/elasticsearch-clients.adoc

@@ -0,0 +1,83 @@
+[[elasticsearch.clients]]
+= Elasticsearch Clients
+
+This chapter illustrates configuration and usage of supported Elasticsearch client implementations.
+
+Spring data Elasticsearch operates upon an Elasticsearch client that is connected to a single Elasticsearch node or a cluster.
+
+WARNING: The well known `TransportClient` is deprecated as of Elasticsearch 7.0.0 and is expected to be removed in Elasticsearch 8.0.
+
+[[elasticsearch.clients.rest]]
+== High Level REST Client
+
+The Java High Level REST Client provides a straight forward replacement for the `TransportClient` as it accepts and returns
+the very same request/response objects and therefore depends on the Elasticsearch core project.
+Asynchronous calls are operated upon a client managed thread pool and require a callback to be notified when the request is done.
+
+.High Level REST Client
+====
+[source,java]
+----
+static class Config {
+
+  @Bean
+  RestHighLevelClient client() {
+
+    ClientConfiguration clientConfiguration = ClientConfiguration.builder() <1>
+      .connectedTo("localhost:9200", "localhost:9201")
+      .build();
+
+    return RestClients.create(clientConfiguration).rest(); <2>
+  }
+}
+
+// ...
+
+IndexRequest request = new IndexRequest("spring-data", "elasticsearch", randomID())
+  .source(singletonMap("feature", "high-level-rest-client"))
+  .setRefreshPolicy(IMMEDIATE);
+
+IndexResponse response = client.index(request);
+----
+<1> Use the builder to provide cluster addresses, set default `HttpHeaders` or enbale SSL.
+<2> Next to the `rest()` client it is also possible to obtain the `lowLevelRest()` client.
+====
+
+[[elasticsearch.clients.reactive]]
+== Reactive Client
+
+The `ReactiveElasticsearchClient` is a non official driver based on `WebClient`.
+It uses the request/response objects provided by the Elasticsearch core project.
+Calls are directly operated on the reactive stack, **not** wrapping async (thread pool bound) responses into reactive types.
+
+.Reactive REST Client
+====
+[source,java]
+----
+static class Config {
+
+  @Bean
+  ReactiveElasticsearchClient client() {
+
+    ClientConfiguration clientConfiguration = ClientConfiguration.builder() <1>
+      .connectedTo("localhost:9200", "localhost:9291")
+      .build();
+
+    return ReactiveRestClients.create(clientConfiguration);
+  }
+}
+
+// ...
+
+Mono<IndexResponse> response = client.index(request ->
+
+  request.index("spring-data")
+    .type("elasticsearch")
+    .id(randomID())
+    .source(singletonMap("feature", "reactive-client"))
+    .setRefreshPolicy(IMMEDIATE);
+);
+----
+<1> Use the builder to provide cluster addresses, set default `HttpHeaders` or enbale SSL.
+====
+

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

@@ -64,7 +64,7 @@
 import org.springframework.data.elasticsearch.client.ClientConfiguration;
 import org.springframework.data.elasticsearch.client.ElasticsearchHost;
 import org.springframework.data.elasticsearch.client.NoReachableHostException;
-import org.springframework.data.elasticsearch.client.reactive.HostProvider.VerificationMode;
+import org.springframework.data.elasticsearch.client.reactive.HostProvider.Verification;
 import org.springframework.data.elasticsearch.client.util.RequestConverters;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpMethod;
@@ -153,9 +153,7 @@ private static WebClientProvider getWebClientProvider(ClientConfiguration client
 
 				Optional<SSLContext> sslContext = clientConfiguration.getSslContext();
 
-				sslContext.ifPresent(it -> {
-					sslConfig.sslContext(new JdkSslContext(it, true, ClientAuth.NONE));
-				});
+				sslContext.ifPresent(it -> sslConfig.sslContext(new JdkSslContext(it, true, ClientAuth.NONE)));
 			}));
 			provider = WebClientProvider.create("https", connector);
 		} else {
@@ -273,13 +271,13 @@ public Flux<SearchHit> search(HttpHeaders headers, SearchRequest searchRequest)
 	@Override
 	public Mono<ClientResponse> execute(ReactiveElasticsearchClientCallback callback) {
 
-		return this.hostProvider.getActive(VerificationMode.LAZY) //
+		return this.hostProvider.getActive(Verification.LAZY) //
 				.flatMap(callback::doWithClient) //
 				.onErrorResume(throwable -> {
 
 					if (throwable instanceof ConnectException) {
 
-						return hostProvider.getActive(VerificationMode.ACTIVE) //
+						return hostProvider.getActive(Verification.ACTIVE) //
 								.flatMap(callback::doWithClient);
 					}
 
@@ -357,9 +355,7 @@ private <T> Publisher<? extends T> readResponseBody(Request request, ClientRespo
 
 		return response.body(BodyExtractors.toMono(byte[].class)) //
 				.map(it -> new String(it, StandardCharsets.UTF_8)) //
-				.flatMap(content -> {
-					return doDecode(response, responseType, content);
-				});
+				.flatMap(content -> doDecode(response, responseType, content));
 	}
 
 	private static <T> Mono<T> doDecode(ClientResponse response, Class<T> responseType, String content) {

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

@@ -31,6 +31,7 @@
  * Default {@link WebClientProvider} that uses cached {@link WebClient} instances per {@code hostAndPort}.
  *
  * @author Mark Paluch
+ * @author Christoph Strobl
  * @since 4.0
  */
 class DefaultWebClientProvider implements WebClientProvider {
@@ -42,19 +43,32 @@ class DefaultWebClientProvider implements WebClientProvider {
 	private final Consumer<Throwable> errorListener;
 	private final HttpHeaders headers;
 
+	/**
+	 * Create new {@link DefaultWebClientProvider} with empty {@link HttpHeaders} and no-op {@literal error listener}.
+	 *
+	 * @param scheme must not be {@literal null}.
+	 * @param connector can be {@literal null}.
+	 */
 	DefaultWebClientProvider(String scheme, @Nullable ClientHttpConnector connector) {
 		this(scheme, connector, e -> {}, HttpHeaders.EMPTY);
 	}
 
+	/**
+	 * Create new {@link DefaultWebClientProvider} with empty {@link HttpHeaders} and no-op {@literal error listener}.
+	 * 
+	 * @param scheme must not be {@literal null}.
+	 * @param connector can be {@literal null}.
+	 * @param errorListener must not be {@literal null}.
+	 * @param headers must not be {@literal null}.
+	 */
 	private DefaultWebClientProvider(String scheme, @Nullable ClientHttpConnector connector,
 			Consumer<Throwable> errorListener, HttpHeaders headers) {
-		this(new ConcurrentHashMap<>(), scheme, connector, errorListener, headers);
-	}
 
-	private DefaultWebClientProvider(Map<InetSocketAddress, WebClient> cachedClients, String scheme,
-			@Nullable ClientHttpConnector connector, Consumer<Throwable> errorListener, HttpHeaders headers) {
+		Assert.notNull(scheme, "Scheme must not be null! A common scheme would be 'http'.");
+		Assert.notNull(errorListener, "ErrorListener must not be null! You may want use a no-op one 'e -> {}' instead.");
+		Assert.notNull(headers, "headers must not be null! Think about using 'HttpHeaders.EMPTY' as an alternative.");
 
-		this.cachedClients = cachedClients;
+		this.cachedClients = new ConcurrentHashMap<>();
 		this.scheme = scheme;
 		this.connector = connector;
 		this.errorListener = errorListener;
@@ -70,18 +84,7 @@ public WebClient get(InetSocketAddress endpoint) {
 
 		Assert.notNull(endpoint, "Endpoint must not be empty!");
 
-		return this.cachedClients.computeIfAbsent(endpoint, key -> {
-
-			Builder builder = WebClient.builder().defaultHeaders(it -> it.addAll(getDefaultHeaders()));
-
-			if (connector != null) {
-				builder.clientConnector(connector);
-			}
-
-			String baseUrl = String.format("%s://%s:%d", this.scheme, key.getHostString(), key.getPort());
-			return builder.baseUrl(baseUrl).filter((request, next) -> next.exchange(request).doOnError(errorListener))
-					.build();
-		});
+		return this.cachedClients.computeIfAbsent(endpoint, this::createWebClientForSocketAddress);
 	}
 
 	/*
@@ -100,7 +103,7 @@ public HttpHeaders getDefaultHeaders() {
 	@Override
 	public WebClientProvider withDefaultHeaders(HttpHeaders headers) {
 
-		Assert.notNull(headers, "HttpHeaders must not be null");
+		Assert.notNull(headers, "HttpHeaders must not be null.");
 
 		HttpHeaders merged = new HttpHeaders();
 		merged.addAll(this.headers);
@@ -125,9 +128,21 @@ public Consumer<Throwable> getErrorListener() {
 	@Override
 	public WebClientProvider withErrorListener(Consumer<Throwable> errorListener) {
 
-		Assert.notNull(errorListener, "Error listener must not be null");
+		Assert.notNull(errorListener, "Error listener must not be null.");
 
 		Consumer<Throwable> listener = this.errorListener.andThen(errorListener);
 		return new DefaultWebClientProvider(this.scheme, this.connector, listener, this.headers);
 	}
+
+	protected WebClient createWebClientForSocketAddress(InetSocketAddress socketAddress) {
+
+		Builder builder = WebClient.builder().defaultHeaders(it -> it.addAll(getDefaultHeaders()));
+
+		if (connector != null) {
+			builder = builder.clientConnector(connector);
+		}
+
+		String baseUrl = String.format("%s://%s:%d", this.scheme, socketAddress.getHostString(), socketAddress.getPort());
+		return builder.baseUrl(baseUrl).filter((request, next) -> next.exchange(request).doOnError(errorListener)).build();
+	}
 }

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

@@ -37,22 +37,41 @@
 public interface HostProvider {
 
 	/**
-	 * Lookup an active host in {@link VerificationMode#LAZY lazy} mode utilizing cached {@link ElasticsearchHost}.
+	 * Create a new {@link HostProvider} best suited for the given {@link WebClientProvider} and number of hosts.
+	 *
+	 * @param clientProvider must not be {@literal null} .
+	 * @param endpoints must not be {@literal null} nor empty.
+	 * @return new instance of {@link HostProvider}.
+	 */
+	static HostProvider provider(WebClientProvider clientProvider, InetSocketAddress... endpoints) {
+
+		Assert.notNull(clientProvider, "WebClientProvider must not be null");
+		Assert.notEmpty(endpoints, "Please provide at least one endpoint to connect to.");
+
+		if (endpoints.length == 1) {
+			return new SingleNodeHostProvider(clientProvider, endpoints[0]);
+		} else {
+			return new MultiNodeHostProvider(clientProvider, endpoints);
+		}
+	}
+
+	/**
+	 * Lookup an active host in {@link Verification#LAZY lazy} mode utilizing cached {@link ElasticsearchHost}.
 	 *
 	 * @return the {@link Mono} emitting the active host or {@link Mono#error(Throwable) an error} if none found.
 	 */
 	default Mono<InetSocketAddress> lookupActiveHost() {
-		return lookupActiveHost(VerificationMode.LAZY);
+		return lookupActiveHost(Verification.LAZY);
 	}
 
 	/**
-	 * Lookup an active host in using the given {@link VerificationMode}.
+	 * Lookup an active host in using the given {@link Verification}.
 	 *
-	 * @param verificationMode
+	 * @param verification
 	 * @return the {@link Mono} emitting the active host or {@link Mono#error(Throwable) an error}
 	 *         ({@link NoReachableHostException}) if none found.
 	 */
-	Mono<InetSocketAddress> lookupActiveHost(VerificationMode verificationMode);
+	Mono<InetSocketAddress> lookupActiveHost(Verification verification);
 
 	/**
 	 * Get the {@link WebClient} connecting to an active host utilizing cached {@link ElasticsearchHost}.
@@ -61,25 +80,25 @@ default Mono<InetSocketAddress> lookupActiveHost() {
 	 *         found.
 	 */
 	default Mono<WebClient> getActive() {
-		return getActive(VerificationMode.LAZY);
+		return getActive(Verification.LAZY);
 	}
 
 	/**
 	 * Get the {@link WebClient} connecting to an active host.
 	 *
-	 * @param verificationMode must not be {@literal null}.
+	 * @param verification must not be {@literal null}.
 	 * @return the {@link Mono} emitting the client for an active host or {@link Mono#error(Throwable) an error} if none
 	 *         found.
 	 */
-	default Mono<WebClient> getActive(VerificationMode verificationMode) {
-		return lookupActiveHost(verificationMode).map(this::createWebClient);
+	default Mono<WebClient> getActive(Verification verification) {
+		return lookupActiveHost(verification).map(this::createWebClient);
 	}
 
 	/**
 	 * Creates a {@link WebClient} for {@link InetSocketAddress endpoint}.
 	 *
-	 * @param baseUrl
-	 * @return
+	 * @param endpoint must not be {@literal null}.
+	 * @return a {@link WebClient} using the the given endpoint as {@literal base url}.
 	 */
 	WebClient createWebClient(InetSocketAddress endpoint);
 
@@ -91,29 +110,12 @@ default Mono<WebClient> getActive(VerificationMode verificationMode) {
 	Mono<ClusterInformation> clusterInfo();
 
 	/**
-	 * Create a new {@link HostProvider} best suited for the given {@link WebClientProvider} and number of hosts.
+	 * {@link Verification} allows to influence the lookup strategy for active hosts.
 	 *
-	 * @param clientProvider must not be {@literal null} .
-	 * @param hosts must not be {@literal null} nor empty.
-	 * @return new instance of {@link HostProvider}.
-	 */
-	static HostProvider provider(WebClientProvider clientProvider, InetSocketAddress... endpoints) {
-
-		Assert.notNull(clientProvider, "WebClientProvider must not be null");
-		Assert.notEmpty(endpoints, "Please provide at least one endpoint to connect to.");
-
-		if (endpoints.length == 1) {
-			return new SingleNodeHostProvider(clientProvider, endpoints[0]);
-		} else {
-			return new MultiNodeHostProvider(clientProvider, endpoints);
-		}
-	}
-
-	/**
 	 * @author Christoph Strobl
 	 * @since 4.0
 	 */
-	enum VerificationMode {
+	enum Verification {
 
 		/**
 		 * Actively check for cluster node health.
@@ -127,9 +129,9 @@ enum VerificationMode {
 	}
 
 	/**
-	 * Value object accumulating information about cluster an Elasticsearch cluster.
+	 * Value object accumulating information about an Elasticsearch cluster.
 	 *
-	 * @author Christoph Strobll
+	 * @author Christoph Strobl
 	 * @since 4.0.
 	 */
 	class ClusterInformation {