DATAES-488 - Add reactive Elasticsearch client support.

Initial implementation of a ReactiveElasticsearchClient using WebClient to connect to cluster nodes.

ReactiveElasticsearchClient client = ElasticsearchClients.createClient()
  .connectedTo("http://localhost:9200", "http://localhost:9201")
  .reactive();
A HostProvider selects active nodes and routes requests.

client.index(request ->

  request.index("spring-data")
    .type("elasticsearch")
    .id(randomUUID().toString())
    .source(singletonMap("feature", "reactive-client"))
    .setRefreshPolicy(IMMEDIATE);
);
This implementation provides the first building block for reactive Template and Repository support to be added subsequently.

Along the lines we upgraded to Elasticsearch 6.5.

Original Pull Request: #226

src/test/java/org/springframework/data/elasticsearch/TestUtils.java

@@ -0,0 +1,63 @@
+/*
+ * Copyright 2018 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
+ *
+ *      http://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.data.elasticsearch;
+
+import lombok.SneakyThrows;
+
+import org.elasticsearch.ElasticsearchStatusException;
+import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
+import org.elasticsearch.client.RequestOptions;
+import org.elasticsearch.client.RestHighLevelClient;
+import org.springframework.data.elasticsearch.client.ElasticsearchClients;
+import org.springframework.data.elasticsearch.client.reactive.ReactiveElasticsearchClient;
+import org.springframework.util.ObjectUtils;
+
+/**
+ * @author Christoph Strobl
+ * @currentRead Fool's Fate - Robin Hobb
+ */
+public final class TestUtils {
+
+	private TestUtils() {}
+
+	public static RestHighLevelClient restHighLevelClient() {
+		return ElasticsearchClients.createClient().connectedToLocalhost().rest();
+	}
+
+	public static ReactiveElasticsearchClient reactiveClient() {
+		return ElasticsearchClients.createClient().connectedToLocalhost().reactive();
+	}
+
+	@SneakyThrows
+	public static void deleteIndex(String... indexes) {
+
+		if (ObjectUtils.isEmpty(indexes)) {
+			return;
+		}
+
+		try (RestHighLevelClient client = restHighLevelClient()) {
+			for (String index : indexes) {
+
+				try {
+					client.indices().delete(new DeleteIndexRequest(index), RequestOptions.DEFAULT);
+				} catch (ElasticsearchStatusException ex) {
+					// just ignore it
+				}
+			}
+		}
+	}
+}

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

@@ -0,0 +1,141 @@
+/*
+ * Copyright 2018 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
+ *
+ *      http://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.data.elasticsearch.client.reactive;
+
+import static org.assertj.core.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.data.elasticsearch.client.reactive.HostProvider.VerificationMode;
+import org.springframework.data.elasticsearch.client.ElasticsearchHost;
+import org.springframework.data.elasticsearch.client.ElasticsearchHost.State;
+import org.springframework.data.elasticsearch.client.reactive.ReactiveMockClientTestsUtils.MockDelegatingElasticsearchHostProvider;
+import org.springframework.data.elasticsearch.client.reactive.ReactiveMockClientTestsUtils.WebClientProvider.Receive;
+import org.springframework.web.reactive.function.client.ClientResponse;
+
+/**
+ * @author Christoph Strobl
+ * @currentRead Golden Fool - Robin Hobb
+ */
+public class MultiNodeHostProviderUnitTests {
+
+	static final String HOST_1 = ":9200";
+	static final String HOST_2 = ":9201";
+	static final String HOST_3 = ":9202";
+
+	MockDelegatingElasticsearchHostProvider<MultiNodeHostProvider> mock;
+	MultiNodeHostProvider provider;
+
+	@Before
+	public void setUp() {
+
+		mock = ReactiveMockClientTestsUtils.multi(HOST_1, HOST_2, HOST_3);
+		provider = mock.getDelegate();
+	}
+
+	@Test // DATAES-488
+	public void refreshHostStateShouldUpdateNodeStateCorrectly() {
+
+		mock.when(HOST_1).receive(Receive::error);
+		mock.when(HOST_2).receive(Receive::ok);
+		mock.when(HOST_3).receive(Receive::ok);
+
+		provider.clusterInfo().as(StepVerifier::create).expectNextCount(1).verifyComplete();
+
+		assertThat(provider.getCachedHostState()).extracting(ElasticsearchHost::getState).containsExactly(State.OFFLINE,
+				State.ONLINE, State.ONLINE);
+	}
+
+	@Test // DATAES-488
+	public void getActiveReturnsFirstActiveHost() {
+
+		mock.when(HOST_1).receive(Receive::error);
+		mock.when(HOST_2).receive(Receive::ok);
+		mock.when(HOST_3).receive(Receive::error);
+
+		provider.getActive().as(StepVerifier::create).expectNext(mock.client(HOST_2)).verifyComplete();
+	}
+
+	@Test // DATAES-488
+	public void getActiveErrorsWhenNoActiveHostFound() {
+
+		mock.when(HOST_1).receive(Receive::error);
+		mock.when(HOST_2).receive(Receive::error);
+		mock.when(HOST_3).receive(Receive::error);
+
+		provider.getActive().as(StepVerifier::create).expectError(IllegalStateException.class);
+	}
+
+	@Test // DATAES-488
+	public void lazyModeDoesNotResolveHostsTwice() {
+
+		mock.when(HOST_1).receive(Receive::error);
+		mock.when(HOST_2).receive(Receive::ok);
+		mock.when(HOST_3).receive(Receive::error);
+
+		provider.clusterInfo().as(StepVerifier::create).expectNextCount(1).verifyComplete();
+
+		provider.getActive(VerificationMode.LAZY).as(StepVerifier::create).expectNext(mock.client(HOST_2)).verifyComplete();
+
+		verify(mock.client(":9201")).head();
+	}
+
+	@Test // DATAES-488
+	public void alwaysModeDoesNotResolveHostsTwice() {
+
+		mock.when(HOST_1).receive(Receive::error);
+		mock.when(HOST_2).receive(Receive::ok);
+		mock.when(HOST_3).receive(Receive::error);
+
+		provider.clusterInfo().as(StepVerifier::create).expectNextCount(1).verifyComplete();
+
+		provider.getActive(VerificationMode.FORCE).as(StepVerifier::create).expectNext(mock.client(HOST_2))
+				.verifyComplete();
+
+		verify(mock.client(HOST_2), times(2)).head();
+	}
+
+	@Test // DATAES-488
+	public void triesDeadHostsIfNoActiveFound() {
+
+		mock.when(HOST_1).receive(Receive::error);
+		mock.when(HOST_2).get(requestHeadersUriSpec -> {
+
+			ClientResponse response1 = mock(ClientResponse.class);
+			Receive.error(response1);
+
+			ClientResponse response2 = mock(ClientResponse.class);
+			Receive.ok(response2);
+
+			when(requestHeadersUriSpec.exchange()).thenReturn(Mono.just(response1), Mono.just(response2));
+		});
+
+		mock.when(HOST_3).receive(Receive::error);
+
+		provider.clusterInfo().as(StepVerifier::create).expectNextCount(1).verifyComplete();
+		assertThat(provider.getCachedHostState()).extracting(ElasticsearchHost::getState).containsExactly(State.OFFLINE,
+				State.OFFLINE, State.OFFLINE);
+
+		provider.getActive().as(StepVerifier::create).expectNext(mock.client(HOST_2)).verifyComplete();
+
+		verify(mock.client(HOST_2), times(2)).head();
+	}
+}

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

