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

DATAES-488 - Polishing.

Browse Source 
Convert spaces to tabs for pom.xml. Switch reactive dependencies to optional. Remove unused commonscollections property. Use managed versions for reactor and Spring dependencies.

Introduce WebClientProvider to avoid reinstantiation of WebClient instances. Introduce ClientConfiguration to encapsulate common Elasticsearch client configuration properties. Split ElasticsearchClients into RestClients and ReactiveRestClients to avoid mandatory dependency on WebFlux/Project Reactor. Adapt tests and code referring to WebClient creation.

Extract response body as byte array instead of Flux of DataBuffer to avoid chunking and to parse an entire response.

Encapsulate hostAndPort string used across configuration/HostProvider with InetSocketAddress. Add parser for InetSocketAddress.

Original Pull Request: #226
This commit is contained in:
Mark Paluch 2018-11-20 09:44:59 +01:00 committed by Christoph Strobl
parent 691a8c57bc
commit 390d7e8273
29 changed files with 1930 additions and 841 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

508
pom.xml
View File

@ -1,285 +1,293 @@
<?xml version="1.0" encoding="UTF-8" standalone="no"?> <?xml version="1.0" encoding="UTF-8" standalone="no"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
<modelVersion>4.0.0</modelVersion> xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>org.springframework.data</groupId> <groupId>org.springframework.data</groupId>
<artifactId>spring-data-elasticsearch</artifactId> <artifactId>spring-data-elasticsearch</artifactId>
<version>3.2.0.BUILD-SNAPSHOT</version> <version>3.2.0.BUILD-SNAPSHOT</version>
<parent> <parent>
<groupId>org.springframework.data.build</groupId> <groupId>org.springframework.data.build</groupId>
<artifactId>spring-data-parent</artifactId> <artifactId>spring-data-parent</artifactId>
<version>2.2.0.BUILD-SNAPSHOT</version> <version>2.2.0.BUILD-SNAPSHOT</version>
</parent> </parent>
<name>Spring Data Elasticsearch</name> <name>Spring Data Elasticsearch</name>
<description>Spring Data Implementation for Elasticsearch</description> <description>Spring Data Implementation for Elasticsearch</description>
<url>https://github.com/spring-projects/spring-data-elasticsearch</url> <url>https://github.com/spring-projects/spring-data-elasticsearch</url>
<properties> <properties>
<commonscollections>3.2.1</commonscollections> <commonslang>2.6</commonslang>
<commonslang>2.6</commonslang> <elasticsearch>6.5.0</elasticsearch>
<elasticsearch>6.5.0</elasticsearch> <log4j>2.9.1</log4j>
<log4j>2.9.1</log4j> <springdata.commons>2.2.0.BUILD-SNAPSHOT</springdata.commons>
<springdata.commons>2.2.0.BUILD-SNAPSHOT</springdata.commons> <java-module-name>spring.data.elasticsearch</java-module-name>
<java-module-name>spring.data.elasticsearch</java-module-name> </properties>
</properties>
<dependencies> <developers>
<developer>
<id>biomedcentral</id>
<name>BioMed Central Development Team</name>
<timezone>+0</timezone>
</developer>
<developer>
<id>cstrobl</id>
<name>Christoph Strobl</name>
<email>cstrobl at pivotal.io</email>
<organization>Pivotal</organization>
<organizationUrl>http://www.pivotal.io</organizationUrl>
<roles>
<role>Developer</role>
</roles>
<timezone>+1</timezone>
</developer>
<developer>
<id>mpaluch</id>
<name>Mark Paluch</name>
<email>mpaluch at pivotal.io</email>
<organization>Pivotal</organization>
<organizationUrl>http://www.pivotal.io</organizationUrl>
<roles>
<role>Developer</role>
</roles>
<timezone>+1</timezone>
</developer>
</developers>
<!-- Spring --> <scm>
<dependency> <url>https://github.com/spring-projects/spring-data-elasticsearch</url>
<groupId>org.springframework</groupId> <connection>scm:git:git://github.com/spring-projects/spring-data-elasticsearch.git</connection>
<artifactId>spring-context</artifactId> <developerConnection>scm:git:ssh://git@github.com/spring-projects/spring-data-elasticsearch.git
<exclusions> </developerConnection>
<exclusion> </scm>
<groupId>commons-logging</groupId>
<artifactId>commons-logging</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency> <ciManagement>
<groupId>org.springframework</groupId> <system>Bamboo</system>
<artifactId>spring-tx</artifactId> <url>https://build.spring.io/browse/SPRINGDATAES</url>
</dependency> </ciManagement>
<!-- SPRING DATA --> <issueManagement>
<dependency> <system>JIRA</system>
<groupId>org.springframework.data</groupId> <url>https://jira.spring.io/browse/DATAES</url>
<artifactId>spring-data-commons</artifactId> </issueManagement>
<version>${springdata.commons}</version>
</dependency>
<dependency> <dependencies>
<groupId>org.springframework</groupId>
<artifactId>spring-webflux</artifactId>
<version>5.1.0.RELEASE</version>
</dependency>
<dependency> <!-- Spring -->
<groupId>io.projectreactor.netty</groupId> <dependency>
<artifactId>reactor-netty</artifactId> <groupId>org.springframework</groupId>
<version>0.8.0.RELEASE</version> <artifactId>spring-context</artifactId>
</dependency> <exclusions>
<exclusion>
<dependency> <groupId>commons-logging</groupId>
<groupId>io.projectreactor</groupId> <artifactId>commons-logging</artifactId>
<artifactId>reactor-test</artifactId> </exclusion>
<version>3.2.0.RELEASE</version> </exclusions>
</dependency>
<!-- APACHE -->
<dependency>
<groupId>commons-lang</groupId>
<artifactId>commons-lang</artifactId>
<version>${commonslang}</version>
<scope>test</scope>
</dependency>
<!-- JODA Time -->
<dependency>
<groupId>joda-time</groupId>
<artifactId>joda-time</artifactId>
<version>${jodatime}</version>
</dependency>
<!-- Elasticsearch -->
<dependency>
<groupId>org.elasticsearch.client</groupId>
<artifactId>transport</artifactId>
<version>${elasticsearch}</version>
<exclusions>
<exclusion>
<groupId>commons-logging</groupId>
<artifactId>commons-logging</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.elasticsearch.client</groupId>
<artifactId>elasticsearch-rest-high-level-client</artifactId>
<version>${elasticsearch}</version>
<exclusions>
<exclusion>
<groupId>commons-logging</groupId>
<artifactId>commons-logging</artifactId>
</exclusion>
</exclusions>
</dependency> </dependency>
<dependency> <dependency>
<groupId>org.slf4j</groupId> <groupId>org.springframework</groupId>
<artifactId>log4j-over-slf4j</artifactId> <artifactId>spring-tx</artifactId>
<version>${slf4j}</version> </dependency>
<scope>test</scope>
</dependency>
<dependency> <!-- SPRING DATA -->
<groupId>org.apache.logging.log4j</groupId> <dependency>
<artifactId>log4j-core</artifactId> <groupId>org.springframework.data</groupId>
<version>${log4j}</version> <artifactId>spring-data-commons</artifactId>
<scope>test</scope> <version>${springdata.commons}</version>
</dependency> </dependency>
<!-- Jackson JSON Mapper --> <!-- Reactive Infrastructure -->
<dependency> <dependency>
<groupId>com.fasterxml.jackson.core</groupId> <groupId>org.springframework</groupId>
<artifactId>jackson-core</artifactId> <artifactId>spring-webflux</artifactId>
</dependency> <optional>true</optional>
<dependency> </dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
</dependency>
<!-- CDI --> <dependency>
<dependency> <groupId>io.projectreactor.netty</groupId>
<groupId>javax.enterprise</groupId> <artifactId>reactor-netty</artifactId>
<artifactId>cdi-api</artifactId> <optional>true</optional>
<version>${cdi}</version> </dependency>
<scope>provided</scope>
<optional>true</optional>
</dependency>
<!-- Test --> <dependency>
<dependency> <groupId>io.projectreactor</groupId>
<groupId>org.springframework</groupId> <artifactId>reactor-test</artifactId>
<artifactId>spring-test</artifactId> <scope>test</scope>
<scope>test</scope> </dependency>
<exclusions>
<exclusion>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency> <!-- APACHE -->
<groupId>org.apache.openwebbeans.test</groupId> <dependency>
<artifactId>cditest-owb</artifactId> <groupId>commons-lang</groupId>
<version>1.2.8</version> <artifactId>commons-lang</artifactId>
<scope>test</scope> <version>${commonslang}</version>
<exclusions> <scope>test</scope>
<exclusion> </dependency>
<groupId>org.apache.geronimo.specs</groupId>
<artifactId>geronimo-jcdi_1.0_spec</artifactId>
</exclusion>
<exclusion>
<groupId>org.apache.geronimo.specs</groupId>
<artifactId>geronimo-atinject_1.0_spec</artifactId>
</exclusion>
</exclusions>
</dependency>
<!-- Upgrade xbean to 4.5 to prevent incompatibilities due to ASM versions --> <!-- JODA Time -->
<dependency> <dependency>
<groupId>org.apache.xbean</groupId> <groupId>joda-time</groupId>
<artifactId>xbean-asm5-shaded</artifactId> <artifactId>joda-time</artifactId>
<version>4.5</version> <version>${jodatime}</version>
<scope>test</scope> </dependency>
</dependency>
<dependency> <!-- Elasticsearch -->
<groupId>javax.servlet</groupId> <dependency>
<artifactId>servlet-api</artifactId> <groupId>org.elasticsearch.client</groupId>
<version>3.0-alpha-1</version> <artifactId>transport</artifactId>
<scope>test</scope> <version>${elasticsearch}</version>
</dependency> <exclusions>
<exclusion>
<groupId>commons-logging</groupId>
<artifactId>commons-logging</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency> <dependency>
<!-- required by elasticsearch --> <!-- required by elasticsearch -->
<groupId>org.elasticsearch.plugin</groupId> <groupId>org.elasticsearch.plugin</groupId>
<artifactId>transport-netty4-client</artifactId> <artifactId>transport-netty4-client</artifactId>
<version>${elasticsearch}</version> <version>${elasticsearch}</version>
<!--<scope>test</scope>--> </dependency>
</dependency>
</dependencies> <dependency>
<groupId>org.elasticsearch.client</groupId>
<artifactId>elasticsearch-rest-high-level-client</artifactId>
<version>${elasticsearch}</version>
<exclusions>
<exclusion>
<groupId>commons-logging</groupId>
<artifactId>commons-logging</artifactId>
</exclusion>
</exclusions>
</dependency>
<build> <dependency>
<plugins> <groupId>org.slf4j</groupId>
<plugin> <artifactId>log4j-over-slf4j</artifactId>
<groupId>org.apache.maven.plugins</groupId> <version>${slf4j}</version>
<artifactId>maven-assembly-plugin</artifactId> <scope>test</scope>
</plugin> </dependency>
<plugin>
<groupId>org.codehaus.mojo</groupId>
<artifactId>wagon-maven-plugin</artifactId>
</plugin>
<plugin>
<groupId>org.asciidoctor</groupId>
<artifactId>asciidoctor-maven-plugin</artifactId>
</plugin>
<!--
please do not remove this configuration for surefire - we need that to avoid issue with jar hell
-->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<configuration>
<useSystemClassLoader>true</useSystemClassLoader>
<useFile>false</useFile>
<includes>
<include>**/*Tests.java</include>
</includes>
</configuration>
</plugin>
</plugins>
</build>
<profiles> <dependency>
<profile> <groupId>org.apache.logging.log4j</groupId>
<id>release</id> <artifactId>log4j-core</artifactId>
<build> <version>${log4j}</version>
<plugins> <scope>test</scope>
<plugin> </dependency>
<groupId>org.jfrog.buildinfo</groupId>
<artifactId>artifactory-maven-plugin</artifactId>
<inherited>false</inherited>
</plugin>
</plugins>
</build>
</profile>
</profiles>
<developers> <!-- Jackson JSON Mapper -->
<developer> <dependency>
<id>biomedcentral</id> <groupId>com.fasterxml.jackson.core</groupId>
<name>BioMed Central Development Team</name> <artifactId>jackson-core</artifactId>
<timezone>+0</timezone> </dependency>
</developer> <dependency>
</developers> <groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
</dependency>
<repositories> <!-- CDI -->
<repository> <dependency>
<id>spring-libs-snapshot</id> <groupId>javax.enterprise</groupId>
<url>https://repo.spring.io/libs-snapshot</url> <artifactId>cdi-api</artifactId>
</repository> <version>${cdi}</version>
</repositories> <scope>provided</scope>
<optional>true</optional>
</dependency>
<pluginRepositories> <!-- Test -->
<pluginRepository> <dependency>
<id>spring-plugins-release</id> <groupId>org.springframework</groupId>
<url>https://repo.spring.io/plugins-release</url> <artifactId>spring-test</artifactId>
</pluginRepository> <scope>test</scope>
</pluginRepositories> <exclusions>
<exclusion>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
</exclusion>
</exclusions>
</dependency>
<scm> <dependency>
<url>https://github.com/spring-projects/spring-data-elasticsearch</url> <groupId>org.apache.openwebbeans.test</groupId>
<connection>scm:git:git://github.com/spring-projects/spring-data-elasticsearch.git</connection> <artifactId>cditest-owb</artifactId>
<developerConnection>scm:git:ssh://git@github.com/spring-projects/spring-data-elasticsearch.git <version>1.2.8</version>
</developerConnection> <scope>test</scope>
</scm> <exclusions>
<exclusion>
<groupId>org.apache.geronimo.specs</groupId>
<artifactId>geronimo-jcdi_1.0_spec</artifactId>
</exclusion>
<exclusion>
<groupId>org.apache.geronimo.specs</groupId>
<artifactId>geronimo-atinject_1.0_spec</artifactId>
</exclusion>
</exclusions>
</dependency>
<ciManagement> <!-- Upgrade xbean to 4.5 to prevent incompatibilities due to ASM versions -->
<system>Bamboo</system> <dependency>
<url>https://build.spring.io/browse/SPRINGDATAES</url> <groupId>org.apache.xbean</groupId>
</ciManagement> <artifactId>xbean-asm5-shaded</artifactId>
<version>4.5</version>
<scope>test</scope>
</dependency>
<issueManagement> <dependency>
<system>JIRA</system> <groupId>javax.servlet</groupId>
<url>https://jira.spring.io/browse/DATAES</url> <artifactId>servlet-api</artifactId>
</issueManagement> <version>3.0-alpha-1</version>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-assembly-plugin</artifactId>
</plugin>
<plugin>
<groupId>org.codehaus.mojo</groupId>
<artifactId>wagon-maven-plugin</artifactId>
</plugin>
<plugin>
<groupId>org.asciidoctor</groupId>
<artifactId>asciidoctor-maven-plugin</artifactId>
</plugin>
<!--
please do not remove this configuration for surefire - we need that to avoid issue with jar hell
-->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<configuration>
<useSystemClassLoader>true</useSystemClassLoader>
<useFile>false</useFile>
<includes>
<include>**/*Tests.java</include>
</includes>
</configuration>
</plugin>
</plugins>
</build>
<repositories>
<repository>
<id>spring-libs-snapshot</id>
<url>https://repo.spring.io/libs-snapshot</url>
</repository>
</repositories>
<pluginRepositories>
<pluginRepository>
<id>spring-plugins-release</id>
<url>https://repo.spring.io/plugins-release</url>
</pluginRepository>
</pluginRepositories>
</project> </project>

189
src/main/java/org/springframework/data/elasticsearch/client/ClientConfiguration.java Normal file
View File

