Skip to content

Commit 7e56c3f

Browse files
mp911demp911de
committed
DATAES-569 - Polishing.
Slightly tweak javadoc. Tweak StepVerifier usage. Add missing assertion. Fix generics. Original pull request: spring-projects#277.
1 parent 2e709a7 commit 7e56c3f

File tree

3 files changed

+41
-47
lines changed

3 files changed

+41
-47
lines changed

‎src/main/java/org/springframework/data/elasticsearch/client/reactive/ReactiveElasticsearchClient.java

Lines changed: 10 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,6 @@
1515
*/
1616
package org.springframework.data.elasticsearch.client.reactive;
1717

18-
import org.elasticsearch.action.admin.indices.flush.FlushRequest;
1918
import reactor.core.publisher.Flux;
2019
import reactor.core.publisher.Mono;
2120

@@ -26,6 +25,7 @@
2625
import org.elasticsearch.action.admin.indices.close.CloseIndexRequest;
2726
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
2827
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
28+
import org.elasticsearch.action.admin.indices.flush.FlushRequest;
2929
import org.elasticsearch.action.admin.indices.get.GetIndexRequest;
3030
import org.elasticsearch.action.admin.indices.mapping.put.PutMappingRequest;
3131
import org.elasticsearch.action.admin.indices.open.OpenIndexRequest;
@@ -45,6 +45,7 @@
4545
import org.elasticsearch.index.reindex.BulkByScrollResponse;
4646
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
4747
import org.elasticsearch.search.SearchHit;
48+
4849
import org.springframework.data.elasticsearch.client.ClientConfiguration;
4950
import org.springframework.data.elasticsearch.client.ElasticsearchHost;
5051
import org.springframework.http.HttpHeaders;
@@ -247,8 +248,8 @@ default Mono<IndexResponse> index(IndexRequest indexRequest) {
247248

248249
/**
249250
* Gain access to index related commands.
250-
*
251-
* @return
251+
*
252+
* @return access to index related commands.
252253
*/
253254
Indices indices();
254255

@@ -788,8 +789,8 @@ default Mono<Void> updateMapping(PutMappingRequest putMappingRequest) {
788789
* @param consumer never {@literal null}.
789790
* @return a {@link Mono} signalling operation completion or an {@link Mono#error(Throwable) error} if eg. the index
790791
* does not exist.
791-
* @see <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/indices-flush.html"> Indices
792-
* Flush API on elastic.co</a>
792+
* @see <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/indices-flush.html"> Indices Flush
793+
* API on elastic.co</a>
793794
*/
794795
default Mono<Void> flushIndex(Consumer<FlushRequest> consumer) {
795796

@@ -804,8 +805,8 @@ default Mono<Void> flushIndex(Consumer<FlushRequest> consumer) {
804805
* @param flushRequest must not be {@literal null}.
805806
* @return a {@link Mono} signalling operation completion or an {@link Mono#error(Throwable) error} if eg. the index
806807
* does not exist.
807-
* @see <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/indices-flush.html"> Indices
808-
* Flush API on elastic.co</a>
808+
* @see <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/indices-flush.html"> Indices Flush
809+
* API on elastic.co</a>
809810
*/
810811
default Mono<Void> flushIndex(FlushRequest flushRequest) {
811812
return flushIndex(HttpHeaders.EMPTY, flushRequest);
@@ -818,8 +819,8 @@ default Mono<Void> flushIndex(FlushRequest flushRequest) {
818819
* @param flushRequest must not be {@literal null}.
819820
* @return a {@link Mono} signalling operation completion or an {@link Mono#error(Throwable) error} if eg. the index
820821
* does not exist.
821-
* @see <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/indices-flush.html"> Indices
822-
* Flush API on elastic.co</a>
822+
* @see <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/indices-flush.html"> Indices Flush
823+
* API on elastic.co</a>
823824
*/
824825
Mono<Void> flushIndex(HttpHeaders headers, FlushRequest flushRequest);
825826
}

‎src/main/java/org/springframework/data/elasticsearch/client/util/RequestConverters.java

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -94,6 +94,8 @@
9494
* (<a href="https://www.elastic.co">https://www.elastic.co</a>) licensed under the Apache License, Version 2.0.
9595
* </p>
9696
* Modified for usage with {@link ReactiveElasticsearchClient}.
97+
* <p>
98+
* Only intended for internal use.
9799
*
98100
* @since 3.2
99101
*/

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

Lines changed: 29 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -42,7 +42,9 @@
4242
import org.elasticsearch.client.RequestOptions;
4343
import org.elasticsearch.client.RestHighLevelClient;
4444
import org.elasticsearch.common.unit.TimeValue;
45+
import org.elasticsearch.index.get.GetResult;
4546
import org.elasticsearch.index.query.QueryBuilders;
47+
import org.elasticsearch.index.reindex.BulkByScrollResponse;
4648
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
4749
import org.elasticsearch.rest.RestStatus;
4850
import org.elasticsearch.search.builder.SearchSourceBuilder;
@@ -51,6 +53,7 @@
5153
import org.junit.Rule;
5254
import org.junit.Test;
5355
import org.junit.runner.RunWith;
56+
5457
import org.springframework.data.elasticsearch.ElasticsearchVersion;
5558
import org.springframework.data.elasticsearch.ElasticsearchVersionRule;
5659
import org.springframework.data.elasticsearch.TestUtils;
@@ -96,6 +99,8 @@ public void setUp() {
9699

97100
syncClient = TestUtils.restHighLevelClient();
98101
client = TestUtils.reactiveClient();
102+
103+
TestUtils.deleteIndex(INDEX_I, INDEX_II);
99104
}
100105

101106
@After
@@ -140,8 +145,10 @@ public void infoShouldReturnClusterInformation() {
140145
@Test // DATAES-519
141146
public void getOnNonExistingIndexShouldThrowException() {
142147

143-
client.get(new GetRequest(INDEX_I, TYPE_I, "nonono")).as(StepVerifier::create)
144-
.expectError(ElasticsearchStatusException.class).verify();
148+
client.get(new GetRequest(INDEX_I, TYPE_I, "nonono")) //
149+
.as(StepVerifier::create) //
150+
.expectError(ElasticsearchStatusException.class) //
151+
.verify();
145152
}
146153

147154
@Test // DATAES-488
@@ -231,13 +238,9 @@ public void multiGetShouldSkipNonExistingDocuments() {
231238
.add(INDEX_I, TYPE_I, id2); //
232239

233240
client.multiGet(request) //
241+
.map(GetResult::getId) //
234242
.as(StepVerifier::create) //
235-
.consumeNextWith(it -> {
236-
assertThat(it.getId()).isEqualTo(id1);
237-
}) //
238-
.consumeNextWith(it -> {
239-
assertThat(it.getId()).isEqualTo(id2);
240-
}) //
243+
.expectNext(id1, id2) //
241244
.verifyComplete();
242245
}
243246

@@ -263,13 +266,9 @@ public void multiGetShouldReturnAllExistingDocumentsFromDifferentCollection() {
263266
.add(INDEX_II, TYPE_II, id2);
264267

265268
client.multiGet(request) //
269+
.map(GetResult::getId) //
266270
.as(StepVerifier::create) //
267-
.consumeNextWith(it -> {
268-
assertThat(it.getId()).isEqualTo(id1);
269-
}) //
270-
.consumeNextWith(it -> {
271-
assertThat(it.getId()).isEqualTo(id2);
272-
}) //
271+
.expectNext(id1, id2) //
273272
.verifyComplete();
274273
}
275274

@@ -320,10 +319,7 @@ public void indexShouldErrorForExistingDocuments() {
320319

321320
client.index(request) //
322321
.as(StepVerifier::create) //
323-
.consumeErrorWith(error -> {
324-
assertThat(error).isInstanceOf(ElasticsearchStatusException.class);
325-
}) //
326-
.verify();
322+
.verifyError(ElasticsearchStatusException.class);
327323
}
328324

329325
@Test // DATAES-488
@@ -372,10 +368,7 @@ public void updateShouldErrorNonExistingDocumentWhenNotUpserted() {
372368

373369
client.update(request) //
374370
.as(StepVerifier::create) //
375-
.consumeErrorWith(error -> {
376-
assertThat(error).isInstanceOf(ElasticsearchStatusException.class);
377-
}) //
378-
.verify();
371+
.verifyError(ElasticsearchStatusException.class);
379372
}
380373

381374
@Test // DATAES-488
@@ -447,11 +440,9 @@ public void deleteByShouldRemoveExistingDocument() {
447440
.setQuery(QueryBuilders.boolQuery().must(QueryBuilders.termQuery("_id", id)));
448441

449442
client.deleteBy(request) //
443+
.map(BulkByScrollResponse::getDeleted) //
450444
.as(StepVerifier::create) //
451-
.consumeNextWith(it -> {
452-
453-
assertThat(it.getDeleted()).isEqualTo(1);
454-
}) //
445+
.expectNext(1L) //
455446
.verifyComplete();
456447
}
457448

@@ -466,10 +457,9 @@ public void deleteByEmitResultWhenNothingRemoved() {
466457
.setQuery(QueryBuilders.boolQuery().must(QueryBuilders.termQuery("_id", "it-was-not-me")));
467458

468459
client.deleteBy(request) //
460+
.map(BulkByScrollResponse::getDeleted) //
469461
.as(StepVerifier::create) //
470-
.consumeNextWith(it -> {
471-
assertThat(it.getDeleted()).isEqualTo(0);
472-
}) //
462+
.expectNext(0L)//
473463
.verifyComplete();
474464
}
475465

@@ -500,7 +490,8 @@ public void scrollShouldReadWhileTakeNotReached() {
500490
request = request.scroll(TimeValue.timeValueMinutes(1));
501491

502492
client.scroll(HttpHeaders.EMPTY, request) //
503-
.take(73).as(StepVerifier::create) //
493+
.take(73) //
494+
.as(StepVerifier::create) //
504495
.expectNextCount(73) //
505496
.verifyComplete();
506497
}
@@ -534,7 +525,7 @@ public void createIndex() throws IOException {
534525
.as(StepVerifier::create) //
535526
.verifyComplete();
536527

537-
syncClient.indices().exists(new GetIndexRequest().indices(INDEX_II), RequestOptions.DEFAULT);
528+
assertThat(syncClient.indices().exists(new GetIndexRequest().indices(INDEX_I), RequestOptions.DEFAULT)).isTrue();
538529
}
539530

540531
@Test // DATAES-569
@@ -663,15 +654,15 @@ public void flushNonExistingIndex() {
663654
.verifyError(ElasticsearchStatusException.class);
664655
}
665656

666-
AddToIndexOfType addSourceDocument() {
657+
private AddToIndexOfType addSourceDocument() {
667658
return add(DOC_SOURCE);
668659
}
669660

670-
AddToIndexOfType add(Map source) {
661+
private AddToIndexOfType add(Map<String, ? extends Object> source) {
671662
return new AddDocument(source);
672663
}
673664

674-
IndexRequest indexRequest(Map source, String index, String type) {
665+
private IndexRequest indexRequest(Map source, String index, String type) {
675666

676667
return new IndexRequest(index, type) //
677668
.id(UUID.randomUUID().toString()) //
@@ -681,7 +672,7 @@ IndexRequest indexRequest(Map source, String index, String type) {
681672
}
682673

683674
@SneakyThrows
684-
String doIndex(Map source, String index, String type) {
675+
private String doIndex(Map<?, ?> source, String index, String type) {
685676
return syncClient.index(indexRequest(source, index, type), RequestOptions.DEFAULT).getId();
686677
}
687678

@@ -695,10 +686,10 @@ interface AddToIndex {
695686

696687
class AddDocument implements AddToIndexOfType {
697688

698-
Map source;
689+
Map<String, ? extends Object> source;
699690
@Nullable String type;
700691

701-
AddDocument(Map source) {
692+
AddDocument(Map<String, ? extends Object> source) {
702693
this.source = source;
703694
}
704695

@@ -711,7 +702,7 @@ public AddToIndex ofType(String type) {
711702

712703
@Override
713704
public String to(String index) {
714-
return doIndex(new LinkedHashMap(source), index, type);
705+
return doIndex(new LinkedHashMap<>(source), index, type);
715706
}
716707
}
717708

0 commit comments

Comments
 (0)