@@ -0,0 +1,467 @@
+/*
+ * Copyright 2018 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
+ *
+ *      http://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.data.elasticsearch.client.reactive;
+
+import static org.assertj.core.api.Assertions.*;
+
+import reactor.test.StepVerifier;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.UUID;
+
+import org.elasticsearch.ElasticsearchStatusException;
+import org.elasticsearch.Version;
+import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
+import org.elasticsearch.action.delete.DeleteRequest;
+import org.elasticsearch.action.get.GetRequest;
+import org.elasticsearch.action.get.MultiGetRequest;
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.action.search.SearchRequest;
+import org.elasticsearch.action.support.WriteRequest.RefreshPolicy;
+import org.elasticsearch.action.update.UpdateRequest;
+import org.elasticsearch.client.RestHighLevelClient;
+import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.rest.RestStatus;
+import org.elasticsearch.search.builder.SearchSourceBuilder;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.data.elasticsearch.TestUtils;
+import org.springframework.http.HttpHeaders;
+import org.springframework.lang.Nullable;
+
+/**
+ * @author Christoph Strobl
+ * @currentRead Fool's Fate - Robin Hobb
+ */
+public class ReactiveElasticsearchClientTests {
+
+	static final String INDEX_I = "idx-1-reactive-client-tests";
+	static final String INDEX_II = "idx-2-reactive-client-tests";
+
+	static final String TYPE_I = "doc-type-1";
+	static final String TYPE_II = "doc-type-2";
+
+	static final Map<String, String> DOC_SOURCE;
+
+	RestHighLevelClient syncClient;
+	ReactiveElasticsearchClient client;
+
+	static {
+
+		Map<String, String> source = new LinkedHashMap<>();
+		source.put("firstname", "chade");
+		source.put("lastname", "fallstar");
+
+		DOC_SOURCE = Collections.unmodifiableMap(source);
+	}
+
+	@Before
+	public void setUp() {
+
+		syncClient = TestUtils.restHighLevelClient();
+		client = TestUtils.reactiveClient();
+	}
+
+	@After
+	public void after() throws IOException {
+
+		TestUtils.deleteIndex(INDEX_I, INDEX_II);
+
+		syncClient.close();
+	}
+
+	@Test // DATAES-488
+	public void pingForActiveHostShouldReturnTrue() {
+		client.ping().as(StepVerifier::create) //
+				.expectNext(true) //
+				.verifyComplete();
+	}
+
+	@Test // DATAES-488
+	public void pingForUnknownHostShouldReturnFalse() {
+
+		DefaultReactiveElasticsearchClient.create(HttpHeaders.EMPTY, "http://localhost:4711").ping() //
+				.as(StepVerifier::create) //
+				.expectNext(false) //
+				.verifyComplete();
+	}
+
+	@Test // DATAES-488
+	public void infoShouldReturnClusterInformation() {
+
+		client.info().as(StepVerifier::create) //
+				.consumeNextWith(it -> {
+
+					assertThat(it.isAvailable()).isTrue();
+					assertThat(it.getVersion()).isGreaterThanOrEqualTo(Version.CURRENT);
+				}) //
+				.verifyComplete();
+	}
+
+	@Test // DATAES-488
+	public void getShouldFetchDocumentById() {
+
+		String id = addSourceDocument().ofType(TYPE_I).to(INDEX_I);
+
+		client.get(new GetRequest(INDEX_I, TYPE_I, id)) //
+				.as(StepVerifier::create) //
+				.consumeNextWith(it -> {
+
+					assertThat(it.getId()).isEqualTo(id);
+					assertThat(it.getSource()).containsAllEntriesOf(DOC_SOURCE);
+				}) //
+				.verifyComplete();
+	}
+
+	@Test // DATAES-488
+	public void getShouldCompleteForNonExistingDocuments() {
+
+		addSourceDocument().ofType(TYPE_I).to(INDEX_I);
+
+		String id = "this-one-does-not-exist";
+		client.get(new GetRequest(INDEX_I, TYPE_I, id)) //
+				.as(StepVerifier::create) //
+				.verifyComplete();
+	}
+
+	@Test // DATAES-488
+	public void getShouldCompleteForNonExistingType() {
+
+		String id = addSourceDocument().ofType(TYPE_I).to(INDEX_I);
+
+		client.get(new GetRequest(INDEX_I, "fantasy-books", id)) //
+				.as(StepVerifier::create) //
+				.verifyComplete();
+	}
+
+	@Test // DATAES-488
+	public void multiGetShouldReturnAllDocumentsFromSameCollection() {
+
+		String id1 = addSourceDocument().ofType(TYPE_I).to(INDEX_I);
+		String id2 = addSourceDocument().ofType(TYPE_I).to(INDEX_I);
+
+		MultiGetRequest request = new MultiGetRequest() //
+				.add(INDEX_I, TYPE_I, id1) //
+				.add(INDEX_I, TYPE_I, id2);
+
+		client.multiGet(request) //
+				.as(StepVerifier::create) //
+				.consumeNextWith(it -> {
+					assertThat(it.getId()).isEqualTo(id1);
+				}) //
+				.consumeNextWith(it -> {
+					assertThat(it.getId()).isEqualTo(id2);
+				}) //
+				.verifyComplete();
+	}
+
+	@Test // DATAES-488
+	public void multiGetShouldReturnAllExistingDocumentsFromSameCollection() {
+
+		String id1 = addSourceDocument().ofType(TYPE_I).to(INDEX_I);
+		addSourceDocument().ofType(TYPE_I).to(INDEX_I);
+
+		MultiGetRequest request = new MultiGetRequest() //
+				.add(INDEX_I, TYPE_I, id1) //
+				.add(INDEX_I, TYPE_I, "this-one-does-not-exist");
+
+		client.multiGet(request) //
+				.as(StepVerifier::create) //
+				.consumeNextWith(it -> {
+					assertThat(it.getId()).isEqualTo(id1);
+				}) //
+				.verifyComplete();
+	}
+
+	@Test // DATAES-488
+	public void multiGetShouldSkipNonExistingDocuments() {
+
+		String id1 = addSourceDocument().ofType(TYPE_I).to(INDEX_I);
+		String id2 = addSourceDocument().ofType(TYPE_I).to(INDEX_I);
+
+		MultiGetRequest request = new MultiGetRequest() //
+				.add(INDEX_I, TYPE_I, id1) //
+				.add(INDEX_I, TYPE_I, "this-one-does-not-exist") //
+				.add(INDEX_I, TYPE_I, id2); //
+
+		client.multiGet(request) //
+				.as(StepVerifier::create) //
+				.consumeNextWith(it -> {
+					assertThat(it.getId()).isEqualTo(id1);
+				}) //
+				.consumeNextWith(it -> {
+					assertThat(it.getId()).isEqualTo(id2);
+				}) //
+				.verifyComplete();
+	}
+
+	@Test // DATAES-488
+	public void multiGetShouldCompleteIfNothingFound() {
+
+		String id1 = addSourceDocument().ofType(TYPE_I).to(INDEX_I);
+		String id2 = addSourceDocument().ofType(TYPE_I).to(INDEX_I);
+
+		client.multiGet(new MultiGetRequest().add(INDEX_II, TYPE_I, id1).add(INDEX_II, TYPE_I, id2)) //
+				.as(StepVerifier::create) //
+				.verifyComplete();
+	}
+
+	@Test // DATAES-488
+	public void multiGetShouldReturnAllExistingDocumentsFromDifferentCollection() {
+
+		String id1 = addSourceDocument().ofType(TYPE_I).to(INDEX_I);
+		String id2 = addSourceDocument().ofType(TYPE_II).to(INDEX_II);
+
+		MultiGetRequest request = new MultiGetRequest() //
+				.add(INDEX_I, TYPE_I, id1) //
+				.add(INDEX_II, TYPE_II, id2);
+
+		client.multiGet(request) //
+				.as(StepVerifier::create) //
+				.consumeNextWith(it -> {
+					assertThat(it.getId()).isEqualTo(id1);
+				}) //
+				.consumeNextWith(it -> {
+					assertThat(it.getId()).isEqualTo(id2);
+				}) //
+				.verifyComplete();
+	}
+
+	@Test // DATAES-488
+	public void existsReturnsTrueForExistingDocuments() {
+
+		String id = addSourceDocument().ofType(TYPE_I).to(INDEX_I);
+
+		client.exists(new GetRequest(INDEX_I, TYPE_I, id)) //
+				.as(StepVerifier::create) //
+				.expectNext(true)//
+				.verifyComplete();
+	}
+
+	@Test // DATAES-488
+	public void existsReturnsFalseForNonExistingDocuments() {
+
+		String id = addSourceDocument().ofType(TYPE_I).to(INDEX_I);
+
+		client.exists(new GetRequest(INDEX_II, TYPE_I, id)) //
+				.as(StepVerifier::create) //
+				.expectNext(false)//
+				.verifyComplete();
+	}
+
+	@Test // DATAES-488
+	public void indexShouldAddDocument() {
+
+		IndexRequest request = indexRequest(DOC_SOURCE, INDEX_I, TYPE_I);
+
+		client.index(request) //
+				.as(StepVerifier::create) //
+				.consumeNextWith(it -> {
+
+					assertThat(it.status()).isEqualTo(RestStatus.CREATED);
+					assertThat(it.getId()).isEqualTo(request.id());
+				})//
+				.verifyComplete();
+	}
+
+	@Test // DATAES-488
+	public void indexShouldErrorForExistingDocuments() {
+
+		String id = addSourceDocument().ofType(TYPE_I).to(INDEX_I);
+
+		IndexRequest request = indexRequest(DOC_SOURCE, INDEX_I, TYPE_I)//
+				.id(id);
+
+		client.index(request) //
+				.as(StepVerifier::create) //
+				.consumeErrorWith(error -> {
+					assertThat(error).isInstanceOf(ElasticsearchStatusException.class);
+				}) //
+				.verify();
+	}
+
+	@Test // DATAES-488
+	public void updateShouldUpsertNonExistingDocumentWhenUsedWithUpsert() {
+
+		String id = UUID.randomUUID().toString();
+		UpdateRequest request = new UpdateRequest(INDEX_I, TYPE_I, id) //
+				.doc(DOC_SOURCE) //
+				.docAsUpsert(true);
+
+		client.update(request) //
+				.as(StepVerifier::create) //
+				.consumeNextWith(it -> {
+
+					assertThat(it.status()).isEqualTo(RestStatus.CREATED);
+					assertThat(it.getId()).isEqualTo(id);
+				}) //
+				.verifyComplete();
+	}
+
+	@Test // DATAES-488
+	public void updateShouldUpdateExistingDocument() {
+
+		String id = addSourceDocument().ofType(TYPE_I).to(INDEX_I);
+
+		UpdateRequest request = new UpdateRequest(INDEX_I, TYPE_I, id) //
+				.doc(Collections.singletonMap("dutiful", "farseer"));
+
+		client.update(request) //
+				.as(StepVerifier::create) //
+				.consumeNextWith(it -> {
+
+					assertThat(it.status()).isEqualTo(RestStatus.OK);
+					assertThat(it.getId()).isEqualTo(id);
+					assertThat(it.getVersion()).isEqualTo(2);
+				}) //
+				.verifyComplete();
+	}
+
+	@Test // DATAES-488
+	public void updateShouldErrorNonExistingDocumentWhenNotUpserted() {
+
+		String id = UUID.randomUUID().toString();
+		UpdateRequest request = new UpdateRequest(INDEX_I, TYPE_I, id) //
+				.doc(DOC_SOURCE);
+
+		client.update(request) //
+				.as(StepVerifier::create) //
+				.consumeErrorWith(error -> {
+					assertThat(error).isInstanceOf(ElasticsearchStatusException.class);
+				}) //
+				.verify();
+	}
+
+	@Test // DATAES-488
+	public void deleteShouldRemoveExistingDocument() {
+
+		String id = addSourceDocument().ofType(TYPE_I).to(INDEX_I);
+
+		DeleteRequest request = new DeleteRequest(INDEX_I, TYPE_I, id);
+
+		client.delete(request) //
+				.as(StepVerifier::create) //
+				.consumeNextWith(it -> {
+					assertThat(it.status()).isEqualTo(RestStatus.OK);
+				}) //
+				.verifyComplete();
+	}
+
+	@Test // DATAES-488
+	public void deleteShouldReturnNotFoundForNonExistingDocument() {
+
+		addSourceDocument().ofType(TYPE_I).to(INDEX_I);
+
+		DeleteRequest request = new DeleteRequest(INDEX_I, TYPE_I, "this-one-does-not-exist");
+
+		client.delete(request) //
+				.as(StepVerifier::create) //
+				.consumeNextWith(it -> {
+					assertThat(it.status()).isEqualTo(RestStatus.NOT_FOUND);
+				}) //
+				.verifyComplete();
+	}
+
+	@Test // DATAES-488
+	public void searchShouldFindExistingDocuments() {
+
+		addSourceDocument().ofType(TYPE_I).to(INDEX_I);
+		addSourceDocument().ofType(TYPE_I).to(INDEX_I);
+
+		SearchRequest request = new SearchRequest(INDEX_I).types(TYPE_I) //
+				.source(new SearchSourceBuilder().query(QueryBuilders.matchAllQuery()));
+
+		client.search(request) //
+				.as(StepVerifier::create) //
+				.expectNextCount(2) //
+				.verifyComplete();
+	}
+
+	@Test // DATAES-488
+	public void searchShouldCompleteIfNothingFound() throws IOException {
+
+		syncClient.indices().create(new CreateIndexRequest(INDEX_I));
+
+		SearchRequest request = new SearchRequest(INDEX_I).types(TYPE_I) //
+				.source(new SearchSourceBuilder().query(QueryBuilders.matchAllQuery()));
+
+		client.search(request) //
+				.as(StepVerifier::create) //
+				.verifyComplete();
+	}
+
+	AddToIndexOfType addSourceDocument() {
+		return add(DOC_SOURCE);
+	}
+
+	AddToIndexOfType add(Map source) {
+		return new AddDocument(source);
+	}
+
+	IndexRequest indexRequest(Map source, String index, String type) {
+
+		return new IndexRequest(index, type) //
+				.id(UUID.randomUUID().toString()) //
+				.source(source) //
+				.setRefreshPolicy(RefreshPolicy.IMMEDIATE) //
+				.create(true);
+	}
+
+	String doIndex(Map source, String index, String type) {
+
+		try {
+			return syncClient.index(indexRequest(source, index, type)).getId();
+		} catch (IOException e) {
+			throw new RuntimeException(e);
+		}
+	}
+
+	interface AddToIndexOfType extends AddToIndex {
+		AddToIndex ofType(String type);
+	}
+
+	interface AddToIndex {
+		String to(String index);
+	}
+
+	class AddDocument implements AddToIndexOfType {
+
+		Map source;
+		@Nullable String type;
+
+		AddDocument(Map source) {
+			this.source = source;
+		}
+
+		@Override
+		public AddToIndex ofType(String type) {
+
+			this.type = type;
+			return this;
+		}
+
+		@Override
+		public String to(String index) {
+			return doIndex(new LinkedHashMap(source), index, type);
+		}
+	}
+
+}

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