@ -0,0 +1,189 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.elasticsearch.client;
import java.net.InetSocketAddress;
import java.util.List;
import java.util.Optional;
import javax.net.ssl.SSLContext;
import org.springframework.http.HttpHeaders;
/**
* Configuration interface exposing common client configuration properties for Elasticsearch clients.
*
* @author Mark Paluch
* @since 4.0
*/
public interface ClientConfiguration {
/**
* Creates a new {@link ClientConfigurationBuilder} instance.
*
* @return a new {@link ClientConfigurationBuilder} instance.
*/
static ClientConfigurationBuilderWithRequiredEndpoint builder() {
return new ClientConfigurationBuilder();
}
/**
* Creates a new {@link ClientConfiguration} instance configured to a single host given {@code hostAndPort}.
* <p/>
* For example given the endpoint http://localhost:9200
*
* <pre class="code">
* ClientConfiguration configuration = ClientConfiguration.create("localhost:9200");
* </pre>
*
* @return a new {@link ClientConfigurationBuilder} instance.
*/
static ClientConfiguration create(String hostAndPort) {
return new ClientConfigurationBuilder().connectedTo(hostAndPort).build();
}
/**
* Creates a new {@link ClientConfiguration} instance configured to a single host given {@link InetSocketAddress}.
* <p/>
* For example given the endpoint http://localhost:9200
*
* <pre class="code">
* ClientConfiguration configuration = ClientConfiguration
* .create(InetSocketAddress.createUnresolved("localhost", 9200));
* </pre>
*
* @return a new {@link ClientConfigurationBuilder} instance.
*/
static ClientConfiguration create(InetSocketAddress socketAddress) {
return new ClientConfigurationBuilder().connectedTo(socketAddress).build();
}
/**
* Returns the configured endpoints.
*
* @return the configured endpoints.
*/
List<InetSocketAddress> getEndpoints();
/**
* Obtain the {@link HttpHeaders} to be used by default.
*
* @return the {@link HttpHeaders} to be used by default.
*/
HttpHeaders getDefaultHeaders();
/**
* Returns {@literal true} when the client should use SSL.
*
* @return {@literal true} when the client should use SSL.
*/
boolean useSsl();
/**
* Returns the {@link SSLContext} to use. Can be {@link Optional#empty()} if unconfigured.
*
* @return the {@link SSLContext} to use. Can be {@link Optional#empty()} if unconfigured.
*/
Optional<SSLContext> getSslContext();
/**
* @author Christoph Strobl
*/
interface ClientConfigurationBuilderWithRequiredEndpoint {
/**
* @param hostAndPort the {@literal host} and {@literal port} formatted as String {@literal host:port}.
* @return the {@link MaybeSecureClientConfigurationBuilder}.
*/
default MaybeSecureClientConfigurationBuilder connectedTo(String hostAndPort) {
return connectedTo(new String[] { hostAndPort });
}
/**
* @param hostAndPorts the list of {@literal host} and {@literal port} combinations formatted as String
* {@literal host:port}.
* @return the {@link MaybeSecureClientConfigurationBuilder}.
*/
MaybeSecureClientConfigurationBuilder connectedTo(String... hostAndPorts);
/**
* @param endpoint the {@literal host} and {@literal port}.
* @return the {@link MaybeSecureClientConfigurationBuilder}.
*/
default MaybeSecureClientConfigurationBuilder connectedTo(InetSocketAddress endpoint) {
return connectedTo(new InetSocketAddress[] { endpoint });
}
/**
* @param endpoints the list of {@literal host} and {@literal port} combinations.
* @return the {@link MaybeSecureClientConfigurationBuilder}.
*/
MaybeSecureClientConfigurationBuilder connectedTo(InetSocketAddress... endpoints);
/**
* Obviously for testing.
*
* @return the {@link MaybeSecureClientConfigurationBuilder}.
*/
default MaybeSecureClientConfigurationBuilder connectedToLocalhost() {
return connectedTo("localhost:9200");
}
}
/**
* @author Christoph Strobl
*/
interface MaybeSecureClientConfigurationBuilder extends ClientConfigurationBuilderWithOptionalDefaultHeaders {
/**
* Connect via {@literal https} <br />
* <strong>NOTE</strong> You need to leave out the protocol in
* {@link ClientConfigurationBuilderWithRequiredEndpoint#connectedTo(String)}.
*
* @return the {@link ClientConfigurationBuilderWithOptionalDefaultHeaders}.
*/
ClientConfigurationBuilderWithOptionalDefaultHeaders usingSsl();
/**
* Connect via {@literal https} using the given {@link SSLContext}.<br />
* <strong>NOTE</strong> You need to leave out the protocol in
* {@link ClientConfigurationBuilderWithRequiredEndpoint#connectedTo(String)}.
*
* @return the {@link ClientConfigurationBuilderWithOptionalDefaultHeaders}.
*/
ClientConfigurationBuilderWithOptionalDefaultHeaders usingSsl(SSLContext sslContext);
}
/**
* @author Christoph Strobl
* @author Mark Paluch
*/
interface ClientConfigurationBuilderWithOptionalDefaultHeaders {
/**
* @param defaultHeaders must not be {@literal null}.
* @return the {@link ClientConfigurationBuilderWithOptionalDefaultHeaders}
*/
ClientConfigurationBuilderWithOptionalDefaultHeaders withDefaultHeaders(HttpHeaders defaultHeaders);
/**
* Build the {@link ClientConfiguration} object.
*
* @return the {@link ClientConfiguration} object.
*/
ClientConfiguration build();
}
}

125
src/main/java/org/springframework/data/elasticsearch/client/ClientConfigurationBuilder.java Normal file
View File

@ -0,0 +1,125 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.elasticsearch.client;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;
import javax.net.ssl.SSLContext;
import org.springframework.data.elasticsearch.client.ClientConfiguration.ClientConfigurationBuilderWithOptionalDefaultHeaders;
import org.springframework.data.elasticsearch.client.ClientConfiguration.ClientConfigurationBuilderWithRequiredEndpoint;
import org.springframework.data.elasticsearch.client.ClientConfiguration.MaybeSecureClientConfigurationBuilder;
import org.springframework.http.HttpHeaders;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
/**
* Default builder implementation for {@link ClientConfiguration}.
*
* @author Christoph Strobl
* @author Mark Paluch
* @since 4.0
*/
class ClientConfigurationBuilder
implements ClientConfigurationBuilderWithRequiredEndpoint, MaybeSecureClientConfigurationBuilder {
private List<InetSocketAddress> hosts = new ArrayList<>();
private HttpHeaders headers = HttpHeaders.EMPTY;
private boolean useSsl;
private @Nullable SSLContext sslContext;
/*
* (non-Javadoc)
* @see org.springframework.data.elasticsearch.client.ClientConfiguration.ClientConfigurationBuilderWithRequiredEndpoint#connectedTo(java.lang.String[])
*/
@Override
public MaybeSecureClientConfigurationBuilder connectedTo(String... hostAndPorts) {
Assert.notEmpty(hostAndPorts, "At least one host is required");
this.hosts.addAll(Arrays.stream(hostAndPorts).map(ClientConfigurationBuilder::parse).collect(Collectors.toList()));
return this;
}
/*
* (non-Javadoc)
* @see org.springframework.data.elasticsearch.client.ClientConfiguration.ClientConfigurationBuilderWithRequiredEndpoint#connectedTo(java.net.InetSocketAddress[])
*/
@Override
public MaybeSecureClientConfigurationBuilder connectedTo(InetSocketAddress... endpoints) {
Assert.notEmpty(endpoints, "At least one endpoint is required");
this.hosts.addAll(Arrays.asList(endpoints));
return this;
}
/*
* (non-Javadoc)
* @see org.springframework.data.elasticsearch.client.ClientConfiguration.MaybeSecureClientConfigurationBuilder#usingSsl()
*/
@Override
public ClientConfigurationBuilderWithOptionalDefaultHeaders usingSsl() {
this.useSsl = true;
return this;
}
/*
* (non-Javadoc)
* @see org.springframework.data.elasticsearch.client.ClientConfiguration.MaybeSecureClientConfigurationBuilder#usingSsl(javax.net.ssl.SSLContext)
*/
@Override
public ClientConfigurationBuilderWithOptionalDefaultHeaders usingSsl(SSLContext sslContext) {
Assert.notNull(sslContext, "SSL Context must not be null");
this.useSsl = true;
this.sslContext = sslContext;
return this;
}
/*
* (non-Javadoc)
* @see org.springframework.data.elasticsearch.client.ClientConfiguration.ClientConfigurationBuilderWithOptionalDefaultHeaders#withDefaultHeaders(org.springframework.http.HttpHeaders)
*/
@Override
public ClientConfigurationBuilderWithOptionalDefaultHeaders withDefaultHeaders(HttpHeaders defaultHeaders) {
Assert.notNull(defaultHeaders, "Default HTTP headers must not be null");
this.headers = defaultHeaders;
return this;
}
/*
* (non-Javadoc)
* @see org.springframework.data.elasticsearch.client.ClientConfiguration.ClientConfigurationBuilderWithOptionalDefaultHeaders#build()
*/
@Override
public ClientConfiguration build() {
return new DefaultClientConfiguration(this.hosts, this.headers, this.useSsl, this.sslContext);
}
private static InetSocketAddress parse(String hostAndPort) {
return InetSocketAddressParser.parse(hostAndPort, ElasticsearchHost.DEFAULT_PORT);
}
}

86
src/main/java/org/springframework/data/elasticsearch/client/DefaultClientConfiguration.java Normal file
View File

@ -0,0 +1,86 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.elasticsearch.client;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
import javax.net.ssl.SSLContext;
import org.springframework.http.HttpHeaders;
import org.springframework.lang.Nullable;
/**
* Default {@link ClientConfiguration} implementation.
*
* @author Mark Paluch
* @since 4.0
*/
class DefaultClientConfiguration implements ClientConfiguration {
private final List<InetSocketAddress> hosts;
private final HttpHeaders headers;
private final boolean useSsl;
private final @Nullable SSLContext sslContext;
DefaultClientConfiguration(List<InetSocketAddress> hosts, HttpHeaders headers, boolean useSsl,
@Nullable SSLContext sslContext) {
this.hosts = Collections.unmodifiableList(new ArrayList<>(hosts));
this.headers = new HttpHeaders(headers);
this.useSsl = useSsl;
this.sslContext = sslContext;
}
/*
* (non-Javadoc)
* @see org.springframework.data.elasticsearch.client.ClientConfiguration#getEndpoints()
*/
@Override
public List<InetSocketAddress> getEndpoints() {
return this.hosts;
}
/*
* (non-Javadoc)
* @see org.springframework.data.elasticsearch.client.ClientConfiguration#getDefaultHeaders()
*/
@Override
public HttpHeaders getDefaultHeaders() {
return this.headers;
}
/*
* (non-Javadoc)
* @see org.springframework.data.elasticsearch.client.ClientConfiguration#useSsl()
*/
@Override
public boolean useSsl() {
return this.useSsl;
}
/*
* (non-Javadoc)
* @see org.springframework.data.elasticsearch.client.ClientConfiguration#getSslContext()
*/
@Override
public Optional<SSLContext> getSslContext() {
return Optional.ofNullable(this.sslContext);
}
}

197
src/main/java/org/springframework/data/elasticsearch/client/ElasticsearchClients.java
View File

@ -1,197 +0,0 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.elasticsearch.client;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;
import org.apache.http.Header;
import org.apache.http.HttpHost;
import org.apache.http.message.BasicHeader;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.data.elasticsearch.client.reactive.DefaultReactiveElasticsearchClient;
import org.springframework.data.elasticsearch.client.reactive.ReactiveElasticsearchClient;
import org.springframework.http.HttpHeaders;
import org.springframework.util.Assert;
/**
* Utility class for common access to Elasticsearch clients. {@link ElasticsearchClients} consolidates set up routines
* for the various drivers into a single place.
*
* @author Christoph Strobl
* @since 4.0
*/
public final class ElasticsearchClients {
private ElasticsearchClients() {}
/**
* Start here to create a new client tailored to your needs.
*
* @return new instance of {@link ClientBuilderWithRequiredHost}.
*/
public static ClientBuilderWithRequiredHost createClient() {
return new ElasticsearchClientBuilderImpl();
}
/**
* @author Christoph Strobl
*/
public interface ElasticsearchClientBuilder {
/**
* Apply the configuration to create a {@link ReactiveElasticsearchClient}.
*
* @return new instance of {@link ReactiveElasticsearchClient}.
*/
ReactiveElasticsearchClient reactive();
/**
* Apply the configuration to create a {@link RestHighLevelClient}.
*
* @return new instance of {@link RestHighLevelClient}.
*/
RestHighLevelClient rest();
/**
* Apply the configuration to create a {@link RestClient}.
*
* @return new instance of {@link RestClient}.
*/
default RestClient lowLevelRest() {
return rest().getLowLevelClient();
}
}
/**
* @author Christoph Strobl
*/
public interface ClientBuilderWithRequiredHost {
/**
* @param host the {@literal host} and {@literal port} formatted as String {@literal host:port}. You may leave out
* {@literal http / https} and use {@link MaybeSecureClientBuilder#viaSsl() viaSsl}.
* @return the {@link MaybeSecureClientBuilder}.
*/
default MaybeSecureClientBuilder connectedTo(String host) {
return connectedTo(new String[] { host });
}
/**
* @param hosts the list of {@literal host} and {@literal port} combinations formatted as String
* {@literal host:port}. You may leave out {@literal http / https} and use
* {@link MaybeSecureClientBuilder#viaSsl() viaSsl}.
* @return the {@link MaybeSecureClientBuilder}.
*/
MaybeSecureClientBuilder connectedTo(String... hosts);
/**
* Obviously for testing.
*
* @return the {@link MaybeSecureClientBuilder}.
*/
default MaybeSecureClientBuilder connectedToLocalhost() {
return connectedTo("localhost:9200");
}
}
/**
* @author Christoph Strobl
*/
public interface MaybeSecureClientBuilder extends ClientBuilderWithOptionalDefaultHeaders {
/**
* Connect via {@literal https} <br />
* <strong>NOTE</strong> You need to leave out the protocol in
* {@link ClientBuilderWithRequiredHost#connectedTo(String)}.
*
* @return the {@link ClientBuilderWithOptionalDefaultHeaders}.
*/
ClientBuilderWithOptionalDefaultHeaders viaSsl();
}
/**
* @author Christoph Strobl
*/
public interface ClientBuilderWithOptionalDefaultHeaders extends ElasticsearchClientBuilder {
/**
* @param defaultHeaders
* @return the {@link ElasticsearchClientBuilder}
*/
ElasticsearchClientBuilder withDefaultHeaders(HttpHeaders defaultHeaders);
}
private static class ElasticsearchClientBuilderImpl
implements ElasticsearchClientBuilder, ClientBuilderWithRequiredHost, MaybeSecureClientBuilder {
private List<String> hosts = new ArrayList<>();
private HttpHeaders headers = HttpHeaders.EMPTY;
private String protocoll = "http";
@Override
public ReactiveElasticsearchClient reactive() {
return DefaultReactiveElasticsearchClient.create(headers, formattedHosts().toArray(new String[0]));
}
@Override
public RestHighLevelClient rest() {
HttpHost[] httpHosts = formattedHosts().stream().map(HttpHost::create).toArray(HttpHost[]::new);
RestClientBuilder builder = RestClient.builder(httpHosts);
if (!headers.isEmpty()) {
Header[] httpHeaders = headers.toSingleValueMap().entrySet().stream()
.map(it -> new BasicHeader(it.getKey(), it.getValue())).toArray(Header[]::new);
builder = builder.setDefaultHeaders(httpHeaders);
}
return new RestHighLevelClient(builder);
}
@Override
public MaybeSecureClientBuilder connectedTo(String... hosts) {
Assert.notEmpty(hosts, "At least one host is required.");
this.hosts.addAll(Arrays.asList(hosts));
return this;
}
@Override
public ClientBuilderWithOptionalDefaultHeaders withDefaultHeaders(HttpHeaders defaultHeaders) {
Assert.notNull(defaultHeaders, "DefaultHeaders must not be null!");
this.headers = defaultHeaders;
return this;
}
List<String> formattedHosts() {
return hosts.stream().map(it -> it.startsWith("http") ? it : protocoll + "://" + it).collect(Collectors.toList());
}
@Override
public ClientBuilderWithOptionalDefaultHeaders viaSsl() {
this.protocoll = "https";
return this;
}
}
}

