Cwikius-Spring/spring-data-elasticsearch
1
0
Fork 0
mirror of https://github.com/spring-projects/spring-data-elasticsearch.git synced 2025-07-12 21:33:27 +00:00
Code Issues Packages Projects Releases Wiki Activity

Use correct overload of Flux.bufferTimeout().

Browse Source 
Original Pull Request #2753
Closes #2607
This commit is contained in:
Peter-Josef Meisch 2023-11-03 13:51:23 +01:00 committed by GitHub
parent d281df74d1
commit 3b93b6aea9
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 78 additions and 74 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

96
src/main/java/org/springframework/data/elasticsearch/core/AbstractReactiveElasticsearchTemplate.java
View File

@ -15,8 +15,6 @@
*/ */
package org.springframework.data.elasticsearch.core; package org.springframework.data.elasticsearch.core;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import reactor.core.publisher.Flux; import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono; import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks; import reactor.core.publisher.Sinks;
@ -73,9 +71,6 @@ import org.springframework.util.Assert;
abstract public class AbstractReactiveElasticsearchTemplate abstract public class AbstractReactiveElasticsearchTemplate
implements ReactiveElasticsearchOperations, ApplicationContextAware { implements ReactiveElasticsearchOperations, ApplicationContextAware {
protected static final Log QUERY_LOGGER = LogFactory
.getLog("org.springframework.data.elasticsearch.core.QUERY");
protected final ElasticsearchConverter converter; protected final ElasticsearchConverter converter;
protected final SimpleElasticsearchMappingContext mappingContext; protected final SimpleElasticsearchMappingContext mappingContext;
protected final EntityOperations entityOperations; protected final EntityOperations entityOperations;
@ -175,11 +170,12 @@ abstract public class AbstractReactiveElasticsearchTemplate
* @return a Mono signalling finished execution * @return a Mono signalling finished execution
* @since 4.3 * @since 4.3
*/ */
@SuppressWarnings("unused")
public Mono<Void> logVersions() { public Mono<Void> logVersions() {
return getVendor() // return getVendor()
.zipWith(getRuntimeLibraryVersion()) // .zipWith(getRuntimeLibraryVersion())
.zipWith(getClusterVersion()) // .zipWith(getClusterVersion())
.doOnNext(objects -> VersionInfo.logVersions(objects.getT1().getT1(), objects.getT1().getT2(), objects.getT2())) .doOnNext(objects -> VersionInfo.logVersions(objects.getT1().getT1(), objects.getT1().getT2(), objects.getT2()))
.then(); .then();
} }
@ -233,11 +229,12 @@ abstract public class AbstractReactiveElasticsearchTemplate
return Flux.defer(() -> { return Flux.defer(() -> {
Sinks.Many<T> sink = Sinks.many().unicast().onBackpressureBuffer(); Sinks.Many<T> sink = Sinks.many().unicast().onBackpressureBuffer();
entities.window(bulkSize) // // noinspection ReactiveStreamsSubscriberImplementation
.concatMap(flux -> flux.collectList()) // entities
.subscribe(new Subscriber<List<T>>() { .bufferTimeout(bulkSize, Duration.ofMillis(200), true)
private Subscription subscription; .subscribe(new Subscriber<>() {
private AtomicBoolean upstreamComplete = new AtomicBoolean(false); @Nullable private Subscription subscription = null;
private final AtomicBoolean upstreamComplete = new AtomicBoolean(false);
@Override @Override
public void onSubscribe(Subscription subscription) { public void onSubscribe(Subscription subscription) {
@ -247,10 +244,13 @@ abstract public class AbstractReactiveElasticsearchTemplate
@Override @Override
public void onNext(List<T> entityList) { public void onNext(List<T> entityList) {
saveAll(entityList, index) // saveAll(entityList, index)
.map(sink::tryEmitNext) // .map(sink::tryEmitNext)
.doOnComplete(() -> { .doOnComplete(() -> {
if (!upstreamComplete.get()) { if (!upstreamComplete.get()) {
if (subscription == null) {
throw new IllegalStateException("no subscription");
}
subscription.request(1); subscription.request(1);
} else { } else {
sink.tryEmitComplete(); sink.tryEmitComplete();
@ -260,7 +260,9 @@ abstract public class AbstractReactiveElasticsearchTemplate
@Override @Override
public void onError(Throwable throwable) { public void onError(Throwable throwable) {
if (subscription != null) {
subscription.cancel(); subscription.cancel();
}
sink.tryEmitError(throwable); sink.tryEmitError(throwable);
} }
@ -324,6 +326,7 @@ abstract public class AbstractReactiveElasticsearchTemplate
.getPersistentEntity(entity.getClass()); .getPersistentEntity(entity.getClass());
if (persistentEntity != null) { if (persistentEntity != null) {
// noinspection DuplicatedCode
PersistentPropertyAccessor<Object> propertyAccessor = persistentEntity.getPropertyAccessor(entity); PersistentPropertyAccessor<Object> propertyAccessor = persistentEntity.getPropertyAccessor(entity);
ElasticsearchPersistentProperty idProperty = persistentEntity.getIdProperty(); ElasticsearchPersistentProperty idProperty = persistentEntity.getIdProperty();
@ -353,8 +356,7 @@ abstract public class AbstractReactiveElasticsearchTemplate
} }
// noinspection unchecked // noinspection unchecked
T updatedEntity = (T) propertyAccessor.getBean(); return (T) propertyAccessor.getBean();
return updatedEntity;
} else { } else {
EntityOperations.AdaptableEntity<T> adaptableEntity = entityOperations.forEntity(entity, EntityOperations.AdaptableEntity<T> adaptableEntity = entityOperations.forEntity(entity,
converter.getConversionService(), routingResolver); converter.getConversionService(), routingResolver);
@ -385,15 +387,15 @@ abstract public class AbstractReactiveElasticsearchTemplate
Assert.notNull(index, "index must not be null"); Assert.notNull(index, "index must not be null");
return maybeCallbackBeforeConvert(entity, index) return maybeCallbackBeforeConvert(entity, index)
.flatMap(entityAfterBeforeConversionCallback -> doIndex(entityAfterBeforeConversionCallback, index)) // .flatMap(entityAfterBeforeConversionCallback -> doIndex(entityAfterBeforeConversionCallback, index))
.map(it -> { .map(it -> {
T savedEntity = it.getT1(); T savedEntity = it.getT1();
IndexResponseMetaData indexResponseMetaData = it.getT2(); IndexResponseMetaData indexResponseMetaData = it.getT2();
return updateIndexedObject(savedEntity, new IndexedObjectInformation( // return updateIndexedObject(savedEntity, new IndexedObjectInformation(
indexResponseMetaData.id(), // indexResponseMetaData.id(),
indexResponseMetaData.index(), // indexResponseMetaData.index(),
indexResponseMetaData.seqNo(), // indexResponseMetaData.seqNo(),
indexResponseMetaData.primaryTerm(), // indexResponseMetaData.primaryTerm(),
indexResponseMetaData.version())); indexResponseMetaData.version()));
}).flatMap(saved -> maybeCallbackAfterSave(saved, index)); }).flatMap(saved -> maybeCallbackAfterSave(saved, index));
} }
@ -478,12 +480,12 @@ abstract public class AbstractReactiveElasticsearchTemplate
SearchDocumentCallback<T> callback = new ReadSearchDocumentCallback<>(resultType, index); SearchDocumentCallback<T> callback = new ReadSearchDocumentCallback<>(resultType, index);
return doFindForResponse(query, entityType, index) // return doFindForResponse(query, entityType, index)
.flatMap(searchDocumentResponse -> Flux.fromIterable(searchDocumentResponse.getSearchDocuments()) // .flatMap(searchDocumentResponse -> Flux.fromIterable(searchDocumentResponse.getSearchDocuments())
.flatMap(callback::toEntity) // .flatMap(callback::toEntity)
.collectList() // .collectList()
.map(entities -> SearchHitMapping.mappingFor(resultType, converter) // .map(entities -> SearchHitMapping.mappingFor(resultType, converter)
.mapHits(searchDocumentResponse, entities))) // .mapHits(searchDocumentResponse, entities)))
.map(searchHits -> SearchHitSupport.searchPageFor(searchHits, query.getPageable())); .map(searchHits -> SearchHitSupport.searchPageFor(searchHits, query.getPageable()));
} }
@ -503,17 +505,18 @@ abstract public class AbstractReactiveElasticsearchTemplate
SearchDocumentCallback<T> callback = new ReadSearchDocumentCallback<>(resultType, index); SearchDocumentCallback<T> callback = new ReadSearchDocumentCallback<>(resultType, index);
return doFindForResponse(query, entityType, index) // return doFindForResponse(query, entityType, index)
.flatMap(searchDocumentResponse -> Flux.fromIterable(searchDocumentResponse.getSearchDocuments()) // .flatMap(searchDocumentResponse -> Flux.fromIterable(searchDocumentResponse.getSearchDocuments())
.flatMap(callback::toEntity) // .flatMap(callback::toEntity)
.collectList() // .collectList()
.map(entities -> SearchHitMapping.mappingFor(resultType, converter) // .map(entities -> SearchHitMapping.mappingFor(resultType, converter)
.mapHits(searchDocumentResponse, entities))) // .mapHits(searchDocumentResponse, entities)))
.map(ReactiveSearchHitSupport::searchHitsFor); .map(ReactiveSearchHitSupport::searchHitsFor);
} }
abstract protected Flux<SearchDocument> doFind(Query query, Class<?> clazz, IndexCoordinates index); abstract protected Flux<SearchDocument> doFind(Query query, Class<?> clazz, IndexCoordinates index);
@SuppressWarnings("unused")
abstract protected <T> Mono<SearchDocumentResponse> doFindForResponse(Query query, Class<?> clazz, abstract protected <T> Mono<SearchDocumentResponse> doFindForResponse(Query query, Class<?> clazz,
IndexCoordinates index); IndexCoordinates index);
@ -639,17 +642,16 @@ abstract public class AbstractReactiveElasticsearchTemplate
return Mono.empty(); return Mono.empty();
} }
return maybeCallbackAfterLoad(document, type, index) // return maybeCallbackAfterLoad(document, type, index)
.flatMap(documentAfterLoad -> { .flatMap(documentAfterLoad -> {
// noinspection DuplicatedCode
T entity = reader.read(type, documentAfterLoad); T entity = reader.read(type, documentAfterLoad);
IndexedObjectInformation indexedObjectInformation = new IndexedObjectInformation(
IndexedObjectInformation indexedObjectInformation = new IndexedObjectInformation( // documentAfterLoad.hasId() ? documentAfterLoad.getId() : null,
documentAfterLoad.hasId() ? documentAfterLoad.getId() : null, // documentAfterLoad.getIndex(),
documentAfterLoad.getIndex(), // documentAfterLoad.hasSeqNo() ? documentAfterLoad.getSeqNo() : null,
documentAfterLoad.hasSeqNo() ? documentAfterLoad.getSeqNo() : null, // documentAfterLoad.hasPrimaryTerm() ? documentAfterLoad.getPrimaryTerm() : null,
documentAfterLoad.hasPrimaryTerm() ? documentAfterLoad.getPrimaryTerm() : null, // documentAfterLoad.hasVersion() ? documentAfterLoad.getVersion() : null);
documentAfterLoad.hasVersion() ? documentAfterLoad.getVersion() : null); //
entity = updateIndexedObject(entity, indexedObjectInformation); entity = updateIndexedObject(entity, indexedObjectInformation);
return maybeCallbackAfterConvert(entity, documentAfterLoad, index); return maybeCallbackAfterConvert(entity, documentAfterLoad, index);
@ -667,7 +669,7 @@ abstract public class AbstractReactiveElasticsearchTemplate
/** /**
* converts a {@link SearchDocument} to an entity * converts a {@link SearchDocument} to an entity
* *
* @param searchDocument * @param searchDocument the document to convert
* @return the entity in a MOno * @return the entity in a MOno
*/ */
Mono<T> toEntity(SearchDocument searchDocument); Mono<T> toEntity(SearchDocument searchDocument);
@ -675,8 +677,8 @@ abstract public class AbstractReactiveElasticsearchTemplate
/** /**
* converts a {@link SearchDocument} into a SearchHit * converts a {@link SearchDocument} into a SearchHit
* *
* @param searchDocument * @param searchDocument the document to convert
* @return * @return the converted SearchHit
*/ */
Mono<SearchHit<T>> toSearchHit(SearchDocument searchDocument); Mono<SearchHit<T>> toSearchHit(SearchDocument searchDocument);
} }

6
src/test/java/org/springframework/data/elasticsearch/core/ReactiveElasticsearchIntegrationTests.java
View File

@ -1189,9 +1189,11 @@ public abstract class ReactiveElasticsearchIntegrationTests {
.mapToObj(SampleEntity::of) // .mapToObj(SampleEntity::of) //
.collect(Collectors.toList()); .collect(Collectors.toList());
// we add a random delay to make suure the underlying implementation handles irregular incoming data // we add a random delay to make sure the underlying implementation handles irregular incoming data
var entities = Flux.fromIterable(entityList).concatMap( var entities = Flux.fromIterable(entityList).concatMap(
entity -> Mono.just(entity).delay(Duration.ofMillis((long) (Math.random() * 10))).thenReturn(entity)); entity -> Mono.just(entity)
.delay(Duration.ofMillis((long) (Math.random() * 10)))
.thenReturn(entity));
operations.save(entities, SampleEntity.class).collectList() // operations.save(entities, SampleEntity.class).collectList() //
.as(StepVerifier::create) // .as(StepVerifier::create) //