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

DATAES-701 - Enable proxy support for the reactive rest client.

Browse Source 
Original PR: #380
This commit is contained in:
Peter-Josef Meisch 2020-01-11 21:13:42 +01:00 committed by GitHub
parent d55cb00d45
commit 513741bcf6
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 12 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -25,6 +25,7 @@ import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink; import reactor.core.publisher.FluxSink;
import reactor.core.publisher.Mono; import reactor.core.publisher.Mono;
import reactor.netty.http.client.HttpClient; import reactor.netty.http.client.HttpClient;
import reactor.netty.tcp.ProxyProvider;
import reactor.netty.tcp.TcpClient; import reactor.netty.tcp.TcpClient;
import java.io.IOException; import java.io.IOException;
@ -198,12 +199,22 @@ public class DefaultReactiveElasticsearchClient implements ReactiveElasticsearch
} }
if (!soTimeout.isNegative()) { if (!soTimeout.isNegative()) {
tcpClient = tcpClient.doOnConnected(connection -> connection // tcpClient = tcpClient.doOnConnected(connection -> connection //
.addHandlerLast(new ReadTimeoutHandler(soTimeout.toMillis(), TimeUnit.MILLISECONDS)) .addHandlerLast(new ReadTimeoutHandler(soTimeout.toMillis(), TimeUnit.MILLISECONDS))
.addHandlerLast(new WriteTimeoutHandler(soTimeout.toMillis(), TimeUnit.MILLISECONDS))); .addHandlerLast(new WriteTimeoutHandler(soTimeout.toMillis(), TimeUnit.MILLISECONDS)));
} }
if (clientConfiguration.getProxy().isPresent()) {
String proxy = clientConfiguration.getProxy().get();
String[] hostPort = proxy.split(":");
if (hostPort.length != 2) {
throw new IllegalArgumentException("invalid proxy configuration " + proxy + ", should be \"host:port\"");
}
tcpClient = tcpClient.proxy(proxyOptions -> proxyOptions.type(ProxyProvider.Proxy.HTTP).host(hostPort[0])
.port(Integer.parseInt(hostPort[1])));
}
String scheme = "http"; String scheme = "http";
HttpClient httpClient = HttpClient.from(tcpClient); HttpClient httpClient = HttpClient.from(tcpClient);