Cwikius-Spring/spring-data-elasticsearch
1
0
Fork 0
mirror of https://github.com/spring-projects/spring-data-elasticsearch.git synced 2025-06-19 02:22:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix reactive save of Flux.

Browse Source 
Original Pull Request #2581
Closes #2576
This commit is contained in:
Peter-Josef Meisch 2023-06-02 15:54:57 +02:00 committed by GitHub
parent 11fc22566e
commit d6b5540614
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 38 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -225,42 +225,42 @@ 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 // entities.window(bulkSize) //
.bufferTimeout(bulkSize, Duration.ofMillis(200)) // .concatMap(flux -> flux.collectList()) //
.subscribe(new Subscriber<List<T>>() { .subscribe(new Subscriber<List<T>>() {
private Subscription subscription; private Subscription subscription;
private AtomicBoolean upstreamComplete = new AtomicBoolean(false); private AtomicBoolean upstreamComplete = new AtomicBoolean(false);
@Override @Override
public void onSubscribe(Subscription subscription) { public void onSubscribe(Subscription subscription) {
this.subscription = subscription; this.subscription = subscription;
subscription.request(1); subscription.request(1);
} }
@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()) {
subscription.request(1); subscription.request(1);
} else { } else {
sink.tryEmitComplete(); sink.tryEmitComplete();
} }
}).subscribe(); }).subscribe();
} }
@Override @Override
public void onError(Throwable throwable) { public void onError(Throwable throwable) {
subscription.cancel(); subscription.cancel();
sink.tryEmitError(throwable); sink.tryEmitError(throwable);
} }
@Override @Override
public void onComplete() { public void onComplete() {
upstreamComplete.set(true); upstreamComplete.set(true);
} }
}); });
return sink.asFlux(); return sink.asFlux();
}); });

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

@ -28,6 +28,7 @@ import java.lang.Boolean;
import java.lang.Integer; import java.lang.Integer;
import java.lang.Long; import java.lang.Long;
import java.lang.Object; import java.lang.Object;
import java.time.Duration;
import java.time.LocalDate; import java.time.LocalDate;
import java.time.format.DateTimeFormatter; import java.time.format.DateTimeFormatter;
import java.util.ArrayList; import java.util.ArrayList;
@ -1171,7 +1172,7 @@ public abstract class ReactiveElasticsearchIntegrationTests {
}).verifyComplete(); }).verifyComplete();
} }
@Test // #2496 @Test // #2496, #2576
@DisplayName("should save data from Flux and return saved data in a flux") @DisplayName("should save data from Flux and return saved data in a flux")
void shouldSaveDataFromFluxAndReturnSavedDataInAFlux() { void shouldSaveDataFromFluxAndReturnSavedDataInAFlux() {
@ -1180,9 +1181,11 @@ public abstract class ReactiveElasticsearchIntegrationTests {
.mapToObj(SampleEntity::of) // .mapToObj(SampleEntity::of) //
.collect(Collectors.toList()); .collect(Collectors.toList());
var entityFlux = Flux.fromIterable(entityList); // we add a random delay to make suure the underlying implementation handles irregular incoming data
var entities = Flux.fromIterable(entityList).concatMap(
entity -> Mono.just(entity).delay(Duration.ofMillis((long) (Math.random() * 10))).thenReturn(entity));
operations.save(entityFlux, SampleEntity.class).collectList() // operations.save(entities, SampleEntity.class).collectList() //
.as(StepVerifier::create) // .as(StepVerifier::create) //
.consumeNextWith(savedEntities -> { .consumeNextWith(savedEntities -> {
assertThat(savedEntities).isEqualTo(entityList); assertThat(savedEntities).isEqualTo(entityList);