@@ -0,0 +1,554 @@
+/*
+ * Copyright 2018 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
+ *
+ *      http://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.data.elasticsearch.client.reactive;
+
+import static org.assertj.core.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.*;
+import static org.springframework.data.elasticsearch.client.reactive.ReactiveMockClientTestsUtils.WebClientProvider.Receive.*;
+
+import reactor.test.StepVerifier;
+
+import java.util.Collections;
+import java.util.Map;
+
+import org.elasticsearch.ElasticsearchStatusException;
+import org.elasticsearch.action.DocWriteResponse.Result;
+import org.elasticsearch.action.delete.DeleteRequest;
+import org.elasticsearch.action.get.GetRequest;
+import org.elasticsearch.action.get.MultiGetRequest;
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.action.search.SearchRequest;
+import org.elasticsearch.action.update.UpdateRequest;
+import org.elasticsearch.common.xcontent.XContentType;
+import org.junit.Before;
+import org.junit.Test;
+import org.reactivestreams.Publisher;
+import org.springframework.data.elasticsearch.client.reactive.ReactiveMockClientTestsUtils.MockDelegatingElasticsearchHostProvider;
+import org.springframework.data.elasticsearch.client.reactive.ReactiveMockClientTestsUtils.WebClientProvider.Receive;
+import org.springframework.http.HttpMethod;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.MediaType;
+
+/**
+ * @author Christoph Strobl
+ * @currentRead Golden Fool - Robin Hobb
+ */
+public class ReactiveElasticsearchClientUnitTests {
+
+	static final String HOST = ":9200";
+
+	MockDelegatingElasticsearchHostProvider<HostProvider> hostProvider;
+	ReactiveElasticsearchClient client;
+
+	@Before
+	public void setUp() {
+
+		hostProvider = ReactiveMockClientTestsUtils.provider(HOST).withActiveDefaultHost(HOST);
+		client = new DefaultReactiveElasticsearchClient(hostProvider);
+	}
+
+	// --> PING
+
+	@Test
+	public void pingShouldHitMainEndpoint() {
+
+		hostProvider.when(HOST) //
+				.receive(Receive::ok);
+
+		client.ping() //
+				.then() //
+				.as(StepVerifier::create) //
+				.verifyComplete();
+
+		hostProvider.when(HOST).exchange(requestBodyUriSpec -> {
+			verify(requestBodyUriSpec).uri(eq("/"), any(Map.class));
+		});
+	}
+
+	@Test // DATAES-488
+	public void pingShouldReturnTrueOnHttp200() {
+
+		hostProvider.when(HOST) //
+				.receive(Receive::ok);
+
+		client.ping() //
+				.as(StepVerifier::create) //
+				.expectNext(true) //
+				.verifyComplete();
+	}
+
+	@Test // DATAES-488
+	public void pingShouldReturnFalseOnNonHttp200() {
+
+		hostProvider.when(HOST) //
+				.receive(Receive::error);
+
+		client.ping() //
+				.as(StepVerifier::create) //
+				.expectNext(false) //
+				.verifyComplete();
+	}
+
+	// --> INFO
+
+	@Test
+	public void infoShouldHitMainEndpoint() {
+
+		hostProvider.when(HOST) //
+				.receiveInfo();
+
+		client.info() //
+				.then() //
+				.as(StepVerifier::create) //
+				.verifyComplete();
+
+		hostProvider.when(HOST).exchange(requestBodyUriSpec -> {
+			verify(requestBodyUriSpec).uri(eq("/"), any(Map.class));
+		});
+	}
+
+	@Test // DATAES-488
+	public void infoShouldReturnResponseCorrectly() {
+
+		hostProvider.when(HOST) //
+				.receiveInfo();
+
+		client.info() //
+				.as(StepVerifier::create) //
+				.consumeNextWith(mainResponse -> {}) //
+				.verifyComplete();
+	}
+
+	// --> GET
+
+	@Test // DATAES-488
+	public void getShouldHitGetEndpoint() {
+
+		hostProvider.when(HOST).receive(clientResponse -> {
+			when(clientResponse.statusCode()).thenReturn(HttpStatus.ACCEPTED, HttpStatus.NOT_FOUND);
+		});
+
+		hostProvider.when(HOST) //
+				.receiveGetByIdNotFound();
+
+		client.get(new GetRequest("twitter").id("1")) //
+				.then() //
+				.as(StepVerifier::create) //
+				.verifyComplete();
+
+		verify(hostProvider.client(HOST)).method(HttpMethod.GET);
+		hostProvider.when(HOST).exchange(requestBodyUriSpec -> {
+			verify(requestBodyUriSpec).uri(eq("/twitter/_all/1"), any(Map.class));
+		});
+	}
+
+	@Test // DATAES-488
+	public void getShouldReturnExistingDocument() {
+
+		hostProvider.when(HOST) //
+				.receiveGetById();
+
+		client.get(new GetRequest("twitter").id("1")) //
+				.as(StepVerifier::create) //
+				.consumeNextWith(result -> {
+
+					assertThat(result.isExists()).isTrue();
+					assertThat(result.getIndex()).isEqualTo("twitter");
+					assertThat(result.getId()).isEqualTo("1");
+					assertThat(result.getSource()) //
+							.containsEntry("user", "kimchy") //
+							.containsEntry("message", "Trying out Elasticsearch, so far so good?") //
+							.containsKey("post_date");
+				}) //
+				.verifyComplete();
+	}
+
+	@Test // DATAES-488
+	public void getShouldReturnEmptyForNonExisting() {
+
+		hostProvider.when(HOST) //
+				.receiveGetByIdNotFound();
+
+		client.get(new GetRequest("twitter").id("1")) //
+				.as(StepVerifier::create) //
+				.verifyComplete();
+	}
+
+	// --> MGET
+
+	@Test // DATAES-488
+	public void multiGetShouldHitMGetEndpoint() {
+
+		hostProvider.when(HOST) //
+				.receiveJsonFromFile("multi-get-ok-2-hits");
+
+		client.multiGet(new MultiGetRequest().add("twitter", "_doc", "1").add("twitter", "_doc", "2")) //
+				.then() //
+				.as(StepVerifier::create) //
+				.verifyComplete();
+
+		verify(hostProvider.client(HOST)).method(HttpMethod.POST);
+
+		hostProvider.when(HOST).exchange(requestBodyUriSpec -> {
+
+			verify(requestBodyUriSpec).uri(eq("/_mget"), any(Map.class));
+			verify(requestBodyUriSpec).body(any(Publisher.class), any(Class.class));
+		});
+	}
+
+	@Test // DATAES-488
+	public void multiGetShouldReturnExistingDocuments() {
+
+		hostProvider.when(HOST) //
+				.receiveJsonFromFile("multi-get-ok-2-hits");
+
+		client.multiGet(new MultiGetRequest().add("twitter", "_doc", "1").add("twitter", "_doc", "2")) //
+				.as(StepVerifier::create) //
+				.consumeNextWith(result -> {
+
+					assertThat(result.isExists()).isTrue();
+					assertThat(result.getIndex()).isEqualTo("twitter");
+					assertThat(result.getId()).isEqualTo("1");
+					assertThat(result.getSource()) //
+							.containsEntry("user", "kimchy") //
+							.containsEntry("message", "Trying out Elasticsearch, so far so good?") //
+							.containsKey("post_date");
+				}) //
+				.consumeNextWith(result -> {
+
+					assertThat(result.isExists()).isTrue();
+					assertThat(result.getIndex()).isEqualTo("twitter");
+					assertThat(result.getId()).isEqualTo("2");
+					assertThat(result.getSource()) //
+							.containsEntry("user", "kimchy") //
+							.containsEntry("message", "Another tweet, will it be indexed?") //
+							.containsKey("post_date");
+				}) //
+				.verifyComplete();
+	}
+
+	@Test // DATAES-488
+	public void multiGetShouldWorkForNonExistingDocuments() {
+
+		hostProvider.when(HOST) //
+				.receiveJsonFromFile("multi-get-ok-2-hits-1-unavailable");
+
+		client.multiGet(new MultiGetRequest().add("twitter", "_doc", "1").add("twitter", "_doc", "2")) //
+				.as(StepVerifier::create) //
+				.consumeNextWith(result -> {
+
+					assertThat(result.isExists()).isTrue();
+					assertThat(result.getIndex()).isEqualTo("twitter");
+					assertThat(result.getId()).isEqualTo("1");
+					assertThat(result.getSource()) //
+							.containsEntry("user", "kimchy") //
+							.containsEntry("message", "Trying out Elasticsearch, so far so good?") //
+							.containsKey("post_date");
+				}) //
+				.consumeNextWith(result -> {
+
+					assertThat(result.isExists()).isTrue();
+					assertThat(result.getIndex()).isEqualTo("twitter");
+					assertThat(result.getId()).isEqualTo("3");
+					assertThat(result.getSource()) //
+							.containsEntry("user", "elastic") //
+							.containsEntry("message", "Building the site, should be kewl") //
+							.containsKey("post_date");
+				}) //
+				.verifyComplete();
+	}
+
+	// --> EXISTS
+
+	@Test // DATAES-488
+	public void existsShouldHitGetEndpoint() {
+
+		hostProvider.when(HOST) //
+				.receiveGetById();
+
+		client.exists(new GetRequest("twitter").id("1")) //
+				.then() //
+				.as(StepVerifier::create) //
+				.verifyComplete();
+
+		verify(hostProvider.client(HOST)).method(HttpMethod.HEAD);
+
+		hostProvider.when(HOST).exchange(requestBodyUriSpec -> {
+			verify(requestBodyUriSpec).uri(eq("/twitter/_all/1"), any(Map.class));
+		});
+	}
+
+	@Test // DATAES-488
+	public void existsShouldReturnTrueIfExists() {
+
+		hostProvider.when(HOST) //
+				.receiveGetById();
+
+		client.exists(new GetRequest("twitter").id("1")) //
+				.as(StepVerifier::create) //
+				.expectNext(true).verifyComplete();
+	}
+
+	@Test // DATAES-488
+	public void existsShouldReturnFalseIfNotExists() {
+
+		hostProvider.when(HOST) //
+				.receiveGetByIdNotFound();
+
+		client.exists(new GetRequest("twitter").id("1")) //
+				.as(StepVerifier::create) //
+				.expectNext(false).verifyComplete();
+	}
+
+	// --> INDEX
+
+	@Test // DATAES-488
+	public void indexNewShouldHitCreateEndpoint() {
+
+		hostProvider.when(HOST) //
+				.receiveIndexCreated();
+
+		client.index(new IndexRequest("twitter").id("10").create(true).source(" { foo : \"bar\" }", XContentType.JSON))
+				.then() //
+				.as(StepVerifier::create) //
+				.verifyComplete();
+
+		verify(hostProvider.client(HOST)).method(HttpMethod.PUT);
+		hostProvider.when(HOST).exchange(requestBodyUriSpec -> {
+
+			verify(requestBodyUriSpec).uri(eq("/twitter/10/_create"), any(Map.class));
+			verify(requestBodyUriSpec).contentType(MediaType.APPLICATION_JSON);
+		});
+	}
+
+	@Test // DATAES-488
+	public void indexExistingShouldHitEndpointCorrectly() {
+
+		hostProvider.when(HOST) //
+				.receiveIndexUpdated();
+
+		client.index(new IndexRequest("twitter").id("10").source(" { foo : \"bar\" }", XContentType.JSON)).then() //
+				.as(StepVerifier::create) //
+				.verifyComplete();
+
+		verify(hostProvider.client(HOST)).method(HttpMethod.PUT);
+		hostProvider.when(HOST).exchange(requestBodyUriSpec -> {
+
+			verify(requestBodyUriSpec).uri(eq("/twitter/10"), any(Map.class));
+			verify(requestBodyUriSpec).contentType(MediaType.APPLICATION_JSON);
+		});
+	}
+
+	@Test // DATAES-488
+	public void indexShouldReturnCreatedWhenNewDocumentIndexed() {
+
+		hostProvider.when(HOST) //
+				.receiveIndexCreated();
+
+		client.index(new IndexRequest("twitter").id("10").create(true).source(" { foo : \"bar\" }", XContentType.JSON))
+				.as(StepVerifier::create) //
+				.consumeNextWith(response -> {
+
+					assertThat(response.getId()).isEqualTo("10");
+					assertThat(response.getIndex()).isEqualTo("twitter");
+					assertThat(response.getResult()).isEqualTo(Result.CREATED);
+				}) //
+				.verifyComplete();
+	}
+
+	@Test // DATAES-488
+	public void indexShouldReturnUpdatedWhenExistingDocumentIndexed() {
+
+		hostProvider.when(HOST) //
+				.receiveIndexUpdated();
+
+		client.index(new IndexRequest("twitter").id("1").source(" { foo : \"bar\" }", XContentType.JSON))
+				.as(StepVerifier::create) //
+				.consumeNextWith(response -> {
+
+					assertThat(response.getId()).isEqualTo("1");
+					assertThat(response.getIndex()).isEqualTo("twitter");
+					assertThat(response.getResult()).isEqualTo(Result.UPDATED);
+				}) //
+				.verifyComplete();
+	}
+
+	// --> UPDATE
+
+	@Test // DATAES-488
+	public void updateShouldHitEndpointCorrectly() {
+
+		hostProvider.when(HOST) //
+				.receiveUpdateOk();
+
+		client.update(new UpdateRequest("twitter", "doc", "1").doc(Collections.singletonMap("user", "cstrobl"))).then() //
+				.as(StepVerifier::create) //
+				.verifyComplete();
+
+		verify(hostProvider.client(HOST)).method(HttpMethod.POST);
+		hostProvider.when(HOST).exchange(requestBodyUriSpec -> {
+
+			verify(requestBodyUriSpec).uri(eq("/twitter/doc/1/_update"), any(Map.class));
+			verify(requestBodyUriSpec).contentType(MediaType.APPLICATION_JSON);
+		});
+	}
+
+	@Test // DATAES-488
+	public void updateShouldEmitResponseCorrectly() {
+
+		hostProvider.when(HOST) //
+				.receiveUpdateOk();
+
+		client.update(new UpdateRequest("twitter", "doc", "1").doc(Collections.singletonMap("user", "cstrobl")))
+				.as(StepVerifier::create) //
+				.consumeNextWith(updateResponse -> {
+
+					assertThat(updateResponse.getResult()).isEqualTo(Result.UPDATED);
+					assertThat(updateResponse.getVersion()).isEqualTo(2);
+					assertThat(updateResponse.getId()).isEqualTo("1");
+					assertThat(updateResponse.getIndex()).isEqualTo("twitter");
+				}) //
+				.verifyComplete();
+	}
+
+	@Test // DATAES-488
+	public void updateShouldEmitErrorWhenNotFound() {
+
+		hostProvider.when(HOST) //
+				.updateFail();
+
+		client.update(new UpdateRequest("twitter", "doc", "1").doc(Collections.singletonMap("user", "cstrobl")))
+				.as(StepVerifier::create) //
+				.expectError(ElasticsearchStatusException.class) //
+				.verify();
+	}
+
+	// --> DELETE
+
+	@Test // DATAES-488
+	public void deleteShouldHitEndpointCorrectly() {
+
+		hostProvider.when(HOST) //
+				.receiveDeleteOk();
+
+		client.delete(new DeleteRequest("twitter", "doc", "1")).then() //
+				.as(StepVerifier::create) //
+				.verifyComplete();
+
+		verify(hostProvider.client(HOST)).method(HttpMethod.DELETE);
+		hostProvider.when(HOST).exchange(requestBodyUriSpec -> {
+			verify(requestBodyUriSpec).uri(eq("/twitter/doc/1"), any(Map.class));
+		});
+	}
+
+	@Test // DATAES-488
+	public void deleteShouldEmitResponseCorrectly() {
+
+		hostProvider.when(HOST) //
+				.receiveDeleteOk();
+
+		client.delete(new DeleteRequest("twitter", "doc", "1")) //
+				.as(StepVerifier::create) //
+				.consumeNextWith(deleteResponse -> {
+
+					assertThat(deleteResponse.getResult()).isEqualTo(Result.DELETED);
+					assertThat(deleteResponse.getVersion()).isEqualTo(1);
+					assertThat(deleteResponse.getId()).isEqualTo("1");
+					assertThat(deleteResponse.getIndex()).isEqualTo("twitter");
+				}) //
+				.verifyComplete();
+	}
+
+	// --> SEARCH
+
+	@Test // DATAES-488
+	public void searchShouldHitSearchEndpoint() {
+
+		hostProvider.when(HOST) //
+				.receiveSearchOk();
+
+		client.search(new SearchRequest("twitter")).as(StepVerifier::create).verifyComplete();
+
+		verify(hostProvider.client(HOST)).method(HttpMethod.POST);
+		hostProvider.when(HOST).exchange(requestBodyUriSpec -> {
+			verify(requestBodyUriSpec).uri(eq("/twitter/_search"), any(Map.class));
+		});
+	}
+
+	@Test // DATAES-488
+	public void searchShouldReturnSingleResultCorrectly() {
+
+		hostProvider.when(HOST) //
+				.receive(Receive::json) //
+				.body(fromPath("search-ok-single-hit"));
+
+		client.search(new SearchRequest("twitter")) //
+				.as(StepVerifier::create) //
+				.consumeNextWith(hit -> {
+
+					assertThat(hit.getId()).isEqualTo("2");
+					assertThat(hit.getIndex()).isEqualTo("twitter");
+					assertThat(hit.getSourceAsMap()) //
+							.containsEntry("user", "kimchy") //
+							.containsEntry("message", "Another tweet, will it be indexed?") //
+							.containsKey("post_date");
+				}).verifyComplete();
+	}
+
+	@Test // DATAES-488
+	public void searchShouldReturnMultipleResultsCorrectly() {
+
+		hostProvider.when(HOST) //
+				.receive(Receive::json) //
+				.body(fromPath("search-ok-multiple-hits"));
+
+		client.search(new SearchRequest("twitter")) //
+				.as(StepVerifier::create) //
+				.consumeNextWith(hit -> {
+
+					assertThat(hit.getId()).isEqualTo("2");
+					assertThat(hit.getIndex()).isEqualTo("twitter");
+					assertThat(hit.getSourceAsMap()) //
+							.containsEntry("user", "kimchy") //
+							.containsEntry("message", "Another tweet, will it be indexed?") //
+							.containsKey("post_date");
+				}) //
+				.consumeNextWith(hit -> {
+
+					assertThat(hit.getId()).isEqualTo("1");
+					assertThat(hit.getIndex()).isEqualTo("twitter");
+					assertThat(hit.getSourceAsMap()) //
+							.containsEntry("user", "kimchy") //
+							.containsEntry("message", "Trying out Elasticsearch, so far so good?") //
+							.containsKey("post_date");
+				}).verifyComplete();
+	}
+
+	@Test // DATAES-488
+	public void searchShouldReturnEmptyFluxIfNothingFound() {
+
+		hostProvider.when(HOST) //
+				.receiveSearchOk();
+
+		client.search(new SearchRequest("twitter")) //
+				.as(StepVerifier::create) //
+				.verifyComplete();
+	}
+
+}

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

