From 60cbb6787757e6c0a41da72fd03693f04fba73f5 Mon Sep 17 00:00:00 2001 From: Peter-Josef Meisch Date: Tue, 21 Apr 2020 21:57:31 +0200 Subject: [PATCH] DATAES-785 - Polishing. --- .../core/AbstractElasticsearchTemplate.java | 2 +- .../core/ReactiveElasticsearchTemplate.java | 45 ++++++++----------- .../event/ReactiveAuditingEntityCallback.java | 2 +- ...iveElasticsearchTemplateCallbackTests.java | 4 +- .../ReactiveAuditingEntityCallbackTests.java | 2 +- 5 files changed, 23 insertions(+), 32 deletions(-) diff --git a/src/main/java/org/springframework/data/elasticsearch/core/AbstractElasticsearchTemplate.java b/src/main/java/org/springframework/data/elasticsearch/core/AbstractElasticsearchTemplate.java index 0fa786821..5aa64604a 100644 --- a/src/main/java/org/springframework/data/elasticsearch/core/AbstractElasticsearchTemplate.java +++ b/src/main/java/org/springframework/data/elasticsearch/core/AbstractElasticsearchTemplate.java @@ -593,7 +593,7 @@ public abstract class AbstractElasticsearchTemplate implements ElasticsearchOper private final Class type; public ReadSearchScrollDocumentResponseCallback(Class type, IndexCoordinates index) { - + Assert.notNull(type, "type is null"); this.delegate = new ReadDocumentCallback<>(elasticsearchConverter, type, index); diff --git a/src/main/java/org/springframework/data/elasticsearch/core/ReactiveElasticsearchTemplate.java b/src/main/java/org/springframework/data/elasticsearch/core/ReactiveElasticsearchTemplate.java index b8054ea5c..8d5e859f7 100644 --- a/src/main/java/org/springframework/data/elasticsearch/core/ReactiveElasticsearchTemplate.java +++ b/src/main/java/org/springframework/data/elasticsearch/core/ReactiveElasticsearchTemplate.java @@ -194,8 +194,7 @@ public class ReactiveElasticsearchTemplate implements ReactiveElasticsearchOpera IndexResponse indexResponse = it.getT2(); AdaptibleEntity adaptableEntity = operations.forEntity(savedEntity, converter.getConversionService()); return adaptableEntity.populateIdIfNecessary(indexResponse.getId()); - }) - .flatMap(saved -> maybeCallAfterSave(saved, index)); + }).flatMap(saved -> maybeCallAfterSave(saved, index)); } @Override @@ -208,31 +207,25 @@ public class ReactiveElasticsearchTemplate implements ReactiveElasticsearchOpera Assert.notNull(entitiesPublisher, "Entities must not be null!"); - return entitiesPublisher - .flatMapMany(entities -> { - return Flux.fromIterable(entities) // - .concatMap(entity -> maybeCallBeforeConvert(entity, index)); - }) - .collectList() - .map(Entities::new) - .flatMapMany(entities -> { - if (entities.isEmpty()) { - return Flux.empty(); - } + return entitiesPublisher.flatMapMany(entities -> { + return Flux.fromIterable(entities) // + .concatMap(entity -> maybeCallBeforeConvert(entity, index)); + }).collectList().map(Entities::new).flatMapMany(entities -> { + if (entities.isEmpty()) { + return Flux.empty(); + } - return doBulkOperation(entities.indexQueries(), BulkOptions.defaultOptions(), index) // - .index() - .flatMap(indexAndResponse -> { - T savedEntity = entities.entityAt(indexAndResponse.getT1()); - BulkItemResponse bulkItemResponse = indexAndResponse.getT2(); + return doBulkOperation(entities.indexQueries(), BulkOptions.defaultOptions(), index) // + .index().flatMap(indexAndResponse -> { + T savedEntity = entities.entityAt(indexAndResponse.getT1()); + BulkItemResponse bulkItemResponse = indexAndResponse.getT2(); - AdaptibleEntity adaptibleEntity = operations.forEntity(savedEntity, - converter.getConversionService()); - adaptibleEntity.populateIdIfNecessary(bulkItemResponse.getResponse().getId()); + AdaptibleEntity adaptibleEntity = operations.forEntity(savedEntity, converter.getConversionService()); + adaptibleEntity.populateIdIfNecessary(bulkItemResponse.getResponse().getId()); - return maybeCallAfterSave(savedEntity, index); - }); - }); + return maybeCallAfterSave(savedEntity, index); + }); + }); } @Override @@ -1017,9 +1010,7 @@ public class ReactiveElasticsearchTemplate implements ReactiveElasticsearchOpera } private List indexQueries() { - return entities.stream() - .map(ReactiveElasticsearchTemplate.this::getIndexQuery) - .collect(Collectors.toList()); + return entities.stream().map(ReactiveElasticsearchTemplate.this::getIndexQuery).collect(Collectors.toList()); } private T entityAt(long index) { diff --git a/src/main/java/org/springframework/data/elasticsearch/core/event/ReactiveAuditingEntityCallback.java b/src/main/java/org/springframework/data/elasticsearch/core/event/ReactiveAuditingEntityCallback.java index 68dedec4b..69a803684 100644 --- a/src/main/java/org/springframework/data/elasticsearch/core/event/ReactiveAuditingEntityCallback.java +++ b/src/main/java/org/springframework/data/elasticsearch/core/event/ReactiveAuditingEntityCallback.java @@ -15,12 +15,12 @@ */ package org.springframework.data.elasticsearch.core.event; -import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates; import reactor.core.publisher.Mono; import org.springframework.beans.factory.ObjectFactory; import org.springframework.core.Ordered; import org.springframework.data.auditing.IsNewAwareAuditingHandler; +import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates; import org.springframework.data.mapping.callback.EntityCallback; import org.springframework.util.Assert; diff --git a/src/test/java/org/springframework/data/elasticsearch/core/ReactiveElasticsearchTemplateCallbackTests.java b/src/test/java/org/springframework/data/elasticsearch/core/ReactiveElasticsearchTemplateCallbackTests.java index 666fcbf9e..93724559a 100644 --- a/src/test/java/org/springframework/data/elasticsearch/core/ReactiveElasticsearchTemplateCallbackTests.java +++ b/src/test/java/org/springframework/data/elasticsearch/core/ReactiveElasticsearchTemplateCallbackTests.java @@ -201,8 +201,8 @@ public class ReactiveElasticsearchTemplateCallbackTests { Person entity1 = new Person("init1", "luke1"); Person entity2 = new Person("init2", "luke2"); - List saved = template.saveAll(Mono.just(Arrays.asList(entity1, entity2)), index) - .toStream().collect(Collectors.toList()); + List saved = template.saveAll(Mono.just(Arrays.asList(entity1, entity2)), index).toStream() + .collect(Collectors.toList()); verify(afterSaveCallback, times(2)).onAfterSave(any(), eq(index)); assertThat(saved.get(0).firstname).isEqualTo("after-save"); diff --git a/src/test/java/org/springframework/data/elasticsearch/core/event/ReactiveAuditingEntityCallbackTests.java b/src/test/java/org/springframework/data/elasticsearch/core/event/ReactiveAuditingEntityCallbackTests.java index 3a5c6d3bd..364b2b9af 100644 --- a/src/test/java/org/springframework/data/elasticsearch/core/event/ReactiveAuditingEntityCallbackTests.java +++ b/src/test/java/org/springframework/data/elasticsearch/core/event/ReactiveAuditingEntityCallbackTests.java @@ -18,7 +18,6 @@ package org.springframework.data.elasticsearch.core.event; import static org.assertj.core.api.Assertions.*; import static org.mockito.Mockito.*; -import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates; import reactor.test.StepVerifier; import java.time.LocalDateTime; @@ -33,6 +32,7 @@ import org.springframework.data.annotation.CreatedDate; import org.springframework.data.annotation.Id; import org.springframework.data.annotation.LastModifiedDate; import org.springframework.data.auditing.IsNewAwareAuditingHandler; +import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates; import org.springframework.data.elasticsearch.core.mapping.SimpleElasticsearchMappingContext; import org.springframework.data.mapping.context.PersistentEntities; import org.springframework.lang.Nullable;