Cwikius-Spring/spring-data-elasticsearch
1
0
Fork 0
mirror of https://github.com/spring-projects/spring-data-elasticsearch.git synced 2025-07-06 18:52:11 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix code not terminating on repository saving an empty flux.

Browse Source 
Original Pull Request #3099
Closes: #3039

Signed-off-by: Peter-Josef Meisch <pj.meisch@sothawo.com>
This commit is contained in:
Peter-Josef Meisch 2025-04-26 10:40:31 +02:00 committed by GitHub
parent 9d025dd469
commit a07ac3c93d
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 26 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -233,6 +233,7 @@ abstract public class AbstractReactiveElasticsearchTemplate
.subscribe(new Subscriber<>() { .subscribe(new Subscriber<>() {
@Nullable private Subscription subscription = null; @Nullable private Subscription subscription = null;
private final AtomicBoolean upstreamComplete = new AtomicBoolean(false); private final AtomicBoolean upstreamComplete = new AtomicBoolean(false);
private final AtomicBoolean onNextHasBeenCalled = new AtomicBoolean(false);
@Override @Override
public void onSubscribe(Subscription subscription) { public void onSubscribe(Subscription subscription) {
@ -242,6 +243,7 @@ abstract public class AbstractReactiveElasticsearchTemplate
@Override @Override
public void onNext(List<T> entityList) { public void onNext(List<T> entityList) {
onNextHasBeenCalled.set(true);
saveAll(entityList, index) saveAll(entityList, index)
.map(sink::tryEmitNext) .map(sink::tryEmitNext)
.doOnComplete(() -> { .doOnComplete(() -> {
@ -267,6 +269,10 @@ abstract public class AbstractReactiveElasticsearchTemplate
@Override @Override
public void onComplete() { public void onComplete() {
upstreamComplete.set(true); upstreamComplete.set(true);
if (!onNextHasBeenCalled.get()) {
// this happens when an empty flux is saved
sink.tryEmitComplete();
}
} }
}); });
return sink.asFlux(); return sink.asFlux();

20
src/test/java/org/springframework/data/elasticsearch/repository/support/SimpleReactiveElasticsearchRepositoryIntegrationTests.java
View File

@ -105,6 +105,26 @@ abstract class SimpleReactiveElasticsearchRepositoryIntegrationTests {
return operations.exists(id, IndexCoordinates.of(indexNameProvider.indexName())); return operations.exists(id, IndexCoordinates.of(indexNameProvider.indexName()));
} }
@Test // #3093
@DisplayName("should save all from empty collection")
void shouldSaveAllFromEmptyCollection() {
repository.saveAll(Collections.emptyList())
.as(StepVerifier::create)
.expectNextCount(0)
.verifyComplete();
}
@Test // #3093
@DisplayName("should save all from empty flux")
void shouldSaveAllFromEmptyFlux() {
repository.saveAll(Flux.empty())
.as(StepVerifier::create)
.expectNextCount(0)
.verifyComplete();
}
@Test // DATAES-519 @Test // DATAES-519
void saveShouldComputeMultipleEntities() { void saveShouldComputeMultipleEntities() {