@@ -0,0 +1,432 @@
+/*
+ * Copyright 2018 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
+ *
+ *      http://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.data.elasticsearch.client.reactive;
+
+import static org.mockito.Mockito.*;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+
+import org.mockito.Mockito;
+import org.springframework.core.io.ClassPathResource;
+import org.springframework.core.io.Resource;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DefaultDataBufferFactory;
+import org.springframework.data.elasticsearch.client.ElasticsearchHost;
+import org.springframework.data.elasticsearch.client.reactive.ReactiveMockClientTestsUtils.WebClientProvider.Send;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.MediaType;
+import org.springframework.lang.Nullable;
+import org.springframework.util.StreamUtils;
+import org.springframework.util.StringUtils;
+import org.springframework.web.reactive.function.client.ClientResponse;
+import org.springframework.web.reactive.function.client.WebClient;
+import org.springframework.web.reactive.function.client.WebClient.RequestBodyUriSpec;
+import org.springframework.web.reactive.function.client.WebClient.RequestHeadersUriSpec;
+
+/**
+ * @author Christoph Strobl
+ * @since 2018/10
+ */
+public class ReactiveMockClientTestsUtils {
+
+	public static MockDelegatingElasticsearchHostProvider<SingleNodeHostProvider> single(String host) {
+		return provider(host);
+	}
+
+	public static MockDelegatingElasticsearchHostProvider<MultiNodeHostProvider> multi(String... hosts) {
+		return provider(hosts);
+	}
+
+	public static <T extends HostProvider> MockDelegatingElasticsearchHostProvider<T> provider(String... hosts) {
+
+		WebClientProvider clientProvider = new WebClientProvider();
+		ErrorCollector errorCollector = new ErrorCollector();
+		HostProvider delegate = null;
+
+		if (hosts.length == 1) {
+
+			delegate = new SingleNodeHostProvider(HttpHeaders.EMPTY, errorCollector, hosts[0]) {
+				@Override // hook in there to modify result
+				public WebClient createWebClient(String host, HttpHeaders headers) {
+					return clientProvider.get(host);
+				}
+			};
+		} else {
+
+			delegate = new MultiNodeHostProvider(HttpHeaders.EMPTY, errorCollector, hosts) {
+				@Override // hook in there to modify result
+				public WebClient createWebClient(String host, HttpHeaders headers) {
+					return clientProvider.get(host);
+				}
+			};
+		}
+
+		return new MockDelegatingElasticsearchHostProvider(HttpHeaders.EMPTY, clientProvider, errorCollector, delegate,
+				null);
+
+	}
+
+	public static class ErrorCollector implements Consumer<Throwable> {
+
+		List<Throwable> errors = new CopyOnWriteArrayList<>();
+
+		@Override
+		public void accept(Throwable throwable) {
+			errors.add(throwable);
+		}
+
+		List<Throwable> captured() {
+			return Collections.unmodifiableList(errors);
+		}
+	}
+
+	public static class MockDelegatingElasticsearchHostProvider<T extends HostProvider> implements HostProvider {
+
+		private final T delegate;
+		private final WebClientProvider clientProvider;
+		private final ErrorCollector errorCollector;
+		private @Nullable String activeDefaultHost;
+
+		public MockDelegatingElasticsearchHostProvider(HttpHeaders httpHeaders, WebClientProvider clientProvider,
+				ErrorCollector errorCollector, T delegate, String activeDefaultHost) {
+
+			this.errorCollector = errorCollector;
+			this.clientProvider = clientProvider;
+			this.delegate = delegate;
+			this.activeDefaultHost = activeDefaultHost;
+		}
+
+		public Mono<String> lookupActiveHost() {
+			return delegate.lookupActiveHost();
+		}
+
+		public Mono<String> lookupActiveHost(VerificationMode verificationMode) {
+
+			if (StringUtils.hasText(activeDefaultHost)) {
+				return Mono.just(activeDefaultHost);
+			}
+
+			return delegate.lookupActiveHost(verificationMode);
+		}
+
+		public Mono<WebClient> getActive() {
+			return delegate.getActive();
+		}
+
+		public Mono<WebClient> getActive(VerificationMode verificationMode) {
+			return delegate.getActive(verificationMode);
+		}
+
+		public Mono<WebClient> getActive(VerificationMode verificationMode, HttpHeaders headers) {
+			return delegate.getActive(verificationMode, headers);
+		}
+
+		public WebClient createWebClient(String host, HttpHeaders headers) {
+			return delegate.createWebClient(host, headers);
+		}
+
+		@Override
+		public Mono<ClusterInformation> clusterInfo() {
+
+			if (StringUtils.hasText(activeDefaultHost)) {
+				return Mono.just(new ClusterInformation(Collections.singleton(ElasticsearchHost.online(activeDefaultHost))));
+			}
+
+			return delegate.clusterInfo();
+		}
+
+		@Override
+		public HttpHeaders getDefaultHeaders() {
+			return delegate.getDefaultHeaders();
+		}
+
+		@Override
+		public HostProvider withDefaultHeaders(HttpHeaders headers) {
+			throw new UnsupportedOperationException();
+		}
+
+		public Send when(String host) {
+			return clientProvider.when(host);
+		}
+
+		public WebClient client(String host) {
+			return clientProvider.when(host).client();
+		}
+
+		public List<Throwable> errors() {
+			return errorCollector.captured();
+		}
+
+		public T getDelegate() {
+			return delegate;
+		}
+
+		@Override
+		public HostProvider withErrorListener(Consumer<Throwable> errorListener) {
+			throw new UnsupportedOperationException();
+		}
+
+		public MockDelegatingElasticsearchHostProvider<T> withActiveDefaultHost(String host) {
+			return new MockDelegatingElasticsearchHostProvider(HttpHeaders.EMPTY, clientProvider, errorCollector, delegate,
+					host);
+		}
+	}
+
+	public static class WebClientProvider {
+
+		private final Object lock = new Object();
+
+		private Map<String, WebClient> clientMap;
+		private Map<String, RequestHeadersUriSpec> headersUriSpecMap;
+		private Map<String, RequestBodyUriSpec> bodyUriSpecMap;
+		private Map<String, ClientResponse> responseMap;
+
+		public WebClientProvider() {
+
+			this.clientMap = new LinkedHashMap<>();
+			this.headersUriSpecMap = new LinkedHashMap<>();
+			this.bodyUriSpecMap = new LinkedHashMap<>();
+			this.responseMap = new LinkedHashMap<>();
+		}
+
+		public WebClient get(String host) {
+
+			synchronized (lock) {
+
+				return clientMap.computeIfAbsent(host, key -> {
+
+					WebClient webClient = mock(WebClient.class);
+
+					RequestHeadersUriSpec headersUriSpec = mock(RequestHeadersUriSpec.class);
+					Mockito.when(webClient.get()).thenReturn(headersUriSpec);
+					Mockito.when(webClient.head()).thenReturn(headersUriSpec);
+
+					Mockito.when(headersUriSpec.uri(any(String.class))).thenReturn(headersUriSpec);
+					Mockito.when(headersUriSpec.uri(any(), any(Map.class))).thenReturn(headersUriSpec);
+					Mockito.when(headersUriSpec.headers(any(Consumer.class))).thenReturn(headersUriSpec);
+
+					RequestBodyUriSpec bodyUriSpec = mock(RequestBodyUriSpec.class);
+					Mockito.when(webClient.method(any())).thenReturn(bodyUriSpec);
+					Mockito.when(bodyUriSpec.body(any())).thenReturn(headersUriSpec);
+					Mockito.when(bodyUriSpec.uri(any(), any(Map.class))).thenReturn(bodyUriSpec);
+					Mockito.when(bodyUriSpec.headers(any(Consumer.class))).thenReturn(bodyUriSpec);
+
+					ClientResponse response = mock(ClientResponse.class);
+					Mockito.when(headersUriSpec.exchange()).thenReturn(Mono.just(response));
+					Mockito.when(bodyUriSpec.exchange()).thenReturn(Mono.just(response));
+					Mockito.when(response.statusCode()).thenReturn(HttpStatus.ACCEPTED);
+
+					headersUriSpecMap.putIfAbsent(host, headersUriSpec);
+					bodyUriSpecMap.putIfAbsent(host, bodyUriSpec);
+					responseMap.putIfAbsent(host, response);
+
+					return webClient;
+				});
+			}
+		}
+
+		public Send when(String host) {
+			return new CallbackImpl(get(host), headersUriSpecMap.get(host), bodyUriSpecMap.get(host), responseMap.get(host));
+		}
+
+		public interface Client {
+			WebClient client();
+		}
+
+		public interface Send extends Receive, Client {
+
+			Receive get(Consumer<RequestHeadersUriSpec> headerSpec);
+
+			Receive exchange(Consumer<RequestBodyUriSpec> bodySpec);
+
+			default Receive receiveJsonFromFile(String file) {
+
+				return receive(Receive::json) //
+						.body(Receive.fromPath(file));
+			}
+
+			default Receive receiveInfo() {
+
+				return receiveJsonFromFile("info") //
+						.receive(Receive::ok);
+
+			}
+
+			default Receive receiveIndexCreated() {
+
+				return receiveJsonFromFile("index-ok-created") //
+						.receive(Receive::ok);
+			}
+
+			default Receive receiveIndexUpdated() {
+
+				return receiveJsonFromFile("index-ok-updated") //
+						.receive(Receive::ok);
+			}
+
+			default Receive receiveSearchOk() {
+
+				return receiveJsonFromFile("search-ok-no-hits") //
+						.receive(Receive::ok);
+			}
+
+			default Receive receiveGetByIdNotFound() {
+
+				return receiveJsonFromFile("get-by-id-no-hit") //
+						.receive(response -> {
+							Mockito.when(response.statusCode()).thenReturn(HttpStatus.ACCEPTED, HttpStatus.NOT_FOUND);
+						});
+			}
+
+			default Receive receiveGetById() {
+
+				return receiveJsonFromFile("get-by-id-ok") //
+						.receive(Receive::ok);
+			}
+
+			default Receive receiveUpdateOk() {
+
+				return receiveJsonFromFile("update-ok-updated") //
+						.receive(Receive::ok);
+			}
+
+			default Receive receiveDeleteOk() {
+
+				return receiveJsonFromFile("update-ok-deleted") //
+						.receive(Receive::ok);
+			}
+
+			default Receive updateFail() {
+
+				return receiveJsonFromFile("update-error-not-found") //
+						.receive(response -> {
+							Mockito.when(response.statusCode()).thenReturn(HttpStatus.ACCEPTED, HttpStatus.NOT_FOUND);
+						});
+			}
+
+		}
+
+		public interface Receive {
+
+			Receive receive(Consumer<ClientResponse> response);
+
+			default Receive body(String json) {
+				return body(() -> json.getBytes(StandardCharsets.UTF_8));
+			}
+
+			default Receive body(Supplier<byte[]> json) {
+				return body(new DefaultDataBufferFactory().wrap(json.get()));
+			}
+
+			default Receive body(Resource resource) {
+
+				return body(() -> {
+					try {
+						return StreamUtils.copyToByteArray(resource.getInputStream());
+					} catch (IOException e) {
+						throw new RuntimeException(e);
+					}
+				});
+			}
+
+			default Receive body(DataBuffer dataBuffer) {
+				return receive(response -> Mockito.when(response.body(any())).thenReturn(Flux.just(dataBuffer)));
+			}
+
+			static void ok(ClientResponse response) {
+				Mockito.when(response.statusCode()).thenReturn(HttpStatus.ACCEPTED);
+			}
+
+			static void error(ClientResponse response) {
+				Mockito.when(response.statusCode()).thenReturn(HttpStatus.INTERNAL_SERVER_ERROR);
+			}
+
+			static void notFound(ClientResponse response) {
+				Mockito.when(response.statusCode()).thenReturn(HttpStatus.NOT_FOUND);
+			}
+
+			static void json(ClientResponse response) {
+
+				ClientResponse.Headers headers = Mockito.mock(ClientResponse.Headers.class);
+				Mockito.when(headers.contentType()).thenReturn(Optional.of(MediaType.APPLICATION_JSON));
+
+				Mockito.when(response.headers()).thenReturn(headers);
+			}
+
+			static Resource fromPath(String filename) {
+				return new ClassPathResource("/org/springframework/data/elasticsearch/client/" + filename + ".json");
+			}
+		}
+
+		class CallbackImpl implements Send, Receive {
+
+			WebClient client;
+			RequestHeadersUriSpec headersUriSpec;
+			RequestBodyUriSpec bodyUriSpec;
+			ClientResponse responseDelegate;
+
+			public CallbackImpl(WebClient client, RequestHeadersUriSpec headersUriSpec, RequestBodyUriSpec bodyUriSpec,
+					ClientResponse responseDelegate) {
+
+				this.client = client;
+				this.headersUriSpec = headersUriSpec;
+				this.bodyUriSpec = bodyUriSpec;
+				this.responseDelegate = responseDelegate;
+			}
+
+			@Override
+			public Receive get(Consumer<RequestHeadersUriSpec> uriSpec) {
+
+				uriSpec.accept(headersUriSpec);
+				return this;
+			}
+
+			@Override
+			public Receive exchange(Consumer<RequestBodyUriSpec> bodySpec) {
+
+				bodySpec.accept(this.bodyUriSpec);
+				return this;
+			}
+
+			@Override
+			public Receive receive(Consumer<ClientResponse> response) {
+
+				response.accept(responseDelegate);
+				return this;
+			}
+
+			@Override
+			public WebClient client() {
+				return client;
+			}
+
+		}
+	}
+}

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