37
src/main/java/org/springframework/data/elasticsearch/client/ElasticsearchHost.java
View File

@ -16,8 +16,11 @@
package org.springframework.data.elasticsearch.client; package org.springframework.data.elasticsearch.client;
import java.net.InetSocketAddress;
import java.time.Instant; import java.time.Instant;
import org.springframework.util.Assert;
/** /**
* Value Object containing information about Elasticsearch cluster nodes. * Value Object containing information about Elasticsearch cluster nodes.
* *
@ -26,13 +29,21 @@ import java.time.Instant;
*/ */
public class ElasticsearchHost { public class ElasticsearchHost {
private final String host; /**
* Default HTTP port for Elasticsearch servers.
*/
public static final int DEFAULT_PORT = 9200;
private final InetSocketAddress endpoint;
private final State state; private final State state;
private final Instant timestamp; private final Instant timestamp;
public ElasticsearchHost(String host, State state) { public ElasticsearchHost(InetSocketAddress endpoint, State state) {
this.host = host; Assert.notNull(endpoint, "Host must not be null");
Assert.notNull(state, "State must not be null");
this.endpoint = endpoint;
this.state = state; this.state = state;
this.timestamp = Instant.now(); this.timestamp = Instant.now();
} }
@ -41,7 +52,7 @@ public class ElasticsearchHost {
* @param host must not be {@literal null}. * @param host must not be {@literal null}.
* @return new instance of {@link ElasticsearchHost}. * @return new instance of {@link ElasticsearchHost}.
*/ */
public static ElasticsearchHost online(String host) { public static ElasticsearchHost online(InetSocketAddress host) {
return new ElasticsearchHost(host, State.ONLINE); return new ElasticsearchHost(host, State.ONLINE);
} }
@ -49,10 +60,20 @@ public class ElasticsearchHost {
* @param host must not be {@literal null}. * @param host must not be {@literal null}.
* @return new instance of {@link ElasticsearchHost}. * @return new instance of {@link ElasticsearchHost}.
*/ */
public static ElasticsearchHost offline(String host) { public static ElasticsearchHost offline(InetSocketAddress host) {
return new ElasticsearchHost(host, State.OFFLINE); return new ElasticsearchHost(host, State.OFFLINE);
} }
/**
* Parse a {@literal hostAndPort} string into a {@link InetSocketAddress}.
*
* @param hostAndPort the string containing host and port or IP address and port in the format {@code host:port}.
* @return the parsed {@link InetSocketAddress}.
*/
public static InetSocketAddress parse(String hostAndPort) {
return InetSocketAddressParser.parse(hostAndPort, DEFAULT_PORT);
}
/** /**
* @return {@literal true} if the last known {@link State} was {@link State#ONLINE} * @return {@literal true} if the last known {@link State} was {@link State#ONLINE}
*/ */
@ -63,8 +84,8 @@ public class ElasticsearchHost {
/** /**
* @return never {@literal null}. * @return never {@literal null}.
*/ */
public String getHost() { public InetSocketAddress getEndpoint() {
return host; return endpoint;
} }
/** /**
@ -83,7 +104,7 @@ public class ElasticsearchHost {
@Override @Override
public String toString() { public String toString() {
return "ElasticsearchHost(" + host + ", " + state.name() + ")"; return "ElasticsearchHost(" + endpoint + ", " + state.name() + ")";
} }
public enum State { public enum State {

117
src/main/java/org/springframework/data/elasticsearch/client/InetSocketAddressParser.java Normal file
View File

@ -0,0 +1,117 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.elasticsearch.client;
import java.net.InetSocketAddress;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
* Utility to parse endpoints in {@code host:port} format into {@link java.net.InetSocketAddress}.
*
* @author Mark Paluch
* @since 4.0
*/
class InetSocketAddressParser {
/**
* Parse a host and port string into a {@link InetSocketAddress}.
*
* @param hostPortString Hostname/IP address and port formatted as {@code host:port} or {@code host}.
* @param defaultPort default port to apply if {@code hostPostString} does not contain a port.
* @return a {@link InetSocketAddress} that is unresolved to avoid DNS lookups.
* @see InetSocketAddress#createUnresolved(String, int)
*/
static InetSocketAddress parse(String hostPortString, int defaultPort) {
Assert.notNull(hostPortString, "HostPortString must not be null");
String host;
String portString = null;
if (hostPortString.startsWith("[")) {
String[] hostAndPort = getHostAndPortFromBracketedHost(hostPortString);
host = hostAndPort[0];
portString = hostAndPort[1];
} else {
int colonPos = hostPortString.indexOf(':');
if (colonPos >= 0 && hostPortString.indexOf(':', colonPos + 1) == -1) {
// Exactly 1 colon. Split into host:port.
host = hostPortString.substring(0, colonPos);
portString = hostPortString.substring(colonPos + 1);
} else {
// 0 or 2+ colons. Bare hostname or IPv6 literal.
host = hostPortString;
}
}
int port = defaultPort;
if (StringUtils.hasText(portString)) {
// Try to parse the whole port string as a number.
Assert.isTrue(!portString.startsWith("+"), String.format("Cannot parse port number: %s", hostPortString));
try {
port = Integer.parseInt(portString);
} catch (NumberFormatException e) {
throw new IllegalArgumentException(String.format("Cannot parse port number: %s", hostPortString));
}
Assert.isTrue(isValidPort(port), String.format("Port number out of range: %s", hostPortString));
}
return InetSocketAddress.createUnresolved(host, port);
}
/**
* Parses a bracketed host-port string, throwing IllegalArgumentException if parsing fails.
*
* @param hostPortString the full bracketed host-port specification. Post might not be specified.
* @return an array with 2 strings: host and port, in that order.
* @throws IllegalArgumentException if parsing the bracketed host-port string fails.
*/
private static String[] getHostAndPortFromBracketedHost(String hostPortString) {
Assert.isTrue(hostPortString.charAt(0) == '[',
String.format("Bracketed host-port string must start with a bracket: %s", hostPortString));
int colonIndex = hostPortString.indexOf(':');
int closeBracketIndex = hostPortString.lastIndexOf(']');
Assert.isTrue(colonIndex > -1 && closeBracketIndex > colonIndex,
String.format("Invalid bracketed host/port: %s", hostPortString));
String host = hostPortString.substring(1, closeBracketIndex);
if (closeBracketIndex + 1 == hostPortString.length()) {
return new String[] { host, "" };
} else {
Assert.isTrue(hostPortString.charAt(closeBracketIndex + 1) == ':',
"Only a colon may follow a close bracket: " + hostPortString);
for (int i = closeBracketIndex + 2; i < hostPortString.length(); ++i) {
Assert.isTrue(Character.isDigit(hostPortString.charAt(i)),
String.format("Port must be numeric: %s", hostPortString));
}
return new String[] { host, hostPortString.substring(closeBracketIndex + 2) };
}
}
/**
* @param port the port number
* @return {@literal true} for valid port numbers.
*/
private static boolean isValidPort(int port) {
return port >= 0 && port <= 65535;
}
}

2
src/main/java/org/springframework/data/elasticsearch/client/NoReachableHostException.java
View File

@ -37,7 +37,7 @@ public class NoReachableHostException extends RuntimeException {
private static String createMessage(Set<ElasticsearchHost> hosts) { private static String createMessage(Set<ElasticsearchHost> hosts) {
if (hosts.size() == 1) { if (hosts.size() == 1) {
return String.format("Host '%s' not reachable. Cluster state is offline.", hosts.iterator().next().getHost()); return String.format("Host '%s' not reachable. Cluster state is offline.", hosts.iterator().next().getEndpoint());
} }
return String.format("No active host found in cluster. (%s) of (%s) nodes offline.", hosts.size(), hosts.size()); return String.format("No active host found in cluster. (%s) of (%s) nodes offline.", hosts.size(), hosts.size());

111
src/main/java/org/springframework/data/elasticsearch/client/RestClients.java Normal file
View File

@ -0,0 +1,111 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.elasticsearch.client;
import java.io.Closeable;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.List;
import java.util.Optional;
import java.util.stream.Collectors;
import javax.net.ssl.SSLContext;
import org.apache.http.Header;
import org.apache.http.HttpHost;
import org.apache.http.message.BasicHeader;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.http.HttpHeaders;
import org.springframework.util.Assert;
/**
* Utility class for common access to Elasticsearch clients. {@link RestClients} consolidates set up routines for the
* various drivers into a single place.
*
* @author Christoph Strobl
* @author Mark Paluch
* @since 4.0
*/
public final class RestClients {
private RestClients() {}
/**
* Start here to create a new client tailored to your needs.
*
* @return new instance of {@link ElasticsearchRestClient}.
*/
public static ElasticsearchRestClient create(ClientConfiguration clientConfiguration) {
Assert.notNull(clientConfiguration, "ClientConfiguration must not be null!");
HttpHost[] httpHosts = formattedHosts(clientConfiguration.getEndpoints(), clientConfiguration.useSsl()).stream()
.map(HttpHost::create).toArray(HttpHost[]::new);
RestClientBuilder builder = RestClient.builder(httpHosts);
HttpHeaders headers = clientConfiguration.getDefaultHeaders();
if (!headers.isEmpty()) {
Header[] httpHeaders = headers.toSingleValueMap().entrySet().stream()
.map(it -> new BasicHeader(it.getKey(), it.getValue())).toArray(Header[]::new);
builder.setDefaultHeaders(httpHeaders);
}
builder.setHttpClientConfigCallback(clientBuilder -> {
Optional<SSLContext> sslContext = clientConfiguration.getSslContext();
sslContext.ifPresent(clientBuilder::setSSLContext);
return clientBuilder;
});
RestHighLevelClient client = new RestHighLevelClient(builder);
return () -> client;
}
private static List<String> formattedHosts(List<InetSocketAddress> hosts, boolean useSsl) {
return hosts.stream().map(it -> (useSsl ? "https" : "http") + "://" + it).collect(Collectors.toList());
}
/**
* @author Christoph Strobl
*/
@FunctionalInterface
public interface ElasticsearchRestClient extends Closeable {
/**
* Apply the configuration to create a {@link RestHighLevelClient}.
*
* @return new instance of {@link RestHighLevelClient}.
*/
RestHighLevelClient rest();
/**
* Apply the configuration to create a {@link RestClient}.
*
* @return new instance of {@link RestClient}.
*/
default RestClient lowLevelRest() {
return rest().getLowLevelClient();
}
@Override
default void close() throws IOException {
rest().close();
}
}
}

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

@ -13,22 +13,25 @@
* See the License for the specific language governing permissions and * See the License for the specific language governing permissions and
* limitations under the License. * limitations under the License.
*/ */
package org.springframework.data.elasticsearch.client.reactive; package org.springframework.data.elasticsearch.client.reactive;
import org.springframework.data.elasticsearch.client.ElasticsearchHost; import io.netty.handler.ssl.ClientAuth;
import org.springframework.data.elasticsearch.client.NoReachableHostException; import io.netty.handler.ssl.JdkSslContext;
import reactor.core.publisher.Flux; import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono; import reactor.core.publisher.Mono;
import reactor.netty.http.client.HttpClient;
import java.io.IOException; import java.io.IOException;
import java.lang.reflect.Method; import java.lang.reflect.Method;
import java.net.ConnectException; import java.net.ConnectException;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets; import java.nio.charset.StandardCharsets;
import java.util.Collection; import java.util.Collection;
import java.util.List; import java.util.Optional;
import java.util.function.Function; import java.util.function.Function;
import javax.net.ssl.SSLContext;
import org.apache.http.util.EntityUtils; import org.apache.http.util.EntityUtils;
import org.elasticsearch.ElasticsearchStatusException; import org.elasticsearch.ElasticsearchStatusException;
import org.elasticsearch.action.ActionRequest; import org.elasticsearch.action.ActionRequest;
@ -57,47 +60,48 @@ import org.elasticsearch.rest.BytesRestResponse;
import org.elasticsearch.rest.RestStatus; import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.SearchHit; import org.elasticsearch.search.SearchHit;
import org.reactivestreams.Publisher; import org.reactivestreams.Publisher;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.dao.DataAccessResourceFailureException; import org.springframework.dao.DataAccessResourceFailureException;
import org.springframework.data.elasticsearch.client.ClientConfiguration;
import org.springframework.data.elasticsearch.client.ElasticsearchHost;
import org.springframework.data.elasticsearch.client.NoReachableHostException;
import org.springframework.data.elasticsearch.client.reactive.HostProvider.VerificationMode; import org.springframework.data.elasticsearch.client.reactive.HostProvider.VerificationMode;
import org.springframework.data.elasticsearch.client.util.RequestConverters; import org.springframework.data.elasticsearch.client.util.RequestConverters;
import org.springframework.http.HttpHeaders; import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod; import org.springframework.http.HttpMethod;
import org.springframework.http.HttpStatus; import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType; import org.springframework.http.MediaType;
import org.springframework.http.ResponseCookie; import org.springframework.http.client.reactive.ReactorClientHttpConnector;
import org.springframework.http.ResponseEntity;
import org.springframework.http.client.reactive.ClientHttpResponse;
import org.springframework.http.codec.HttpMessageReader;
import org.springframework.util.Assert; import org.springframework.util.Assert;
import org.springframework.util.MultiValueMap;
import org.springframework.util.ReflectionUtils; import org.springframework.util.ReflectionUtils;
import org.springframework.util.StreamUtils; import org.springframework.web.client.HttpServerErrorException;
import org.springframework.web.client.HttpClientErrorException;
import org.springframework.web.reactive.function.BodyExtractor;
import org.springframework.web.reactive.function.BodyExtractors; import org.springframework.web.reactive.function.BodyExtractors;
import org.springframework.web.reactive.function.client.ClientResponse; import org.springframework.web.reactive.function.client.ClientResponse;
import org.springframework.web.reactive.function.client.ExchangeStrategies;
import org.springframework.web.reactive.function.client.WebClient; import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.reactive.function.client.WebClient.RequestBodySpec; import org.springframework.web.reactive.function.client.WebClient.RequestBodySpec;
import org.springframework.web.reactive.function.client.WebClientException;
/** /**
* A {@link WebClient} based {@link ReactiveElasticsearchClient} that connects to an Elasticsearch cluster through HTTP. * A {@link WebClient} based {@link ReactiveElasticsearchClient} that connects to an Elasticsearch cluster using HTTP.
* *
* @author Christoph Strobl * @author Christoph Strobl
* @author Mark Paluch
* @since 4.0 * @since 4.0
* @see ClientConfiguration
* @see ReactiveRestClients
*/ */
public class DefaultReactiveElasticsearchClient implements ReactiveElasticsearchClient { public class DefaultReactiveElasticsearchClient implements ReactiveElasticsearchClient {
private final HostProvider hostProvider; private final HostProvider hostProvider;
/** /**
* Create a new {@link DefaultReactiveElasticsearchClient} using the given hostProvider to obtain server connections. * Create a new {@link DefaultReactiveElasticsearchClient} using the given {@link HostProvider} to obtain server
* connections.
* *
* @param hostProvider must not be {@literal null}. * @param hostProvider must not be {@literal null}.
*/ */
public DefaultReactiveElasticsearchClient(HostProvider hostProvider) { public DefaultReactiveElasticsearchClient(HostProvider hostProvider) {
Assert.notNull(hostProvider, "HostProvider must not be null");
this.hostProvider = hostProvider; this.hostProvider = hostProvider;
} }
@ -112,11 +116,53 @@ public class DefaultReactiveElasticsearchClient implements ReactiveElasticsearch
*/ */
public static ReactiveElasticsearchClient create(HttpHeaders headers, String... hosts) { public static ReactiveElasticsearchClient create(HttpHeaders headers, String... hosts) {
Assert.notNull(headers, "HttpHeaders must not be null");
Assert.notEmpty(hosts, "Elasticsearch Cluster needs to consist of at least one host"); Assert.notEmpty(hosts, "Elasticsearch Cluster needs to consist of at least one host");
HostProvider hostProvider = HostProvider.provider(hosts); ClientConfiguration clientConfiguration = ClientConfiguration.builder().connectedTo(hosts)
return new DefaultReactiveElasticsearchClient( .withDefaultHeaders(headers).build();
headers.isEmpty() ? hostProvider : hostProvider.withDefaultHeaders(headers)); return create(clientConfiguration);
}
/**
* Create a new {@link DefaultReactiveElasticsearchClient} given {@link ClientConfiguration}. <br />
* <strong>NOTE</strong> If the cluster requires authentication be sure to provide the according {@link HttpHeaders}
* correctly.
*
* @param clientConfiguration Client configuration. Must not be {@literal null}.
* @return new instance of {@link DefaultReactiveElasticsearchClient}.
*/
public static ReactiveElasticsearchClient create(ClientConfiguration clientConfiguration) {
Assert.notNull(clientConfiguration, "ClientConfiguration must not be null");
WebClientProvider provider = getWebClientProvider(clientConfiguration);
HostProvider hostProvider = HostProvider.provider(provider,
clientConfiguration.getEndpoints().toArray(new InetSocketAddress[0]));
return new DefaultReactiveElasticsearchClient(hostProvider);
}
private static WebClientProvider getWebClientProvider(ClientConfiguration clientConfiguration) {
WebClientProvider provider;
if (clientConfiguration.useSsl()) {
ReactorClientHttpConnector connector = new ReactorClientHttpConnector(HttpClient.create().secure(sslConfig -> {
Optional<SSLContext> sslContext = clientConfiguration.getSslContext();
sslContext.ifPresent(it -> {
sslConfig.sslContext(new JdkSslContext(it, true, ClientAuth.NONE));
});
}));
provider = WebClientProvider.create("https", connector);
} else {
provider = WebClientProvider.create("http");
}
return provider.withDefaultHeaders(clientConfiguration.getDefaultHeaders());
} }
/* /*
@ -144,7 +190,7 @@ public class DefaultReactiveElasticsearchClient implements ReactiveElasticsearch
/* /*
* (non-Javadoc) * (non-Javadoc)
* @see org.springframework.data.elasticsearch.client.reactive.ReactiveElasticsearchClient#get(org.springframework.http.HttpHeaderss, org.elasticsearch.action.get.GetRequest) * @see org.springframework.data.elasticsearch.client.reactive.ReactiveElasticsearchClient#get(org.springframework.http.HttpHeaders, org.elasticsearch.action.get.GetRequest)
*/ */
@Override @Override
public Mono<GetResult> get(HttpHeaders headers, GetRequest getRequest) { public Mono<GetResult> get(HttpHeaders headers, GetRequest getRequest) {
@ -177,20 +223,7 @@ public class DefaultReactiveElasticsearchClient implements ReactiveElasticsearch
public Mono<Boolean> exists(HttpHeaders headers, GetRequest getRequest) { public Mono<Boolean> exists(HttpHeaders headers, GetRequest getRequest) {
return sendRequest(getRequest, RequestCreator.exists(), RawActionResponse.class, headers) // return sendRequest(getRequest, RequestCreator.exists(), RawActionResponse.class, headers) //
.map(response -> { .map(response -> response.statusCode().is2xxSuccessful()) //
if (response.statusCode().is2xxSuccessful()) {
return true;
}
if (response.statusCode().is5xxServerError()) {
throw new HttpClientErrorException(response.statusCode(), String.format(
"Exists request (%s) returned error code %s.", getRequest.toString(), response.statusCode().value()));
}
return false;
}) //
.next(); .next();
} }
@ -241,13 +274,13 @@ public class DefaultReactiveElasticsearchClient implements ReactiveElasticsearch
public Mono<ClientResponse> execute(ReactiveElasticsearchClientCallback callback) { public Mono<ClientResponse> execute(ReactiveElasticsearchClientCallback callback) {
return this.hostProvider.getActive(VerificationMode.LAZY) // return this.hostProvider.getActive(VerificationMode.LAZY) //
.flatMap(it -> callback.doWithClient(it)) // .flatMap(callback::doWithClient) //
.onErrorResume(throwable -> { .onErrorResume(throwable -> {
if (throwable instanceof ConnectException) { if (throwable instanceof ConnectException) {
return hostProvider.getActive(VerificationMode.FORCE) // return hostProvider.getActive(VerificationMode.ACTIVE) //
.flatMap(webClient -> callback.doWithClient(webClient)); .flatMap(callback::doWithClient);
} }
return Mono.error(throwable); return Mono.error(throwable);
@ -315,55 +348,60 @@ public class DefaultReactiveElasticsearchClient implements ReactiveElasticsearch
private <T> Publisher<? extends T> readResponseBody(Request request, ClientResponse response, Class<T> responseType) { private <T> Publisher<? extends T> readResponseBody(Request request, ClientResponse response, Class<T> responseType) {
if (RawActionResponse.class.equals(responseType)) { if (RawActionResponse.class.equals(responseType)) {
return Mono.just((T) new RawActionResponse(response)); return Mono.just(responseType.cast(RawActionResponse.create(response)));
} }
if (response.statusCode().is5xxServerError()) { if (response.statusCode().is5xxServerError()) {
return handleServerError(request, response);
throw new HttpClientErrorException(response.statusCode(),
String.format("%s request to %s returned error code %s.", request.getMethod(), request.getEndpoint(),
response.statusCode().value()));
} }
return response.body(BodyExtractors.toDataBuffers()).flatMap(it -> { return response.body(BodyExtractors.toMono(byte[].class)) //
.map(it -> new String(it, StandardCharsets.UTF_8)) //
.flatMap(content -> {
return doDecode(response, responseType, content);
});
}
private static <T> Mono<T> doDecode(ClientResponse response, Class<T> responseType, String content) {
String mediaType = response.headers().contentType().map(MediaType::toString).orElse(XContentType.JSON.mediaType());
try {
XContentParser contentParser = createParser(mediaType, content);
try { try {
String content = StreamUtils.copyToString(it.asInputStream(true), StandardCharsets.UTF_8); Method fromXContent = ReflectionUtils.findMethod(responseType, "fromXContent", XContentParser.class);
return Mono
.justOrEmpty(responseType.cast(ReflectionUtils.invokeMethod(fromXContent, responseType, contentParser)));
} catch (Exception errorParseFailure) {
try { try {
return Mono.error(BytesRestResponse.errorFromXContent(contentParser));
XContentParser contentParser = XContentType } catch (Exception e) {
.fromMediaTypeOrFormat( // return Mono.error to avoid ElasticsearchStatusException to be caught by outer catch.
response.headers().contentType().map(MediaType::toString).orElse(XContentType.JSON.mediaType())) return Mono.error(new ElasticsearchStatusException("Unable to parse response body",
.xContent() RestStatus.fromCode(response.statusCode().value())));
.createParser(NamedXContentRegistry.EMPTY, DeprecationHandler.THROW_UNSUPPORTED_OPERATION, content);
Method fromXContent = ReflectionUtils.findMethod(responseType, "fromXContent", XContentParser.class);
return Mono.just((T) ReflectionUtils.invokeMethod(fromXContent, responseType, contentParser));
} catch (Exception parseFailure) {
try {
XContentParser errorParser = XContentType
.fromMediaTypeOrFormat(
response.headers().contentType().map(MediaType::toString).orElse(XContentType.JSON.mediaType()))
.xContent()
.createParser(NamedXContentRegistry.EMPTY, DeprecationHandler.THROW_UNSUPPORTED_OPERATION, content);
// return Mono.error to avoid ElasticsearchStatusException to be caught by outer catch.
return Mono.error(BytesRestResponse.errorFromXContent(errorParser));
} catch (Exception errorParseFailure) {
// return Mono.error to avoid ElasticsearchStatusException to be caught by outer catch.
return Mono.error(new ElasticsearchStatusException("Unable to parse response body",
RestStatus.fromCode(response.statusCode().value())));
}
} }
} catch (IOException e) {
throw new DataAccessResourceFailureException("Error parsing XContent.", e);
} }
});
} catch (IOException e) {
return Mono.error(new DataAccessResourceFailureException("Error parsing XContent.", e));
}
}
private static XContentParser createParser(String mediaType, String content) throws IOException {
return XContentType.fromMediaTypeOrFormat(mediaType) //
.xContent() //
.createParser(NamedXContentRegistry.EMPTY, DeprecationHandler.THROW_UNSUPPORTED_OPERATION, content);
}
private static <T> Publisher<? extends T> handleServerError(Request request, ClientResponse response) {
return Mono.error(
new HttpServerErrorException(response.statusCode(), String.format("%s request to %s returned error code %s.",
request.getMethod(), request.getEndpoint(), response.statusCode().value())));
} }
static class RequestCreator { static class RequestCreator {
@ -405,97 +443,9 @@ public class DefaultReactiveElasticsearchClient implements ReactiveElasticsearch
} }
} }
public static class RequestBodyEncodingException extends WebClientException {
RequestBodyEncodingException(String msg, Throwable ex) {
super(msg, ex);
}
}
static class RawActionResponse extends ActionResponse implements ClientResponse {
final ClientResponse delegate;
RawActionResponse(ClientResponse delegate) {
this.delegate = delegate;
}
public HttpStatus statusCode() {
return delegate.statusCode();
}
public int rawStatusCode() {
return delegate.rawStatusCode();
}
public Headers headers() {
return delegate.headers();
}
public MultiValueMap<String, ResponseCookie> cookies() {
return delegate.cookies();
}
public ExchangeStrategies strategies() {
return delegate.strategies();
}
public <T> T body(BodyExtractor<T, ? super ClientHttpResponse> extractor) {
return delegate.body(extractor);
}
public <T> Mono<T> bodyToMono(Class<? extends T> elementClass) {
return delegate.bodyToMono(elementClass);
}
public <T> Mono<T> bodyToMono(ParameterizedTypeReference<T> typeReference) {
return delegate.bodyToMono(typeReference);
}
public <T> Flux<T> bodyToFlux(Class<? extends T> elementClass) {
return delegate.bodyToFlux(elementClass);
}
public <T> Flux<T> bodyToFlux(ParameterizedTypeReference<T> typeReference) {
return delegate.bodyToFlux(typeReference);
}
public <T> Mono<ResponseEntity<T>> toEntity(Class<T> bodyType) {
return delegate.toEntity(bodyType);
}
public <T> Mono<ResponseEntity<T>> toEntity(ParameterizedTypeReference<T> typeReference) {
return delegate.toEntity(typeReference);
}
public <T> Mono<ResponseEntity<List<T>>> toEntityList(Class<T> elementType) {
return delegate.toEntityList(elementType);
}
public <T> Mono<ResponseEntity<List<T>>> toEntityList(ParameterizedTypeReference<T> typeReference) {
return delegate.toEntityList(typeReference);
}
public static Builder from(ClientResponse other) {
return ClientResponse.from(other);
}
public static Builder create(HttpStatus statusCode) {
return ClientResponse.create(statusCode);
}
public static Builder create(HttpStatus statusCode, ExchangeStrategies strategies) {
return ClientResponse.create(statusCode, strategies);
}
public static Builder create(HttpStatus statusCode, List<HttpMessageReader<?>> messageReaders) {
return ClientResponse.create(statusCode, messageReaders);
}
}
/** /**
* Reactive client {@link ReactiveElasticsearchClient.Status} implementation. * Reactive client {@link ReactiveElasticsearchClient.Status} implementation.
* *
* @author Christoph Strobl * @author Christoph Strobl
*/ */
class ClientStatus implements Status { class ClientStatus implements Status {

133
src/main/java/org/springframework/data/elasticsearch/client/reactive/DefaultWebClientProvider.java Normal file
View File

@ -0,0 +1,133 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.elasticsearch.client.reactive;
import java.net.InetSocketAddress;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;
import org.springframework.http.HttpHeaders;
import org.springframework.http.client.reactive.ClientHttpConnector;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.reactive.function.client.WebClient.Builder;
/**
* Default {@link WebClientProvider} that uses cached {@link WebClient} instances per {@code hostAndPort}.
*
* @author Mark Paluch
* @since 4.0
*/
class DefaultWebClientProvider implements WebClientProvider {
private final Map<InetSocketAddress, WebClient> cachedClients;
private final String scheme;
private final @Nullable ClientHttpConnector connector;
private final Consumer<Throwable> errorListener;
private final HttpHeaders headers;
DefaultWebClientProvider(String scheme, @Nullable ClientHttpConnector connector) {
this(scheme, connector, e -> {}, HttpHeaders.EMPTY);
}
private DefaultWebClientProvider(String scheme, @Nullable ClientHttpConnector connector,
Consumer<Throwable> errorListener, HttpHeaders headers) {
this(new ConcurrentHashMap<>(), scheme, connector, errorListener, headers);
}
private DefaultWebClientProvider(Map<InetSocketAddress, WebClient> cachedClients, String scheme,
@Nullable ClientHttpConnector connector, Consumer<Throwable> errorListener, HttpHeaders headers) {
this.cachedClients = cachedClients;
this.scheme = scheme;
this.connector = connector;
this.errorListener = errorListener;
this.headers = headers;
}
/*
* (non-Javadoc)
* @see org.springframework.data.elasticsearch.client.reactive.WebClientProvider#get(java.net.InetSocketAddress)
*/
@Override
public WebClient get(InetSocketAddress endpoint) {
Assert.notNull(endpoint, "Endpoint must not be empty!");
return this.cachedClients.computeIfAbsent(endpoint, key -> {
Builder builder = WebClient.builder().defaultHeaders(it -> it.addAll(getDefaultHeaders()));
if (connector != null) {
builder.clientConnector(connector);
}
String baseUrl = String.format("%s://%s:%d", this.scheme, key.getHostString(), key.getPort());
return builder.baseUrl(baseUrl).filter((request, next) -> next.exchange(request).doOnError(errorListener))
.build();
});
}
/*
* (non-Javadoc)
* @see org.springframework.data.elasticsearch.client.reactive.WebClientProvider#getDefaultHeaders()
*/
@Override
public HttpHeaders getDefaultHeaders() {
return headers;
}
/*
* (non-Javadoc)
* @see org.springframework.data.elasticsearch.client.reactive.WebClientProvider#withDefaultHeaders(org.springframework.http.HttpHeaders)
*/
@Override
public WebClientProvider withDefaultHeaders(HttpHeaders headers) {
Assert.notNull(headers, "HttpHeaders must not be null");
HttpHeaders merged = new HttpHeaders();
merged.addAll(this.headers);
merged.addAll(headers);
return new DefaultWebClientProvider(this.scheme, this.connector, errorListener, merged);
}
/*
* (non-Javadoc)
* @see org.springframework.data.elasticsearch.client.reactive.WebClientProvider#getErrorListener()
*/
@Override
public Consumer<Throwable> getErrorListener() {
return this.errorListener;
}
/*
* (non-Javadoc)
* @see org.springframework.data.elasticsearch.client.reactive.WebClientProvider#withErrorListener(java.util.function.Consumer)
*/
@Override
public WebClientProvider withErrorListener(Consumer<Throwable> errorListener) {
Assert.notNull(errorListener, "Error listener must not be null");
Consumer<Throwable> listener = this.errorListener.andThen(errorListener);
return new DefaultWebClientProvider(this.scheme, this.connector, listener, this.headers);
}
}

73
src/main/java/org/springframework/data/elasticsearch/client/reactive/HostProvider.java
View File

@ -13,18 +13,16 @@
* See the License for the specific language governing permissions and * See the License for the specific language governing permissions and
* limitations under the License. * limitations under the License.
*/ */
package org.springframework.data.elasticsearch.client.reactive; package org.springframework.data.elasticsearch.client.reactive;
import org.springframework.data.elasticsearch.client.NoReachableHostException;
import reactor.core.publisher.Mono; import reactor.core.publisher.Mono;
import java.net.InetSocketAddress;
import java.util.Collections; import java.util.Collections;
import java.util.Set; import java.util.Set;
import java.util.function.Consumer;
import org.springframework.data.elasticsearch.client.ElasticsearchHost; import org.springframework.data.elasticsearch.client.ElasticsearchHost;
import org.springframework.http.HttpHeaders; import org.springframework.data.elasticsearch.client.NoReachableHostException;
import org.springframework.util.Assert; import org.springframework.util.Assert;
import org.springframework.web.reactive.function.client.WebClient; import org.springframework.web.reactive.function.client.WebClient;
@ -33,6 +31,7 @@ import org.springframework.web.reactive.function.client.WebClient;
* active ones. * active ones.
* *
* @author Christoph Strobl * @author Christoph Strobl
* @author Mark Paluch
* @since 4.0 * @since 4.0
*/ */
public interface HostProvider { public interface HostProvider {
@ -42,7 +41,7 @@ public interface HostProvider {
* *
* @return the {@link Mono} emitting the active host or {@link Mono#error(Throwable) an error} if none found. * @return the {@link Mono} emitting the active host or {@link Mono#error(Throwable) an error} if none found.
*/ */
default Mono<String> lookupActiveHost() { default Mono<InetSocketAddress> lookupActiveHost() {
return lookupActiveHost(VerificationMode.LAZY); return lookupActiveHost(VerificationMode.LAZY);
} }
@ -53,7 +52,7 @@ public interface HostProvider {
* @return the {@link Mono} emitting the active host or {@link Mono#error(Throwable) an error} * @return the {@link Mono} emitting the active host or {@link Mono#error(Throwable) an error}
* ({@link NoReachableHostException}) if none found. * ({@link NoReachableHostException}) if none found.
*/ */
Mono<String> lookupActiveHost(VerificationMode verificationMode); Mono<InetSocketAddress> lookupActiveHost(VerificationMode verificationMode);
/** /**
* Get the {@link WebClient} connecting to an active host utilizing cached {@link ElasticsearchHost}. * Get the {@link WebClient} connecting to an active host utilizing cached {@link ElasticsearchHost}.
@ -73,31 +72,16 @@ public interface HostProvider {
* found. * found.
*/ */
default Mono<WebClient> getActive(VerificationMode verificationMode) { default Mono<WebClient> getActive(VerificationMode verificationMode) {
return getActive(verificationMode, getDefaultHeaders()); return lookupActiveHost(verificationMode).map(this::createWebClient);
} }
/** /**
* Get the {@link WebClient} with default {@link HttpHeaders} connecting to an active host. * Creates a {@link WebClient} for {@link InetSocketAddress endpoint}.
* *
* @param verificationMode must not be {@literal null}. * @param baseUrl
* @param headers must not be {@literal null}.
* @return the {@link Mono} emitting the client for an active host or {@link Mono#error(Throwable) an error} if none
* found.
*/
default Mono<WebClient> getActive(VerificationMode verificationMode, HttpHeaders headers) {
return lookupActiveHost(verificationMode).map(host -> createWebClient(host, headers));
}
/**
* Get the {@link WebClient} with default {@link HttpHeaders} connecting to the given host.
*
* @param host must not be {@literal null}.
* @param headers must not be {@literal null}.
* @return * @return
*/ */
default WebClient createWebClient(String host, HttpHeaders headers) { WebClient createWebClient(InetSocketAddress endpoint);
return WebClient.builder().baseUrl(host).defaultHeaders(defaultHeaders -> defaultHeaders.putAll(headers)).build();
}
/** /**
* Obtain information about known cluster nodes. * Obtain information about known cluster nodes.
@ -107,42 +91,21 @@ public interface HostProvider {
Mono<ClusterInformation> clusterInfo(); Mono<ClusterInformation> clusterInfo();
/** /**
* Obtain the {@link HttpHeaders} to be used by default. * Create a new {@link HostProvider} best suited for the given {@link WebClientProvider} and number of hosts.
*
* @return never {@literal null}. {@link HttpHeaders#EMPTY} by default.
*/
HttpHeaders getDefaultHeaders();
/**
* Create a new instance of {@link HostProvider} applying the given headers by default.
*
* @param headers must not be {@literal null}.
* @return new instance of {@link HostProvider}.
*/
HostProvider withDefaultHeaders(HttpHeaders headers);
/**
* Create a new instance of {@link HostProvider} calling the given {@link Consumer} on error.
*
* @param errorListener must not be {@literal null}.
* @return new instance of {@link HostProvider}.
*/
HostProvider withErrorListener(Consumer<Throwable> errorListener);
/**
* Create a new {@link HostProvider} best suited for the given number of hosts.
* *
* @param clientProvider must not be {@literal null} .
* @param hosts must not be {@literal null} nor empty. * @param hosts must not be {@literal null} nor empty.
* @return new instance of {@link HostProvider}. * @return new instance of {@link HostProvider}.
*/ */
static HostProvider provider(String... hosts) { static HostProvider provider(WebClientProvider clientProvider, InetSocketAddress... endpoints) {
Assert.notEmpty(hosts, "Please provide at least one host to connect to."); Assert.notNull(clientProvider, "WebClientProvider must not be null");
Assert.notEmpty(endpoints, "Please provide at least one endpoint to connect to.");
if (hosts.length == 1) { if (endpoints.length == 1) {
return new SingleNodeHostProvider(HttpHeaders.EMPTY, (err) -> {}, hosts[0]); return new SingleNodeHostProvider(clientProvider, endpoints[0]);
} else { } else {
return new MultiNodeHostProvider(HttpHeaders.EMPTY, (err) -> {}, hosts); return new MultiNodeHostProvider(clientProvider, endpoints);
} }
} }
@ -155,7 +118,7 @@ public interface HostProvider {
/** /**
* Actively check for cluster node health. * Actively check for cluster node health.
*/ */
FORCE, ACTIVE,
/** /**
* Use cached data for cluster node health. * Use cached data for cluster node health.

83
src/main/java/org/springframework/data/elasticsearch/client/reactive/MultiNodeHostProvider.java
View File

@ -13,13 +13,13 @@
* See the License for the specific language governing permissions and * See the License for the specific language governing permissions and
* limitations under the License. * limitations under the License.
*/ */
package org.springframework.data.elasticsearch.client.reactive; package org.springframework.data.elasticsearch.client.reactive;
import reactor.core.publisher.Flux; import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono; import reactor.core.publisher.Mono;
import reactor.util.function.Tuple2; import reactor.util.function.Tuple2;
import java.net.InetSocketAddress;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Collection; import java.util.Collection;
import java.util.Collections; import java.util.Collections;
@ -27,56 +27,68 @@ import java.util.LinkedHashSet;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;
import org.springframework.data.elasticsearch.client.ElasticsearchHost; import org.springframework.data.elasticsearch.client.ElasticsearchHost;
import org.springframework.data.elasticsearch.client.ElasticsearchHost.State; import org.springframework.data.elasticsearch.client.ElasticsearchHost.State;
import org.springframework.data.elasticsearch.client.NoReachableHostException; import org.springframework.data.elasticsearch.client.NoReachableHostException;
import org.springframework.http.HttpHeaders;
import org.springframework.lang.Nullable; import org.springframework.lang.Nullable;
import org.springframework.web.reactive.function.client.ClientResponse; import org.springframework.web.reactive.function.client.ClientResponse;
import org.springframework.web.reactive.function.client.WebClient;
/** /**
* {@link HostProvider} for a cluster of nodes.
*
* @author Christoph Strobl * @author Christoph Strobl
* @author Mark Paluch
* @since 4.0 * @since 4.0
*/ */
class MultiNodeHostProvider implements HostProvider { class MultiNodeHostProvider implements HostProvider {
private final HttpHeaders headers; private final WebClientProvider clientProvider;
private final Consumer<Throwable> errorListener; private final Map<InetSocketAddress, ElasticsearchHost> hosts;
private final Map<String, ElasticsearchHost> hosts;
MultiNodeHostProvider(HttpHeaders headers, Consumer<Throwable> errorListener, String... hosts) { MultiNodeHostProvider(WebClientProvider clientProvider, InetSocketAddress... endpoints) {
this.headers = headers; this.clientProvider = clientProvider;
this.errorListener = errorListener;
this.hosts = new ConcurrentHashMap<>(); this.hosts = new ConcurrentHashMap<>();
for (String host : hosts) { for (InetSocketAddress endpoint : endpoints) {
this.hosts.put(host, new ElasticsearchHost(host, State.UNKNOWN)); this.hosts.put(endpoint, new ElasticsearchHost(endpoint, State.UNKNOWN));
} }
} }
/*
* (non-Javadoc)
* @see org.springframework.data.elasticsearch.client.reactive.HostProvider#clusterInfo()
*/
public Mono<ClusterInformation> clusterInfo() { public Mono<ClusterInformation> clusterInfo() {
return nodes(null).map(this::updateNodeState).buffer(hosts.size()) return nodes(null).map(this::updateNodeState).buffer(hosts.size())
.then(Mono.just(new ClusterInformation(new LinkedHashSet<>(this.hosts.values())))); .then(Mono.just(new ClusterInformation(new LinkedHashSet<>(this.hosts.values()))));
} }
/*
* (non-Javadoc)
* @see org.springframework.data.elasticsearch.client.reactive.HostProvider#createWebClient(java.net.InetSocketAddress)
*/
@Override
public WebClient createWebClient(InetSocketAddress endpoint) {
return this.clientProvider.get(endpoint);
}
Collection<ElasticsearchHost> getCachedHostState() { Collection<ElasticsearchHost> getCachedHostState() {
return hosts.values(); return hosts.values();
} }
/*
* (non-Javadoc)
* @see org.springframework.data.elasticsearch.client.reactive.HostProvider#lookupActiveHost(org.springframework.data.elasticsearch.client.reactive.HostProvider.VerificationMode)
*/
@Override @Override
public HttpHeaders getDefaultHeaders() { public Mono<InetSocketAddress> lookupActiveHost(VerificationMode verificationMode) {
return this.headers;
}
@Override
public Mono<String> lookupActiveHost(VerificationMode verificationMode) {
if (VerificationMode.LAZY.equals(verificationMode)) { if (VerificationMode.LAZY.equals(verificationMode)) {
for (ElasticsearchHost entry : hosts()) { for (ElasticsearchHost entry : hosts()) {
if (entry.isOnline()) { if (entry.isOnline()) {
return Mono.just(entry.getHost()); return Mono.just(entry.getEndpoint());
} }
} }
} }
@ -87,33 +99,24 @@ class MultiNodeHostProvider implements HostProvider {
.switchIfEmpty(Mono.error(() -> new NoReachableHostException(new LinkedHashSet<>(getCachedHostState())))); .switchIfEmpty(Mono.error(() -> new NoReachableHostException(new LinkedHashSet<>(getCachedHostState()))));
} }
@Override private Mono<InetSocketAddress> findActiveHostInKnownActives() {
public HostProvider withDefaultHeaders(HttpHeaders headers) {
return new MultiNodeHostProvider(headers, errorListener, hosts.keySet().toArray(new String[0]));
}
@Override
public HostProvider withErrorListener(Consumer<Throwable> errorListener) {
return new MultiNodeHostProvider(headers, errorListener, hosts.keySet().toArray(new String[0]));
}
private Mono<String> findActiveHostInKnownActives() {
return findActiveForSate(State.ONLINE); return findActiveForSate(State.ONLINE);
} }
private Mono<String> findActiveHostInUnresolved() { private Mono<InetSocketAddress> findActiveHostInUnresolved() {
return findActiveForSate(State.UNKNOWN); return findActiveForSate(State.UNKNOWN);
} }
private Mono<String> findActiveHostInDead() { private Mono<InetSocketAddress> findActiveHostInDead() {
return findActiveForSate(State.OFFLINE); return findActiveForSate(State.OFFLINE);
} }
private Mono<String> findActiveForSate(State state) { private Mono<InetSocketAddress> findActiveForSate(State state) {
return nodes(state).map(this::updateNodeState).filter(ElasticsearchHost::isOnline).map(it -> it.getHost()).next(); return nodes(state).map(this::updateNodeState).filter(ElasticsearchHost::isOnline)
.map(ElasticsearchHost::getEndpoint).next();
} }
private ElasticsearchHost updateNodeState(Tuple2<String, ClientResponse> tuple2) { private ElasticsearchHost updateNodeState(Tuple2<InetSocketAddress, ClientResponse> tuple2) {
State state = tuple2.getT2().statusCode().isError() ? State.OFFLINE : State.ONLINE; State state = tuple2.getT2().statusCode().isError() ? State.OFFLINE : State.ONLINE;
ElasticsearchHost elasticsearchHost = new ElasticsearchHost(tuple2.getT1(), state); ElasticsearchHost elasticsearchHost = new ElasticsearchHost(tuple2.getT1(), state);
@ -121,24 +124,24 @@ class MultiNodeHostProvider implements HostProvider {
return elasticsearchHost; return elasticsearchHost;
} }
private Flux<Tuple2<String, ClientResponse>> nodes(@Nullable State state) { private Flux<Tuple2<InetSocketAddress, ClientResponse>> nodes(@Nullable State state) {
return Flux.fromIterable(hosts()) // return Flux.fromIterable(hosts()) //
.filter(entry -> state != null ? entry.getState().equals(state) : true) // .filter(entry -> state == null || entry.getState().equals(state)) //
.map(ElasticsearchHost::getHost) // .map(ElasticsearchHost::getEndpoint) //
.flatMap(host -> { .flatMap(host -> {
Mono<ClientResponse> exchange = createWebClient(host, headers) // Mono<ClientResponse> exchange = createWebClient(host) //
.head().uri("/").exchange().doOnError(throwable -> { .head().uri("/").exchange().doOnError(throwable -> {
hosts.put(host, new ElasticsearchHost(host, State.OFFLINE)); hosts.put(host, new ElasticsearchHost(host, State.OFFLINE));
errorListener.accept(throwable); clientProvider.getErrorListener().accept(throwable);
}); });
return Mono.just(host).zipWith(exchange); return Mono.just(host).zipWith(exchange);
}) // }) //
.onErrorContinue((throwable, o) -> { .onErrorContinue((throwable, o) -> {
errorListener.accept(throwable); clientProvider.getErrorListener().accept(throwable);
}); });
} }

177
src/main/java/org/springframework/data/elasticsearch/client/reactive/RawActionResponse.java Normal file
View File

@ -0,0 +1,177 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.elasticsearch.client.reactive;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.util.List;
import org.elasticsearch.action.ActionResponse;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseCookie;
import org.springframework.http.ResponseEntity;
import org.springframework.http.client.reactive.ClientHttpResponse;
import org.springframework.http.codec.HttpMessageReader;
import org.springframework.util.MultiValueMap;
import org.springframework.web.reactive.function.BodyExtractor;
import org.springframework.web.reactive.function.client.ClientResponse;
import org.springframework.web.reactive.function.client.ExchangeStrategies;
/**
* Extension to {@link ActionResponse} that also implements {@link ClientResponse}.
*/
class RawActionResponse extends ActionResponse implements ClientResponse {
private final ClientResponse delegate;
private RawActionResponse(ClientResponse delegate) {
this.delegate = delegate;
}
public static RawActionResponse create(ClientResponse response) {
return new RawActionResponse(response);
}
public static Builder builder(ClientResponse other) {
return ClientResponse.from(other);
}
public static Builder builder(HttpStatus statusCode) {
return ClientResponse.create(statusCode);
}
public static Builder builder(HttpStatus statusCode, ExchangeStrategies strategies) {
return ClientResponse.create(statusCode, strategies);
}
public static Builder builder(HttpStatus statusCode, List<HttpMessageReader<?>> messageReaders) {
return ClientResponse.create(statusCode, messageReaders);
}
/*
* (non-Javadoc)
* @see org.springframework.web.reactive.function.client.ClientResponse#statusCode()
*/
public HttpStatus statusCode() {
return delegate.statusCode();
}
/*
* (non-Javadoc)
* @see org.springframework.web.reactive.function.client.ClientResponse#rawStatusCode()
*/
public int rawStatusCode() {
return delegate.rawStatusCode();
}
/*
* (non-Javadoc)
* @see org.springframework.web.reactive.function.client.ClientResponse#headers()
*/
public Headers headers() {
return delegate.headers();
}
/*
* (non-Javadoc)
* @see org.springframework.web.reactive.function.client.ClientResponse#cookies()
*/
public MultiValueMap<String, ResponseCookie> cookies() {
return delegate.cookies();
}
/*
* (non-Javadoc)
* @see org.springframework.web.reactive.function.client.ClientResponse#strategies()
*/
public ExchangeStrategies strategies() {
return delegate.strategies();
}
/*
* (non-Javadoc)
* @see org.springframework.web.reactive.function.client.ClientResponse#body(org.springframework.web.reactive.function.BodyExtractor)
*/
public <T> T body(BodyExtractor<T, ? super ClientHttpResponse> extractor) {
return delegate.body(extractor);
}
/*
* (non-Javadoc)
* @see org.springframework.web.reactive.function.client.ClientResponse#bodyToMono(java.lang.Class)
*/
public <T> Mono<T> bodyToMono(Class<? extends T> elementClass) {
return delegate.bodyToMono(elementClass);
}
/*
* (non-Javadoc)
* @see org.springframework.web.reactive.function.client.ClientResponse#bodyToMono(org.springframework.core.ParameterizedTypeReference)
*/
public <T> Mono<T> bodyToMono(ParameterizedTypeReference<T> typeReference) {
return delegate.bodyToMono(typeReference);
}
/*
* (non-Javadoc)
* @see org.springframework.web.reactive.function.client.ClientResponse#bodyToFlux(java.lang.Class)
*/
public <T> Flux<T> bodyToFlux(Class<? extends T> elementClass) {
return delegate.bodyToFlux(elementClass);
}
/*
* (non-Javadoc)
* @see org.springframework.web.reactive.function.client.ClientResponse#bodyToFlux(org.springframework.core.ParameterizedTypeReference)
*/
public <T> Flux<T> bodyToFlux(ParameterizedTypeReference<T> typeReference) {
return delegate.bodyToFlux(typeReference);
}
/*
* (non-Javadoc)
* @see org.springframework.web.reactive.function.client.ClientResponse#toEntity(java.lang.Class)
*/
public <T> Mono<ResponseEntity<T>> toEntity(Class<T> bodyType) {
return delegate.toEntity(bodyType);
}
/*
* (non-Javadoc)
* @see org.springframework.web.reactive.function.client.ClientResponse#toEntity(org.springframework.core.ParameterizedTypeReference)
*/
public <T> Mono<ResponseEntity<T>> toEntity(ParameterizedTypeReference<T> typeReference) {
return delegate.toEntity(typeReference);
}
/*
* (non-Javadoc)
* @see org.springframework.web.reactive.function.client.ClientResponse#toEntityList(java.lang.Class)
*/
public <T> Mono<ResponseEntity<List<T>>> toEntityList(Class<T> elementType) {
return delegate.toEntityList(elementType);
}
/*
* (non-Javadoc)
* @see org.springframework.web.reactive.function.client.ClientResponse#toEntityList(org.springframework.core.ParameterizedTypeReference)
*/
public <T> Mono<ResponseEntity<List<T>>> toEntityList(ParameterizedTypeReference<T> typeReference) {
return delegate.toEntityList(typeReference);
}
}

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

@ -13,7 +13,6 @@
* See the License for the specific language governing permissions and * See the License for the specific language governing permissions and
* limitations under the License. * limitations under the License.
*/ */
package org.springframework.data.elasticsearch.client.reactive; package org.springframework.data.elasticsearch.client.reactive;
import reactor.core.publisher.Flux; import reactor.core.publisher.Flux;
@ -35,6 +34,7 @@ import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse; import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.index.get.GetResult; import org.elasticsearch.index.get.GetResult;
import org.elasticsearch.search.SearchHit; import org.elasticsearch.search.SearchHit;
import org.springframework.data.elasticsearch.client.ClientConfiguration;
import org.springframework.data.elasticsearch.client.ElasticsearchHost; import org.springframework.data.elasticsearch.client.ElasticsearchHost;
import org.springframework.http.HttpHeaders; import org.springframework.http.HttpHeaders;
import org.springframework.util.CollectionUtils; import org.springframework.util.CollectionUtils;
@ -42,10 +42,13 @@ import org.springframework.web.reactive.function.client.ClientResponse;
import org.springframework.web.reactive.function.client.WebClient; import org.springframework.web.reactive.function.client.WebClient;
/** /**
* A reactive client to connect to Elasticsearch. <br /> * A reactive client to connect to Elasticsearch.
* *
* @author Christoph Strobl * @author Christoph Strobl
* @author Mark Paluch
* @since 4.0 * @since 4.0
* @see ClientConfiguration
* @see ReactiveRestClients
*/ */
public interface ReactiveElasticsearchClient { public interface ReactiveElasticsearchClient {
@ -385,7 +388,7 @@ public interface ReactiveElasticsearchClient {
/** /**
* Get the list of known hosts and their getCachedHostState. * Get the list of known hosts and their getCachedHostState.
* *
* @return never {@literal null}. * @return never {@literal null}.
*/ */
Collection<ElasticsearchHost> hosts(); Collection<ElasticsearchHost> hosts();
@ -401,7 +404,7 @@ public interface ReactiveElasticsearchClient {
return false; return false;
} }
return !hosts().stream().filter(it -> !it.isOnline()).findFirst().isPresent(); return hosts().stream().anyMatch(ElasticsearchHost::isOnline);
} }
} }
} }

45
src/main/java/org/springframework/data/elasticsearch/client/reactive/ReactiveRestClients.java Normal file
View File

@ -0,0 +1,45 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.elasticsearch.client.reactive;
import org.springframework.data.elasticsearch.client.ClientConfiguration;
import org.springframework.data.elasticsearch.client.RestClients.ElasticsearchRestClient;
import org.springframework.util.Assert;
/**
* Utility class for common access to reactive Elasticsearch clients. {@link ReactiveRestClients} consolidates set up
* routines for the various drivers into a single place.
*
* @author Christoph Strobl
* @author Mark Paluch
* @since 4.0
*/
public final class ReactiveRestClients {
private ReactiveRestClients() {}
/**
* Start here to create a new client tailored to your needs.
*
* @return new instance of {@link ElasticsearchRestClient}.
*/
public static ReactiveElasticsearchClient create(ClientConfiguration clientConfiguration) {
Assert.notNull(clientConfiguration, "ClientConfiguration must not be null!");
return DefaultReactiveElasticsearchClient.create(clientConfiguration);
}
}

40
src/main/java/org/springframework/data/elasticsearch/client/reactive/RequestBodyEncodingException.java Normal file
View File

@ -0,0 +1,40 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.elasticsearch.client.reactive;
import org.springframework.web.reactive.function.client.WebClientException;
/**
* Exception thrown if the request body cannot be properly encoded.
*
* @author Christoph Strobl
* @author Mark Paluch
* @since 4.0
*/
public class RequestBodyEncodingException extends WebClientException {
private static final long serialVersionUID = 472776714118912855L;
/**
* Construct a new instance of {@link RequestBodyEncodingException} with the given message and exception.
*
* @param msg the message
* @param ex the exception
*/
public RequestBodyEncodingException(String msg, Throwable ex) {
super(msg, ex);
}
}

73
src/main/java/org/springframework/data/elasticsearch/client/reactive/SingleNodeHostProvider.java
View File

@ -13,95 +13,96 @@
* See the License for the specific language governing permissions and * See the License for the specific language governing permissions and
* limitations under the License. * limitations under the License.
*/ */
package org.springframework.data.elasticsearch.client.reactive; package org.springframework.data.elasticsearch.client.reactive;
import reactor.core.publisher.Mono; import reactor.core.publisher.Mono;
import java.net.InetSocketAddress;
import java.util.Collections; import java.util.Collections;
import java.util.function.Consumer;
import org.springframework.data.elasticsearch.client.ElasticsearchHost; import org.springframework.data.elasticsearch.client.ElasticsearchHost;
import org.springframework.data.elasticsearch.client.ElasticsearchHost.State; import org.springframework.data.elasticsearch.client.ElasticsearchHost.State;
import org.springframework.data.elasticsearch.client.NoReachableHostException; import org.springframework.data.elasticsearch.client.NoReachableHostException;
import org.springframework.http.HttpHeaders; import org.springframework.web.reactive.function.client.WebClient;
/** /**
* {@link HostProvider} for a single host.
*
* @author Christoph Strobl * @author Christoph Strobl
* @author Mark Paluch
* @since 4.0 * @since 4.0
*/ */
class SingleNodeHostProvider implements HostProvider { class SingleNodeHostProvider implements HostProvider {
private final HttpHeaders headers; private final WebClientProvider clientProvider;
private final Consumer<Throwable> errorListener; private final InetSocketAddress endpoint;
private final String hostname;
private volatile ElasticsearchHost state; private volatile ElasticsearchHost state;
SingleNodeHostProvider(HttpHeaders headers, Consumer<Throwable> errorListener, String host) { SingleNodeHostProvider(WebClientProvider clientProvider, InetSocketAddress endpoint) {
this.headers = headers; this.clientProvider = clientProvider;
this.errorListener = errorListener; this.endpoint = endpoint;
this.hostname = host; this.state = new ElasticsearchHost(this.endpoint, State.UNKNOWN);
this.state = new ElasticsearchHost(hostname, State.UNKNOWN);
} }
/*
* (non-Javadoc)
* @see org.springframework.data.elasticsearch.client.reactive.HostProvider#clusterInfo()
*/
@Override @Override
public Mono<ClusterInformation> clusterInfo() { public Mono<ClusterInformation> clusterInfo() {
return createWebClient(hostname, headers) // return createWebClient(endpoint) //
.head().uri("/").exchange() // .head().uri("/").exchange() //
.flatMap(it -> { .flatMap(it -> {
if(it.statusCode().isError()) { if (it.statusCode().isError()) {
state = ElasticsearchHost.offline(hostname); state = ElasticsearchHost.offline(endpoint);
} else { } else {
state = ElasticsearchHost.online(hostname); state = ElasticsearchHost.online(endpoint);
} }
return Mono.just(state); return Mono.just(state);
}).onErrorResume(throwable -> { }).onErrorResume(throwable -> {
state = ElasticsearchHost.offline(hostname); state = ElasticsearchHost.offline(endpoint);
errorListener.accept(throwable); clientProvider.getErrorListener().accept(throwable);
return Mono.just(state); return Mono.just(state);
}) // }) //
.flatMap(it -> Mono.just(new ClusterInformation(Collections.singleton(it)))); .flatMap(it -> Mono.just(new ClusterInformation(Collections.singleton(it))));
} }
/*
* (non-Javadoc)
* @see org.springframework.data.elasticsearch.client.reactive.HostProvider#createWebClient(java.net.InetSocketAddress)
*/
@Override @Override
public Mono<String> lookupActiveHost(VerificationMode verificationMode) { public WebClient createWebClient(InetSocketAddress endpoint) {
return this.clientProvider.get(endpoint);
}
/*
* (non-Javadoc)
* @see org.springframework.data.elasticsearch.client.reactive.HostProvider#lookupActiveHost(org.springframework.data.elasticsearch.client.reactive.HostProvider.VerificationMode)
*/
@Override
public Mono<InetSocketAddress> lookupActiveHost(VerificationMode verificationMode) {
if (VerificationMode.LAZY.equals(verificationMode) && state.isOnline()) { if (VerificationMode.LAZY.equals(verificationMode) && state.isOnline()) {
return Mono.just(hostname); return Mono.just(endpoint);
} }
return clusterInfo().flatMap(it -> { return clusterInfo().flatMap(it -> {
ElasticsearchHost host = it.getNodes().iterator().next(); ElasticsearchHost host = it.getNodes().iterator().next();
if (host.isOnline()) { if (host.isOnline()) {
return Mono.just(host.getHost()); return Mono.just(host.getEndpoint());
} }
return Mono.error(() -> new NoReachableHostException(Collections.singleton(host))); return Mono.error(() -> new NoReachableHostException(Collections.singleton(host)));
}); });
} }
@Override
public HttpHeaders getDefaultHeaders() {
return this.headers;
}
@Override
public HostProvider withDefaultHeaders(HttpHeaders headers) {
return new SingleNodeHostProvider(headers, errorListener, hostname);
}
@Override
public HostProvider withErrorListener(Consumer<Throwable> errorListener) {
return new SingleNodeHostProvider(headers, errorListener, hostname);
}
ElasticsearchHost getCachedHostState() { ElasticsearchHost getCachedHostState() {
return state; return state;
} }
} }

104
src/main/java/org/springframework/data/elasticsearch/client/reactive/WebClientProvider.java Normal file
View File

@ -0,0 +1,104 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.elasticsearch.client.reactive;
import java.net.InetSocketAddress;
import java.util.function.Consumer;
import org.springframework.http.HttpHeaders;
import org.springframework.http.client.reactive.ClientHttpConnector;
import org.springframework.util.Assert;
import org.springframework.web.reactive.function.client.WebClient;
/**
* Provider for {@link WebClient}s using a pre-configured {@code scheme}. This class returns {@link WebClient} for a
* specific {@link InetSocketAddress endpoint} and encapsulates common configuration aspects of {@link WebClient} so
* that code using {@link WebClient} is not required to apply further configuration to the actual client.
* <p/>
* Client instances are typically cached allowing reuse of pooled connections if configured on the
* {@link ClientHttpConnector}.
*
* @author Christoph Strobl
* @author Mark Paluch
* @since 4.0
*/
public interface WebClientProvider {
/**
* Creates a new {@link WebClientProvider} using the given {@code scheme} and a default {@link ClientHttpConnector}.
*
* @param scheme protocol scheme such as {@literal http} or {@literal https}.
* @return the resulting {@link WebClientProvider}.
*/
static WebClientProvider create(String scheme) {
Assert.hasText(scheme, "Protocol scheme must not be empty");
return new DefaultWebClientProvider(scheme, null);
}
/**
* Creates a new {@link WebClientProvider} given {@code scheme} and {@link ClientHttpConnector}.
*
* @param scheme protocol scheme such as {@literal http} or {@literal https}.
* @param connector the HTTP connector to use.
* @return the resulting {@link WebClientProvider}.
*/
static WebClientProvider create(String scheme, ClientHttpConnector connector) {
Assert.hasText(scheme, "Protocol scheme must not be empty");
return new DefaultWebClientProvider(scheme, connector);
}
/**
* Obtain the {@link WebClient} configured with {@link #withDefaultHeaders(HttpHeaders) default HTTP headers} and
* {@link Consumer} error callback for a given {@link InetSocketAddress endpoint}.
*
* @return the {@link WebClient} for the given {@link InetSocketAddress endpoint}.
*/
WebClient get(InetSocketAddress endpoint);
/**
* Obtain the {@link HttpHeaders} to be used by default.
*
* @return never {@literal null}. {@link HttpHeaders#EMPTY} by default.
*/
HttpHeaders getDefaultHeaders();
/**
* Create a new instance of {@link WebClientProvider} applying the given headers by default.
*
* @param headers must not be {@literal null}.
* @return new instance of {@link WebClientProvider}.
*/
WebClientProvider withDefaultHeaders(HttpHeaders headers);
/**
* Obtain the {@link Consumer error listener} to be used;
*
* @return never {@literal null}.
*/
Consumer<Throwable> getErrorListener();
/**
* Create a new instance of {@link WebClientProvider} calling the given {@link Consumer} on error.
*
* @param errorListener must not be {@literal null}.
* @return new instance of {@link WebClientProvider}.
*/
WebClientProvider withErrorListener(Consumer<Throwable> errorListener);
}

9
src/main/java/org/springframework/data/elasticsearch/core/ReactiveElasticsearchTemplate.java
View File

@ -13,12 +13,10 @@
* See the License for the specific language governing permissions and * See the License for the specific language governing permissions and
* limitations under the License. * limitations under the License.
*/ */
package org.springframework.data.elasticsearch.core; package org.springframework.data.elasticsearch.core;
import static org.elasticsearch.index.VersionType.*; import static org.elasticsearch.index.VersionType.*;
import org.elasticsearch.action.support.WriteRequest.RefreshPolicy;
import reactor.core.publisher.Flux; import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono; import reactor.core.publisher.Mono;
@ -31,6 +29,7 @@ import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.index.IndexRequest; import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse; import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest; import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.support.WriteRequest.RefreshPolicy;
import org.elasticsearch.client.Requests; import org.elasticsearch.client.Requests;
import org.elasticsearch.index.get.GetResult; import org.elasticsearch.index.get.GetResult;
import org.elasticsearch.index.query.QueryBuilder; import org.elasticsearch.index.query.QueryBuilder;
@ -130,7 +129,7 @@ public class ReactiveElasticsearchTemplate {
ElasticsearchPersistentEntity<?> persistentEntity = lookupPersistentEntity(resultType); ElasticsearchPersistentEntity<?> persistentEntity = lookupPersistentEntity(resultType);
GetRequest request = new GetRequest(persistentEntity.getIndexName(), persistentEntity.getIndexType(), id); GetRequest request = new GetRequest(persistentEntity.getIndexName(), persistentEntity.getIndexType(), id);
return goGet(id, persistentEntity, index, type).map(it -> mapper.mapEntity(it.sourceAsString(), resultType)); return doGet(id, persistentEntity, index, type).map(it -> mapper.mapEntity(it.sourceAsString(), resultType));
} }
/** /**
@ -198,7 +197,7 @@ public class ReactiveElasticsearchTemplate {
// Customization Hooks // Customization Hooks
protected Mono<GetResult> goGet(String id, ElasticsearchPersistentEntity<?> entity, @Nullable String index, protected Mono<GetResult> doGet(String id, ElasticsearchPersistentEntity<?> entity, @Nullable String index,
@Nullable String type) { @Nullable String type) {
String indexToUse = indexName(index, entity); String indexToUse = indexName(index, entity);
@ -342,7 +341,7 @@ public class ReactiveElasticsearchTemplate {
} }
// Additional types // Additional types
public interface ClientCallback<T extends Publisher> { public interface ClientCallback<T extends Publisher<?>> {
T doWithClient(ReactiveElasticsearchClient client); T doWithClient(ReactiveElasticsearchClient client);
} }

10
src/test/java/org/springframework/data/elasticsearch/TestUtils.java
View File

@ -13,7 +13,6 @@
* See the License for the specific language governing permissions and * See the License for the specific language governing permissions and
* limitations under the License. * limitations under the License.
*/ */
package org.springframework.data.elasticsearch; package org.springframework.data.elasticsearch;
import lombok.SneakyThrows; import lombok.SneakyThrows;
@ -22,12 +21,15 @@ import org.elasticsearch.ElasticsearchStatusException;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest; import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.client.RequestOptions; import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient; import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.data.elasticsearch.client.ElasticsearchClients; import org.springframework.data.elasticsearch.client.ClientConfiguration;
import org.springframework.data.elasticsearch.client.RestClients;
import org.springframework.data.elasticsearch.client.reactive.ReactiveElasticsearchClient; import org.springframework.data.elasticsearch.client.reactive.ReactiveElasticsearchClient;
import org.springframework.data.elasticsearch.client.reactive.ReactiveRestClients;
import org.springframework.util.ObjectUtils; import org.springframework.util.ObjectUtils;
/** /**
* @author Christoph Strobl * @author Christoph Strobl
* @author Mark Paluch
* @currentRead Fool's Fate - Robin Hobb * @currentRead Fool's Fate - Robin Hobb
*/ */
public final class TestUtils { public final class TestUtils {
@ -35,11 +37,11 @@ public final class TestUtils {
private TestUtils() {} private TestUtils() {}
public static RestHighLevelClient restHighLevelClient() { public static RestHighLevelClient restHighLevelClient() {
return ElasticsearchClients.createClient().connectedToLocalhost().rest(); return RestClients.create(ClientConfiguration.create("localhost:9200")).rest();
} }
public static ReactiveElasticsearchClient reactiveClient() { public static ReactiveElasticsearchClient reactiveClient() {
return ElasticsearchClients.createClient().connectedToLocalhost().reactive(); return ReactiveRestClients.create(ClientConfiguration.create("localhost:9200"));
} }
@SneakyThrows @SneakyThrows

76
src/test/java/org/springframework/data/elasticsearch/client/ClientConfigurationUnitTests.java Normal file
View File

@ -0,0 +1,76 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.elasticsearch.client;
import static org.assertj.core.api.Assertions.*;
import static org.mockito.Mockito.*;
import java.net.InetSocketAddress;
import javax.net.ssl.SSLContext;
import org.junit.Test;
import org.springframework.http.HttpHeaders;
/**
* Unit tests for {@link ClientConfiguration}.
*
* @author Mark Paluch
*/
public class ClientConfigurationUnitTests {
@Test // DATAES-488
public void shouldCreateSimpleConfiguration() {
ClientConfiguration clientConfiguration = ClientConfiguration.create("localhost:9200");
assertThat(clientConfiguration.getEndpoints()).containsOnly(InetSocketAddress.createUnresolved("localhost", 9200));
}
@Test // DATAES-488
public void shouldCreateCustomizedConfiguration() {
HttpHeaders headers = new HttpHeaders();
headers.set("foo", "bar");
ClientConfiguration clientConfiguration = ClientConfiguration.builder() //
.connectedTo("foo", "bar") //
.usingSsl() //
.withDefaultHeaders(headers) //
.build();
assertThat(clientConfiguration.getEndpoints()).containsOnly(InetSocketAddress.createUnresolved("foo", 9200),
InetSocketAddress.createUnresolved("bar", 9200));
assertThat(clientConfiguration.useSsl()).isTrue();
assertThat(clientConfiguration.getDefaultHeaders().get("foo")).containsOnly("bar");
}
@Test // DATAES-488
public void shouldCreateSslConfiguration() {
SSLContext sslContext = mock(SSLContext.class);
ClientConfiguration clientConfiguration = ClientConfiguration.builder() //
.connectedTo("foo", "bar") //
.usingSsl(sslContext) //
.build();
assertThat(clientConfiguration.getEndpoints()).containsOnly(InetSocketAddress.createUnresolved("foo", 9200),
InetSocketAddress.createUnresolved("bar", 9200));
assertThat(clientConfiguration.useSsl()).isTrue();
assertThat(clientConfiguration.getSslContext()).contains(sslContext);
}
}

122
src/test/java/org/springframework/data/elasticsearch/client/InetSocketAddressParserUnitTests.java Normal file
View File

@ -0,0 +1,122 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.elasticsearch.client;
import static org.assertj.core.api.Assertions.*;
import java.net.InetSocketAddress;
import org.junit.Test;
/**
* Unit tests for {@link InetSocketAddressParser}.
*
* @author Mark Paluch
*/
public class InetSocketAddressParserUnitTests {
@Test
public void testFromStringWellFormed() {
// Well-formed inputs.
checkFromStringCase("pivotal.io", 80, "pivotal.io", 80, false);
checkFromStringCase("pivotal.io", 80, "pivotal.io", 80, false);
checkFromStringCase("192.0.2.1", 82, "192.0.2.1", 82, false);
checkFromStringCase("[2001::1]", 84, "2001::1", 84, false);
checkFromStringCase("2001::3", 86, "2001::3", 86, false);
checkFromStringCase("host:", 80, "host", 80, false);
}
@Test
public void testFromStringBadDefaultPort() {
// Well-formed strings with bad default ports.
checkFromStringCase("gmail.com:81", -1, "gmail.com", 81, true);
checkFromStringCase("192.0.2.2:83", -1, "192.0.2.2", 83, true);
checkFromStringCase("[2001::2]:85", -1, "2001::2", 85, true);
checkFromStringCase("goo.gl:65535", 65536, "goo.gl", 65535, true);
// No port, bad default.
checkFromStringCase("pivotal.io", -1, null, -1, false);
checkFromStringCase("192.0.2.1", 65536, null, -1, false);
checkFromStringCase("[2001::1]", -1, null, -1, false);
checkFromStringCase("2001::3", 65536, null, -1, false);
}
@Test
public void testFromStringUnusedDefaultPort() {
// Default port, but unused.
checkFromStringCase("gmail.com:81", 77, "gmail.com", 81, true);
checkFromStringCase("192.0.2.2:83", 77, "192.0.2.2", 83, true);
checkFromStringCase("[2001::2]:85", 77, "2001::2", 85, true);
}
@Test
public void testFromStringBadPort() {
// Out-of-range ports.
checkFromStringCase("pivotal.io:65536", 1, null, 99, false);
checkFromStringCase("pivotal.io:9999999999", 1, null, 99, false);
// Invalid port parts.
checkFromStringCase("pivotal.io:port", 1, null, 99, false);
checkFromStringCase("pivotal.io:-25", 1, null, 99, false);
checkFromStringCase("pivotal.io:+25", 1, null, 99, false);
checkFromStringCase("pivotal.io:25 ", 1, null, 99, false);
checkFromStringCase("pivotal.io:25\t", 1, null, 99, false);
checkFromStringCase("pivotal.io:0x25 ", 1, null, 99, false);
}
@Test
public void testFromStringUnparseableNonsense() {
// Some nonsense that causes parse failures.
checkFromStringCase("[goo.gl]", 1, null, 99, false);
checkFromStringCase("[goo.gl]:80", 1, null, 99, false);
checkFromStringCase("[", 1, null, 99, false);
checkFromStringCase("[]:", 1, null, 99, false);
checkFromStringCase("[]:80", 1, null, 99, false);
checkFromStringCase("[]bad", 1, null, 99, false);
}
@Test
public void testFromStringParseableNonsense() {
// Examples of nonsense that gets through.
checkFromStringCase("[[:]]", 86, "[:]", 86, false);
checkFromStringCase("x:y:z", 87, "x:y:z", 87, false);
checkFromStringCase("", 88, "", 88, false);
checkFromStringCase(":", 99, "", 99, false);
checkFromStringCase(":123", -1, "", 123, true);
checkFromStringCase("\nOMG\t", 89, "\nOMG\t", 89, false);
}
private static void checkFromStringCase(String hpString, int defaultPort, String expectHost, int expectPort,
boolean expectHasExplicitPort) {
InetSocketAddress hp;
try {
hp = InetSocketAddressParser.parse(hpString, defaultPort);
} catch (IllegalArgumentException e) {
// Make sure we expected this.
assertThat(expectHost).isNull();
return;
}
assertThat(expectHost).isNotNull();
if (expectHasExplicitPort) {
assertThat(hp.getPort()).isEqualTo(expectPort);
} else {
assertThat(hp.getPort()).isEqualTo(defaultPort);
}
assertThat(hp.getHostString()).isEqualTo(expectHost);
}
}

7
src/test/java/org/springframework/data/elasticsearch/client/reactive/MultiNodeHostProviderUnitTests.java
View File

@ -13,7 +13,6 @@
* See the License for the specific language governing permissions and * See the License for the specific language governing permissions and
* limitations under the License. * limitations under the License.
*/ */
package org.springframework.data.elasticsearch.client.reactive; package org.springframework.data.elasticsearch.client.reactive;
import static org.assertj.core.api.Assertions.*; import static org.assertj.core.api.Assertions.*;
@ -24,11 +23,11 @@ import reactor.test.StepVerifier;
import org.junit.Before; import org.junit.Before;
import org.junit.Test; import org.junit.Test;
import org.springframework.data.elasticsearch.client.reactive.HostProvider.VerificationMode;
import org.springframework.data.elasticsearch.client.ElasticsearchHost; import org.springframework.data.elasticsearch.client.ElasticsearchHost;
import org.springframework.data.elasticsearch.client.ElasticsearchHost.State; import org.springframework.data.elasticsearch.client.ElasticsearchHost.State;
import org.springframework.data.elasticsearch.client.reactive.HostProvider.VerificationMode;
import org.springframework.data.elasticsearch.client.reactive.ReactiveMockClientTestsUtils.MockDelegatingElasticsearchHostProvider; import org.springframework.data.elasticsearch.client.reactive.ReactiveMockClientTestsUtils.MockDelegatingElasticsearchHostProvider;
import org.springframework.data.elasticsearch.client.reactive.ReactiveMockClientTestsUtils.WebClientProvider.Receive; import org.springframework.data.elasticsearch.client.reactive.ReactiveMockClientTestsUtils.MockWebClientProvider.Receive;
import org.springframework.web.reactive.function.client.ClientResponse; import org.springframework.web.reactive.function.client.ClientResponse;
/** /**
@ -107,7 +106,7 @@ public class MultiNodeHostProviderUnitTests {
provider.clusterInfo().as(StepVerifier::create).expectNextCount(1).verifyComplete(); provider.clusterInfo().as(StepVerifier::create).expectNextCount(1).verifyComplete();
provider.getActive(VerificationMode.FORCE).as(StepVerifier::create).expectNext(mock.client(HOST_2)) provider.getActive(VerificationMode.ACTIVE).as(StepVerifier::create).expectNext(mock.client(HOST_2))
.verifyComplete(); .verifyComplete();
verify(mock.client(HOST_2), times(2)).head(); verify(mock.client(HOST_2), times(2)).head();

7
src/test/java/org/springframework/data/elasticsearch/client/reactive/ReactiveElasticsearchClientTests.java
View File

@ -13,7 +13,6 @@
* See the License for the specific language governing permissions and * See the License for the specific language governing permissions and
* limitations under the License. * limitations under the License.
*/ */
package org.springframework.data.elasticsearch.client.reactive; package org.springframework.data.elasticsearch.client.reactive;
import static org.assertj.core.api.Assertions.*; import static org.assertj.core.api.Assertions.*;
@ -43,14 +42,20 @@ import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.junit.After; import org.junit.After;
import org.junit.Before; import org.junit.Before;
import org.junit.Test; import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.data.elasticsearch.TestUtils; import org.springframework.data.elasticsearch.TestUtils;
import org.springframework.http.HttpHeaders; import org.springframework.http.HttpHeaders;
import org.springframework.lang.Nullable; import org.springframework.lang.Nullable;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringRunner;
/** /**
* @author Christoph Strobl * @author Christoph Strobl
* @author Mark Paluch
* @currentRead Fool's Fate - Robin Hobb * @currentRead Fool's Fate - Robin Hobb
*/ */
@RunWith(SpringRunner.class)
@ContextConfiguration("classpath:infrastructure.xml")
public class ReactiveElasticsearchClientTests { public class ReactiveElasticsearchClientTests {
static final String INDEX_I = "idx-1-reactive-client-tests"; static final String INDEX_I = "idx-1-reactive-client-tests";

8
src/test/java/org/springframework/data/elasticsearch/client/reactive/ReactiveElasticsearchClientUnitTests.java
View File

@ -13,14 +13,12 @@
* See the License for the specific language governing permissions and * See the License for the specific language governing permissions and
* limitations under the License. * limitations under the License.
*/ */
package org.springframework.data.elasticsearch.client.reactive; package org.springframework.data.elasticsearch.client.reactive;
import static org.assertj.core.api.Assertions.*; import static org.assertj.core.api.Assertions.*;
import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.*;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.*; import static org.mockito.Mockito.*;
import static org.springframework.data.elasticsearch.client.reactive.ReactiveMockClientTestsUtils.WebClientProvider.Receive.*; import static org.springframework.data.elasticsearch.client.reactive.ReactiveMockClientTestsUtils.MockWebClientProvider.Receive.*;
import reactor.test.StepVerifier; import reactor.test.StepVerifier;
@ -40,7 +38,7 @@ import org.junit.Before;
import org.junit.Test; import org.junit.Test;
import org.reactivestreams.Publisher; import org.reactivestreams.Publisher;
import org.springframework.data.elasticsearch.client.reactive.ReactiveMockClientTestsUtils.MockDelegatingElasticsearchHostProvider; import org.springframework.data.elasticsearch.client.reactive.ReactiveMockClientTestsUtils.MockDelegatingElasticsearchHostProvider;
import org.springframework.data.elasticsearch.client.reactive.ReactiveMockClientTestsUtils.WebClientProvider.Receive; import org.springframework.data.elasticsearch.client.reactive.ReactiveMockClientTestsUtils.MockWebClientProvider.Receive;
import org.springframework.http.HttpMethod; import org.springframework.http.HttpMethod;
import org.springframework.http.HttpStatus; import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType; import org.springframework.http.MediaType;

125
src/test/java/org/springframework/data/elasticsearch/client/reactive/ReactiveMockClientTestsUtils.java
View File

@ -13,21 +13,22 @@
* See the License for the specific language governing permissions and * See the License for the specific language governing permissions and
* limitations under the License. * limitations under the License.
*/ */
package org.springframework.data.elasticsearch.client.reactive; package org.springframework.data.elasticsearch.client.reactive;
import static org.mockito.Mockito.*; import static org.mockito.Mockito.*;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono; import reactor.core.publisher.Mono;
import java.io.IOException; import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets; import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Collections; import java.util.Collections;
import java.util.LinkedHashMap; import java.util.LinkedHashMap;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
import java.util.Optional; import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CopyOnWriteArrayList;
import java.util.function.Consumer; import java.util.function.Consumer;
import java.util.function.Supplier; import java.util.function.Supplier;
@ -35,10 +36,8 @@ import java.util.function.Supplier;
import org.mockito.Mockito; import org.mockito.Mockito;
import org.springframework.core.io.ClassPathResource; import org.springframework.core.io.ClassPathResource;
import org.springframework.core.io.Resource; import org.springframework.core.io.Resource;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DefaultDataBufferFactory;
import org.springframework.data.elasticsearch.client.ElasticsearchHost; import org.springframework.data.elasticsearch.client.ElasticsearchHost;
import org.springframework.data.elasticsearch.client.reactive.ReactiveMockClientTestsUtils.WebClientProvider.Send; import org.springframework.data.elasticsearch.client.reactive.ReactiveMockClientTestsUtils.MockWebClientProvider.Send;
import org.springframework.http.HttpHeaders; import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus; import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType; import org.springframework.http.MediaType;
@ -52,10 +51,11 @@ import org.springframework.web.reactive.function.client.WebClient.RequestHeaders
/** /**
* @author Christoph Strobl * @author Christoph Strobl
* @since 2018/10
*/ */
public class ReactiveMockClientTestsUtils { public class ReactiveMockClientTestsUtils {
private final static Map<String, InetSocketAddress> ADDRESS_CACHE = new ConcurrentHashMap<>();
public static MockDelegatingElasticsearchHostProvider<SingleNodeHostProvider> single(String host) { public static MockDelegatingElasticsearchHostProvider<SingleNodeHostProvider> single(String host) {
return provider(host); return provider(host);
} }
@ -66,26 +66,17 @@ public class ReactiveMockClientTestsUtils {
public static <T extends HostProvider> MockDelegatingElasticsearchHostProvider<T> provider(String... hosts) { public static <T extends HostProvider> MockDelegatingElasticsearchHostProvider<T> provider(String... hosts) {
WebClientProvider clientProvider = new WebClientProvider();
ErrorCollector errorCollector = new ErrorCollector(); ErrorCollector errorCollector = new ErrorCollector();
MockWebClientProvider clientProvider = new MockWebClientProvider(errorCollector);
HostProvider delegate = null; HostProvider delegate = null;
if (hosts.length == 1) { if (hosts.length == 1) {
delegate = new SingleNodeHostProvider(HttpHeaders.EMPTY, errorCollector, hosts[0]) { delegate = new SingleNodeHostProvider(clientProvider, getInetSocketAddress(hosts[0])) {};
@Override // hook in there to modify result
public WebClient createWebClient(String host, HttpHeaders headers) {
return clientProvider.get(host);
}
};
} else { } else {
delegate = new MultiNodeHostProvider(HttpHeaders.EMPTY, errorCollector, hosts) { delegate = new MultiNodeHostProvider(clientProvider, Arrays.stream(hosts)
@Override // hook in there to modify result .map(ReactiveMockClientTestsUtils::getInetSocketAddress).toArray(InetSocketAddress[]::new)) {};
public WebClient createWebClient(String host, HttpHeaders headers) {
return clientProvider.get(host);
}
};
} }
return new MockDelegatingElasticsearchHostProvider(HttpHeaders.EMPTY, clientProvider, errorCollector, delegate, return new MockDelegatingElasticsearchHostProvider(HttpHeaders.EMPTY, clientProvider, errorCollector, delegate,
@ -93,6 +84,10 @@ public class ReactiveMockClientTestsUtils {
} }
private static InetSocketAddress getInetSocketAddress(String hostAndPort) {
return ADDRESS_CACHE.computeIfAbsent(hostAndPort, ElasticsearchHost::parse);
}
public static class ErrorCollector implements Consumer<Throwable> { public static class ErrorCollector implements Consumer<Throwable> {
List<Throwable> errors = new CopyOnWriteArrayList<>(); List<Throwable> errors = new CopyOnWriteArrayList<>();
@ -110,11 +105,11 @@ public class ReactiveMockClientTestsUtils {
public static class MockDelegatingElasticsearchHostProvider<T extends HostProvider> implements HostProvider { public static class MockDelegatingElasticsearchHostProvider<T extends HostProvider> implements HostProvider {
private final T delegate; private final T delegate;
private final WebClientProvider clientProvider; private final MockWebClientProvider clientProvider;
private final ErrorCollector errorCollector; private final ErrorCollector errorCollector;
private @Nullable String activeDefaultHost; private @Nullable String activeDefaultHost;
public MockDelegatingElasticsearchHostProvider(HttpHeaders httpHeaders, WebClientProvider clientProvider, public MockDelegatingElasticsearchHostProvider(HttpHeaders httpHeaders, MockWebClientProvider clientProvider,
ErrorCollector errorCollector, T delegate, String activeDefaultHost) { ErrorCollector errorCollector, T delegate, String activeDefaultHost) {
this.errorCollector = errorCollector; this.errorCollector = errorCollector;
@ -123,14 +118,14 @@ public class ReactiveMockClientTestsUtils {
this.activeDefaultHost = activeDefaultHost; this.activeDefaultHost = activeDefaultHost;
} }
public Mono<String> lookupActiveHost() { public Mono<InetSocketAddress> lookupActiveHost() {
return delegate.lookupActiveHost(); return delegate.lookupActiveHost();
} }
public Mono<String> lookupActiveHost(VerificationMode verificationMode) { public Mono<InetSocketAddress> lookupActiveHost(VerificationMode verificationMode) {
if (StringUtils.hasText(activeDefaultHost)) { if (StringUtils.hasText(activeDefaultHost)) {
return Mono.just(activeDefaultHost); return Mono.just(getInetSocketAddress(activeDefaultHost));
} }
return delegate.lookupActiveHost(verificationMode); return delegate.lookupActiveHost(verificationMode);
@ -144,34 +139,21 @@ public class ReactiveMockClientTestsUtils {
return delegate.getActive(verificationMode); return delegate.getActive(verificationMode);
} }
public Mono<WebClient> getActive(VerificationMode verificationMode, HttpHeaders headers) { public WebClient createWebClient(InetSocketAddress endpoint) {
return delegate.getActive(verificationMode, headers); return delegate.createWebClient(endpoint);
}
public WebClient createWebClient(String host, HttpHeaders headers) {
return delegate.createWebClient(host, headers);
} }
@Override @Override
public Mono<ClusterInformation> clusterInfo() { public Mono<ClusterInformation> clusterInfo() {
if (StringUtils.hasText(activeDefaultHost)) { if (StringUtils.hasText(activeDefaultHost)) {
return Mono.just(new ClusterInformation(Collections.singleton(ElasticsearchHost.online(activeDefaultHost)))); return Mono.just(new ClusterInformation(
Collections.singleton(ElasticsearchHost.online(getInetSocketAddress(activeDefaultHost)))));
} }
return delegate.clusterInfo(); return delegate.clusterInfo();
} }
@Override
public HttpHeaders getDefaultHeaders() {
return delegate.getDefaultHeaders();
}
@Override
public HostProvider withDefaultHeaders(HttpHeaders headers) {
throw new UnsupportedOperationException();
}
public Send when(String host) { public Send when(String host) {
return clientProvider.when(host); return clientProvider.when(host);
} }
@ -188,28 +170,25 @@ public class ReactiveMockClientTestsUtils {
return delegate; return delegate;
} }
@Override
public HostProvider withErrorListener(Consumer<Throwable> errorListener) {
throw new UnsupportedOperationException();
}
public MockDelegatingElasticsearchHostProvider<T> withActiveDefaultHost(String host) { public MockDelegatingElasticsearchHostProvider<T> withActiveDefaultHost(String host) {
return new MockDelegatingElasticsearchHostProvider(HttpHeaders.EMPTY, clientProvider, errorCollector, delegate, return new MockDelegatingElasticsearchHostProvider(HttpHeaders.EMPTY, clientProvider, errorCollector, delegate,
host); host);
} }
} }
public static class WebClientProvider { public static class MockWebClientProvider implements WebClientProvider {
private final Object lock = new Object(); private final Object lock = new Object();
private final Consumer<Throwable> errorListener;
private Map<String, WebClient> clientMap; private Map<InetSocketAddress, WebClient> clientMap;
private Map<String, RequestHeadersUriSpec> headersUriSpecMap; private Map<InetSocketAddress, RequestHeadersUriSpec> headersUriSpecMap;
private Map<String, RequestBodyUriSpec> bodyUriSpecMap; private Map<InetSocketAddress, RequestBodyUriSpec> bodyUriSpecMap;
private Map<String, ClientResponse> responseMap; private Map<InetSocketAddress, ClientResponse> responseMap;
public WebClientProvider() { public MockWebClientProvider(Consumer<Throwable> errorListener) {
this.errorListener = errorListener;
this.clientMap = new LinkedHashMap<>(); this.clientMap = new LinkedHashMap<>();
this.headersUriSpecMap = new LinkedHashMap<>(); this.headersUriSpecMap = new LinkedHashMap<>();
this.bodyUriSpecMap = new LinkedHashMap<>(); this.bodyUriSpecMap = new LinkedHashMap<>();
@ -217,10 +196,14 @@ public class ReactiveMockClientTestsUtils {
} }
public WebClient get(String host) { public WebClient get(String host) {
return get(getInetSocketAddress(host));
}
public WebClient get(InetSocketAddress endpoint) {
synchronized (lock) { synchronized (lock) {
return clientMap.computeIfAbsent(host, key -> { return clientMap.computeIfAbsent(endpoint, key -> {
WebClient webClient = mock(WebClient.class); WebClient webClient = mock(WebClient.class);
@ -243,17 +226,39 @@ public class ReactiveMockClientTestsUtils {
Mockito.when(bodyUriSpec.exchange()).thenReturn(Mono.just(response)); Mockito.when(bodyUriSpec.exchange()).thenReturn(Mono.just(response));
Mockito.when(response.statusCode()).thenReturn(HttpStatus.ACCEPTED); Mockito.when(response.statusCode()).thenReturn(HttpStatus.ACCEPTED);
headersUriSpecMap.putIfAbsent(host, headersUriSpec); headersUriSpecMap.putIfAbsent(key, headersUriSpec);
bodyUriSpecMap.putIfAbsent(host, bodyUriSpec); bodyUriSpecMap.putIfAbsent(key, bodyUriSpec);
responseMap.putIfAbsent(host, response); responseMap.putIfAbsent(key, response);
return webClient; return webClient;
}); });
} }
} }
@Override
public HttpHeaders getDefaultHeaders() {
return HttpHeaders.EMPTY;
}
@Override
public WebClientProvider withDefaultHeaders(HttpHeaders headers) {
throw new UnsupportedOperationException();
}
@Override
public Consumer<Throwable> getErrorListener() {
return errorListener;
}
@Override
public WebClientProvider withErrorListener(Consumer<Throwable> errorListener) {
throw new UnsupportedOperationException();
}
public Send when(String host) { public Send when(String host) {
return new CallbackImpl(get(host), headersUriSpecMap.get(host), bodyUriSpecMap.get(host), responseMap.get(host)); InetSocketAddress inetSocketAddress = getInetSocketAddress(host);
return new CallbackImpl(get(host), headersUriSpecMap.get(inetSocketAddress),
bodyUriSpecMap.get(inetSocketAddress), responseMap.get(inetSocketAddress));
} }
public interface Client { public interface Client {
@ -342,7 +347,7 @@ public class ReactiveMockClientTestsUtils {
} }
default Receive body(Supplier<byte[]> json) { default Receive body(Supplier<byte[]> json) {
return body(new DefaultDataBufferFactory().wrap(json.get())); return body(json.get());
} }
default Receive body(Resource resource) { default Receive body(Resource resource) {
@ -356,8 +361,8 @@ public class ReactiveMockClientTestsUtils {
}); });
} }
default Receive body(DataBuffer dataBuffer) { default Receive body(byte[] bytes) {
return receive(response -> Mockito.when(response.body(any())).thenReturn(Flux.just(dataBuffer))); return receive(response -> Mockito.when(response.body(any())).thenReturn(Mono.just(bytes)));
} }
static void ok(ClientResponse response) { static void ok(ClientResponse response) {

6
src/test/java/org/springframework/data/elasticsearch/client/reactive/SingleNodeHostProviderUnitTests.java
View File

@ -16,17 +16,17 @@
package org.springframework.data.elasticsearch.client.reactive; package org.springframework.data.elasticsearch.client.reactive;
import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.*;
import org.springframework.data.elasticsearch.client.NoReachableHostException;
import reactor.test.StepVerifier; import reactor.test.StepVerifier;
import org.junit.Before; import org.junit.Before;
import org.junit.Test; import org.junit.Test;
import org.springframework.data.elasticsearch.client.ElasticsearchHost; import org.springframework.data.elasticsearch.client.ElasticsearchHost;
import org.springframework.data.elasticsearch.client.ElasticsearchHost.State; import org.springframework.data.elasticsearch.client.ElasticsearchHost.State;
import org.springframework.data.elasticsearch.client.NoReachableHostException;
import org.springframework.data.elasticsearch.client.reactive.ReactiveMockClientTestsUtils.MockDelegatingElasticsearchHostProvider; import org.springframework.data.elasticsearch.client.reactive.ReactiveMockClientTestsUtils.MockDelegatingElasticsearchHostProvider;
import org.springframework.data.elasticsearch.client.reactive.ReactiveMockClientTestsUtils.WebClientProvider.Receive; import org.springframework.data.elasticsearch.client.reactive.ReactiveMockClientTestsUtils.MockWebClientProvider.Receive;
/** /**
* @author Christoph Strobl * @author Christoph Strobl

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

@ -13,7 +13,6 @@
* See the License for the specific language governing permissions and * See the License for the specific language governing permissions and
* limitations under the License. * limitations under the License.
*/ */
package org.springframework.data.elasticsearch.core; package org.springframework.data.elasticsearch.core;
import static org.apache.commons.lang.RandomStringUtils.*; import static org.apache.commons.lang.RandomStringUtils.*;
@ -25,17 +24,22 @@ import java.util.List;
import org.junit.Before; import org.junit.Before;
import org.junit.Test; import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.data.elasticsearch.TestUtils; import org.springframework.data.elasticsearch.TestUtils;
import org.springframework.data.elasticsearch.core.query.Criteria; import org.springframework.data.elasticsearch.core.query.Criteria;
import org.springframework.data.elasticsearch.core.query.CriteriaQuery; import org.springframework.data.elasticsearch.core.query.CriteriaQuery;
import org.springframework.data.elasticsearch.core.query.IndexQuery; import org.springframework.data.elasticsearch.core.query.IndexQuery;
import org.springframework.data.elasticsearch.core.query.IndexQueryBuilder; import org.springframework.data.elasticsearch.core.query.IndexQueryBuilder;
import org.springframework.data.elasticsearch.entities.SampleEntity; import org.springframework.data.elasticsearch.entities.SampleEntity;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringRunner;
/** /**
* @author Christoph Strobl * @author Christoph Strobl
* @currentRead Golden Fool - Robin Hobb * @currentRead Golden Fool - Robin Hobb
*/ */
@RunWith(SpringRunner.class)
@ContextConfiguration("classpath:infrastructure.xml")
public class ReactiveElasticsearchTemplateTests { public class ReactiveElasticsearchTemplateTests {
private ElasticsearchRestTemplate restTemplate; private ElasticsearchRestTemplate restTemplate;