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

DATAES-894 - Adapt to changes in Reactor

Browse Source 
This commit updates the `DefaultReactiveElasticsearchClient` after
changes in Reactor around `Processors` and `Sinks`.

Original pull request: #498.
This commit is contained in:
Brian Clozel 2020-08-05 14:36:41 +02:00 committed by Mark Paluch
parent b30f12503d
commit 103bf9f1b9
No known key found for this signature in database
GPG Key ID: 51A00FA751B91849
1 changed files with 7 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
src/main/java/org/springframework/data/elasticsearch/client/reactive/DefaultReactiveElasticsearchClient.java
View File

@ -23,9 +23,8 @@ import io.netty.handler.ssl.JdkSslContext;
import io.netty.handler.timeout.ReadTimeoutHandler; import io.netty.handler.timeout.ReadTimeoutHandler;
import io.netty.handler.timeout.WriteTimeoutHandler; import io.netty.handler.timeout.WriteTimeoutHandler;
import reactor.core.publisher.Flux; import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxIdentityProcessor; import reactor.core.publisher.FluxProcessor;
import reactor.core.publisher.Mono; import reactor.core.publisher.Mono;
import reactor.core.publisher.Processors;
import reactor.core.publisher.Sinks; import reactor.core.publisher.Sinks;
import reactor.netty.http.client.HttpClient; import reactor.netty.http.client.HttpClient;
import reactor.netty.transport.ProxyProvider; import reactor.netty.transport.ProxyProvider;
@ -467,8 +466,9 @@ public class DefaultReactiveElasticsearchClient implements ReactiveElasticsearch
searchRequest.scroll(scrollTimeout); searchRequest.scroll(scrollTimeout);
} }
Sinks.StandaloneFluxSink<ActionRequest> request = Sinks.unicast(); Sinks.Many<ActionRequest> request = Sinks.many().unicast().onBackpressureBuffer();
FluxIdentityProcessor<SearchResponse> inbound = Processors.unicast();
FluxProcessor<SearchResponse, SearchResponse> inbound = FluxProcessor.fromSink(Sinks.many().unicast().onBackpressureBuffer());
Flux<SearchResponse> exchange = request.asFlux().flatMap(it -> { Flux<SearchResponse> exchange = request.asFlux().flatMap(it -> {
@ -495,7 +495,7 @@ public class DefaultReactiveElasticsearchClient implements ReactiveElasticsearch
if (isEmpty(searchResponse.getHits())) { if (isEmpty(searchResponse.getHits())) {
inbound.onComplete(); inbound.onComplete();
request.complete(); request.emitComplete();
} else { } else {
@ -503,7 +503,7 @@ public class DefaultReactiveElasticsearchClient implements ReactiveElasticsearch
SearchScrollRequest searchScrollRequest = new SearchScrollRequest(scrollState.getScrollId()) SearchScrollRequest searchScrollRequest = new SearchScrollRequest(scrollState.getScrollId())
.scroll(scrollTimeout); .scroll(scrollTimeout);
request.next(searchScrollRequest); request.emitNext(searchScrollRequest);
} }
}).map(SearchResponse::getHits) // }).map(SearchResponse::getHits) //
@ -511,7 +511,7 @@ public class DefaultReactiveElasticsearchClient implements ReactiveElasticsearch
return searchHits.doOnSubscribe(ignore -> { return searchHits.doOnSubscribe(ignore -> {
exchange.subscribe(inbound); exchange.subscribe(inbound);
request.next(searchRequest); request.emitNext(searchRequest);
}); });
}, state -> cleanupScroll(headers, state), // }, state -> cleanupScroll(headers, state), //