@@ -0,0 +1,76 @@
+/*
+ * Copyright 2018 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
+ *
+ *      http://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.data.elasticsearch.client.reactive;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.springframework.data.elasticsearch.client.NoReachableHostException;
+import reactor.test.StepVerifier;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.data.elasticsearch.client.ElasticsearchHost;
+import org.springframework.data.elasticsearch.client.ElasticsearchHost.State;
+import org.springframework.data.elasticsearch.client.reactive.ReactiveMockClientTestsUtils.MockDelegatingElasticsearchHostProvider;
+import org.springframework.data.elasticsearch.client.reactive.ReactiveMockClientTestsUtils.WebClientProvider.Receive;
+
+/**
+ * @author Christoph Strobl
+ * @currentRead Golden Fool - Robin Hobb
+ */
+public class SingleNodeHostProviderUnitTests {
+
+	static final String HOST_1 = ":9200";
+
+	MockDelegatingElasticsearchHostProvider<SingleNodeHostProvider> mock;
+	SingleNodeHostProvider provider;
+
+	@Before
+	public void setUp() {
+
+		mock = ReactiveMockClientTestsUtils.single(HOST_1);
+		provider = mock.getDelegate();
+	}
+
+	@Test // DATAES-488
+	public void refreshHostStateShouldUpdateNodeStateCorrectly() {
+
+		mock.when(HOST_1).receive(Receive::error);
+
+		provider.clusterInfo().as(StepVerifier::create).expectNextCount(1).verifyComplete();
+
+		assertThat(provider.getCachedHostState()).extracting(ElasticsearchHost::getState).isEqualTo(State.OFFLINE);
+	}
+
+	@Test // DATAES-488
+	public void getActiveReturnsFirstActiveHost() {
+
+		mock.when(HOST_1).receive(Receive::ok);
+
+		provider.clusterInfo().as(StepVerifier::create).expectNextCount(1).verifyComplete();
+
+		assertThat(provider.getCachedHostState()).extracting(ElasticsearchHost::getState).isEqualTo(State.ONLINE);
+	}
+
+	@Test // DATAES-488
+	public void getActiveErrorsWhenNoActiveHostFound() {
+
+		mock.when(HOST_1).receive(Receive::error);
+
+		provider.getActive().as(StepVerifier::create).expectError(NoReachableHostException.class);
+	}
+}

src/test/java/org/springframework/data/elasticsearch/core/ReactiveElasticsearchTemplateTests.java

@@ -0,0 +1,148 @@
+/*
+ * Copyright 2018 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
+ *
+ *      http://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.data.elasticsearch.core;
+
+import static org.apache.commons.lang.RandomStringUtils.*;
+import static org.assertj.core.api.Assertions.*;
+
+import reactor.test.StepVerifier;
+
+import java.util.List;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.data.elasticsearch.TestUtils;
+import org.springframework.data.elasticsearch.core.query.Criteria;
+import org.springframework.data.elasticsearch.core.query.CriteriaQuery;
+import org.springframework.data.elasticsearch.core.query.IndexQuery;
+import org.springframework.data.elasticsearch.core.query.IndexQueryBuilder;
+import org.springframework.data.elasticsearch.entities.SampleEntity;
+
+/**
+ * @author Christoph Strobl
+ * @currentRead Golden Fool - Robin Hobb
+ */
+public class ReactiveElasticsearchTemplateTests {
+
+	private ElasticsearchRestTemplate restTemplate;
+	private ReactiveElasticsearchTemplate template;
+
+	@Before
+	public void setUp() {
+
+		restTemplate = new ElasticsearchRestTemplate(TestUtils.restHighLevelClient());
+
+		TestUtils.deleteIndex("test-index-sample");
+
+		restTemplate.createIndex(SampleEntity.class);
+		restTemplate.putMapping(SampleEntity.class);
+		restTemplate.refresh(SampleEntity.class);
+
+		template = new ReactiveElasticsearchTemplate(TestUtils.reactiveClient());
+	}
+
+	@Test // DATAES-488
+	public void indexWithIdShouldWork() {
+
+		String documentId = randomNumeric(5);
+		SampleEntity sampleEntity = SampleEntity.builder().id(documentId).message("foo bar")
+				.version(System.currentTimeMillis()).build();
+
+		template.index(sampleEntity).as(StepVerifier::create).expectNextCount(1).verifyComplete();
+
+		restTemplate.refresh(SampleEntity.class);
+
+		List<SampleEntity> result = restTemplate
+				.queryForList(new CriteriaQuery(Criteria.where("message").is(sampleEntity.getMessage())), SampleEntity.class);
+		assertThat(result).hasSize(1);
+	}
+
+	@Test // DATAES-488
+	public void getShouldReturnEntity() {
+
+		String documentId = randomNumeric(5);
+		SampleEntity sampleEntity = SampleEntity.builder().id(documentId).message("some message")
+				.version(System.currentTimeMillis()).build();
+
+		IndexQuery indexQuery = getIndexQuery(sampleEntity);
+		restTemplate.index(indexQuery);
+		restTemplate.refresh(SampleEntity.class);
+
+		template.get(documentId, SampleEntity.class) //
+				.as(StepVerifier::create) //
+				.expectNext(sampleEntity) //
+				.verifyComplete();
+	}
+
+	@Test // DATAES-488
+	public void getForNothing() {
+
+		String documentId = randomNumeric(5);
+		SampleEntity sampleEntity = SampleEntity.builder().id(documentId).message("some message")
+				.version(System.currentTimeMillis()).build();
+
+		IndexQuery indexQuery = getIndexQuery(sampleEntity);
+		restTemplate.index(indexQuery);
+		restTemplate.refresh(SampleEntity.class);
+
+		template.get("foo", SampleEntity.class) //
+				.as(StepVerifier::create) //
+				.verifyComplete();
+	}
+
+	@Test // DATAES-488
+	public void findShouldApplyCriteria() {
+
+		String documentId = randomNumeric(5);
+		SampleEntity sampleEntity = SampleEntity.builder().id(documentId).message("some message")
+				.version(System.currentTimeMillis()).build();
+
+		IndexQuery indexQuery = getIndexQuery(sampleEntity);
+		restTemplate.index(indexQuery);
+		restTemplate.refresh(SampleEntity.class);
+
+		CriteriaQuery criteriaQuery = new CriteriaQuery(Criteria.where("message").is("some message"));
+
+		template.query(criteriaQuery, SampleEntity.class) //
+				.as(StepVerifier::create) //
+				.expectNext(sampleEntity) //
+				.verifyComplete();
+	}
+
+	@Test // DATAES-488
+	public void findShouldReturnEmptyFluxIfNothingFound() {
+
+		String documentId = randomNumeric(5);
+		SampleEntity sampleEntity = SampleEntity.builder().id(documentId).message("some message")
+				.version(System.currentTimeMillis()).build();
+
+		IndexQuery indexQuery = getIndexQuery(sampleEntity);
+		restTemplate.index(indexQuery);
+		restTemplate.refresh(SampleEntity.class);
+
+		CriteriaQuery criteriaQuery = new CriteriaQuery(Criteria.where("message").is("foo"));
+
+		template.query(criteriaQuery, SampleEntity.class) //
+				.as(StepVerifier::create) //
+				.verifyComplete();
+	}
+
+	private IndexQuery getIndexQuery(SampleEntity sampleEntity) {
+		return new IndexQueryBuilder().withId(sampleEntity.getId()).withObject(sampleEntity)
+				.withVersion(sampleEntity.getVersion()).build();
+	}
+}

src/test/java/org/springframework/data/elasticsearch/repositories/synonym/SynonymRepositoryTests.java

@@ -59,7 +59,7 @@ public class SynonymRepositoryTests {
 	public void shouldDo() {
 		//given
 		SynonymEntity entry1 = new SynonymEntity();
-		entry1.setText("Elizabeth is the English queen");
+		entry1.setText("Elizabeth is the english queen");
 		SynonymEntity entry2 = new SynonymEntity();
 		entry2.setText("Other text");