You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@james.apache.org by bt...@apache.org on 2023/01/06 03:02:39 UTC
[james-project] branch master updated (bf448b4d93 -> 19984d6765)
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
from bf448b4d93 JAMES-3756 MailboxSession should have a loggedInUser field set to the loggedIn user (#1373)
new c7cb996ee9 JAMES-3771 Migrate backend opensearch high level to new java client
new 092c357b18 JAMES-3771 Migrate mailbox opensearch to new java client
new efbfae7833 JAMES-3771 Migrate quota-search-opensearch module to new java client
new 34f6cc44cc JAMES-3771 Migrate rest of the code to opensearch new java client
new 1fe4f82507 JAMES-3771 DockerElasticSearchExtension should cleanup its data after each test
new 7e1c35fa64 JAMES-3771 Upgrade to opensearch java client 2.1.0
new dc6d29eedf JAMES-3771 Add more memory to OpenSearch docker container for tests
new 5da1eafca4 JAMES-3771 Upgrade to OpenSearch 2.4
new 5934ecaa4d JAMES-3771 Refactor extract operations from hit in OpenSearchListeningMessageSearchIndex
new 49c358a39d JAMES-3771 Prevent SPI lookups on the HTTP eventloop
new 250a79dcf0 JAMES-3771 Fix thread leak issue within DeleteAllIndexDocumentsCleanupStrategy
new 19984d6765 JAMES-3771 Date searching should align IMAP4rev2 specifications
The 12 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
backends-common/opensearch/pom.xml | 17 +-
.../james/backends/opensearch/ClientProvider.java | 34 +-
.../opensearch/DeleteByQueryPerformer.java | 41 +-
.../james/backends/opensearch/DocumentId.java | 3 +-
.../backends/opensearch/IndexCreationFactory.java | 254 +++++-----
.../backends/opensearch/OpenSearchHealthCheck.java | 38 +-
.../backends/opensearch/OpenSearchIndexer.java | 121 +++--
.../opensearch/ReactorOpenSearchClient.java | 181 +++----
.../james/backends/opensearch/RoutingKey.java | 3 +-
.../backends/opensearch/UpdatedRepresentation.java | 3 +-
.../opensearch/json/JsonpDeserializerBase.java | 397 ++++++++++++++++
.../backends/opensearch/json/JsonpMapperBase.java | 102 ++++
.../james/backends/opensearch/json/JsonpUtils.java | 207 ++++++++
.../json/UnexpectedJsonEventException.java | 25 +-
.../json/jackson/JacksonJsonProvider.java | 298 ++++++++++++
.../json/jackson/JacksonJsonpGenerator.java | 374 +++++++++++++++
.../json/jackson/JacksonJsonpLocation.java | 41 +-
.../json/jackson/JacksonJsonpMapper.java | 123 +++++
.../json/jackson/JacksonJsonpParser.java | 313 +++++++++++++
.../opensearch/json/jackson/JacksonUtils.java | 24 +-
.../opensearch/json/jackson/JsonValueParser.java | 108 +++++
.../backends/opensearch/json/package-info.java | 14 +-
.../backends/opensearch/search/ScrolledSearch.java | 74 +--
.../ClientProviderImplConnectionContract.java | 12 +-
.../backends/opensearch/DockerOpenSearch.java | 18 +-
.../opensearch/DockerOpenSearchExtension.java | 64 ++-
.../opensearch/IndexCreationFactoryTest.java | 213 ++++-----
.../opensearch/OpenSearchHealthCheckTest.java | 52 +--
.../backends/opensearch/OpenSearchIndexerTest.java | 76 +--
.../opensearch/search/ScrolledSearchTest.java | 135 +++---
.../mailbox/opensearch/MailboxMappingFactory.java | 518 ++++++++++-----------
.../OpenSearchListeningMessageSearchIndex.java | 66 +--
.../opensearch/query/CriterionConverter.java | 384 +++++++++++----
.../mailbox/opensearch/query/QueryConverter.java | 37 +-
.../mailbox/opensearch/query/SortConverter.java | 42 +-
.../opensearch/search/OpenSearchSearcher.java | 42 +-
.../opensearch/OpenSearchIntegrationTest.java | 40 +-
.../OpenSearchListeningMessageSearchIndexTest.java | 64 +--
.../opensearch/search/OpenSearchSearcherTest.java | 21 +-
.../search/opensearch/OpenSearchQuotaSearcher.java | 59 +--
.../search/opensearch/QuotaQueryConverter.java | 56 ++-
.../opensearch/QuotaRatioMappingFactory.java | 63 ++-
.../opensearch/QuotaSearchIndexCreationUtil.java | 4 +-
.../OpenSearchQuotaConfigurationTest.java | 24 +-
.../search/opensearch/QuotaQueryConverterTest.java | 73 ++-
.../events/OpenSearchQuotaMailboxListenerTest.java | 15 +-
.../json/QuotaRatioToOpenSearchJsonTest.java | 3 +-
.../apache/james/DockerOpenSearchExtension.java | 13 +-
.../james/JamesServerWithRetryConnectionTest.java | 4 +-
...esWithNonCompatibleElasticSearchServerTest.java | 2 +-
.../apache/james/DockerOpenSearchExtension.java | 13 +-
.../modules/mailbox/OpenSearchStartUpCheck.java | 21 +-
.../DistributedThreadGetMethodTest.java | 17 +-
...RabbitMQReindexingWithEventDeadLettersTest.java | 4 +-
.../java/org/apache/james/util/docker/Images.java | 2 +-
55 files changed, 3578 insertions(+), 1374 deletions(-)
create mode 100644 backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/JsonpDeserializerBase.java
create mode 100644 backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/JsonpMapperBase.java
create mode 100644 backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/JsonpUtils.java
copy server/protocols/jmap-draft/src/main/java/org/apache/james/jmap/draft/exceptions/BlobNotFoundException.java => backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/UnexpectedJsonEventException.java (60%)
create mode 100644 backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/jackson/JacksonJsonProvider.java
create mode 100644 backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/jackson/JacksonJsonpGenerator.java
copy protocols/imap/src/main/java/org/apache/james/imap/message/response/FlagsResponse.java => backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/jackson/JacksonJsonpLocation.java (61%)
create mode 100644 backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/jackson/JacksonJsonpMapper.java
create mode 100644 backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/jackson/JacksonJsonpParser.java
copy server/protocols/jmap-draft/src/main/java/org/apache/james/jmap/draft/methods/JmapRequestParser.java => backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/jackson/JacksonUtils.java (62%)
create mode 100644 backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/jackson/JsonValueParser.java
copy mdn/src/main/java/org/apache/james/mdn/fields/Field.java => backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/package-info.java (72%)
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 05/12: JAMES-3771 DockerElasticSearchExtension should cleanup its data after each test
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 1fe4f82507da7c2611232c425c411b85addeb4c4
Author: Rene Cordier <rc...@linagora.com>
AuthorDate: Fri Jun 24 16:58:09 2022 +0700
JAMES-3771 DockerElasticSearchExtension should cleanup its data after each test
---
.../java/org/apache/james/DockerOpenSearchExtension.java | 13 +++++++------
.../apache/james/JamesServerWithRetryConnectionTest.java | 4 ++--
.../java/org/apache/james/DockerOpenSearchExtension.java | 13 +++++++------
.../RabbitMQReindexingWithEventDeadLettersTest.java | 4 ++--
4 files changed, 18 insertions(+), 16 deletions(-)
diff --git a/server/apps/cassandra-app/src/test/java/org/apache/james/DockerOpenSearchExtension.java b/server/apps/cassandra-app/src/test/java/org/apache/james/DockerOpenSearchExtension.java
index c9114ef5c3..3dceac582b 100644
--- a/server/apps/cassandra-app/src/test/java/org/apache/james/DockerOpenSearchExtension.java
+++ b/server/apps/cassandra-app/src/test/java/org/apache/james/DockerOpenSearchExtension.java
@@ -50,19 +50,20 @@ public class DockerOpenSearchExtension implements GuiceModuleTestExtension {
@Override
public void beforeAll(ExtensionContext extensionContext) {
- getDockerES().start();
+ getDockerOS().start();
}
@Override
public void beforeEach(ExtensionContext extensionContext) {
- if (!getDockerES().isRunning()) {
- getDockerES().unpause();
+ if (!getDockerOS().isRunning()) {
+ getDockerOS().unpause();
}
await();
}
@Override
public void afterEach(ExtensionContext extensionContext) {
+ getDockerOS().cleanUpData();
}
@Override
@@ -73,17 +74,17 @@ public class DockerOpenSearchExtension implements GuiceModuleTestExtension {
@Override
public void await() {
- getDockerES().flushIndices();
+ getDockerOS().flushIndices();
}
private OpenSearchConfiguration getOpenSearchConfigurationForDocker() {
return OpenSearchConfiguration.builder()
- .addHost(getDockerES().getHttpHost())
+ .addHost(getDockerOS().getHttpHost())
.requestTimeout(requestTimeout)
.build();
}
- public DockerOpenSearch getDockerES() {
+ public DockerOpenSearch getDockerOS() {
return dockerOpenSearch;
}
}
diff --git a/server/apps/cassandra-app/src/test/java/org/apache/james/JamesServerWithRetryConnectionTest.java b/server/apps/cassandra-app/src/test/java/org/apache/james/JamesServerWithRetryConnectionTest.java
index bcea7a3b67..f5afa48acb 100644
--- a/server/apps/cassandra-app/src/test/java/org/apache/james/JamesServerWithRetryConnectionTest.java
+++ b/server/apps/cassandra-app/src/test/java/org/apache/james/JamesServerWithRetryConnectionTest.java
@@ -86,9 +86,9 @@ class JamesServerWithRetryConnectionTest {
@Test
void serverShouldRetryToConnectToOpenSearchWhenStartService(GuiceJamesServer server) throws Exception {
- dockerOpenSearch.getDockerES().pause();
+ dockerOpenSearch.getDockerOS().pause();
- waitToStartContainer(WAITING_TIME, dockerOpenSearch.getDockerES()::unpause);
+ waitToStartContainer(WAITING_TIME, dockerOpenSearch.getDockerOS()::unpause);
assertThatServerStartCorrectly(server);
}
diff --git a/server/apps/scaling-pulsar-smtp/src/test/java/org/apache/james/DockerOpenSearchExtension.java b/server/apps/scaling-pulsar-smtp/src/test/java/org/apache/james/DockerOpenSearchExtension.java
index c9114ef5c3..3dceac582b 100644
--- a/server/apps/scaling-pulsar-smtp/src/test/java/org/apache/james/DockerOpenSearchExtension.java
+++ b/server/apps/scaling-pulsar-smtp/src/test/java/org/apache/james/DockerOpenSearchExtension.java
@@ -50,19 +50,20 @@ public class DockerOpenSearchExtension implements GuiceModuleTestExtension {
@Override
public void beforeAll(ExtensionContext extensionContext) {
- getDockerES().start();
+ getDockerOS().start();
}
@Override
public void beforeEach(ExtensionContext extensionContext) {
- if (!getDockerES().isRunning()) {
- getDockerES().unpause();
+ if (!getDockerOS().isRunning()) {
+ getDockerOS().unpause();
}
await();
}
@Override
public void afterEach(ExtensionContext extensionContext) {
+ getDockerOS().cleanUpData();
}
@Override
@@ -73,17 +74,17 @@ public class DockerOpenSearchExtension implements GuiceModuleTestExtension {
@Override
public void await() {
- getDockerES().flushIndices();
+ getDockerOS().flushIndices();
}
private OpenSearchConfiguration getOpenSearchConfigurationForDocker() {
return OpenSearchConfiguration.builder()
- .addHost(getDockerES().getHttpHost())
+ .addHost(getDockerOS().getHttpHost())
.requestTimeout(requestTimeout)
.build();
}
- public DockerOpenSearch getDockerES() {
+ public DockerOpenSearch getDockerOS() {
return dockerOpenSearch;
}
}
diff --git a/server/protocols/webadmin-integration-test/distributed-webadmin-integration-test/src/test/java/org/apache/james/webadmin/integration/rabbitmq/RabbitMQReindexingWithEventDeadLettersTest.java b/server/protocols/webadmin-integration-test/distributed-webadmin-integration-test/src/test/java/org/apache/james/webadmin/integration/rabbitmq/RabbitMQReindexingWithEventDeadLettersTest.java
index 7be41f7073..360c40128b 100644
--- a/server/protocols/webadmin-integration-test/distributed-webadmin-integration-test/src/test/java/org/apache/james/webadmin/integration/rabbitmq/RabbitMQReindexingWithEventDeadLettersTest.java
+++ b/server/protocols/webadmin-integration-test/distributed-webadmin-integration-test/src/test/java/org/apache/james/webadmin/integration/rabbitmq/RabbitMQReindexingWithEventDeadLettersTest.java
@@ -125,7 +125,7 @@ class RabbitMQReindexingWithEventDeadLettersTest {
aliceAccessToken = authenticateJamesUser(LocalHostURIBuilder.baseUri(jmapPort), ALICE, ALICE_PASSWORD);
- dockerOpenSearch.getDockerES().pause();
+ dockerOpenSearch.getDockerOS().pause();
Thread.sleep(Duration.ofSeconds(2).toMillis()); // Docker pause is asynchronous and we found no way to poll for it
}
@@ -153,7 +153,7 @@ class RabbitMQReindexingWithEventDeadLettersTest {
}
private void unpauseOpenSearch() throws Exception {
- dockerOpenSearch.getDockerES().unpause();
+ dockerOpenSearch.getDockerOS().unpause();
Thread.sleep(Duration.ofSeconds(2).toMillis()); // Docker unpause is asynchronous and we found no way to poll for it
}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 08/12: JAMES-3771 Upgrade to OpenSearch 2.4
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 5da1eafca441314a5907a8dd52fb03dff9b4c85d
Author: Rene Cordier <rc...@linagora.com>
AuthorDate: Fri Dec 9 14:45:31 2022 +0700
JAMES-3771 Upgrade to OpenSearch 2.4
---
backends-common/opensearch/pom.xml | 2 +-
server/testing/src/main/java/org/apache/james/util/docker/Images.java | 2 +-
2 files changed, 2 insertions(+), 2 deletions(-)
diff --git a/backends-common/opensearch/pom.xml b/backends-common/opensearch/pom.xml
index 86237bd857..46339a3b2e 100644
--- a/backends-common/opensearch/pom.xml
+++ b/backends-common/opensearch/pom.xml
@@ -88,7 +88,7 @@
<dependency>
<groupId>org.opensearch.client</groupId>
<artifactId>opensearch-rest-client</artifactId>
- <version>2.3.0</version>
+ <version>2.4.0</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
diff --git a/server/testing/src/main/java/org/apache/james/util/docker/Images.java b/server/testing/src/main/java/org/apache/james/util/docker/Images.java
index 9b227cab8e..13c11ad521 100644
--- a/server/testing/src/main/java/org/apache/james/util/docker/Images.java
+++ b/server/testing/src/main/java/org/apache/james/util/docker/Images.java
@@ -25,7 +25,7 @@ public interface Images {
String ELASTICSEARCH_2 = "elasticsearch:2.4.6";
String ELASTICSEARCH_6 = "docker.elastic.co/elasticsearch/elasticsearch:6.3.2";
String ELASTICSEARCH_7 = "docker.elastic.co/elasticsearch/elasticsearch:7.10.2";
- String OPENSEARCH = "opensearchproject/opensearch:2.1.0";
+ String OPENSEARCH = "opensearchproject/opensearch:2.4.0";
String TIKA = "apache/tika:1.28.2";
String MOCK_SMTP_SERVER = "linagora/mock-smtp-server:0.6";
}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 07/12: JAMES-3771 Add more memory to OpenSearch docker container for tests
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit dc6d29eedf172f6fec0e8f023984e2830c74e446
Author: Rene Cordier <rc...@linagora.com>
AuthorDate: Fri Dec 9 14:44:56 2022 +0700
JAMES-3771 Add more memory to OpenSearch docker container for tests
---
.../james/backends/opensearch/DockerOpenSearch.java | 18 +++++++++---------
1 file changed, 9 insertions(+), 9 deletions(-)
diff --git a/backends-common/opensearch/src/test/java/org/apache/james/backends/opensearch/DockerOpenSearch.java b/backends-common/opensearch/src/test/java/org/apache/james/backends/opensearch/DockerOpenSearch.java
index 4116d7c007..e803d57fb2 100644
--- a/backends-common/opensearch/src/test/java/org/apache/james/backends/opensearch/DockerOpenSearch.java
+++ b/backends-common/opensearch/src/test/java/org/apache/james/backends/opensearch/DockerOpenSearch.java
@@ -19,8 +19,8 @@
package org.apache.james.backends.opensearch;
-import static org.apache.james.backends.opensearch.DockerOpenSearch.Fixture.ES_HTTP_PORT;
-import static org.apache.james.backends.opensearch.DockerOpenSearch.Fixture.ES_MEMORY;
+import static org.apache.james.backends.opensearch.DockerOpenSearch.Fixture.OS_HTTP_PORT;
+import static org.apache.james.backends.opensearch.DockerOpenSearch.Fixture.OS_MEMORY;
import java.net.MalformedURLException;
import java.net.URL;
@@ -125,8 +125,8 @@ public interface DockerOpenSearch {
}
interface Fixture {
- int ES_HTTP_PORT = 9200;
- int ES_MEMORY = 620;
+ int OS_HTTP_PORT = 9200;
+ int OS_MEMORY = 1024;
}
class NoAuth implements DockerOpenSearch {
@@ -134,11 +134,11 @@ public interface DockerOpenSearch {
static DockerContainer defaultContainer(String imageName) {
return DockerContainer.fromName(imageName)
.withTmpFs(ImmutableMap.of("/usr/share/opensearch/data/nodes/0", "rw,size=200m"))
- .withExposedPorts(ES_HTTP_PORT)
+ .withExposedPorts(OS_HTTP_PORT)
.withEnv("discovery.type", "single-node")
.withEnv("DISABLE_INSTALL_DEMO_CONFIG", "true")
.withEnv("DISABLE_SECURITY_PLUGIN", "true")
- .withEnv("ES_JAVA_OPTS", "-Xms" + ES_MEMORY + "m -Xmx" + ES_MEMORY + "m")
+ .withEnv("ES_JAVA_OPTS", "-Xms" + OS_MEMORY + "m -Xmx" + OS_MEMORY + "m")
.withAffinityToContainer()
.withName("james-testing-opensearch-" + UUID.randomUUID())
.waitingFor(new HostPortWaitStrategy().withRateLimiter(RateLimiters.TWENTIES_PER_SECOND));
@@ -169,7 +169,7 @@ public interface DockerOpenSearch {
}
public int getHttpPort() {
- return eSContainer.getMappedPort(ES_HTTP_PORT);
+ return eSContainer.getMappedPort(OS_HTTP_PORT);
}
public String getIp() {
@@ -227,7 +227,7 @@ public interface DockerOpenSearch {
.withFileFromClasspath("conf/default.crt", "auth-es/default.crt")
.withFileFromClasspath("conf/default.key", "auth-es/default.key")
.withFileFromClasspath("Dockerfile", "auth-es/NginxDockerfile")))
- .withExposedPorts(ES_HTTP_PORT)
+ .withExposedPorts(OS_HTTP_PORT)
.withTmpFs(ImmutableMap.of("/usr/share/opensearch/data/nodes/0", "rw,size=200m"))
.withLogConsumer(frame -> LOGGER.debug("[NGINX] " + frame.getUtf8String()))
.withNetwork(network)
@@ -246,7 +246,7 @@ public interface DockerOpenSearch {
}
public int getHttpPort() {
- return nginx.getMappedPort(ES_HTTP_PORT);
+ return nginx.getMappedPort(OS_HTTP_PORT);
}
public String getIp() {
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 04/12: JAMES-3771 Migrate rest of the code to opensearch new java client
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 34f6cc44cc88e8f28df22374a057bb8a87d82bbd
Author: Rene Cordier <rc...@linagora.com>
AuthorDate: Wed Jun 22 11:10:51 2022 +0700
JAMES-3771 Migrate rest of the code to opensearch new java client
---
...mesWithNonCompatibleElasticSearchServerTest.java | 2 +-
.../modules/mailbox/OpenSearchStartUpCheck.java | 21 +++++++++++----------
.../distributed/DistributedThreadGetMethodTest.java | 17 ++++++++---------
3 files changed, 20 insertions(+), 20 deletions(-)
diff --git a/server/apps/cassandra-app/src/test/java/org/apache/james/JamesWithNonCompatibleElasticSearchServerTest.java b/server/apps/cassandra-app/src/test/java/org/apache/james/JamesWithNonCompatibleElasticSearchServerTest.java
index 078782885c..31699b491f 100644
--- a/server/apps/cassandra-app/src/test/java/org/apache/james/JamesWithNonCompatibleElasticSearchServerTest.java
+++ b/server/apps/cassandra-app/src/test/java/org/apache/james/JamesWithNonCompatibleElasticSearchServerTest.java
@@ -59,7 +59,7 @@ class JamesWithNonCompatibleElasticSearchServerTest {
.containsOnly(CheckResult.builder()
.checkName(OpenSearchStartUpCheck.CHECK_NAME)
.resultType(StartUpCheck.ResultType.BAD)
- .description("ES version(6.3.2) is not compatible with the recommendation(2.0.0)")
+ .description("Error when checking ES version: Missing required property 'OpenSearchVersionInfo.distribution'")
.build()));
assertThat(server.isStarted())
diff --git a/server/container/guice/opensearch/src/main/java/org/apache/james/modules/mailbox/OpenSearchStartUpCheck.java b/server/container/guice/opensearch/src/main/java/org/apache/james/modules/mailbox/OpenSearchStartUpCheck.java
index dba5f904b1..4d1e6db11b 100644
--- a/server/container/guice/opensearch/src/main/java/org/apache/james/modules/mailbox/OpenSearchStartUpCheck.java
+++ b/server/container/guice/opensearch/src/main/java/org/apache/james/modules/mailbox/OpenSearchStartUpCheck.java
@@ -19,21 +19,18 @@
package org.apache.james.modules.mailbox;
-import java.io.IOException;
-
import javax.inject.Inject;
import org.apache.james.backends.opensearch.ReactorOpenSearchClient;
import org.apache.james.lifecycle.api.StartUpCheck;
-import org.opensearch.Version;
-import org.opensearch.client.RequestOptions;
+import org.opensearch.client.transport.Version;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class OpenSearchStartUpCheck implements StartUpCheck {
private static final Logger LOGGER = LoggerFactory.getLogger(OpenSearchStartUpCheck.class);
- private static final Version RECOMMENDED_ES_VERSION = Version.V_2_0_0;
+ private static final Version RECOMMENDED_OS_VERSION = Version.parse("2.0.0");
private static final String VERSION_CHECKING_ERROR_MESSAGE = "Error when checking ES version";
public static final String CHECK_NAME = "OpenSearchStartUpCheck";
@@ -48,10 +45,9 @@ public class OpenSearchStartUpCheck implements StartUpCheck {
@Override
public CheckResult check() {
try {
+ Version esVersion = Version.parse(client.info().block().version().number());
- Version esVersion = Version.fromString(client.info(RequestOptions.DEFAULT).getVersion().getNumber());
-
- if (esVersion.isCompatible(RECOMMENDED_ES_VERSION)) {
+ if (isCompatible(esVersion)) {
return CheckResult.builder()
.checkName(checkName())
.resultType(ResultType.GOOD)
@@ -61,7 +57,7 @@ public class OpenSearchStartUpCheck implements StartUpCheck {
String esVersionCompatibilityWarn = String.format(
"ES version(%s) is not compatible with the recommendation(%s)",
esVersion.toString(),
- RECOMMENDED_ES_VERSION.toString());
+ RECOMMENDED_OS_VERSION.toString());
LOGGER.warn(esVersionCompatibilityWarn);
return CheckResult.builder()
@@ -69,7 +65,7 @@ public class OpenSearchStartUpCheck implements StartUpCheck {
.resultType(ResultType.BAD)
.description(esVersionCompatibilityWarn)
.build();
- } catch (IOException e) {
+ } catch (Exception e) {
LOGGER.error(VERSION_CHECKING_ERROR_MESSAGE, e);
return CheckResult.builder()
.checkName(checkName())
@@ -83,4 +79,9 @@ public class OpenSearchStartUpCheck implements StartUpCheck {
public String checkName() {
return CHECK_NAME;
}
+
+ private boolean isCompatible(Version usedVersion) {
+ return usedVersion.major() > RECOMMENDED_OS_VERSION.major()
+ || usedVersion.major() == RECOMMENDED_OS_VERSION.major() && usedVersion.minor() >= RECOMMENDED_OS_VERSION.minor();
+ }
}
diff --git a/server/protocols/jmap-rfc-8621-integration-tests/distributed-jmap-rfc-8621-integration-tests/src/test/java/org/apache/james/jmap/rfc8621/distributed/DistributedThreadGetMethodTest.java b/server/protocols/jmap-rfc-8621-integration-tests/distributed-jmap-rfc-8621-integration-tests/src/test/java/org/apache/james/jmap/rfc8621/distributed/DistributedThreadGetMethodTest.java
index cc507c9e4c..58e6e1476f 100644
--- a/server/protocols/jmap-rfc-8621-integration-tests/distributed-jmap-rfc-8621-integration-tests/src/test/java/org/apache/james/jmap/rfc8621/distributed/DistributedThreadGetMethodTest.java
+++ b/server/protocols/jmap-rfc-8621-integration-tests/distributed-jmap-rfc-8621-integration-tests/src/test/java/org/apache/james/jmap/rfc8621/distributed/DistributedThreadGetMethodTest.java
@@ -49,10 +49,8 @@ import org.awaitility.Durations;
import org.awaitility.core.ConditionFactory;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.extension.RegisterExtension;
-import org.opensearch.action.search.SearchRequest;
-import org.opensearch.client.RequestOptions;
-import org.opensearch.index.query.QueryBuilder;
-import org.opensearch.search.builder.SearchSourceBuilder;
+import org.opensearch.client.opensearch._types.query_dsl.Query;
+import org.opensearch.client.opensearch.core.SearchRequest;
public class DistributedThreadGetMethodTest implements ThreadGetContract {
private static final ConditionFactory CALMLY_AWAIT = Awaitility
@@ -103,13 +101,14 @@ public class DistributedThreadGetMethodTest implements ThreadGetContract {
openSearch.getDockerOpenSearch().configuration());
}
- private void awaitForOpenSearch(QueryBuilder query, long totalHits) {
+ private void awaitForOpenSearch(Query query, long totalHits) {
CALMLY_AWAIT.atMost(Durations.TEN_SECONDS)
.untilAsserted(() -> assertThat(client.search(
- new SearchRequest(MailboxOpenSearchConstants.DEFAULT_MAILBOX_INDEX.getValue())
- .source(new SearchSourceBuilder().query(query)),
- RequestOptions.DEFAULT)
+ new SearchRequest.Builder()
+ .index(MailboxOpenSearchConstants.DEFAULT_MAILBOX_INDEX.getValue())
+ .query(query)
+ .build())
.block()
- .getHits().getTotalHits().value).isEqualTo(totalHits));
+ .hits().total().value()).isEqualTo(totalHits));
}
}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 03/12: JAMES-3771 Migrate quota-search-opensearch module to new java client
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit efbfae7833710a416800f3d83a8c591d84c4d0b1
Author: Rene Cordier <rc...@linagora.com>
AuthorDate: Tue Jun 21 16:53:40 2022 +0700
JAMES-3771 Migrate quota-search-opensearch module to new java client
---
.../search/opensearch/OpenSearchQuotaSearcher.java | 59 +++++++++--------
.../search/opensearch/QuotaQueryConverter.java | 56 ++++++++++-------
.../opensearch/QuotaRatioMappingFactory.java | 63 +++++++++----------
.../opensearch/QuotaSearchIndexCreationUtil.java | 4 +-
.../OpenSearchQuotaConfigurationTest.java | 24 +++----
.../search/opensearch/QuotaQueryConverterTest.java | 73 +++++++++++++++-------
.../events/OpenSearchQuotaMailboxListenerTest.java | 15 +++--
.../json/QuotaRatioToOpenSearchJsonTest.java | 3 +-
8 files changed, 167 insertions(+), 130 deletions(-)
diff --git a/mailbox/plugin/quota-search-opensearch/src/main/java/org/apache/james/quota/search/opensearch/OpenSearchQuotaSearcher.java b/mailbox/plugin/quota-search-opensearch/src/main/java/org/apache/james/quota/search/opensearch/OpenSearchQuotaSearcher.java
index f26c2faa55..30654067c7 100644
--- a/mailbox/plugin/quota-search-opensearch/src/main/java/org/apache/james/quota/search/opensearch/OpenSearchQuotaSearcher.java
+++ b/mailbox/plugin/quota-search-opensearch/src/main/java/org/apache/james/quota/search/opensearch/OpenSearchQuotaSearcher.java
@@ -19,6 +19,9 @@
package org.apache.james.quota.search.opensearch;
+import static org.apache.james.quota.search.opensearch.json.JsonMessageConstants.USER;
+
+import java.io.IOException;
import java.util.List;
import org.apache.james.backends.opensearch.AliasName;
@@ -28,21 +31,20 @@ import org.apache.james.backends.opensearch.search.ScrolledSearch;
import org.apache.james.core.Username;
import org.apache.james.quota.search.QuotaQuery;
import org.apache.james.quota.search.QuotaSearcher;
-import org.apache.james.quota.search.opensearch.json.JsonMessageConstants;
-import org.opensearch.action.search.SearchRequest;
-import org.opensearch.client.RequestOptions;
-import org.opensearch.common.unit.TimeValue;
-import org.opensearch.search.SearchHit;
-import org.opensearch.search.builder.SearchSourceBuilder;
-import org.opensearch.search.sort.SortBuilders;
-import org.opensearch.search.sort.SortOrder;
+import org.opensearch.client.opensearch._types.FieldSort;
+import org.opensearch.client.opensearch._types.SortOptions;
+import org.opensearch.client.opensearch._types.SortOrder;
+import org.opensearch.client.opensearch._types.Time;
+import org.opensearch.client.opensearch.core.SearchRequest;
+import org.opensearch.client.opensearch.core.search.Hit;
+import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.collect.ImmutableList;
import reactor.core.publisher.Flux;
public class OpenSearchQuotaSearcher implements QuotaSearcher {
- private static final TimeValue TIMEOUT = TimeValue.timeValueMinutes(1);
+ private static final Time TIMEOUT = new Time.Builder().time("1m").build();
private final ReactorOpenSearchClient client;
private final AliasName readAlias;
@@ -58,7 +60,7 @@ public class OpenSearchQuotaSearcher implements QuotaSearcher {
public List<Username> search(QuotaQuery query) {
try {
return searchHits(query)
- .map(SearchHit::getId)
+ .map(Hit::id)
.map(Username::of)
.collect(ImmutableList.toImmutableList())
.block();
@@ -67,7 +69,7 @@ public class OpenSearchQuotaSearcher implements QuotaSearcher {
}
}
- private Flux<SearchHit> searchHits(QuotaQuery query) {
+ private Flux<Hit<ObjectNode>> searchHits(QuotaQuery query) throws IOException {
if (query.getLimit().isLimited()) {
return executeSingleSearch(query);
} else {
@@ -75,31 +77,36 @@ public class OpenSearchQuotaSearcher implements QuotaSearcher {
}
}
- private Flux<SearchHit> executeSingleSearch(QuotaQuery query) {
- SearchSourceBuilder searchSourceBuilder = searchSourceBuilder(query)
+ private Flux<Hit<ObjectNode>> executeSingleSearch(QuotaQuery query) throws IOException {
+ SearchRequest.Builder searchRequest = searchRequestBuilder(query)
+ .index(readAlias.getValue())
.from(query.getOffset().getValue());
- query.getLimit().getValue()
- .ifPresent(searchSourceBuilder::size);
- SearchRequest searchRequest = new SearchRequest(readAlias.getValue())
- .source(searchSourceBuilder);
+ query.getLimit().getValue()
+ .ifPresent(searchRequest::size);
- return client.search(searchRequest, RequestOptions.DEFAULT)
- .flatMapIterable(searchResponse -> ImmutableList.copyOf(searchResponse.getHits().getHits()));
+ return client.search(searchRequest.build())
+ .flatMapIterable(searchResponse -> ImmutableList.copyOf(searchResponse.hits().hits()));
}
- private Flux<SearchHit> executeScrolledSearch(QuotaQuery query) {
+ private Flux<Hit<ObjectNode>> executeScrolledSearch(QuotaQuery query) {
return new ScrolledSearch(client,
- new SearchRequest(readAlias.getValue())
- .source(searchSourceBuilder(query))
- .scroll(TIMEOUT))
+ searchRequestBuilder(query)
+ .index(readAlias.getValue())
+ .scroll(TIMEOUT)
+ .build())
.searchHits()
.skip(query.getOffset().getValue());
}
- private SearchSourceBuilder searchSourceBuilder(QuotaQuery query) {
- return new SearchSourceBuilder()
+ private SearchRequest.Builder searchRequestBuilder(QuotaQuery query) {
+ return new SearchRequest.Builder()
.query(quotaQueryConverter.from(query))
- .sort(SortBuilders.fieldSort(JsonMessageConstants.USER).order(SortOrder.ASC));
+ .sort(new SortOptions.Builder()
+ .field(new FieldSort.Builder()
+ .field(USER)
+ .order(SortOrder.Asc)
+ .build())
+ .build());
}
}
\ No newline at end of file
diff --git a/mailbox/plugin/quota-search-opensearch/src/main/java/org/apache/james/quota/search/opensearch/QuotaQueryConverter.java b/mailbox/plugin/quota-search-opensearch/src/main/java/org/apache/james/quota/search/opensearch/QuotaQueryConverter.java
index 3454ef2eb8..6e037e1d9c 100644
--- a/mailbox/plugin/quota-search-opensearch/src/main/java/org/apache/james/quota/search/opensearch/QuotaQueryConverter.java
+++ b/mailbox/plugin/quota-search-opensearch/src/main/java/org/apache/james/quota/search/opensearch/QuotaQueryConverter.java
@@ -21,9 +21,6 @@ package org.apache.james.quota.search.opensearch;
import static org.apache.james.quota.search.opensearch.json.JsonMessageConstants.DOMAIN;
import static org.apache.james.quota.search.opensearch.json.JsonMessageConstants.QUOTA_RATIO;
-import static org.opensearch.index.query.QueryBuilders.matchAllQuery;
-import static org.opensearch.index.query.QueryBuilders.rangeQuery;
-import static org.opensearch.index.query.QueryBuilders.termQuery;
import java.util.List;
import java.util.Map;
@@ -35,19 +32,22 @@ import org.apache.james.quota.search.QuotaClause.HasDomain;
import org.apache.james.quota.search.QuotaClause.LessThan;
import org.apache.james.quota.search.QuotaClause.MoreThan;
import org.apache.james.quota.search.QuotaQuery;
-import org.opensearch.index.query.BoolQueryBuilder;
-import org.opensearch.index.query.QueryBuilder;
-import org.opensearch.index.query.RangeQueryBuilder;
-import org.opensearch.index.query.TermQueryBuilder;
+import org.opensearch.client.json.JsonData;
+import org.opensearch.client.opensearch._types.FieldValue;
+import org.opensearch.client.opensearch._types.query_dsl.BoolQuery;
+import org.opensearch.client.opensearch._types.query_dsl.MatchAllQuery;
+import org.opensearch.client.opensearch._types.query_dsl.Query;
+import org.opensearch.client.opensearch._types.query_dsl.RangeQuery;
+import org.opensearch.client.opensearch._types.query_dsl.TermQuery;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableMap.Builder;
class QuotaQueryConverter {
- private final Map<Class<? extends QuotaClause>, Function<QuotaClause, QueryBuilder>> clauseConverter;
+ private final Map<Class<? extends QuotaClause>, Function<QuotaClause, Query>> clauseConverter;
QuotaQueryConverter() {
- Builder<Class<? extends QuotaClause>, Function<QuotaClause, QueryBuilder>> builder = ImmutableMap.builder();
+ Builder<Class<? extends QuotaClause>, Function<QuotaClause, Query>> builder = ImmutableMap.builder();
builder.put(HasDomain.class, this::convertHasDomain);
builder.put(And.class, this::disableNestedAnd);
@@ -57,10 +57,10 @@ class QuotaQueryConverter {
clauseConverter = builder.build();
}
- QueryBuilder from(QuotaQuery query) {
+ Query from(QuotaQuery query) {
List<QuotaClause> clauses = query.getClause().getClauses();
if (clauses.isEmpty()) {
- return matchAllQuery();
+ return new MatchAllQuery.Builder().build()._toQuery();
}
if (clauses.size() == 1) {
return singleClauseAsESQuery(clauses.get(0));
@@ -69,34 +69,46 @@ class QuotaQueryConverter {
return clausesAsAndESQuery(clauses);
}
- private BoolQueryBuilder clausesAsAndESQuery(List<QuotaClause> clauses) {
- BoolQueryBuilder boolQueryBuilder = new BoolQueryBuilder();
+ private Query clausesAsAndESQuery(List<QuotaClause> clauses) {
+ BoolQuery.Builder boolQueryBuilder = new BoolQuery.Builder();
clauses.stream()
.map(this::singleClauseAsESQuery)
.forEach(boolQueryBuilder::must);
- return boolQueryBuilder;
+ return boolQueryBuilder.build()._toQuery();
}
- private QueryBuilder disableNestedAnd(QuotaClause clause) {
+ private Query disableNestedAnd(QuotaClause clause) {
throw new IllegalArgumentException("Nested \"And\" clauses are not supported");
}
- private TermQueryBuilder convertHasDomain(QuotaClause clause) {
+ private Query convertHasDomain(QuotaClause clause) {
HasDomain hasDomain = (HasDomain) clause;
- return termQuery(DOMAIN, hasDomain.getDomain().asString());
+ return new TermQuery.Builder()
+ .field(DOMAIN)
+ .value(new FieldValue.Builder().stringValue(hasDomain.getDomain().asString()).build())
+ .build()
+ ._toQuery();
}
- private RangeQueryBuilder convertMoreThan(QuotaClause clause) {
+ private Query convertMoreThan(QuotaClause clause) {
MoreThan moreThan = (MoreThan) clause;
- return rangeQuery(QUOTA_RATIO).gte(moreThan.getQuotaBoundary().getRatio());
+ return new RangeQuery.Builder()
+ .field(QUOTA_RATIO)
+ .gte(JsonData.of(moreThan.getQuotaBoundary().getRatio()))
+ .build()
+ ._toQuery();
}
- private RangeQueryBuilder convertLessThan(QuotaClause clause) {
+ private Query convertLessThan(QuotaClause clause) {
LessThan lessThan = (LessThan) clause;
- return rangeQuery(QUOTA_RATIO).lte(lessThan.getQuotaBoundary().getRatio());
+ return new RangeQuery.Builder()
+ .field(QUOTA_RATIO)
+ .lte(JsonData.of(lessThan.getQuotaBoundary().getRatio()))
+ .build()
+ ._toQuery();
}
- private QueryBuilder singleClauseAsESQuery(QuotaClause clause) {
+ private Query singleClauseAsESQuery(QuotaClause clause) {
return clauseConverter.get(clause.getClass()).apply(clause);
}
diff --git a/mailbox/plugin/quota-search-opensearch/src/main/java/org/apache/james/quota/search/opensearch/QuotaRatioMappingFactory.java b/mailbox/plugin/quota-search-opensearch/src/main/java/org/apache/james/quota/search/opensearch/QuotaRatioMappingFactory.java
index 68b73bc737..662ba7c488 100644
--- a/mailbox/plugin/quota-search-opensearch/src/main/java/org/apache/james/quota/search/opensearch/QuotaRatioMappingFactory.java
+++ b/mailbox/plugin/quota-search-opensearch/src/main/java/org/apache/james/quota/search/opensearch/QuotaRatioMappingFactory.java
@@ -19,48 +19,41 @@
package org.apache.james.quota.search.opensearch;
-import static org.apache.james.backends.opensearch.IndexCreationFactory.DOUBLE;
-import static org.apache.james.backends.opensearch.IndexCreationFactory.KEYWORD;
-import static org.apache.james.backends.opensearch.IndexCreationFactory.PROPERTIES;
-import static org.apache.james.backends.opensearch.IndexCreationFactory.REQUIRED;
-import static org.apache.james.backends.opensearch.IndexCreationFactory.ROUTING;
-import static org.apache.james.backends.opensearch.IndexCreationFactory.TYPE;
import static org.apache.james.quota.search.opensearch.json.JsonMessageConstants.DOMAIN;
import static org.apache.james.quota.search.opensearch.json.JsonMessageConstants.QUOTA_RATIO;
import static org.apache.james.quota.search.opensearch.json.JsonMessageConstants.USER;
-import static org.opensearch.common.xcontent.XContentFactory.jsonBuilder;
-import java.io.IOException;
+import java.util.Map;
-import org.opensearch.common.xcontent.XContentBuilder;
+import org.opensearch.client.opensearch._types.mapping.DoubleNumberProperty;
+import org.opensearch.client.opensearch._types.mapping.KeywordProperty;
+import org.opensearch.client.opensearch._types.mapping.Property;
+import org.opensearch.client.opensearch._types.mapping.RoutingField;
+import org.opensearch.client.opensearch._types.mapping.TypeMapping;
-class QuotaRatioMappingFactory {
-
- public static XContentBuilder getMappingContent() {
- try {
- return jsonBuilder()
- .startObject()
- .startObject(ROUTING)
- .field(REQUIRED, true)
- .endObject()
-
- .startObject(PROPERTIES)
+import com.google.common.collect.ImmutableMap;
- .startObject(USER)
- .field(TYPE, KEYWORD)
- .endObject()
-
- .startObject(DOMAIN)
- .field(TYPE, KEYWORD)
- .endObject()
+class QuotaRatioMappingFactory {
+ public static TypeMapping getMappingContent() {
+ return new TypeMapping.Builder()
+ .routing(new RoutingField.Builder()
+ .required(true)
+ .build())
+ .properties(generateProperties())
+ .build();
+ }
- .startObject(QUOTA_RATIO)
- .field(TYPE, DOUBLE)
- .endObject()
- .endObject()
- .endObject();
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
+ private static Map<String, Property> generateProperties() {
+ return new ImmutableMap.Builder<String, Property>()
+ .put(USER, new Property.Builder()
+ .keyword(new KeywordProperty.Builder().build())
+ .build())
+ .put(DOMAIN, new Property.Builder()
+ .keyword(new KeywordProperty.Builder().build())
+ .build())
+ .put(QUOTA_RATIO, new Property.Builder()
+ .double_(new DoubleNumberProperty.Builder().build())
+ .build())
+ .build();
}
}
\ No newline at end of file
diff --git a/mailbox/plugin/quota-search-opensearch/src/main/java/org/apache/james/quota/search/opensearch/QuotaSearchIndexCreationUtil.java b/mailbox/plugin/quota-search-opensearch/src/main/java/org/apache/james/quota/search/opensearch/QuotaSearchIndexCreationUtil.java
index 21baee11a2..9d55f50a5a 100644
--- a/mailbox/plugin/quota-search-opensearch/src/main/java/org/apache/james/quota/search/opensearch/QuotaSearchIndexCreationUtil.java
+++ b/mailbox/plugin/quota-search-opensearch/src/main/java/org/apache/james/quota/search/opensearch/QuotaSearchIndexCreationUtil.java
@@ -19,8 +19,6 @@
package org.apache.james.quota.search.opensearch;
-import java.io.IOException;
-
import org.apache.james.backends.opensearch.AliasName;
import org.apache.james.backends.opensearch.IndexCreationFactory;
import org.apache.james.backends.opensearch.IndexName;
@@ -41,7 +39,7 @@ public class QuotaSearchIndexCreationUtil {
.createIndexAndAliases(client, QuotaRatioMappingFactory.getMappingContent());
}
- public static ReactorOpenSearchClient prepareDefaultClient(ReactorOpenSearchClient client, OpenSearchConfiguration configuration) throws IOException {
+ public static ReactorOpenSearchClient prepareDefaultClient(ReactorOpenSearchClient client, OpenSearchConfiguration configuration) {
return prepareClient(client,
QuotaRatioOpenSearchConstants.DEFAULT_QUOTA_RATIO_READ_ALIAS,
QuotaRatioOpenSearchConstants.DEFAULT_QUOTA_RATIO_WRITE_ALIAS,
diff --git a/mailbox/plugin/quota-search-opensearch/src/test/java/org/apache/james/quota/search/opensearch/OpenSearchQuotaConfigurationTest.java b/mailbox/plugin/quota-search-opensearch/src/test/java/org/apache/james/quota/search/opensearch/OpenSearchQuotaConfigurationTest.java
index 4b964f991f..a0c42922e2 100644
--- a/mailbox/plugin/quota-search-opensearch/src/test/java/org/apache/james/quota/search/opensearch/OpenSearchQuotaConfigurationTest.java
+++ b/mailbox/plugin/quota-search-opensearch/src/test/java/org/apache/james/quota/search/opensearch/OpenSearchQuotaConfigurationTest.java
@@ -36,9 +36,9 @@ class OpenSearchQuotaConfigurationTest {
configuration.addProperty("opensearch.alias.read.quota.ratio.name", name);
configuration.addProperty("opensearch.hosts", "127.0.0.1");
- OpenSearchQuotaConfiguration elasticSearchConfiguration = OpenSearchQuotaConfiguration.fromProperties(configuration);
+ OpenSearchQuotaConfiguration openSearchQuotaConfiguration = OpenSearchQuotaConfiguration.fromProperties(configuration);
- assertThat(elasticSearchConfiguration.getReadAliasQuotaRatioName())
+ assertThat(openSearchQuotaConfiguration.getReadAliasQuotaRatioName())
.isEqualTo(new ReadAliasName(name));
}
@@ -47,9 +47,9 @@ class OpenSearchQuotaConfigurationTest {
PropertiesConfiguration configuration = new PropertiesConfiguration();
configuration.addProperty("opensearch.hosts", "127.0.0.1");
- OpenSearchQuotaConfiguration elasticSearchConfiguration = OpenSearchQuotaConfiguration.fromProperties(configuration);
+ OpenSearchQuotaConfiguration openSearchConfiguration = OpenSearchQuotaConfiguration.fromProperties(configuration);
- assertThat(elasticSearchConfiguration.getReadAliasQuotaRatioName())
+ assertThat(openSearchConfiguration.getReadAliasQuotaRatioName())
.isEqualTo(QuotaRatioOpenSearchConstants.DEFAULT_QUOTA_RATIO_READ_ALIAS);
}
@@ -60,9 +60,9 @@ class OpenSearchQuotaConfigurationTest {
configuration.addProperty("opensearch.alias.write.quota.ratio.name", name);
configuration.addProperty("opensearch.hosts", "127.0.0.1");
- OpenSearchQuotaConfiguration elasticSearchConfiguration = OpenSearchQuotaConfiguration.fromProperties(configuration);
+ OpenSearchQuotaConfiguration openSearchConfiguration = OpenSearchQuotaConfiguration.fromProperties(configuration);
- assertThat(elasticSearchConfiguration.getWriteAliasQuotaRatioName())
+ assertThat(openSearchConfiguration.getWriteAliasQuotaRatioName())
.isEqualTo(new WriteAliasName(name));
}
@@ -71,9 +71,9 @@ class OpenSearchQuotaConfigurationTest {
PropertiesConfiguration configuration = new PropertiesConfiguration();
configuration.addProperty("opensearch.hosts", "127.0.0.1");
- OpenSearchQuotaConfiguration elasticSearchConfiguration = OpenSearchQuotaConfiguration.fromProperties(configuration);
+ OpenSearchQuotaConfiguration openSearchConfiguration = OpenSearchQuotaConfiguration.fromProperties(configuration);
- assertThat(elasticSearchConfiguration.getWriteAliasQuotaRatioName())
+ assertThat(openSearchConfiguration.getWriteAliasQuotaRatioName())
.isEqualTo(QuotaRatioOpenSearchConstants.DEFAULT_QUOTA_RATIO_WRITE_ALIAS);
}
@@ -84,9 +84,9 @@ class OpenSearchQuotaConfigurationTest {
configuration.addProperty("opensearch.index.quota.ratio.name", name);
configuration.addProperty("opensearch.hosts", "127.0.0.1");
- OpenSearchQuotaConfiguration elasticSearchConfiguration = OpenSearchQuotaConfiguration.fromProperties(configuration);
+ OpenSearchQuotaConfiguration openSearchConfiguration = OpenSearchQuotaConfiguration.fromProperties(configuration);
- assertThat(elasticSearchConfiguration.getIndexQuotaRatioName())
+ assertThat(openSearchConfiguration.getIndexQuotaRatioName())
.isEqualTo(new IndexName(name));
}
@@ -95,9 +95,9 @@ class OpenSearchQuotaConfigurationTest {
PropertiesConfiguration configuration = new PropertiesConfiguration();
configuration.addProperty("opensearch.hosts", "127.0.0.1");
- OpenSearchQuotaConfiguration elasticSearchConfiguration = OpenSearchQuotaConfiguration.fromProperties(configuration);
+ OpenSearchQuotaConfiguration openSearchConfiguration = OpenSearchQuotaConfiguration.fromProperties(configuration);
- assertThat(elasticSearchConfiguration.getIndexQuotaRatioName())
+ assertThat(openSearchConfiguration.getIndexQuotaRatioName())
.isEqualTo(QuotaRatioOpenSearchConstants.DEFAULT_QUOTA_RATIO_INDEX);
}
}
\ No newline at end of file
diff --git a/mailbox/plugin/quota-search-opensearch/src/test/java/org/apache/james/quota/search/opensearch/QuotaQueryConverterTest.java b/mailbox/plugin/quota-search-opensearch/src/test/java/org/apache/james/quota/search/opensearch/QuotaQueryConverterTest.java
index eef42098eb..c3fdee3eb1 100644
--- a/mailbox/plugin/quota-search-opensearch/src/test/java/org/apache/james/quota/search/opensearch/QuotaQueryConverterTest.java
+++ b/mailbox/plugin/quota-search-opensearch/src/test/java/org/apache/james/quota/search/opensearch/QuotaQueryConverterTest.java
@@ -18,19 +18,22 @@
****************************************************************/
package org.apache.james.quota.search.opensearch;
+import static org.apache.james.quota.search.opensearch.json.JsonMessageConstants.DOMAIN;
import static org.apache.james.quota.search.opensearch.json.JsonMessageConstants.QUOTA_RATIO;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.opensearch.index.query.QueryBuilders.matchAllQuery;
-import static org.opensearch.index.query.QueryBuilders.rangeQuery;
-import static org.opensearch.index.query.QueryBuilders.termQuery;
import org.apache.james.core.Domain;
import org.apache.james.quota.search.QuotaBoundary;
import org.apache.james.quota.search.QuotaQuery;
-import org.apache.james.quota.search.opensearch.json.JsonMessageConstants;
+import org.assertj.core.api.SoftAssertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-import org.opensearch.index.query.QueryBuilder;
+import org.opensearch.client.json.JsonData;
+import org.opensearch.client.opensearch._types.FieldValue;
+import org.opensearch.client.opensearch._types.query_dsl.MatchAllQuery;
+import org.opensearch.client.opensearch._types.query_dsl.Query;
+import org.opensearch.client.opensearch._types.query_dsl.RangeQuery;
+import org.opensearch.client.opensearch._types.query_dsl.TermQuery;
class QuotaQueryConverterTest {
QuotaQueryConverter testee;
@@ -43,41 +46,65 @@ class QuotaQueryConverterTest {
@Test
void fromShouldReturnMatchAllWhenEmptyClauses() {
QuotaQuery query = QuotaQuery.builder().build();
- QueryBuilder expected = matchAllQuery();
+ Query expected = new MatchAllQuery.Builder().build()._toQuery();
- QueryBuilder actual = testee.from(query);
+ Query actual = testee.from(query);
- assertThat(actual).isEqualToComparingFieldByField(expected);
+ SoftAssertions.assertSoftly(softly -> assertThat(actual._kind()).isEqualTo(expected._kind()));
}
@Test
void fromShouldReturnDomainMatchWhenOnlyDomain() {
QuotaQuery query = QuotaQuery.builder().hasDomain(Domain.of("my.tld")).build();
- QueryBuilder expected = termQuery(JsonMessageConstants.DOMAIN, "my.tld");
-
- QueryBuilder actual = testee.from(query);
-
- assertThat(actual).isEqualToComparingFieldByField(expected);
+ Query expected = new TermQuery.Builder()
+ .field(DOMAIN)
+ .value(new FieldValue.Builder().stringValue("my.tld").build())
+ .build()
+ ._toQuery();
+
+ Query actual = testee.from(query);
+
+ SoftAssertions.assertSoftly(softly -> {
+ assertThat(actual._kind()).isEqualTo(expected._kind());
+ assertThat(actual.term().field()).isEqualTo(expected.term().field());
+ assertThat(actual.term().value().stringValue()).isEqualTo(expected.term().value().stringValue());
+ });
}
@Test
void fromShouldReturnQuotaRatioMatchWhenLessThan() {
QuotaQuery query = QuotaQuery.builder().lessThan(new QuotaBoundary(0.1)).build();
- QueryBuilder expected = rangeQuery(QUOTA_RATIO).lte(0.1);
-
- QueryBuilder actual = testee.from(query);
-
- assertThat(actual).isEqualToComparingFieldByField(expected);
+ Query expected = new RangeQuery.Builder()
+ .field(QUOTA_RATIO)
+ .lte(JsonData.of(0.1))
+ .build()
+ ._toQuery();
+
+ Query actual = testee.from(query);
+
+ SoftAssertions.assertSoftly(softly -> {
+ assertThat(actual._kind()).isEqualTo(expected._kind());
+ assertThat(actual.range().field()).isEqualTo(expected.range().field());
+ assertThat(actual.range().lte().to(Double.class)).isEqualTo(expected.range().lte().to(Double.class));
+ });
}
@Test
void fromShouldReturnQuotaRatioMatchWhenMoreThan() {
QuotaQuery query = QuotaQuery.builder().moreThan(new QuotaBoundary(0.1)).build();
- QueryBuilder expected = rangeQuery(QUOTA_RATIO).gte(0.1);
-
- QueryBuilder actual = testee.from(query);
-
- assertThat(actual).isEqualToComparingFieldByField(expected);
+ Query expected = new RangeQuery.Builder()
+ .field(QUOTA_RATIO)
+ .gte(JsonData.of(0.1))
+ .build()
+ ._toQuery();
+
+ Query actual = testee.from(query);
+
+ SoftAssertions.assertSoftly(softly -> {
+ assertThat(actual._kind()).isEqualTo(expected._kind());
+ assertThat(actual.range().field()).isEqualTo(expected.range().field());
+ assertThat(actual.range().gte().to(Double.class)).isEqualTo(expected.range().gte().to(Double.class));
+ });
}
}
diff --git a/mailbox/plugin/quota-search-opensearch/src/test/java/org/apache/james/quota/search/opensearch/events/OpenSearchQuotaMailboxListenerTest.java b/mailbox/plugin/quota-search-opensearch/src/test/java/org/apache/james/quota/search/opensearch/events/OpenSearchQuotaMailboxListenerTest.java
index 3acf31f804..cf31d3da62 100644
--- a/mailbox/plugin/quota-search-opensearch/src/test/java/org/apache/james/quota/search/opensearch/events/OpenSearchQuotaMailboxListenerTest.java
+++ b/mailbox/plugin/quota-search-opensearch/src/test/java/org/apache/james/quota/search/opensearch/events/OpenSearchQuotaMailboxListenerTest.java
@@ -53,10 +53,8 @@ import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
-import org.opensearch.action.search.SearchRequest;
-import org.opensearch.client.RequestOptions;
-import org.opensearch.index.query.QueryBuilders;
-import org.opensearch.search.builder.SearchSourceBuilder;
+import org.opensearch.client.opensearch._types.query_dsl.MatchAllQuery;
+import org.opensearch.client.opensearch.core.SearchRequest;
class OpenSearchQuotaMailboxListenerTest {
private static final ConditionFactory CALMLY_AWAIT = Awaitility
@@ -117,10 +115,11 @@ class OpenSearchQuotaMailboxListenerTest {
CALMLY_AWAIT.atMost(Durations.TEN_SECONDS)
.untilAsserted(() -> assertThat(client.search(
- new SearchRequest(QuotaRatioOpenSearchConstants.DEFAULT_QUOTA_RATIO_READ_ALIAS.getValue())
- .source(new SearchSourceBuilder().query(QueryBuilders.matchAllQuery())),
- RequestOptions.DEFAULT)
+ new SearchRequest.Builder()
+ .index(QuotaRatioOpenSearchConstants.DEFAULT_QUOTA_RATIO_READ_ALIAS.getValue())
+ .query(new MatchAllQuery.Builder().build()._toQuery())
+ .build())
.block()
- .getHits().getTotalHits().value).isEqualTo(1));
+ .hits().total().value()).isEqualTo(1));
}
}
\ No newline at end of file
diff --git a/mailbox/plugin/quota-search-opensearch/src/test/java/org/apache/james/quota/search/opensearch/json/QuotaRatioToOpenSearchJsonTest.java b/mailbox/plugin/quota-search-opensearch/src/test/java/org/apache/james/quota/search/opensearch/json/QuotaRatioToOpenSearchJsonTest.java
index cf731a6f65..c84335bc6c 100644
--- a/mailbox/plugin/quota-search-opensearch/src/test/java/org/apache/james/quota/search/opensearch/json/QuotaRatioToOpenSearchJsonTest.java
+++ b/mailbox/plugin/quota-search-opensearch/src/test/java/org/apache/james/quota/search/opensearch/json/QuotaRatioToOpenSearchJsonTest.java
@@ -41,7 +41,8 @@ class QuotaRatioToOpenSearchJsonTest {
@Test
void quotaRatioShouldBeWellConvertedToJson() throws IOException {
- String user = "user@domain.org"; DefaultUserQuotaRootResolver quotaRootResolver = new DefaultUserQuotaRootResolver(mock(SessionProvider.class), mock(MailboxSessionMapperFactory.class));
+ String user = "user@domain.org";
+ DefaultUserQuotaRootResolver quotaRootResolver = new DefaultUserQuotaRootResolver(mock(SessionProvider.class), mock(MailboxSessionMapperFactory.class));
Username username = Username.of(user);
QuotaUsageUpdatedEvent event = EventFactory.quotaUpdated()
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 02/12: JAMES-3771 Migrate mailbox opensearch to new java client
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 092c357b18b325c457f4d1529208cb013c4055a7
Author: Rene Cordier <rc...@linagora.com>
AuthorDate: Tue Jun 21 16:19:16 2022 +0700
JAMES-3771 Migrate mailbox opensearch to new java client
---
.../opensearch/DockerOpenSearchExtension.java | 7 +-
.../mailbox/opensearch/MailboxMappingFactory.java | 518 ++++++++++-----------
.../OpenSearchListeningMessageSearchIndex.java | 61 +--
.../opensearch/query/CriterionConverter.java | 384 +++++++++++----
.../mailbox/opensearch/query/QueryConverter.java | 37 +-
.../mailbox/opensearch/query/SortConverter.java | 42 +-
.../opensearch/search/OpenSearchSearcher.java | 42 +-
.../opensearch/OpenSearchIntegrationTest.java | 46 +-
.../OpenSearchListeningMessageSearchIndexTest.java | 64 +--
.../opensearch/search/OpenSearchSearcherTest.java | 21 +-
10 files changed, 690 insertions(+), 532 deletions(-)
diff --git a/backends-common/opensearch/src/test/java/org/apache/james/backends/opensearch/DockerOpenSearchExtension.java b/backends-common/opensearch/src/test/java/org/apache/james/backends/opensearch/DockerOpenSearchExtension.java
index 021a60d790..aee8790003 100644
--- a/backends-common/opensearch/src/test/java/org/apache/james/backends/opensearch/DockerOpenSearchExtension.java
+++ b/backends-common/opensearch/src/test/java/org/apache/james/backends/opensearch/DockerOpenSearchExtension.java
@@ -35,13 +35,13 @@ public class DockerOpenSearchExtension implements AfterEachCallback, BeforeEachC
interface CleanupStrategy {
CleanupStrategy NONE = any -> {};
- void clean(DockerOpenSearch elasticSearch);
+ void clean(DockerOpenSearch openSearch);
}
public static class DefaultCleanupStrategy implements CleanupStrategy {
@Override
- public void clean(DockerOpenSearch elasticSearch) {
- elasticSearch.cleanUpData();
+ public void clean(DockerOpenSearch openSearch) {
+ openSearch.cleanUpData();
}
}
@@ -55,6 +55,7 @@ public class DockerOpenSearchExtension implements AfterEachCallback, BeforeEachC
@Override
public void clean(DockerOpenSearch openSearch) {
Awaitility.await()
+ .ignoreExceptions()
.until(() -> {
openSearch.flushIndices();
ReactorOpenSearchClient client = openSearch.clientProvider().get();
diff --git a/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/MailboxMappingFactory.java b/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/MailboxMappingFactory.java
index d9813d738c..2964a8738f 100644
--- a/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/MailboxMappingFactory.java
+++ b/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/MailboxMappingFactory.java
@@ -19,298 +19,256 @@
package org.apache.james.mailbox.opensearch;
-import static org.apache.james.backends.opensearch.IndexCreationFactory.ANALYZER;
-import static org.apache.james.backends.opensearch.IndexCreationFactory.BOOLEAN;
import static org.apache.james.backends.opensearch.IndexCreationFactory.CASE_INSENSITIVE;
-import static org.apache.james.backends.opensearch.IndexCreationFactory.FIELDS;
-import static org.apache.james.backends.opensearch.IndexCreationFactory.FORMAT;
import static org.apache.james.backends.opensearch.IndexCreationFactory.KEEP_MAIL_AND_URL;
import static org.apache.james.backends.opensearch.IndexCreationFactory.KEYWORD;
-import static org.apache.james.backends.opensearch.IndexCreationFactory.LONG;
-import static org.apache.james.backends.opensearch.IndexCreationFactory.NESTED;
-import static org.apache.james.backends.opensearch.IndexCreationFactory.NORMALIZER;
-import static org.apache.james.backends.opensearch.IndexCreationFactory.PROPERTIES;
import static org.apache.james.backends.opensearch.IndexCreationFactory.RAW;
-import static org.apache.james.backends.opensearch.IndexCreationFactory.REQUIRED;
-import static org.apache.james.backends.opensearch.IndexCreationFactory.ROUTING;
-import static org.apache.james.backends.opensearch.IndexCreationFactory.SEARCH_ANALYZER;
-import static org.apache.james.backends.opensearch.IndexCreationFactory.TYPE;
-import static org.opensearch.common.xcontent.XContentFactory.jsonBuilder;
-import java.io.IOException;
+import java.util.Map;
-import org.apache.james.backends.opensearch.IndexCreationFactory;
import org.apache.james.mailbox.opensearch.json.JsonMessageConstants;
-import org.opensearch.common.xcontent.XContentBuilder;
+import org.opensearch.client.opensearch._types.mapping.BooleanProperty;
+import org.opensearch.client.opensearch._types.mapping.DateProperty;
+import org.opensearch.client.opensearch._types.mapping.DynamicMapping;
+import org.opensearch.client.opensearch._types.mapping.KeywordProperty;
+import org.opensearch.client.opensearch._types.mapping.LongNumberProperty;
+import org.opensearch.client.opensearch._types.mapping.NestedProperty;
+import org.opensearch.client.opensearch._types.mapping.ObjectProperty;
+import org.opensearch.client.opensearch._types.mapping.Property;
+import org.opensearch.client.opensearch._types.mapping.RoutingField;
+import org.opensearch.client.opensearch._types.mapping.TextProperty;
+import org.opensearch.client.opensearch._types.mapping.TypeMapping;
+
+import com.google.common.collect.ImmutableMap;
public class MailboxMappingFactory {
private static final String STANDARD = "standard";
private static final String SIMPLE = "simple";
- private static final String STORE = "store";
- public static XContentBuilder getMappingContent() {
- try {
- return jsonBuilder()
- .startObject()
-
- .field("dynamic", "strict")
-
- .startObject(ROUTING)
- .field(REQUIRED, true)
- .endObject()
-
- .startObject(PROPERTIES)
-
- .startObject(JsonMessageConstants.MESSAGE_ID)
- .field(TYPE, KEYWORD)
- .field(STORE, true)
- .endObject()
-
- .startObject(JsonMessageConstants.THREAD_ID)
- .field(TYPE, KEYWORD)
- .endObject()
-
- .startObject(JsonMessageConstants.UID)
- .field(TYPE, LONG)
- .field(STORE, true)
- .endObject()
-
- .startObject(JsonMessageConstants.MODSEQ)
- .field(TYPE, LONG)
- .endObject()
-
- .startObject(JsonMessageConstants.SIZE)
- .field(TYPE, LONG)
- .endObject()
-
- .startObject(JsonMessageConstants.IS_ANSWERED)
- .field(TYPE, BOOLEAN)
- .endObject()
-
- .startObject(JsonMessageConstants.IS_DELETED)
- .field(TYPE, BOOLEAN)
- .endObject()
-
- .startObject(JsonMessageConstants.IS_DRAFT)
- .field(TYPE, BOOLEAN)
- .endObject()
-
- .startObject(JsonMessageConstants.IS_FLAGGED)
- .field(TYPE, BOOLEAN)
- .endObject()
-
- .startObject(JsonMessageConstants.IS_RECENT)
- .field(TYPE, BOOLEAN)
- .endObject()
-
- .startObject(JsonMessageConstants.IS_UNREAD)
- .field(TYPE, BOOLEAN)
- .endObject()
-
- .startObject(JsonMessageConstants.DATE)
- .field(TYPE, IndexCreationFactory.DATE)
- .field(FORMAT, "uuuu-MM-dd'T'HH:mm:ssX||uuuu-MM-dd'T'HH:mm:ssXXX||uuuu-MM-dd'T'HH:mm:ssXXXXX")
- .endObject()
-
- .startObject(JsonMessageConstants.SENT_DATE)
- .field(TYPE, IndexCreationFactory.DATE)
- .field(FORMAT, "uuuu-MM-dd'T'HH:mm:ssX||uuuu-MM-dd'T'HH:mm:ssXXX||uuuu-MM-dd'T'HH:mm:ssXXXXX")
- .endObject()
-
- .startObject(JsonMessageConstants.SAVE_DATE)
- .field(TYPE, IndexCreationFactory.DATE)
- .field(FORMAT, "uuuu-MM-dd'T'HH:mm:ssX||uuuu-MM-dd'T'HH:mm:ssXXX||uuuu-MM-dd'T'HH:mm:ssXXXXX")
- .endObject()
-
- .startObject(JsonMessageConstants.USER_FLAGS)
- .field(TYPE, KEYWORD)
- .field(NORMALIZER, CASE_INSENSITIVE)
- .endObject()
-
- .startObject(JsonMessageConstants.MEDIA_TYPE)
- .field(TYPE, KEYWORD)
- .endObject()
-
- .startObject(JsonMessageConstants.SUBTYPE)
- .field(TYPE, KEYWORD)
- .endObject()
-
- .startObject(JsonMessageConstants.FROM)
- .startObject(PROPERTIES)
- .startObject(JsonMessageConstants.EMailer.NAME)
- .field(TYPE, JsonMessageConstants.TEXT)
- .field(ANALYZER, KEEP_MAIL_AND_URL)
- .endObject()
- .startObject(JsonMessageConstants.EMailer.DOMAIN)
- .field(TYPE, JsonMessageConstants.TEXT)
- .field(ANALYZER, SIMPLE)
- .field(SEARCH_ANALYZER, KEYWORD)
- .endObject()
- .startObject(JsonMessageConstants.EMailer.ADDRESS)
- .field(TYPE, JsonMessageConstants.TEXT)
- .field(ANALYZER, STANDARD)
- .field(SEARCH_ANALYZER, KEEP_MAIL_AND_URL)
- .startObject(FIELDS)
- .startObject(RAW)
- .field(TYPE, KEYWORD)
- .field(NORMALIZER, CASE_INSENSITIVE)
- .endObject()
- .endObject()
- .endObject()
- .endObject()
- .endObject()
-
- .startObject(JsonMessageConstants.HEADERS)
- .field(TYPE, NESTED)
- .startObject(PROPERTIES)
- .startObject(JsonMessageConstants.HEADER.NAME)
- .field(TYPE, KEYWORD)
- .endObject()
- .startObject(JsonMessageConstants.HEADER.VALUE)
- .field(TYPE, JsonMessageConstants.TEXT)
- .field(ANALYZER, KEEP_MAIL_AND_URL)
- .endObject()
- .endObject()
- .endObject()
-
- .startObject(JsonMessageConstants.SUBJECT)
- .field(TYPE, JsonMessageConstants.TEXT)
- .field(ANALYZER, KEEP_MAIL_AND_URL)
- .startObject(FIELDS)
- .startObject(RAW)
- .field(TYPE, KEYWORD)
- .field(NORMALIZER, CASE_INSENSITIVE)
- .endObject()
- .endObject()
- .endObject()
-
- .startObject(JsonMessageConstants.TO)
- .startObject(PROPERTIES)
- .startObject(JsonMessageConstants.EMailer.NAME)
- .field(TYPE, JsonMessageConstants.TEXT)
- .field(ANALYZER, KEEP_MAIL_AND_URL)
- .endObject()
- .startObject(JsonMessageConstants.EMailer.DOMAIN)
- .field(TYPE, JsonMessageConstants.TEXT)
- .field(ANALYZER, SIMPLE)
- .field(SEARCH_ANALYZER, KEYWORD)
- .endObject()
- .startObject(JsonMessageConstants.EMailer.ADDRESS)
- .field(TYPE, JsonMessageConstants.TEXT)
- .field(ANALYZER, STANDARD)
- .field(SEARCH_ANALYZER, KEEP_MAIL_AND_URL)
- .startObject(FIELDS)
- .startObject(RAW)
- .field(TYPE, KEYWORD)
- .field(NORMALIZER, CASE_INSENSITIVE)
- .endObject()
- .endObject()
- .endObject()
- .endObject()
- .endObject()
-
- .startObject(JsonMessageConstants.CC)
- .startObject(PROPERTIES)
- .startObject(JsonMessageConstants.EMailer.NAME)
- .field(TYPE, JsonMessageConstants.TEXT)
- .field(ANALYZER, KEEP_MAIL_AND_URL)
- .endObject()
- .startObject(JsonMessageConstants.EMailer.DOMAIN)
- .field(TYPE, JsonMessageConstants.TEXT)
- .field(ANALYZER, SIMPLE)
- .field(SEARCH_ANALYZER, KEYWORD)
- .endObject()
- .startObject(JsonMessageConstants.EMailer.ADDRESS)
- .field(TYPE, JsonMessageConstants.TEXT)
- .field(ANALYZER, STANDARD)
- .field(SEARCH_ANALYZER, KEEP_MAIL_AND_URL)
- .startObject(FIELDS)
- .startObject(RAW)
- .field(TYPE, KEYWORD)
- .field(NORMALIZER, CASE_INSENSITIVE)
- .endObject()
- .endObject()
- .endObject()
- .endObject()
- .endObject()
-
- .startObject(JsonMessageConstants.BCC)
- .startObject(PROPERTIES)
- .startObject(JsonMessageConstants.EMailer.NAME)
- .field(TYPE, JsonMessageConstants.TEXT)
- .field(ANALYZER, KEEP_MAIL_AND_URL)
- .endObject()
- .startObject(JsonMessageConstants.EMailer.DOMAIN)
- .field(TYPE, JsonMessageConstants.TEXT)
- .field(ANALYZER, SIMPLE)
- .field(SEARCH_ANALYZER, KEYWORD)
- .endObject()
- .startObject(JsonMessageConstants.EMailer.ADDRESS)
- .field(TYPE, JsonMessageConstants.TEXT)
- .field(ANALYZER, STANDARD)
- .field(SEARCH_ANALYZER, KEEP_MAIL_AND_URL)
- .startObject(FIELDS)
- .startObject(RAW)
- .field(TYPE, KEYWORD)
- .field(NORMALIZER, CASE_INSENSITIVE)
- .endObject()
- .endObject()
- .endObject()
- .endObject()
- .endObject()
-
- .startObject(JsonMessageConstants.MAILBOX_ID)
- .field(TYPE, KEYWORD)
- .field(STORE, true)
- .endObject()
-
- .startObject(JsonMessageConstants.MIME_MESSAGE_ID)
- .field(TYPE, KEYWORD)
- .endObject()
-
- .startObject(JsonMessageConstants.TEXT_BODY)
- .field(TYPE, JsonMessageConstants.TEXT)
- .field(ANALYZER, STANDARD)
- .endObject()
-
- .startObject(JsonMessageConstants.HTML_BODY)
- .field(TYPE, JsonMessageConstants.TEXT)
- .field(ANALYZER, STANDARD)
- .endObject()
-
- .startObject(JsonMessageConstants.HAS_ATTACHMENT)
- .field(TYPE, BOOLEAN)
- .endObject()
-
- .startObject(JsonMessageConstants.ATTACHMENTS)
- .startObject(PROPERTIES)
- .startObject(JsonMessageConstants.Attachment.FILENAME)
- .field(TYPE, JsonMessageConstants.TEXT)
- .field(ANALYZER, STANDARD)
- .endObject()
- .startObject(JsonMessageConstants.Attachment.TEXT_CONTENT)
- .field(TYPE, JsonMessageConstants.TEXT)
- .field(ANALYZER, STANDARD)
- .endObject()
- .startObject(JsonMessageConstants.Attachment.MEDIA_TYPE)
- .field(TYPE, KEYWORD)
- .endObject()
- .startObject(JsonMessageConstants.Attachment.SUBTYPE)
- .field(TYPE, KEYWORD)
- .endObject()
- .startObject(JsonMessageConstants.Attachment.FILE_EXTENSION)
- .field(TYPE, KEYWORD)
- .endObject()
- .startObject(JsonMessageConstants.Attachment.CONTENT_DISPOSITION)
- .field(TYPE, KEYWORD)
- .endObject()
- .endObject()
- .endObject()
+ public static TypeMapping getMappingContent() {
+ return new TypeMapping.Builder()
+ .dynamic(DynamicMapping.Strict)
+ .routing(new RoutingField.Builder()
+ .required(true)
+ .build())
+ .properties(generateProperties())
+ .build();
+ }
- .endObject()
- .endObject();
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
+ private static Map<String, Property> generateProperties() {
+ return new ImmutableMap.Builder<String, Property>()
+ .put(JsonMessageConstants.MESSAGE_ID, new Property.Builder()
+ .keyword(new KeywordProperty.Builder().store(true).build())
+ .build())
+ .put(JsonMessageConstants.THREAD_ID, new Property.Builder()
+ .keyword(new KeywordProperty.Builder().build())
+ .build())
+ .put(JsonMessageConstants.UID, new Property.Builder()
+ .long_(new LongNumberProperty.Builder().store(true).build())
+ .build())
+ .put(JsonMessageConstants.MODSEQ, new Property.Builder()
+ .long_(new LongNumberProperty.Builder().build())
+ .build())
+ .put(JsonMessageConstants.SIZE, new Property.Builder()
+ .long_(new LongNumberProperty.Builder().build())
+ .build())
+ .put(JsonMessageConstants.IS_ANSWERED, new Property.Builder()
+ .boolean_(new BooleanProperty.Builder().build())
+ .build())
+ .put(JsonMessageConstants.IS_DELETED, new Property.Builder()
+ .boolean_(new BooleanProperty.Builder().build())
+ .build())
+ .put(JsonMessageConstants.IS_DRAFT, new Property.Builder()
+ .boolean_(new BooleanProperty.Builder().build())
+ .build())
+ .put(JsonMessageConstants.IS_FLAGGED, new Property.Builder()
+ .boolean_(new BooleanProperty.Builder().build())
+ .build())
+ .put(JsonMessageConstants.IS_RECENT, new Property.Builder()
+ .boolean_(new BooleanProperty.Builder().build())
+ .build())
+ .put(JsonMessageConstants.IS_UNREAD, new Property.Builder()
+ .boolean_(new BooleanProperty.Builder().build())
+ .build())
+ .put(JsonMessageConstants.DATE, new Property.Builder()
+ .date(new DateProperty.Builder()
+ .format("uuuu-MM-dd'T'HH:mm:ssX||uuuu-MM-dd'T'HH:mm:ssXXX||uuuu-MM-dd'T'HH:mm:ssXXXXX")
+ .build())
+ .build())
+ .put(JsonMessageConstants.SENT_DATE, new Property.Builder()
+ .date(new DateProperty.Builder()
+ .format("uuuu-MM-dd'T'HH:mm:ssX||uuuu-MM-dd'T'HH:mm:ssXXX||uuuu-MM-dd'T'HH:mm:ssXXXXX")
+ .build())
+ .build())
+ .put(JsonMessageConstants.SAVE_DATE, new Property.Builder()
+ .date(new DateProperty.Builder()
+ .format("uuuu-MM-dd'T'HH:mm:ssX||uuuu-MM-dd'T'HH:mm:ssXXX||uuuu-MM-dd'T'HH:mm:ssXXXXX")
+ .build())
+ .build())
+ .put(JsonMessageConstants.USER_FLAGS, new Property.Builder()
+ .keyword(new KeywordProperty.Builder().normalizer(CASE_INSENSITIVE).build())
+ .build())
+ .put(JsonMessageConstants.MEDIA_TYPE, new Property.Builder()
+ .keyword(new KeywordProperty.Builder().build())
+ .build())
+ .put(JsonMessageConstants.SUBTYPE, new Property.Builder()
+ .keyword(new KeywordProperty.Builder().build())
+ .build())
+ .put(JsonMessageConstants.FROM, new Property.Builder()
+ .object(new ObjectProperty.Builder()
+ .properties(ImmutableMap.of(
+ JsonMessageConstants.EMailer.NAME, new Property.Builder()
+ .text(new TextProperty.Builder().analyzer(KEEP_MAIL_AND_URL).build())
+ .build(),
+ JsonMessageConstants.EMailer.DOMAIN, new Property.Builder()
+ .text(new TextProperty.Builder()
+ .analyzer(SIMPLE)
+ .searchAnalyzer(KEYWORD)
+ .build())
+ .build(),
+ JsonMessageConstants.EMailer.ADDRESS, new Property.Builder()
+ .text(new TextProperty.Builder()
+ .analyzer(STANDARD)
+ .searchAnalyzer(KEEP_MAIL_AND_URL)
+ .fields(RAW, new Property.Builder()
+ .keyword(new KeywordProperty.Builder().normalizer(CASE_INSENSITIVE).build())
+ .build())
+ .build())
+ .build()
+ ))
+ .build())
+ .build())
+ .put(JsonMessageConstants.HEADERS, new Property.Builder()
+ .nested(new NestedProperty.Builder()
+ .properties(ImmutableMap.of(
+ JsonMessageConstants.HEADER.NAME, new Property.Builder()
+ .keyword(new KeywordProperty.Builder().build())
+ .build(),
+ JsonMessageConstants.HEADER.VALUE, new Property.Builder()
+ .text(new TextProperty.Builder().analyzer(KEEP_MAIL_AND_URL).build())
+ .build()
+ ))
+ .build())
+ .build())
+ .put(JsonMessageConstants.SUBJECT, new Property.Builder()
+ .text(new TextProperty.Builder()
+ .analyzer(KEEP_MAIL_AND_URL)
+ .fields(RAW, new Property.Builder()
+ .keyword(new KeywordProperty.Builder().normalizer(CASE_INSENSITIVE).build())
+ .build())
+ .build())
+ .build())
+ .put(JsonMessageConstants.TO, new Property.Builder()
+ .object(new ObjectProperty.Builder()
+ .properties(ImmutableMap.of(
+ JsonMessageConstants.EMailer.NAME, new Property.Builder()
+ .text(new TextProperty.Builder().analyzer(KEEP_MAIL_AND_URL).build())
+ .build(),
+ JsonMessageConstants.EMailer.DOMAIN, new Property.Builder()
+ .text(new TextProperty.Builder()
+ .analyzer(SIMPLE)
+ .searchAnalyzer(KEYWORD)
+ .build())
+ .build(),
+ JsonMessageConstants.EMailer.ADDRESS, new Property.Builder()
+ .text(new TextProperty.Builder()
+ .analyzer(STANDARD)
+ .searchAnalyzer(KEEP_MAIL_AND_URL)
+ .fields(RAW, new Property.Builder()
+ .keyword(new KeywordProperty.Builder().normalizer(CASE_INSENSITIVE).build())
+ .build())
+ .build())
+ .build()
+ ))
+ .build())
+ .build())
+ .put(JsonMessageConstants.CC, new Property.Builder()
+ .object(new ObjectProperty.Builder()
+ .properties(ImmutableMap.of(
+ JsonMessageConstants.EMailer.NAME, new Property.Builder()
+ .text(new TextProperty.Builder().analyzer(KEEP_MAIL_AND_URL).build())
+ .build(),
+ JsonMessageConstants.EMailer.DOMAIN, new Property.Builder()
+ .text(new TextProperty.Builder()
+ .analyzer(SIMPLE)
+ .searchAnalyzer(KEYWORD)
+ .build())
+ .build(),
+ JsonMessageConstants.EMailer.ADDRESS, new Property.Builder()
+ .text(new TextProperty.Builder()
+ .analyzer(STANDARD)
+ .searchAnalyzer(KEEP_MAIL_AND_URL)
+ .fields(RAW, new Property.Builder()
+ .keyword(new KeywordProperty.Builder().normalizer(CASE_INSENSITIVE).build())
+ .build())
+ .build())
+ .build()
+ ))
+ .build())
+ .build())
+ .put(JsonMessageConstants.BCC, new Property.Builder()
+ .object(new ObjectProperty.Builder()
+ .properties(ImmutableMap.of(
+ JsonMessageConstants.EMailer.NAME, new Property.Builder()
+ .text(new TextProperty.Builder().analyzer(KEEP_MAIL_AND_URL).build())
+ .build(),
+ JsonMessageConstants.EMailer.DOMAIN, new Property.Builder()
+ .text(new TextProperty.Builder()
+ .analyzer(SIMPLE)
+ .searchAnalyzer(KEYWORD)
+ .build())
+ .build(),
+ JsonMessageConstants.EMailer.ADDRESS, new Property.Builder()
+ .text(new TextProperty.Builder()
+ .analyzer(STANDARD)
+ .searchAnalyzer(KEEP_MAIL_AND_URL)
+ .fields(RAW, new Property.Builder()
+ .keyword(new KeywordProperty.Builder().normalizer(CASE_INSENSITIVE).build())
+ .build())
+ .build())
+ .build()
+ ))
+ .build())
+ .build())
+ .put(JsonMessageConstants.MAILBOX_ID, new Property.Builder()
+ .keyword(new KeywordProperty.Builder().store(true).build())
+ .build())
+ .put(JsonMessageConstants.MIME_MESSAGE_ID, new Property.Builder()
+ .keyword(new KeywordProperty.Builder().build())
+ .build())
+ .put(JsonMessageConstants.TEXT_BODY, new Property.Builder()
+ .text(new TextProperty.Builder().analyzer(STANDARD).build())
+ .build())
+ .put(JsonMessageConstants.HTML_BODY, new Property.Builder()
+ .text(new TextProperty.Builder().analyzer(STANDARD).build())
+ .build())
+ .put(JsonMessageConstants.HAS_ATTACHMENT, new Property.Builder()
+ .boolean_(new BooleanProperty.Builder().build())
+ .build())
+ .put(JsonMessageConstants.ATTACHMENTS, new Property.Builder()
+ .object(new ObjectProperty.Builder()
+ .properties(ImmutableMap.of(
+ JsonMessageConstants.Attachment.FILENAME, new Property.Builder()
+ .text(new TextProperty.Builder().analyzer(STANDARD).build())
+ .build(),
+ JsonMessageConstants.Attachment.TEXT_CONTENT, new Property.Builder()
+ .text(new TextProperty.Builder().analyzer(STANDARD).build())
+ .build(),
+ JsonMessageConstants.Attachment.MEDIA_TYPE, new Property.Builder()
+ .keyword(new KeywordProperty.Builder().build())
+ .build(),
+ JsonMessageConstants.Attachment.SUBTYPE, new Property.Builder()
+ .keyword(new KeywordProperty.Builder().build())
+ .build(),
+ JsonMessageConstants.Attachment.FILE_EXTENSION, new Property.Builder()
+ .keyword(new KeywordProperty.Builder().build())
+ .build(),
+ JsonMessageConstants.Attachment.CONTENT_DISPOSITION, new Property.Builder()
+ .keyword(new KeywordProperty.Builder().build())
+ .build()
+ ))
+ .build())
+ .build())
+ .build();
}
}
diff --git a/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/events/OpenSearchListeningMessageSearchIndex.java b/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/events/OpenSearchListeningMessageSearchIndex.java
index 6b78c920d3..420d734d31 100644
--- a/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/events/OpenSearchListeningMessageSearchIndex.java
+++ b/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/events/OpenSearchListeningMessageSearchIndex.java
@@ -28,12 +28,10 @@ import static org.apache.james.mailbox.opensearch.json.JsonMessageConstants.IS_U
import static org.apache.james.mailbox.opensearch.json.JsonMessageConstants.MAILBOX_ID;
import static org.apache.james.mailbox.opensearch.json.JsonMessageConstants.MESSAGE_ID;
import static org.apache.james.mailbox.opensearch.json.JsonMessageConstants.UID;
-import static org.opensearch.index.query.QueryBuilders.termQuery;
import java.util.Collection;
import java.util.EnumSet;
import java.util.List;
-import java.util.Map;
import java.util.Optional;
import java.util.Set;
@@ -63,14 +61,20 @@ import org.apache.james.mailbox.opensearch.search.OpenSearchSearcher;
import org.apache.james.mailbox.store.MailboxSessionMapperFactory;
import org.apache.james.mailbox.store.mail.model.MailboxMessage;
import org.apache.james.mailbox.store.search.ListeningMessageSearchIndex;
-import org.opensearch.action.get.GetResponse;
-import org.opensearch.common.document.DocumentField;
-import org.opensearch.index.query.TermQueryBuilder;
-import org.opensearch.search.SearchHit;
+import org.opensearch.client.json.JsonData;
+import org.opensearch.client.json.JsonpDeserializer;
+import org.opensearch.client.opensearch._types.FieldValue;
+import org.opensearch.client.opensearch._types.query_dsl.Query;
+import org.opensearch.client.opensearch._types.query_dsl.TermQuery;
+import org.opensearch.client.opensearch.core.GetResponse;
+import org.opensearch.client.opensearch.core.search.Hit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
import com.github.fge.lambdas.Throwing;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
@@ -193,12 +197,12 @@ public class OpenSearchListeningMessageSearchIndex extends ListeningMessageSearc
@Override
public Mono<Void> deleteAll(MailboxSession session, MailboxId mailboxId) {
- TermQueryBuilder queryBuilder = termQuery(
- MAILBOX_ID,
- mailboxId.serialize());
+ Query query = TermQuery.of(t -> t
+ .field(MAILBOX_ID)
+ .value(new FieldValue.Builder().stringValue(mailboxId.serialize()).build()))._toQuery();
return openSearchIndexer
- .deleteAllMatchingQuery(queryBuilder, routingKeyFactory.from(mailboxId));
+ .deleteAllMatchingQuery(query, routingKeyFactory.from(mailboxId));
}
@Override
@@ -237,12 +241,12 @@ public class OpenSearchListeningMessageSearchIndex extends ListeningMessageSearc
RoutingKey routingKey = routingKeyFactory.from(mailbox.getMailboxId());
return openSearchIndexer.get(indexIdFor(mailbox.getMailboxId(), uid), routingKey)
- .filter(GetResponse::isExists)
- .map(GetResponse::getSourceAsMap)
+ .filter(GetResponse::found)
+ .map(GetResponse::source)
.map(this::extractFlags);
}
- private Flags extractFlags(Map<String, Object> source) {
+ private Flags extractFlags(ObjectNode source) {
FlagsBuilder flagsBuilder = FlagsBuilder.builder()
.isAnswered(extractFlag(source, IS_ANSWERED))
.isDeleted(extractFlag(source, IS_DELETED))
@@ -251,37 +255,38 @@ public class OpenSearchListeningMessageSearchIndex extends ListeningMessageSearc
.isRecent(extractFlag(source, IS_RECENT))
.isSeen(!extractFlag(source, IS_UNREAD));
- for (String userFlag : extractUserFlags(source)) {
- flagsBuilder.add(userFlag);
+ for (JsonNode userFlag : extractUserFlags(source)) {
+ flagsBuilder.add(userFlag.textValue());
}
return flagsBuilder.build();
}
- private boolean extractFlag(Map<String, Object> source, String flag) {
- return (Boolean) source.get(flag);
+ private boolean extractFlag(ObjectNode source, String flag) {
+ return source.get(flag).asBoolean();
}
- private List<String> extractUserFlags(Map<String, Object> source) {
- return (List<String>) source.get("userFlags");
+ private ArrayNode extractUserFlags(ObjectNode source) {
+ return source.withArray("userFlags");
}
- private void extractMessageIdFromHit(SearchHit hit, SynchronousSink<MessageId> sink) {
- DocumentField messageId = hit.field(MESSAGE_ID);
+ private void extractMessageIdFromHit(Hit<ObjectNode> hit, SynchronousSink<MessageId> sink) {
+ JsonData messageId = hit.fields().get(MESSAGE_ID);
if (messageId != null) {
- sink.next(messageIdFactory.fromString(messageId.getValue()));
+ List<String> extractMessageId = messageId.deserialize(JsonpDeserializer.arrayDeserializer(JsonpDeserializer.stringDeserializer()));
+ sink.next(messageIdFactory.fromString(extractMessageId.get(0)));
} else {
- LOGGER.warn("Can not extract UID, MessageID and/or MailboxId for search result {}", hit.getId());
+ LOGGER.warn("Can not extract UID, MessageID and/or MailboxId for search result {}", hit.id());
}
}
- private void extractUidFromHit(SearchHit hit, SynchronousSink<MessageUid> sink) {
- DocumentField uid = hit.field(UID);
+ private void extractUidFromHit(Hit<ObjectNode> hit, SynchronousSink<MessageUid> sink) {
+ JsonData uid = hit.fields().get(UID);
if (uid != null) {
- Number uidAsNumber = uid.getValue();
- sink.next(MessageUid.of(uidAsNumber.longValue()));
+ List<Number> uidAsNumber = uid.deserialize(JsonpDeserializer.arrayDeserializer(JsonpDeserializer.numberDeserializer()));
+ sink.next(MessageUid.of(uidAsNumber.get(0).longValue()));
} else {
- LOGGER.warn("Can not extract UID, MessageID and/or MailboxId for search result {}", hit.getId());
+ LOGGER.warn("Can not extract UID, MessageID and/or MailboxId for search result {}", hit.id());
}
}
}
diff --git a/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/query/CriterionConverter.java b/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/query/CriterionConverter.java
index 7513fda265..d7b019c8ce 100644
--- a/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/query/CriterionConverter.java
+++ b/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/query/CriterionConverter.java
@@ -20,12 +20,6 @@
package org.apache.james.mailbox.opensearch.query;
import static org.apache.james.backends.opensearch.IndexCreationFactory.RAW;
-import static org.opensearch.index.query.QueryBuilders.boolQuery;
-import static org.opensearch.index.query.QueryBuilders.matchAllQuery;
-import static org.opensearch.index.query.QueryBuilders.matchQuery;
-import static org.opensearch.index.query.QueryBuilders.nestedQuery;
-import static org.opensearch.index.query.QueryBuilders.rangeQuery;
-import static org.opensearch.index.query.QueryBuilders.termQuery;
import java.time.format.DateTimeFormatter;
import java.util.Arrays;
@@ -34,7 +28,6 @@ import java.util.Locale;
import java.util.Map;
import java.util.function.BiFunction;
import java.util.function.Function;
-import java.util.stream.Collector;
import java.util.stream.Stream;
import javax.mail.Flags;
@@ -44,15 +37,21 @@ import org.apache.james.mailbox.model.SearchQuery.Criterion;
import org.apache.james.mailbox.model.SearchQuery.HeaderOperator;
import org.apache.james.mailbox.opensearch.json.HeaderCollection;
import org.apache.james.mailbox.opensearch.json.JsonMessageConstants;
-import org.apache.lucene.search.join.ScoreMode;
-import org.opensearch.index.query.BoolQueryBuilder;
-import org.opensearch.index.query.QueryBuilder;
-import org.opensearch.index.query.QueryBuilders;
+import org.opensearch.client.json.JsonData;
+import org.opensearch.client.opensearch._types.FieldValue;
+import org.opensearch.client.opensearch._types.query_dsl.BoolQuery;
+import org.opensearch.client.opensearch._types.query_dsl.ChildScoreMode;
+import org.opensearch.client.opensearch._types.query_dsl.MatchAllQuery;
+import org.opensearch.client.opensearch._types.query_dsl.MatchQuery;
+import org.opensearch.client.opensearch._types.query_dsl.NestedQuery;
+import org.opensearch.client.opensearch._types.query_dsl.Query;
+import org.opensearch.client.opensearch._types.query_dsl.RangeQuery;
+import org.opensearch.client.opensearch._types.query_dsl.TermQuery;
public class CriterionConverter {
- private final Map<Class<?>, Function<Criterion, QueryBuilder>> criterionConverterMap;
- private final Map<Class<?>, BiFunction<String, HeaderOperator, QueryBuilder>> headerOperatorConverterMap;
+ private final Map<Class<?>, Function<Criterion, Query>> criterionConverterMap;
+ private final Map<Class<?>, BiFunction<String, HeaderOperator, Query>> headerOperatorConverterMap;
public CriterionConverter() {
criterionConverterMap = new HashMap<>();
@@ -72,7 +71,7 @@ public class CriterionConverter {
registerCriterionConverter(SearchQuery.CustomFlagCriterion.class, this::convertCustomFlagCriterion);
registerCriterionConverter(SearchQuery.AllCriterion.class,
- criterion -> matchAllQuery());
+ criterion -> new MatchAllQuery.Builder().build()._toQuery());
registerCriterionConverter(SearchQuery.ModSeqCriterion.class,
criterion -> createNumericFilter(JsonMessageConstants.MODSEQ, criterion.getOperator()));
@@ -92,17 +91,23 @@ public class CriterionConverter {
}
@SuppressWarnings("unchecked")
- private <T extends Criterion> void registerCriterionConverter(Class<T> type, Function<T, QueryBuilder> f) {
- criterionConverterMap.put(type, (Function<Criterion, QueryBuilder>) f);
+ private <T extends Criterion> void registerCriterionConverter(Class<T> type, Function<T, Query> f) {
+ criterionConverterMap.put(type, (Function<Criterion, Query>) f);
}
private void registerHeaderOperatorConverters() {
registerHeaderOperatorConverter(
SearchQuery.ExistsOperator.class,
- (headerName, operator) ->
- nestedQuery(JsonMessageConstants.HEADERS,
- termQuery(JsonMessageConstants.HEADERS + "." + JsonMessageConstants.HEADER.NAME, headerName),
- ScoreMode.Avg));
+ (headerName, operator) -> new NestedQuery.Builder()
+ .path(JsonMessageConstants.HEADERS)
+ .query(new TermQuery.Builder()
+ .field(JsonMessageConstants.HEADERS + "." + JsonMessageConstants.HEADER.NAME)
+ .value(new FieldValue.Builder().stringValue(headerName).build())
+ .build()
+ ._toQuery())
+ .scoreMode(ChildScoreMode.Avg)
+ .build()
+ ._toQuery());
registerHeaderOperatorConverter(
SearchQuery.AddressOperator.class,
@@ -114,71 +119,136 @@ public class CriterionConverter {
registerHeaderOperatorConverter(
SearchQuery.ContainsOperator.class,
- (headerName, operator) ->
- nestedQuery(JsonMessageConstants.HEADERS,
- boolQuery()
- .must(termQuery(JsonMessageConstants.HEADERS + "." + JsonMessageConstants.HEADER.NAME, headerName))
- .must(matchQuery(JsonMessageConstants.HEADERS + "." + JsonMessageConstants.HEADER.VALUE, operator.getValue())),
- ScoreMode.Avg));
+ (headerName, operator) -> new NestedQuery.Builder()
+ .path(JsonMessageConstants.HEADERS)
+ .query(new BoolQuery.Builder()
+ .must(new TermQuery.Builder()
+ .field(JsonMessageConstants.HEADERS + "." + JsonMessageConstants.HEADER.NAME)
+ .value(new FieldValue.Builder().stringValue(headerName).build())
+ .build()
+ ._toQuery())
+ .must(new MatchQuery.Builder()
+ .field(JsonMessageConstants.HEADERS + "." + JsonMessageConstants.HEADER.VALUE)
+ .query(new FieldValue.Builder().stringValue(operator.getValue()).build())
+ .build()
+ ._toQuery())
+ .build()
+ ._toQuery())
+ .scoreMode(ChildScoreMode.Avg)
+ .build()
+ ._toQuery());
}
@SuppressWarnings("unchecked")
- private <T extends HeaderOperator> void registerHeaderOperatorConverter(Class<T> type, BiFunction<String, T, QueryBuilder> f) {
- headerOperatorConverterMap.put(type, (BiFunction<String, HeaderOperator, QueryBuilder>) f);
+ private <T extends HeaderOperator> void registerHeaderOperatorConverter(Class<T> type, BiFunction<String, T, Query> f) {
+ headerOperatorConverterMap.put(type, (BiFunction<String, HeaderOperator, Query>) f);
}
- public QueryBuilder convertCriterion(Criterion criterion) {
+ public Query convertCriterion(Criterion criterion) {
return criterionConverterMap.get(criterion.getClass()).apply(criterion);
}
- private QueryBuilder convertAttachmentCriterion(SearchQuery.AttachmentCriterion criterion) {
- return termQuery(JsonMessageConstants.HAS_ATTACHMENT, criterion.getOperator().isSet());
+ private Query convertAttachmentCriterion(SearchQuery.AttachmentCriterion criterion) {
+ return new TermQuery.Builder()
+ .field(JsonMessageConstants.HAS_ATTACHMENT)
+ .value(new FieldValue.Builder().booleanValue(criterion.getOperator().isSet()).build())
+ .build()
+ ._toQuery();
}
- private QueryBuilder convertMimeMessageIDCriterion(SearchQuery.MimeMessageIDCriterion criterion) {
- return termQuery(JsonMessageConstants.MIME_MESSAGE_ID, criterion.getMessageID());
+ private Query convertMimeMessageIDCriterion(SearchQuery.MimeMessageIDCriterion criterion) {
+ return new TermQuery.Builder()
+ .field(JsonMessageConstants.MIME_MESSAGE_ID)
+ .value(new FieldValue.Builder().stringValue(criterion.getMessageID()).build())
+ .build()
+ ._toQuery();
}
- private QueryBuilder convertThreadIdCriterion(SearchQuery.ThreadIdCriterion criterion) {
- return termQuery(JsonMessageConstants.THREAD_ID, criterion.getThreadId().serialize());
+ private Query convertThreadIdCriterion(SearchQuery.ThreadIdCriterion criterion) {
+ return new TermQuery.Builder()
+ .field(JsonMessageConstants.THREAD_ID)
+ .value(new FieldValue.Builder().stringValue(criterion.getThreadId().serialize()).build())
+ .build()
+ ._toQuery();
}
- private QueryBuilder convertCustomFlagCriterion(SearchQuery.CustomFlagCriterion criterion) {
- QueryBuilder termQueryBuilder = termQuery(JsonMessageConstants.USER_FLAGS, criterion.getFlag());
+ private Query convertCustomFlagCriterion(SearchQuery.CustomFlagCriterion criterion) {
+ Query termQuery = new TermQuery.Builder()
+ .field(JsonMessageConstants.USER_FLAGS)
+ .value(new FieldValue.Builder().stringValue(criterion.getFlag()).build())
+ .build()
+ ._toQuery();
if (criterion.getOperator().isSet()) {
- return termQueryBuilder;
+ return termQuery;
} else {
- return boolQuery().mustNot(termQueryBuilder);
+ return new BoolQuery.Builder()
+ .mustNot(termQuery)
+ .build()
+ ._toQuery();
}
}
- private QueryBuilder convertTextCriterion(SearchQuery.TextCriterion textCriterion) {
+ private Query convertTextCriterion(SearchQuery.TextCriterion textCriterion) {
switch (textCriterion.getType()) {
case BODY:
- return boolQuery()
- .should(matchQuery(JsonMessageConstants.TEXT_BODY, textCriterion.getOperator().getValue()))
- .should(matchQuery(JsonMessageConstants.HTML_BODY, textCriterion.getOperator().getValue()));
+ return new BoolQuery.Builder()
+ .should(new MatchQuery.Builder()
+ .field(JsonMessageConstants.TEXT_BODY)
+ .query(new FieldValue.Builder().stringValue(textCriterion.getOperator().getValue()).build())
+ .build()
+ ._toQuery())
+ .should(new MatchQuery.Builder()
+ .field(JsonMessageConstants.HTML_BODY)
+ .query(new FieldValue.Builder().stringValue(textCriterion.getOperator().getValue()).build())
+ .build()
+ ._toQuery())
+ .build()
+ ._toQuery();
case FULL:
- return boolQuery()
- .should(matchQuery(JsonMessageConstants.TEXT_BODY, textCriterion.getOperator().getValue()))
- .should(matchQuery(JsonMessageConstants.HTML_BODY, textCriterion.getOperator().getValue()))
- .should(matchQuery(JsonMessageConstants.ATTACHMENTS + "." + JsonMessageConstants.Attachment.TEXT_CONTENT,
- textCriterion.getOperator().getValue()));
+ return new BoolQuery.Builder()
+ .should(new MatchQuery.Builder()
+ .field(JsonMessageConstants.TEXT_BODY)
+ .query(new FieldValue.Builder().stringValue(textCriterion.getOperator().getValue()).build())
+ .build()
+ ._toQuery())
+ .should(new MatchQuery.Builder()
+ .field(JsonMessageConstants.HTML_BODY)
+ .query(new FieldValue.Builder().stringValue(textCriterion.getOperator().getValue()).build())
+ .build()
+ ._toQuery())
+ .should(new MatchQuery.Builder()
+ .field(JsonMessageConstants.ATTACHMENTS + "." + JsonMessageConstants.Attachment.TEXT_CONTENT)
+ .query(new FieldValue.Builder().stringValue(textCriterion.getOperator().getValue()).build())
+ .build()
+ ._toQuery())
+ .build()
+ ._toQuery();
case ATTACHMENTS:
- return boolQuery()
- .should(matchQuery(JsonMessageConstants.ATTACHMENTS + "." + JsonMessageConstants.Attachment.TEXT_CONTENT,
- textCriterion.getOperator().getValue()));
+ return new BoolQuery.Builder()
+ .should(new MatchQuery.Builder()
+ .field(JsonMessageConstants.ATTACHMENTS + "." + JsonMessageConstants.Attachment.TEXT_CONTENT)
+ .query(new FieldValue.Builder().stringValue(textCriterion.getOperator().getValue()).build())
+ .build()
+ ._toQuery())
+ .build()
+ ._toQuery();
case ATTACHMENT_FILE_NAME:
- return boolQuery()
- .should(termQuery(JsonMessageConstants.ATTACHMENTS + "." + JsonMessageConstants.Attachment.FILENAME,
- textCriterion.getOperator().getValue()));
+ return new BoolQuery.Builder()
+ .should(new MatchQuery.Builder()
+ .field(JsonMessageConstants.ATTACHMENTS + "." + JsonMessageConstants.Attachment.FILENAME)
+ .query(new FieldValue.Builder().stringValue(textCriterion.getOperator().getValue()).build())
+ .build()
+ ._toQuery())
+ .build()
+ ._toQuery();
+ default:
+ throw new RuntimeException("Unknown SCOPE for text criterion");
}
- throw new RuntimeException("Unknown SCOPE for text criterion");
}
- private QueryBuilder dateRangeFilter(String field, SearchQuery.DateOperator dateOperator) {
- return boolQuery().filter(
- convertDateOperator(field,
+ private Query dateRangeFilter(String field, SearchQuery.DateOperator dateOperator) {
+ return new BoolQuery.Builder()
+ .filter(convertDateOperator(field,
dateOperator.getType(),
DateTimeFormatter.ISO_OFFSET_DATE_TIME.format(
DateResolutionFormatter.computeLowerDate(
@@ -187,104 +257,197 @@ public class CriterionConverter {
DateTimeFormatter.ISO_OFFSET_DATE_TIME.format(
DateResolutionFormatter.computeUpperDate(
DateResolutionFormatter.convertDateToZonedDateTime(dateOperator.getDate()),
- dateOperator.getDateResultion()))));
+ dateOperator.getDateResultion()))))
+ .build()
+ ._toQuery();
}
- private BoolQueryBuilder convertConjunction(SearchQuery.ConjunctionCriterion criterion) {
+ private Query convertConjunction(SearchQuery.ConjunctionCriterion criterion) {
return convertToBoolQuery(criterion.getCriteria().stream().map(this::convertCriterion),
convertConjunctionType(criterion.getType()));
}
- private BiFunction<BoolQueryBuilder, QueryBuilder, BoolQueryBuilder> convertConjunctionType(SearchQuery.Conjunction type) {
+ private BiFunction<BoolQuery.Builder, Query, BoolQuery.Builder> convertConjunctionType(SearchQuery.Conjunction type) {
switch (type) {
case AND:
- return BoolQueryBuilder::must;
+ return BoolQuery.Builder::must;
case OR:
- return BoolQueryBuilder::should;
+ return BoolQuery.Builder::should;
case NOR:
- return BoolQueryBuilder::mustNot;
+ return BoolQuery.Builder::mustNot;
default:
throw new RuntimeException("Unexpected conjunction criteria " + type);
}
}
@SuppressWarnings("ReturnValueIgnored")
- private BoolQueryBuilder convertToBoolQuery(Stream<QueryBuilder> stream, BiFunction<BoolQueryBuilder, QueryBuilder, BoolQueryBuilder> addCriterionToBoolQuery) {
- return stream.collect(Collector.of(QueryBuilders::boolQuery,
- addCriterionToBoolQuery::apply,
- addCriterionToBoolQuery::apply));
+ private Query convertToBoolQuery(Stream<Query> stream, BiFunction<BoolQuery.Builder, Query, BoolQuery.Builder> addCriterionToBoolQuery) {
+ BoolQuery.Builder builder = new BoolQuery.Builder();
+ stream.forEach(query -> addCriterionToBoolQuery.apply(builder, query));
+ return builder.build()._toQuery();
}
- private QueryBuilder convertFlag(SearchQuery.FlagCriterion flagCriterion) {
+ private Query convertFlag(SearchQuery.FlagCriterion flagCriterion) {
SearchQuery.BooleanOperator operator = flagCriterion.getOperator();
Flags.Flag flag = flagCriterion.getFlag();
if (flag.equals(Flags.Flag.DELETED)) {
- return boolQuery().filter(termQuery(JsonMessageConstants.IS_DELETED, operator.isSet()));
+ return new BoolQuery.Builder()
+ .filter(new TermQuery.Builder()
+ .field(JsonMessageConstants.IS_DELETED)
+ .value(new FieldValue.Builder().booleanValue(operator.isSet()).build())
+ .build()
+ ._toQuery())
+ .build()
+ ._toQuery();
}
if (flag.equals(Flags.Flag.ANSWERED)) {
- return boolQuery().filter(termQuery(JsonMessageConstants.IS_ANSWERED, operator.isSet()));
+ return new BoolQuery.Builder()
+ .filter(new TermQuery.Builder()
+ .field(JsonMessageConstants.IS_ANSWERED)
+ .value(new FieldValue.Builder().booleanValue(operator.isSet()).build())
+ .build()
+ ._toQuery())
+ .build()
+ ._toQuery();
}
if (flag.equals(Flags.Flag.DRAFT)) {
- return boolQuery().filter(termQuery(JsonMessageConstants.IS_DRAFT, operator.isSet()));
+ return new BoolQuery.Builder()
+ .filter(new TermQuery.Builder()
+ .field(JsonMessageConstants.IS_DRAFT)
+ .value(new FieldValue.Builder().booleanValue(operator.isSet()).build())
+ .build()
+ ._toQuery())
+ .build()
+ ._toQuery();
}
if (flag.equals(Flags.Flag.SEEN)) {
- return boolQuery().filter(termQuery(JsonMessageConstants.IS_UNREAD, !operator.isSet()));
+ return new BoolQuery.Builder()
+ .filter(new TermQuery.Builder()
+ .field(JsonMessageConstants.IS_UNREAD)
+ .value(new FieldValue.Builder().booleanValue(!operator.isSet()).build())
+ .build()
+ ._toQuery())
+ .build()
+ ._toQuery();
}
if (flag.equals(Flags.Flag.RECENT)) {
- return boolQuery().filter(termQuery(JsonMessageConstants.IS_RECENT, operator.isSet()));
+ return new BoolQuery.Builder()
+ .filter(new TermQuery.Builder()
+ .field(JsonMessageConstants.IS_RECENT)
+ .value(new FieldValue.Builder().booleanValue(operator.isSet()).build())
+ .build()
+ ._toQuery())
+ .build()
+ ._toQuery();
}
if (flag.equals(Flags.Flag.FLAGGED)) {
- return boolQuery().filter(termQuery(JsonMessageConstants.IS_FLAGGED, operator.isSet()));
+ return new BoolQuery.Builder()
+ .filter(new TermQuery.Builder()
+ .field(JsonMessageConstants.IS_FLAGGED)
+ .value(new FieldValue.Builder().booleanValue(operator.isSet()).build())
+ .build()
+ ._toQuery())
+ .build()
+ ._toQuery();
}
throw new RuntimeException("Unknown flag used in Flag search criterion");
}
- private QueryBuilder createNumericFilter(String fieldName, SearchQuery.NumericOperator operator) {
+ private Query createNumericFilter(String fieldName, SearchQuery.NumericOperator operator) {
switch (operator.getType()) {
case EQUALS:
- return boolQuery().filter(rangeQuery(fieldName).gte(operator.getValue()).lte(operator.getValue()));
+ return new BoolQuery.Builder()
+ .filter(new RangeQuery.Builder()
+ .field(fieldName)
+ .gte(JsonData.of(operator.getValue()))
+ .lte(JsonData.of(operator.getValue()))
+ .build()
+ ._toQuery())
+ .build()
+ ._toQuery();
case GREATER_THAN:
- return boolQuery().filter(rangeQuery(fieldName).gt(operator.getValue()));
+ return new BoolQuery.Builder()
+ .filter(new RangeQuery.Builder()
+ .field(fieldName)
+ .gt(JsonData.of(operator.getValue()))
+ .build()
+ ._toQuery())
+ .build()
+ ._toQuery();
case LESS_THAN:
- return boolQuery().filter(rangeQuery(fieldName).lt(operator.getValue()));
+ return new BoolQuery.Builder()
+ .filter(new RangeQuery.Builder()
+ .field(fieldName)
+ .lt(JsonData.of(operator.getValue()))
+ .build()
+ ._toQuery())
+ .build()
+ ._toQuery();
default:
throw new RuntimeException("A non existing numeric operator was triggered");
}
}
- private BoolQueryBuilder convertUid(SearchQuery.UidCriterion uidCriterion) {
+ private Query convertUid(SearchQuery.UidCriterion uidCriterion) {
if (uidCriterion.getOperator().getRange().length == 0) {
- return boolQuery();
+ return new BoolQuery.Builder().build()._toQuery();
}
- return boolQuery().filter(
- convertToBoolQuery(
+ return new BoolQuery.Builder()
+ .filter(convertToBoolQuery(
Arrays.stream(uidCriterion.getOperator().getRange())
- .map(this::uidRangeFilter), BoolQueryBuilder::should));
+ .map(this::uidRangeFilter), BoolQuery.Builder::should))
+ .build()
+ ._toQuery();
}
- private QueryBuilder convertMessageId(SearchQuery.MessageIdCriterion messageIdCriterion) {
- return termQuery(JsonMessageConstants.MESSAGE_ID, messageIdCriterion.getMessageId().serialize());
+ private Query convertMessageId(SearchQuery.MessageIdCriterion messageIdCriterion) {
+ return new TermQuery.Builder()
+ .field(JsonMessageConstants.MESSAGE_ID)
+ .value(new FieldValue.Builder().stringValue(messageIdCriterion.getMessageId().serialize()).build())
+ .build()
+ ._toQuery();
}
- private QueryBuilder uidRangeFilter(SearchQuery.UidRange numericRange) {
- return rangeQuery(JsonMessageConstants.UID)
- .lte(numericRange.getHighValue().asLong())
- .gte(numericRange.getLowValue().asLong());
+ private Query uidRangeFilter(SearchQuery.UidRange numericRange) {
+ return new RangeQuery.Builder()
+ .field(JsonMessageConstants.UID)
+ .lte(JsonData.of(numericRange.getHighValue().asLong()))
+ .gte(JsonData.of(numericRange.getLowValue().asLong()))
+ .build()
+ ._toQuery();
}
- private QueryBuilder convertHeader(SearchQuery.HeaderCriterion headerCriterion) {
+ private Query convertHeader(SearchQuery.HeaderCriterion headerCriterion) {
return headerOperatorConverterMap.get(headerCriterion.getOperator().getClass())
.apply(
headerCriterion.getHeaderName().toLowerCase(Locale.US),
headerCriterion.getOperator());
}
- private QueryBuilder manageAddressFields(String headerName, String value) {
- return boolQuery()
- .should(matchQuery(getFieldNameFromHeaderName(headerName) + "." + JsonMessageConstants.EMailer.NAME, value))
- .should(matchQuery(getFieldNameFromHeaderName(headerName) + "." + JsonMessageConstants.EMailer.ADDRESS, value))
- .should(matchQuery(getFieldNameFromHeaderName(headerName) + "." + JsonMessageConstants.EMailer.DOMAIN, value))
- .should(matchQuery(getFieldNameFromHeaderName(headerName) + "." + JsonMessageConstants.EMailer.ADDRESS + "." + RAW, value));
+ private Query manageAddressFields(String headerName, String value) {
+ return new BoolQuery.Builder()
+ .should(new MatchQuery.Builder()
+ .field(getFieldNameFromHeaderName(headerName) + "." + JsonMessageConstants.EMailer.NAME)
+ .query(new FieldValue.Builder().stringValue(value).build())
+ .build()
+ ._toQuery())
+ .should(new MatchQuery.Builder()
+ .field(getFieldNameFromHeaderName(headerName) + "." + JsonMessageConstants.EMailer.ADDRESS)
+ .query(new FieldValue.Builder().stringValue(value).build())
+ .build()
+ ._toQuery())
+ .should(new MatchQuery.Builder()
+ .field(getFieldNameFromHeaderName(headerName) + "." + JsonMessageConstants.EMailer.DOMAIN)
+ .query(new FieldValue.Builder().stringValue(value).build())
+ .build()
+ ._toQuery())
+ .should(new MatchQuery.Builder()
+ .field(getFieldNameFromHeaderName(headerName) + "." + JsonMessageConstants.EMailer.ADDRESS + "." + RAW)
+ .query(new FieldValue.Builder().stringValue(value).build())
+ .build()
+ ._toQuery())
+ .build()
+ ._toQuery();
}
private String getFieldNameFromHeaderName(String headerName) {
@@ -297,20 +460,35 @@ public class CriterionConverter {
return JsonMessageConstants.BCC;
case HeaderCollection.FROM:
return JsonMessageConstants.FROM;
+ default:
+ throw new RuntimeException("Header not recognized as Addess Header : " + headerName);
}
- throw new RuntimeException("Header not recognized as Addess Header : " + headerName);
}
- private QueryBuilder convertDateOperator(String field, SearchQuery.DateComparator dateComparator, String lowDateString, String upDateString) {
+ private Query convertDateOperator(String field, SearchQuery.DateComparator dateComparator, String lowDateString, String upDateString) {
switch (dateComparator) {
case BEFORE:
- return rangeQuery(field).lt(lowDateString); // less than start of the current day
+ return new RangeQuery.Builder()
+ .field(field)
+ .lt(JsonData.of(upDateString))
+ .build()
+ ._toQuery();
case AFTER:
- return rangeQuery(field).gte(upDateString); // start of next day + greater than that
+ return new RangeQuery.Builder()
+ .field(field)
+ .gte(JsonData.of(lowDateString))
+ .build()
+ ._toQuery();
case ON:
- return rangeQuery(field).lt(upDateString).gte(lowDateString);
+ return new RangeQuery.Builder()
+ .field(field)
+ .lt(JsonData.of(upDateString))
+ .gte(JsonData.of(lowDateString))
+ .build()
+ ._toQuery();
+ default:
+ throw new RuntimeException("Unknown date operator");
}
- throw new RuntimeException("Unknown date operator");
}
}
diff --git a/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/query/QueryConverter.java b/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/query/QueryConverter.java
index 1f6a34cf7e..e2d451a469 100644
--- a/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/query/QueryConverter.java
+++ b/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/query/QueryConverter.java
@@ -19,9 +19,6 @@
package org.apache.james.mailbox.opensearch.query;
-import static org.opensearch.index.query.QueryBuilders.boolQuery;
-import static org.opensearch.index.query.QueryBuilders.termsQuery;
-
import java.util.Collection;
import java.util.List;
import java.util.Optional;
@@ -31,14 +28,15 @@ import javax.inject.Inject;
import org.apache.james.mailbox.model.MailboxId;
import org.apache.james.mailbox.model.SearchQuery;
import org.apache.james.mailbox.opensearch.json.JsonMessageConstants;
-import org.opensearch.index.query.BoolQueryBuilder;
-import org.opensearch.index.query.QueryBuilder;
+import org.opensearch.client.opensearch._types.FieldValue;
+import org.opensearch.client.opensearch._types.query_dsl.BoolQuery;
+import org.opensearch.client.opensearch._types.query_dsl.Query;
+import org.opensearch.client.opensearch._types.query_dsl.TermsQuery;
+import org.opensearch.client.opensearch._types.query_dsl.TermsQueryField;
import com.google.common.collect.ImmutableList;
public class QueryConverter {
-
-
private final CriterionConverter criterionConverter;
@Inject
@@ -46,15 +44,15 @@ public class QueryConverter {
this.criterionConverter = criterionConverter;
}
- public QueryBuilder from(Collection<MailboxId> mailboxIds, SearchQuery query) {
- BoolQueryBuilder boolQueryBuilder = boolQuery()
+ public Query from(Collection<MailboxId> mailboxIds, SearchQuery query) {
+ BoolQuery.Builder boolQueryBuilder = new BoolQuery.Builder()
.must(generateQueryBuilder(query));
mailboxesQuery(mailboxIds).map(boolQueryBuilder::filter);
- return boolQueryBuilder;
+ return boolQueryBuilder.build()._toQuery();
}
- private QueryBuilder generateQueryBuilder(SearchQuery searchQuery) {
+ private Query generateQueryBuilder(SearchQuery searchQuery) {
List<SearchQuery.Criterion> criteria = searchQuery.getCriteria();
if (criteria.isEmpty()) {
return criterionConverter.convertCriterion(SearchQuery.all());
@@ -65,14 +63,21 @@ public class QueryConverter {
}
}
- private Optional<QueryBuilder> mailboxesQuery(Collection<MailboxId> mailboxIds) {
+ private Optional<Query> mailboxesQuery(Collection<MailboxId> mailboxIds) {
if (mailboxIds.isEmpty()) {
return Optional.empty();
}
- ImmutableList<String> ids = mailboxIds.stream()
- .map(MailboxId::serialize)
- .collect(ImmutableList.toImmutableList());
- return Optional.of(termsQuery(JsonMessageConstants.MAILBOX_ID, ids));
+ ImmutableList<FieldValue> ids = mailboxIds.stream()
+ .map(MailboxId::serialize)
+ .map(id -> new FieldValue.Builder().stringValue(id).build())
+ .collect(ImmutableList.toImmutableList());
+ return Optional.of(new TermsQuery.Builder()
+ .field(JsonMessageConstants.MAILBOX_ID)
+ .terms(new TermsQueryField.Builder()
+ .value(ids)
+ .build())
+ .build()
+ ._toQuery());
}
}
diff --git a/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/query/SortConverter.java b/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/query/SortConverter.java
index e1216a951f..cf4bb8d91e 100644
--- a/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/query/SortConverter.java
+++ b/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/query/SortConverter.java
@@ -22,44 +22,44 @@ package org.apache.james.mailbox.opensearch.query;
import org.apache.james.backends.opensearch.IndexCreationFactory;
import org.apache.james.mailbox.model.SearchQuery;
import org.apache.james.mailbox.opensearch.json.JsonMessageConstants;
-import org.opensearch.search.sort.FieldSortBuilder;
-import org.opensearch.search.sort.SortBuilders;
-import org.opensearch.search.sort.SortMode;
-import org.opensearch.search.sort.SortOrder;
+import org.opensearch.client.opensearch._types.FieldSort;
+import org.opensearch.client.opensearch._types.SortMode;
+import org.opensearch.client.opensearch._types.SortOrder;
public class SortConverter {
private static final String PATH_SEPARATOR = ".";
- public static FieldSortBuilder convertSort(SearchQuery.Sort sort) {
+ public static FieldSort convertSort(SearchQuery.Sort sort) {
return getSortClause(sort.getSortClause())
.order(getOrder(sort))
- .sortMode(SortMode.MIN);
+ .mode(SortMode.Min)
+ .build();
}
- private static FieldSortBuilder getSortClause(SearchQuery.Sort.SortClause clause) {
+ private static FieldSort.Builder getSortClause(SearchQuery.Sort.SortClause clause) {
switch (clause) {
case Arrival :
- return SortBuilders.fieldSort(JsonMessageConstants.DATE);
+ return new FieldSort.Builder().field(JsonMessageConstants.DATE);
case MailboxCc :
- return SortBuilders.fieldSort(JsonMessageConstants.CC + PATH_SEPARATOR + JsonMessageConstants.EMailer.ADDRESS
- + PATH_SEPARATOR + IndexCreationFactory.RAW);
+ return new FieldSort.Builder().field(JsonMessageConstants.CC + PATH_SEPARATOR
+ + JsonMessageConstants.EMailer.ADDRESS + PATH_SEPARATOR + IndexCreationFactory.RAW);
case MailboxFrom :
- return SortBuilders.fieldSort(JsonMessageConstants.FROM + PATH_SEPARATOR + JsonMessageConstants.EMailer.ADDRESS
- + PATH_SEPARATOR + IndexCreationFactory.RAW);
+ return new FieldSort.Builder().field(JsonMessageConstants.FROM + PATH_SEPARATOR
+ + JsonMessageConstants.EMailer.ADDRESS + PATH_SEPARATOR + IndexCreationFactory.RAW);
case MailboxTo :
- return SortBuilders.fieldSort(JsonMessageConstants.TO + PATH_SEPARATOR + JsonMessageConstants.EMailer.ADDRESS
- + PATH_SEPARATOR + IndexCreationFactory.RAW);
+ return new FieldSort.Builder().field(JsonMessageConstants.TO + PATH_SEPARATOR
+ + JsonMessageConstants.EMailer.ADDRESS + PATH_SEPARATOR + IndexCreationFactory.RAW);
case BaseSubject :
- return SortBuilders.fieldSort(JsonMessageConstants.SUBJECT + PATH_SEPARATOR + IndexCreationFactory.RAW);
+ return new FieldSort.Builder().field(JsonMessageConstants.SUBJECT + PATH_SEPARATOR + IndexCreationFactory.RAW);
case Size :
- return SortBuilders.fieldSort(JsonMessageConstants.SIZE);
+ return new FieldSort.Builder().field(JsonMessageConstants.SIZE);
case SentDate :
- return SortBuilders.fieldSort(JsonMessageConstants.SENT_DATE);
+ return new FieldSort.Builder().field(JsonMessageConstants.SENT_DATE);
case Uid :
- return SortBuilders.fieldSort(JsonMessageConstants.UID);
+ return new FieldSort.Builder().field(JsonMessageConstants.UID);
case Id:
- return SortBuilders.fieldSort(JsonMessageConstants.MESSAGE_ID);
+ return new FieldSort.Builder().field(JsonMessageConstants.MESSAGE_ID);
default:
throw new RuntimeException("Sort is not implemented");
}
@@ -67,9 +67,9 @@ public class SortConverter {
private static SortOrder getOrder(SearchQuery.Sort sort) {
if (sort.isReverse()) {
- return SortOrder.DESC;
+ return SortOrder.Desc;
} else {
- return SortOrder.ASC;
+ return SortOrder.Asc;
}
}
}
diff --git a/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/search/OpenSearchSearcher.java b/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/search/OpenSearchSearcher.java
index f1bfcecfc6..7b7f4acd97 100644
--- a/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/search/OpenSearchSearcher.java
+++ b/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/search/OpenSearchSearcher.java
@@ -22,6 +22,7 @@ package org.apache.james.mailbox.opensearch.search;
import java.util.Collection;
import java.util.List;
import java.util.Optional;
+import java.util.stream.Collectors;
import org.apache.james.backends.opensearch.AliasName;
import org.apache.james.backends.opensearch.ReactorOpenSearchClient;
@@ -32,16 +33,18 @@ import org.apache.james.mailbox.model.MailboxId;
import org.apache.james.mailbox.model.SearchQuery;
import org.apache.james.mailbox.opensearch.query.QueryConverter;
import org.apache.james.mailbox.opensearch.query.SortConverter;
-import org.opensearch.action.search.SearchRequest;
-import org.opensearch.common.unit.TimeValue;
-import org.opensearch.search.SearchHit;
-import org.opensearch.search.builder.SearchSourceBuilder;
+import org.opensearch.client.opensearch._types.SortOptions;
+import org.opensearch.client.opensearch._types.Time;
+import org.opensearch.client.opensearch.core.SearchRequest;
+import org.opensearch.client.opensearch.core.search.Hit;
+
+import com.fasterxml.jackson.databind.node.ObjectNode;
import reactor.core.publisher.Flux;
public class OpenSearchSearcher {
public static final int DEFAULT_SEARCH_SIZE = 100;
- private static final TimeValue TIMEOUT = TimeValue.timeValueMinutes(1);
+ private static final Time TIMEOUT = new Time.Builder().time("1m").build();
private static final int MAX_ROUTING_KEY = 5;
private final ReactorOpenSearchClient client;
@@ -59,39 +62,40 @@ public class OpenSearchSearcher {
this.routingKeyFactory = routingKeyFactory;
}
- public Flux<SearchHit> search(Collection<MailboxId> mailboxIds, SearchQuery query,
- Optional<Integer> limit, List<String> fields) {
+ public Flux<Hit<ObjectNode>> search(Collection<MailboxId> mailboxIds, SearchQuery query,
+ Optional<Integer> limit, List<String> fields) {
SearchRequest searchRequest = prepareSearch(mailboxIds, query, limit, fields);
return new ScrolledSearch(client, searchRequest)
.searchHits();
}
private SearchRequest prepareSearch(Collection<MailboxId> mailboxIds, SearchQuery query, Optional<Integer> limit, List<String> fields) {
- SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder()
- .query(queryConverter.from(mailboxIds, query))
- .size(computeRequiredSize(limit))
- .storedFields(fields);
-
- query.getSorts()
+ List<SortOptions> sorts = query.getSorts()
.stream()
.map(SortConverter::convertSort)
- .forEach(searchSourceBuilder::sort);
+ .map(fieldSort -> new SortOptions.Builder().field(fieldSort).build())
+ .collect(Collectors.toList());
- SearchRequest request = new SearchRequest(aliasName.getValue())
+ SearchRequest.Builder request = new SearchRequest.Builder()
+ .index(aliasName.getValue())
.scroll(TIMEOUT)
- .source(searchSourceBuilder);
+ .query(queryConverter.from(mailboxIds, query))
+ .size(computeRequiredSize(limit))
+ .storedFields(fields)
+ .sort(sorts);
return toRoutingKey(mailboxIds)
.map(request::routing)
- .orElse(request);
+ .orElse(request)
+ .build();
}
- private Optional<String[]> toRoutingKey(Collection<MailboxId> mailboxIds) {
+ private Optional<String> toRoutingKey(Collection<MailboxId> mailboxIds) {
if (mailboxIds.size() < MAX_ROUTING_KEY) {
return Optional.of(mailboxIds.stream()
.map(routingKeyFactory::from)
.map(RoutingKey::asString)
- .toArray(String[]::new));
+ .collect(Collectors.joining(",")));
}
return Optional.empty();
}
diff --git a/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/OpenSearchIntegrationTest.java b/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/OpenSearchIntegrationTest.java
index 19f05f1c78..c87203b756 100644
--- a/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/OpenSearchIntegrationTest.java
+++ b/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/OpenSearchIntegrationTest.java
@@ -63,11 +63,9 @@ import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
-import org.opensearch.action.search.SearchRequest;
-import org.opensearch.client.RequestOptions;
-import org.opensearch.index.query.QueryBuilder;
-import org.opensearch.index.query.QueryBuilders;
-import org.opensearch.search.builder.SearchSourceBuilder;
+import org.opensearch.client.opensearch._types.query_dsl.Query;
+import org.opensearch.client.opensearch._types.query_dsl.QueryBuilders;
+import org.opensearch.client.opensearch.core.SearchRequest;
import com.google.common.base.Strings;
import com.google.common.collect.ImmutableList;
@@ -172,7 +170,7 @@ class OpenSearchIntegrationTest extends AbstractMessageSearchIndexTest {
.setBody(Strings.repeat("0à 2345678é", 3200), StandardCharsets.UTF_8)),
session).getId();
- awaitForOpenSearch(QueryBuilders.matchAllQuery(), 14);
+ awaitForOpenSearch(QueryBuilders.matchAll().build()._toQuery(), 14);
assertThat(Flux.from(messageManager.search(SearchQuery.of(SearchQuery.address(SearchQuery.AddressType.To, recipient)), session)).toStream())
.containsExactly(composedMessageId.getUid());
@@ -193,11 +191,12 @@ class OpenSearchIntegrationTest extends AbstractMessageSearchIndexTest {
CALMLY_AWAIT.atMost(Durations.TEN_SECONDS)
.untilAsserted(() -> assertThat(client.search(
- new SearchRequest(MailboxOpenSearchConstants.DEFAULT_MAILBOX_INDEX.getValue())
- .source(new SearchSourceBuilder().query(QueryBuilders.matchAllQuery())),
- RequestOptions.DEFAULT)
+ new SearchRequest.Builder()
+ .index(MailboxOpenSearchConstants.DEFAULT_MAILBOX_INDEX.getValue())
+ .query(QueryBuilders.matchAll().build()._toQuery())
+ .build())
.block()
- .getHits().getTotalHits().value).isGreaterThanOrEqualTo(14));
+ .hits().total().value()).isEqualTo(14));
assertThat(Flux.from(messageManager.search(SearchQuery.of(SearchQuery.address(SearchQuery.AddressType.To, recipient)), session)).toStream())
.containsExactly(composedMessageId.getUid());
@@ -216,7 +215,7 @@ class OpenSearchIntegrationTest extends AbstractMessageSearchIndexTest {
.setBody(Strings.repeat("0123456789 ", 5000), StandardCharsets.UTF_8)),
session).getId();
- awaitForOpenSearch(QueryBuilders.matchAllQuery(), 14);
+ awaitForOpenSearch(QueryBuilders.matchAll().build()._toQuery(), 14);
assertThat(Flux.from(messageManager.search(SearchQuery.of(SearchQuery.bodyContains("0123456789")), session)).toStream())
.containsExactly(composedMessageId.getUid());
@@ -235,7 +234,7 @@ class OpenSearchIntegrationTest extends AbstractMessageSearchIndexTest {
.setBody(Strings.repeat("0123456789 ", 5000) + " matchMe", StandardCharsets.UTF_8)),
session).getId();
- awaitForOpenSearch(QueryBuilders.matchAllQuery(), 14);
+ awaitForOpenSearch(QueryBuilders.matchAll().build()._toQuery(), 14);
assertThat(Flux.from(messageManager.search(SearchQuery.of(SearchQuery.bodyContains("matchMe")), session)).toStream())
.containsExactly(composedMessageId.getUid());
@@ -277,7 +276,7 @@ class OpenSearchIntegrationTest extends AbstractMessageSearchIndexTest {
.build(ClassLoaderUtils.getSystemResourceAsSharedStream("eml/mailCustomStringHeader.eml")),
session).getId();
- awaitForOpenSearch(QueryBuilders.matchAllQuery(), 15);
+ awaitForOpenSearch(QueryBuilders.matchAll().build()._toQuery(), 15);
assertThat(Flux.from(messageManager.search(SearchQuery.of(SearchQuery.headerExists("Custom-header")), session)).toStream())
.containsExactly(customDateHeaderMessageId.getUid(), customStringHeaderMessageId.getUid());
@@ -330,7 +329,7 @@ class OpenSearchIntegrationTest extends AbstractMessageSearchIndexTest {
.build()),
session).getId();
- awaitForOpenSearch(QueryBuilders.matchAllQuery(), 15);
+ awaitForOpenSearch(QueryBuilders.matchAll().build()._toQuery(), 15);
assertThat(Flux.from(messageManager.search(SearchQuery.of(SearchQuery.address(SearchQuery.AddressType.To, "bob@other.tld")), session)).toStream())
.containsOnly(messageId2.getUid());
@@ -361,13 +360,19 @@ class OpenSearchIntegrationTest extends AbstractMessageSearchIndexTest {
.build()),
session).getId();
- awaitForOpenSearch(QueryBuilders.matchAllQuery(), 15);
+ awaitForOpenSearch(QueryBuilders.matchAll().build()._toQuery(), 15);
Thread.sleep(500);
assertThat(Flux.from(messageManager.search(SearchQuery.of(SearchQuery.address(SearchQuery.AddressType.To, "other")), session)).toStream())
.containsOnly(messageId2.getUid());
}
+ @Disabled("JAMES-3771 Waiting for a fix on opensearch client to be merged: https://github.com/opensearch-project/opensearch-java/pull/169")
+ @Test
+ public void sortOnCcShouldWork() throws Exception {
+
+ }
+
@Disabled("MAILBOX-403 Relaxed the matching constraints for email addresses in text bodies to reduce OpenSearch disk space usage")
@Test
public void textShouldNotMatchOtherAddressesOfTheSameDomain() {
@@ -470,13 +475,14 @@ class OpenSearchIntegrationTest extends AbstractMessageSearchIndexTest {
.containsOnly(messageId1.getUid());
}
- private void awaitForOpenSearch(QueryBuilder query, long totalHits) {
+ private void awaitForOpenSearch(Query query, long totalHits) {
CALMLY_AWAIT.atMost(Durations.TEN_SECONDS)
.untilAsserted(() -> assertThat(client.search(
- new SearchRequest(MailboxOpenSearchConstants.DEFAULT_MAILBOX_INDEX.getValue())
- .source(new SearchSourceBuilder().query(query)),
- RequestOptions.DEFAULT)
+ new SearchRequest.Builder()
+ .index(MailboxOpenSearchConstants.DEFAULT_MAILBOX_INDEX.getValue())
+ .query(query)
+ .build())
.block()
- .getHits().getTotalHits().value).isEqualTo(totalHits));
+ .hits().total().value()).isEqualTo(totalHits));
}
}
\ No newline at end of file
diff --git a/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/events/OpenSearchListeningMessageSearchIndexTest.java b/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/events/OpenSearchListeningMessageSearchIndexTest.java
index ce34fbe26b..bf3025cbf2 100644
--- a/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/events/OpenSearchListeningMessageSearchIndexTest.java
+++ b/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/events/OpenSearchListeningMessageSearchIndexTest.java
@@ -86,11 +86,10 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Nested;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
-import org.opensearch.action.search.SearchRequest;
-import org.opensearch.client.RequestOptions;
-import org.opensearch.index.query.QueryBuilder;
-import org.opensearch.index.query.QueryBuilders;
-import org.opensearch.search.builder.SearchSourceBuilder;
+import org.opensearch.client.opensearch._types.FieldValue;
+import org.opensearch.client.opensearch._types.query_dsl.Query;
+import org.opensearch.client.opensearch._types.query_dsl.QueryBuilders;
+import org.opensearch.client.opensearch.core.SearchRequest;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
@@ -219,7 +218,7 @@ class OpenSearchListeningMessageSearchIndexTest {
@Test
void addShouldIndexMessageWithoutAttachment() throws Exception {
testee.add(session, mailbox, MESSAGE_1).block();
- awaitForOpenSearch(QueryBuilders.matchAllQuery(), 1L);
+ awaitForOpenSearch(QueryBuilders.matchAll().build()._toQuery(), 1L);
SearchQuery query = SearchQuery.of(SearchQuery.all());
assertThat(testee.search(session, mailbox, query).toStream())
@@ -229,7 +228,7 @@ class OpenSearchListeningMessageSearchIndexTest {
@Test
void addShouldIndexMessageWithAttachment() throws Exception {
testee.add(session, mailbox, MESSAGE_WITH_ATTACHMENT).block();
- awaitForOpenSearch(QueryBuilders.matchAllQuery(), 1L);
+ awaitForOpenSearch(QueryBuilders.matchAll().build()._toQuery(), 1L);
SearchQuery query = SearchQuery.of(SearchQuery.all());
assertThat(testee.search(session, mailbox, query).toStream())
@@ -241,7 +240,7 @@ class OpenSearchListeningMessageSearchIndexTest {
testee.add(session, mailbox, MESSAGE_1).block();
testee.add(session, mailbox, MESSAGE_1).block();
- awaitForOpenSearch(QueryBuilders.matchAllQuery(), 1L);
+ awaitForOpenSearch(QueryBuilders.matchAll().build()._toQuery(), 1L);
SearchQuery query = SearchQuery.of(SearchQuery.all());
assertThat(testee.search(session, mailbox, query).toStream())
@@ -253,7 +252,7 @@ class OpenSearchListeningMessageSearchIndexTest {
testee.add(session, mailbox, MESSAGE_1).block();
testee.add(session, mailbox, MESSAGE_2).block();
- awaitForOpenSearch(QueryBuilders.matchAllQuery(), 2L);
+ awaitForOpenSearch(QueryBuilders.matchAll().build()._toQuery(), 2L);
SearchQuery query = SearchQuery.of(SearchQuery.all());
assertThat(testee.search(session, mailbox, query).toStream())
@@ -272,7 +271,7 @@ class OpenSearchListeningMessageSearchIndexTest {
messageToOpenSearchJson, sessionProvider, new MailboxIdRoutingKeyFactory(), new InMemoryMessageId.Factory());
testee.add(session, mailbox, MESSAGE_WITH_ATTACHMENT).block();
- awaitForOpenSearch(QueryBuilders.matchAllQuery(), 1L);
+ awaitForOpenSearch(QueryBuilders.matchAll().build()._toQuery(), 1L);
SearchQuery query = SearchQuery.of(SearchQuery.all());
assertThat(testee.search(session, mailbox, query).toStream())
@@ -293,10 +292,10 @@ class OpenSearchListeningMessageSearchIndexTest {
@Test
void deleteShouldRemoveIndex() throws Exception {
testee.add(session, mailbox, MESSAGE_1).block();
- awaitForOpenSearch(QueryBuilders.matchAllQuery(), 1L);
+ awaitForOpenSearch(QueryBuilders.matchAll().build()._toQuery(), 1L);
testee.delete(session, mailbox.getMailboxId(), Lists.newArrayList(MESSAGE_UID_1)).block();
- awaitForOpenSearch(QueryBuilders.matchAllQuery(), 0L);
+ awaitForOpenSearch(QueryBuilders.matchAll().build()._toQuery(), 0L);
SearchQuery query = SearchQuery.of(SearchQuery.all());
assertThat(testee.search(session, mailbox, query).toStream())
@@ -307,10 +306,10 @@ class OpenSearchListeningMessageSearchIndexTest {
void deleteShouldOnlyRemoveIndexesPassedAsArguments() throws Exception {
testee.add(session, mailbox, MESSAGE_1).block();
testee.add(session, mailbox, MESSAGE_2).block();
- awaitForOpenSearch(QueryBuilders.matchAllQuery(), 2L);
+ awaitForOpenSearch(QueryBuilders.matchAll().build()._toQuery(), 2L);
testee.delete(session, mailbox.getMailboxId(), Lists.newArrayList(MESSAGE_UID_1)).block();
- awaitForOpenSearch(QueryBuilders.matchAllQuery(), 1L);
+ awaitForOpenSearch(QueryBuilders.matchAll().build()._toQuery(), 1L);
SearchQuery query = SearchQuery.of(SearchQuery.all());
assertThat(testee.search(session, mailbox, query).toStream())
@@ -321,10 +320,10 @@ class OpenSearchListeningMessageSearchIndexTest {
void deleteShouldRemoveMultipleIndexes() throws Exception {
testee.add(session, mailbox, MESSAGE_1).block();
testee.add(session, mailbox, MESSAGE_2).block();
- awaitForOpenSearch(QueryBuilders.matchAllQuery(), 2L);
+ awaitForOpenSearch(QueryBuilders.matchAll().build()._toQuery(), 2L);
testee.delete(session, mailbox.getMailboxId(), Lists.newArrayList(MESSAGE_UID_1, MESSAGE_UID_2)).block();
- awaitForOpenSearch(QueryBuilders.matchAllQuery(), 0L);
+ awaitForOpenSearch(QueryBuilders.matchAll().build()._toQuery(), 0L);
SearchQuery query = SearchQuery.of(SearchQuery.all());
assertThat(testee.search(session, mailbox, query).toStream())
@@ -334,11 +333,11 @@ class OpenSearchListeningMessageSearchIndexTest {
@Test
void deleteShouldBeIdempotent() throws Exception {
testee.add(session, mailbox, MESSAGE_1).block();
- awaitForOpenSearch(QueryBuilders.matchAllQuery(), 1L);
+ awaitForOpenSearch(QueryBuilders.matchAll().build()._toQuery(), 1L);
testee.delete(session, mailbox.getMailboxId(), Lists.newArrayList(MESSAGE_UID_1)).block();
testee.delete(session, mailbox.getMailboxId(), Lists.newArrayList(MESSAGE_UID_1)).block();
- awaitForOpenSearch(QueryBuilders.matchAllQuery(), 0L);
+ awaitForOpenSearch(QueryBuilders.matchAll().build()._toQuery(), 0L);
SearchQuery query = SearchQuery.of(SearchQuery.all());
assertThat(testee.search(session, mailbox, query).toStream())
@@ -365,7 +364,7 @@ class OpenSearchListeningMessageSearchIndexTest {
@Test
void updateShouldUpdateIndex() throws Exception {
testee.add(session, mailbox, MESSAGE_1).block();
- awaitForOpenSearch(QueryBuilders.matchAllQuery(), 1L);
+ awaitForOpenSearch(QueryBuilders.matchAll().build()._toQuery(), 1L);
Flags newFlags = new Flags(Flags.Flag.ANSWERED);
UpdatedFlags updatedFlags = UpdatedFlags.builder()
@@ -376,7 +375,7 @@ class OpenSearchListeningMessageSearchIndexTest {
.build();
testee.update(session, mailbox.getMailboxId(), Lists.newArrayList(updatedFlags)).block();
- awaitForOpenSearch(QueryBuilders.termQuery("isAnswered", true), 1L);
+ awaitForOpenSearch(QueryBuilders.term().field("isAnswered").value(FieldValue.of(true)).build()._toQuery(), 1L);
SearchQuery query = SearchQuery.of(SearchQuery.flagIsSet(Flags.Flag.ANSWERED));
assertThat(testee.search(session, mailbox, query).toStream())
@@ -384,9 +383,9 @@ class OpenSearchListeningMessageSearchIndexTest {
}
@Test
- void updateShouldNotUpdateNorThrowOnUnknownMessageUid() throws Exception {
+ void updateShouldThrowOnUnknownMessageUid() throws Exception {
testee.add(session, mailbox, MESSAGE_1).block();
- awaitForOpenSearch(QueryBuilders.matchAllQuery(), 1L);
+ awaitForOpenSearch(QueryBuilders.matchAll().build()._toQuery(), 1L);
Flags newFlags = new Flags(Flags.Flag.ANSWERED);
UpdatedFlags updatedFlags = UpdatedFlags.builder()
@@ -406,7 +405,7 @@ class OpenSearchListeningMessageSearchIndexTest {
@Test
void updateShouldBeIdempotent() throws Exception {
testee.add(session, mailbox, MESSAGE_1).block();
- awaitForOpenSearch(QueryBuilders.matchAllQuery(), 1L);
+ awaitForOpenSearch(QueryBuilders.matchAll().build()._toQuery(), 1L);
Flags newFlags = new Flags(Flags.Flag.ANSWERED);
UpdatedFlags updatedFlags = UpdatedFlags.builder()
@@ -418,7 +417,7 @@ class OpenSearchListeningMessageSearchIndexTest {
testee.update(session, mailbox.getMailboxId(), Lists.newArrayList(updatedFlags)).block();
testee.update(session, mailbox.getMailboxId(), Lists.newArrayList(updatedFlags)).block();
- awaitForOpenSearch(QueryBuilders.termQuery("isAnswered", true), 1L);
+ awaitForOpenSearch(QueryBuilders.term().field("isAnswered").value(FieldValue.of(true)).build()._toQuery(), 1L);
SearchQuery query = SearchQuery.of(SearchQuery.flagIsSet(Flags.Flag.ANSWERED));
assertThat(testee.search(session, mailbox, query).toStream())
@@ -448,10 +447,10 @@ class OpenSearchListeningMessageSearchIndexTest {
void deleteAllShouldRemoveAllIndexes() throws Exception {
testee.add(session, mailbox, MESSAGE_1).block();
testee.add(session, mailbox, MESSAGE_2).block();
- awaitForOpenSearch(QueryBuilders.matchAllQuery(), 2L);
+ awaitForOpenSearch(QueryBuilders.matchAll().build()._toQuery(), 2L);
testee.deleteAll(session, mailbox.getMailboxId()).block();
- awaitForOpenSearch(QueryBuilders.matchAllQuery(), 0L);
+ awaitForOpenSearch(QueryBuilders.matchAll().build()._toQuery(), 0L);
SearchQuery query = SearchQuery.of(SearchQuery.all());
assertThat(testee.search(session, mailbox, query).toStream())
@@ -483,18 +482,21 @@ class OpenSearchListeningMessageSearchIndexTest {
@Test
void retrieveIndexedFlagsShouldReturnEmptyWhenNotFound() {
+ testee.add(session, mailbox, MESSAGE_1).block();
+
assertThat(testee.retrieveIndexedFlags(mailbox, MESSAGE_UID_4).blockOptional())
.isEmpty();
}
}
- private void awaitForOpenSearch(QueryBuilder query, long totalHits) {
+ private void awaitForOpenSearch(Query query, long totalHits) {
CALMLY_AWAIT.atMost(Durations.TEN_SECONDS)
.untilAsserted(() -> assertThat(client.search(
- new SearchRequest(MailboxOpenSearchConstants.DEFAULT_MAILBOX_INDEX.getValue())
- .source(new SearchSourceBuilder().query(query)),
- RequestOptions.DEFAULT)
+ new SearchRequest.Builder()
+ .index(MailboxOpenSearchConstants.DEFAULT_MAILBOX_INDEX.getValue())
+ .query(query)
+ .build())
.block()
- .getHits().getTotalHits().value).isEqualTo(totalHits));
+ .hits().total().value()).isEqualTo(totalHits));
}
}
\ No newline at end of file
diff --git a/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/search/OpenSearchSearcherTest.java b/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/search/OpenSearchSearcherTest.java
index 9c24556392..70fbde8e8b 100644
--- a/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/search/OpenSearchSearcherTest.java
+++ b/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/search/OpenSearchSearcherTest.java
@@ -67,11 +67,9 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
-import org.opensearch.action.search.SearchRequest;
-import org.opensearch.client.RequestOptions;
-import org.opensearch.index.query.QueryBuilder;
-import org.opensearch.index.query.QueryBuilders;
-import org.opensearch.search.builder.SearchSourceBuilder;
+import org.opensearch.client.opensearch._types.query_dsl.Query;
+import org.opensearch.client.opensearch._types.query_dsl.QueryBuilders;
+import org.opensearch.client.opensearch.core.SearchRequest;
import com.github.fge.lambdas.Throwing;
import com.google.common.collect.ImmutableList;
@@ -157,7 +155,7 @@ class OpenSearchSearcherTest {
.map(Throwing.<MailboxPath, ComposedMessageId>function(mailboxPath -> addMessage(session, mailboxPath)).sneakyThrow())
.collect(ImmutableList.toImmutableList());
- awaitForOpenSearch(QueryBuilders.matchAllQuery(), composedMessageIds.size());
+ awaitForOpenSearch(QueryBuilders.matchAll().build()._toQuery(), composedMessageIds.size());
MultimailboxesSearchQuery multimailboxesSearchQuery = MultimailboxesSearchQuery
.from(SearchQuery.of(SearchQuery.all()))
@@ -184,13 +182,14 @@ class OpenSearchSearcherTest {
.getId();
}
- private void awaitForOpenSearch(QueryBuilder query, long totalHits) {
+ private void awaitForOpenSearch(Query query, long totalHits) {
CALMLY_AWAIT.atMost(Durations.TEN_SECONDS)
.untilAsserted(() -> assertThat(client.search(
- new SearchRequest(MailboxOpenSearchConstants.DEFAULT_MAILBOX_INDEX.getValue())
- .source(new SearchSourceBuilder().query(query)),
- RequestOptions.DEFAULT)
+ new SearchRequest.Builder()
+ .index(MailboxOpenSearchConstants.DEFAULT_MAILBOX_INDEX.getValue())
+ .query(query)
+ .build())
.block()
- .getHits().getTotalHits().value).isEqualTo(totalHits));
+ .hits().total().value()).isEqualTo(totalHits));
}
}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 01/12: JAMES-3771 Migrate backend opensearch high level to new java client
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit c7cb996ee941b649f4d270ce4365aa7eaf910ed6
Author: Rene Cordier <rc...@linagora.com>
AuthorDate: Tue Jun 21 15:03:11 2022 +0700
JAMES-3771 Migrate backend opensearch high level to new java client
---
backends-common/opensearch/pom.xml | 15 +-
.../james/backends/opensearch/ClientProvider.java | 34 ++-
.../opensearch/DeleteByQueryPerformer.java | 41 ++--
.../james/backends/opensearch/DocumentId.java | 3 +-
.../backends/opensearch/IndexCreationFactory.java | 254 ++++++++++++---------
.../backends/opensearch/OpenSearchHealthCheck.java | 38 +--
.../backends/opensearch/OpenSearchIndexer.java | 121 ++++++----
.../opensearch/ReactorOpenSearchClient.java | 181 ++++++---------
.../james/backends/opensearch/RoutingKey.java | 3 +-
.../backends/opensearch/UpdatedRepresentation.java | 3 +-
.../backends/opensearch/search/ScrolledSearch.java | 74 +++---
.../ClientProviderImplConnectionContract.java | 12 +-
.../opensearch/DockerOpenSearchExtension.java | 42 ++--
.../opensearch/IndexCreationFactoryTest.java | 213 ++++++++---------
.../opensearch/OpenSearchHealthCheckTest.java | 52 ++---
.../backends/opensearch/OpenSearchIndexerTest.java | 76 +++---
.../opensearch/search/ScrolledSearchTest.java | 135 ++++++-----
17 files changed, 690 insertions(+), 607 deletions(-)
diff --git a/backends-common/opensearch/pom.xml b/backends-common/opensearch/pom.xml
index ce40d7dca4..a53014a685 100644
--- a/backends-common/opensearch/pom.xml
+++ b/backends-common/opensearch/pom.xml
@@ -47,14 +47,10 @@
<artifactId>testing-base</artifactId>
<scope>test</scope>
</dependency>
-
- <!-- Prevents https://cve.mitre.org/cgi-bin/cvename.cgi?name=CVE-2020-28491 -->
<dependency>
- <groupId>com.fasterxml.jackson.dataformat</groupId>
- <artifactId>jackson-dataformat-cbor</artifactId>
- <version>2.13.4</version>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
</dependency>
-
<dependency>
<groupId>com.github.fge</groupId>
<artifactId>throwing-lambdas</artifactId>
@@ -86,7 +82,12 @@
</dependency>
<dependency>
<groupId>org.opensearch.client</groupId>
- <artifactId>opensearch-rest-high-level-client</artifactId>
+ <artifactId>opensearch-java</artifactId>
+ <version>2.0.0</version>
+ </dependency>
+ <dependency>
+ <groupId>org.opensearch.client</groupId>
+ <artifactId>opensearch-rest-client</artifactId>
<version>2.3.0</version>
</dependency>
<dependency>
diff --git a/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/ClientProvider.java b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/ClientProvider.java
index 52bf765741..89a104bb8b 100644
--- a/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/ClientProvider.java
+++ b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/ClientProvider.java
@@ -46,7 +46,9 @@ import org.apache.http.ssl.SSLContextBuilder;
import org.apache.http.ssl.TrustStrategy;
import org.apache.james.util.concurrent.NamedThreadFactory;
import org.opensearch.client.RestClient;
-import org.opensearch.client.RestHighLevelClient;
+import org.opensearch.client.json.jackson.JacksonJsonpMapper;
+import org.opensearch.client.opensearch.OpenSearchAsyncClient;
+import org.opensearch.client.transport.rest_client.RestClientTransport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -181,7 +183,8 @@ public class ClientProvider implements Provider<ReactorOpenSearchClient> {
private static final Logger LOGGER = LoggerFactory.getLogger(ClientProvider.class);
private final OpenSearchConfiguration configuration;
- private final RestHighLevelClient openSearchRestHighLevelClient;
+ private final RestClient lowLevelRestClient;
+ private final OpenSearchAsyncClient openSearchClient;
private final HttpAsyncClientConfigurer httpAsyncClientConfigurer;
private final ReactorOpenSearchClient client;
@@ -189,28 +192,35 @@ public class ClientProvider implements Provider<ReactorOpenSearchClient> {
public ClientProvider(OpenSearchConfiguration configuration) {
this.httpAsyncClientConfigurer = new HttpAsyncClientConfigurer(configuration);
this.configuration = configuration;
- this.openSearchRestHighLevelClient = connect(configuration);
- this.client = new ReactorOpenSearchClient(this.openSearchRestHighLevelClient);
+ this.lowLevelRestClient = buildRestClient();
+ this.openSearchClient = connect();
+ this.client = new ReactorOpenSearchClient(this.openSearchClient, lowLevelRestClient);
}
- private RestHighLevelClient connect(OpenSearchConfiguration configuration) {
+ private RestClient buildRestClient() {
+ return RestClient.builder(hostsToHttpHosts())
+ .setHttpClientConfigCallback(httpAsyncClientConfigurer::configure)
+ .build();
+ }
+
+ private OpenSearchAsyncClient connect() {
Duration waitDelay = Duration.ofMillis(configuration.getMinDelay());
boolean suppressLeadingZeroElements = true;
boolean suppressTrailingZeroElements = true;
return Mono.fromCallable(this::connectToCluster)
.doOnError(e -> LOGGER.warn("Error establishing OpenSearch connection. Next retry scheduled in {}",
DurationFormatUtils.formatDurationWords(waitDelay.toMillis(), suppressLeadingZeroElements, suppressTrailingZeroElements), e))
- .retryWhen(Retry.backoff(configuration.getMaxRetries(), waitDelay).scheduler(Schedulers.boundedElastic()))
+ .retryWhen(Retry.backoff(configuration.getMaxRetries(), waitDelay).scheduler(Schedulers.elastic()))
+ .publishOn(Schedulers.elastic())
.block();
}
- private RestHighLevelClient connectToCluster() {
+ private OpenSearchAsyncClient connectToCluster() {
LOGGER.info("Trying to connect to OpenSearch service at {}", LocalDateTime.now());
- return new RestHighLevelClient(
- RestClient
- .builder(hostsToHttpHosts())
- .setHttpClientConfigCallback(httpAsyncClientConfigurer::configure));
+ RestClientTransport transport = new RestClientTransport(lowLevelRestClient, new JacksonJsonpMapper());
+
+ return new OpenSearchAsyncClient(transport);
}
private HttpHost[] hostsToHttpHosts() {
@@ -226,6 +236,6 @@ public class ClientProvider implements Provider<ReactorOpenSearchClient> {
@PreDestroy
public void close() throws IOException {
- openSearchRestHighLevelClient.close();
+ lowLevelRestClient.close();
}
}
diff --git a/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/DeleteByQueryPerformer.java b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/DeleteByQueryPerformer.java
index a0261badd8..7ea40f627e 100644
--- a/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/DeleteByQueryPerformer.java
+++ b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/DeleteByQueryPerformer.java
@@ -19,9 +19,10 @@
package org.apache.james.backends.opensearch;
-import org.opensearch.client.RequestOptions;
-import org.opensearch.index.query.QueryBuilder;
-import org.opensearch.index.reindex.DeleteByQueryRequest;
+import java.io.IOException;
+
+import org.opensearch.client.opensearch._types.query_dsl.Query;
+import org.opensearch.client.opensearch.core.DeleteByQueryRequest;
import reactor.core.publisher.Mono;
@@ -35,20 +36,32 @@ public class DeleteByQueryPerformer {
this.aliasName = aliasName;
}
- public Mono<Void> perform(QueryBuilder queryBuilder, RoutingKey routingKey) {
- DeleteByQueryRequest deleteRequest = new DeleteByQueryRequest(aliasName.getValue());
- deleteRequest.setQuery(queryBuilder);
- deleteRequest.setRouting(routingKey.asString());
+ public Mono<Void> perform(Query query, RoutingKey routingKey) {
+ DeleteByQueryRequest deleteRequest = new DeleteByQueryRequest.Builder()
+ .index(aliasName.getValue())
+ .query(query)
+ .routing(routingKey.asString())
+ .build();
- return client.deleteByQuery(deleteRequest, RequestOptions.DEFAULT)
- .then();
+ try {
+ return client.deleteByQuery(deleteRequest)
+ .then();
+ } catch (IOException e) {
+ return Mono.error(e);
+ }
}
- public Mono<Void> perform(QueryBuilder queryBuilder) {
- DeleteByQueryRequest deleteRequest = new DeleteByQueryRequest(aliasName.getValue());
- deleteRequest.setQuery(queryBuilder);
+ public Mono<Void> perform(Query query) {
+ DeleteByQueryRequest deleteRequest = new DeleteByQueryRequest.Builder()
+ .index(aliasName.getValue())
+ .query(query)
+ .build();
- return client.deleteByQuery(deleteRequest, RequestOptions.DEFAULT)
- .then();
+ try {
+ return client.deleteByQuery(deleteRequest)
+ .then();
+ } catch (IOException e) {
+ return Mono.error(e);
+ }
}
}
diff --git a/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/DocumentId.java b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/DocumentId.java
index 42f650b03b..be9a994539 100644
--- a/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/DocumentId.java
+++ b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/DocumentId.java
@@ -21,9 +21,8 @@ package org.apache.james.backends.opensearch;
import java.util.Objects;
-import org.opensearch.common.Strings;
-
import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
public class DocumentId {
diff --git a/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/IndexCreationFactory.java b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/IndexCreationFactory.java
index 48756d104c..f314b373cc 100644
--- a/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/IndexCreationFactory.java
+++ b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/IndexCreationFactory.java
@@ -19,58 +19,76 @@
package org.apache.james.backends.opensearch;
-import static org.opensearch.action.admin.indices.alias.IndicesAliasesRequest.AliasActions;
-import static org.opensearch.common.xcontent.XContentFactory.jsonBuilder;
-
-import java.io.ByteArrayInputStream;
import java.io.IOException;
-import java.io.InputStream;
-import java.nio.charset.StandardCharsets;
import java.util.Collection;
+import java.util.Map;
import java.util.Optional;
+import java.util.stream.Collectors;
import javax.inject.Inject;
-import org.opensearch.OpenSearchStatusException;
-import org.opensearch.action.admin.indices.alias.IndicesAliasesRequest;
-import org.opensearch.action.admin.indices.alias.get.GetAliasesRequest;
-import org.opensearch.client.RequestOptions;
-import org.opensearch.client.indices.CreateIndexRequest;
-import org.opensearch.client.indices.GetIndexRequest;
-import org.opensearch.common.Strings;
-import org.opensearch.common.xcontent.XContentBuilder;
-import org.opensearch.common.xcontent.XContentType;
+import org.opensearch.client.opensearch._types.OpenSearchException;
+import org.opensearch.client.opensearch._types.WaitForActiveShards;
+import org.opensearch.client.opensearch._types.analysis.Analyzer;
+import org.opensearch.client.opensearch._types.analysis.CustomAnalyzer;
+import org.opensearch.client.opensearch._types.analysis.CustomNormalizer;
+import org.opensearch.client.opensearch._types.analysis.Normalizer;
+import org.opensearch.client.opensearch._types.analysis.SnowballLanguage;
+import org.opensearch.client.opensearch._types.analysis.SnowballTokenFilter;
+import org.opensearch.client.opensearch._types.analysis.Tokenizer;
+import org.opensearch.client.opensearch._types.mapping.TypeMapping;
+import org.opensearch.client.opensearch.indices.CreateIndexRequest;
+import org.opensearch.client.opensearch.indices.ExistsAliasRequest;
+import org.opensearch.client.opensearch.indices.ExistsRequest;
+import org.opensearch.client.opensearch.indices.IndexSettings;
+import org.opensearch.client.opensearch.indices.IndexSettingsAnalysis;
+import org.opensearch.client.opensearch.indices.UpdateAliasesRequest;
+import org.opensearch.client.opensearch.indices.update_aliases.Action;
+import org.opensearch.client.opensearch.indices.update_aliases.AddAction;
+import org.opensearch.client.transport.endpoints.BooleanResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
import com.github.fge.lambdas.Throwing;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
public class IndexCreationFactory {
- public static class IndexCreationCustomElement {
- public static IndexCreationCustomElement EMPTY = from("{}");
+ public static class IndexCreationCustomAnalyzer {
+ private final String key;
+ private final Analyzer analyzer;
- public static IndexCreationCustomElement from(String value) {
- try {
- new ObjectMapper().readTree(value);
- } catch (JsonProcessingException e) {
- throw new IllegalArgumentException("value must be a valid json");
- }
- return new IndexCreationCustomElement(value);
+ public IndexCreationCustomAnalyzer(String key, Analyzer analyzer) {
+ this.key = key;
+ this.analyzer = analyzer;
+ }
+
+ public String getKey() {
+ return key;
+ }
+
+ public Analyzer getAnalyzer() {
+ return analyzer;
}
+ }
- private final String payload;
+ public static class IndexCreationCustomTokenizer {
+ private final String key;
+ private final Tokenizer tokenizer;
- IndexCreationCustomElement(String payload) {
- this.payload = payload;
+ public IndexCreationCustomTokenizer(String key, Tokenizer tokenizer) {
+ this.key = key;
+ this.tokenizer = tokenizer;
}
- public String getPayload() {
- return payload;
+ public String getKey() {
+ return key;
+ }
+
+ public Tokenizer getTokenizer() {
+ return tokenizer;
}
}
@@ -103,8 +121,8 @@ public class IndexCreationFactory {
private final int waitForActiveShards;
private final IndexName indexName;
private final ImmutableList.Builder<AliasName> aliases;
- private Optional<IndexCreationCustomElement> customAnalyzers;
- private Optional<IndexCreationCustomElement> customTokenizers;
+ private final ImmutableList.Builder<IndexCreationCustomAnalyzer> customAnalyzers;
+ private final ImmutableList.Builder<IndexCreationCustomTokenizer> customTokenizers;
FinalStage(int nbShards, int nbReplica, int waitForActiveShards, IndexName indexName) {
this.nbShards = nbShards;
@@ -112,8 +130,8 @@ public class IndexCreationFactory {
this.waitForActiveShards = waitForActiveShards;
this.indexName = indexName;
this.aliases = ImmutableList.builder();
- this.customAnalyzers = Optional.empty();
- this.customTokenizers = Optional.empty();
+ this.customAnalyzers = ImmutableList.builder();
+ this.customTokenizers = ImmutableList.builder();
}
public FinalStage addAlias(AliasName... aliases) {
@@ -126,29 +144,29 @@ public class IndexCreationFactory {
return this;
}
- public FinalStage customAnalyzers(IndexCreationCustomElement customAnalyzers) {
- this.customAnalyzers = Optional.of(customAnalyzers);
+ public FinalStage customAnalyzers(IndexCreationCustomAnalyzer... customAnalyzers) {
+ this.customAnalyzers.add(customAnalyzers);
return this;
}
- public FinalStage customTokenizers(IndexCreationCustomElement customTokenizers) {
- this.customTokenizers = Optional.of(customTokenizers);
+ public FinalStage customTokenizers(IndexCreationCustomTokenizer... customTokenizers) {
+ this.customTokenizers.add(customTokenizers);
return this;
}
public IndexCreationPerformer build() {
- return new IndexCreationPerformer(nbShards, nbReplica, waitForActiveShards, indexName, aliases.build(), customAnalyzers, customTokenizers);
+ return new IndexCreationPerformer(nbShards, nbReplica, waitForActiveShards, indexName, aliases.build(), customAnalyzers.build(), customTokenizers.build());
}
public ReactorOpenSearchClient createIndexAndAliases(ReactorOpenSearchClient client) {
return build().createIndexAndAliases(client, Optional.empty(), Optional.empty());
}
- public ReactorOpenSearchClient createIndexAndAliases(ReactorOpenSearchClient client, XContentBuilder mappingContent) {
+ public ReactorOpenSearchClient createIndexAndAliases(ReactorOpenSearchClient client, TypeMapping mappingContent) {
return build().createIndexAndAliases(client, Optional.empty(), Optional.of(mappingContent));
}
- public ReactorOpenSearchClient createIndexAndAliases(ReactorOpenSearchClient client, Optional<XContentBuilder> indexSettings, Optional<XContentBuilder> mappingContent) {
+ public ReactorOpenSearchClient createIndexAndAliases(ReactorOpenSearchClient client, Optional<IndexSettings> indexSettings, Optional<TypeMapping> mappingContent) {
return build().createIndexAndAliases(client, indexSettings, mappingContent);
}
}
@@ -164,11 +182,12 @@ public class IndexCreationFactory {
private final int waitForActiveShards;
private final IndexName indexName;
private final ImmutableList<AliasName> aliases;
- private final Optional<IndexCreationCustomElement> customAnalyzers;
- private final Optional<IndexCreationCustomElement> customTokenizers;
+ private final ImmutableList<IndexCreationCustomAnalyzer> customAnalyzers;
+ private final ImmutableList<IndexCreationCustomTokenizer> customTokenizers;
- private IndexCreationPerformer(int nbShards, int nbReplica, int waitForActiveShards, IndexName indexName, ImmutableList<AliasName> aliases,
- Optional<IndexCreationCustomElement> customAnalyzers, Optional<IndexCreationCustomElement> customTokenizers) {
+ private IndexCreationPerformer(int nbShards, int nbReplica, int waitForActiveShards, IndexName indexName,
+ ImmutableList<AliasName> aliases, ImmutableList<IndexCreationCustomAnalyzer> customAnalyzers,
+ ImmutableList<IndexCreationCustomTokenizer> customTokenizers) {
this.nbShards = nbShards;
this.nbReplica = nbReplica;
this.waitForActiveShards = waitForActiveShards;
@@ -178,8 +197,9 @@ public class IndexCreationFactory {
this.customTokenizers = customTokenizers;
}
- public ReactorOpenSearchClient createIndexAndAliases(ReactorOpenSearchClient client, Optional<XContentBuilder> indexSettings,
- Optional<XContentBuilder> mappingContent) {
+ public ReactorOpenSearchClient createIndexAndAliases(ReactorOpenSearchClient client,
+ Optional<IndexSettings> indexSettings,
+ Optional<TypeMapping> mappingContent) {
Preconditions.checkNotNull(indexName);
try {
createIndexIfNeeded(client, indexName, indexSettings.orElse(generateSetting()), mappingContent);
@@ -193,31 +213,41 @@ public class IndexCreationFactory {
private void createAliasIfNeeded(ReactorOpenSearchClient client, IndexName indexName, AliasName aliasName) throws IOException {
if (!aliasExist(client, aliasName)) {
- client.indices()
- .updateAliases(
- new IndicesAliasesRequest().addAliasAction(
- new AliasActions(AliasActions.Type.ADD)
+ client.updateAliases(
+ new UpdateAliasesRequest.Builder()
+ .actions(new Action.Builder()
+ .add(new AddAction.Builder()
.index(indexName.getValue())
- .alias(aliasName.getValue())),
- RequestOptions.DEFAULT);
+ .alias(aliasName.getValue())
+ .build())
+ .build())
+ .build())
+ .block();
}
}
private boolean aliasExist(ReactorOpenSearchClient client, AliasName aliasName) throws IOException {
- return client.indices()
- .existsAlias(new GetAliasesRequest().aliases(aliasName.getValue()), RequestOptions.DEFAULT);
+ return client.aliasExists(new ExistsAliasRequest.Builder()
+ .name(aliasName.getValue())
+ .build())
+ .map(BooleanResponse::value)
+ .block();
}
- private void createIndexIfNeeded(ReactorOpenSearchClient client, IndexName indexName, XContentBuilder settings, Optional<XContentBuilder> mappingContent) throws IOException {
+ private void createIndexIfNeeded(ReactorOpenSearchClient client, IndexName indexName, IndexSettings settings, Optional<TypeMapping> mappingContent) throws IOException {
try {
if (!indexExists(client, indexName)) {
- CreateIndexRequest request = new CreateIndexRequest(indexName.getValue()).source(settings);
- mappingContent.ifPresent(request::mapping);
- client.indices().create(
- request,
- RequestOptions.DEFAULT);
+ CreateIndexRequest.Builder request = new CreateIndexRequest.Builder()
+ .index(indexName.getValue())
+ .waitForActiveShards(new WaitForActiveShards.Builder()
+ .count(waitForActiveShards)
+ .build())
+ .settings(settings);
+ mappingContent.ifPresent(request::mappings);
+ client.createIndex(request.build())
+ .block();
}
- } catch (OpenSearchStatusException exception) {
+ } catch (OpenSearchException exception) {
if (exception.getMessage().contains(INDEX_ALREADY_EXISTS_EXCEPTION_MESSAGE)) {
LOGGER.info("Index [{}] already exists", indexName.getValue());
} else {
@@ -227,60 +257,66 @@ public class IndexCreationFactory {
}
private boolean indexExists(ReactorOpenSearchClient client, IndexName indexName) throws IOException {
- return client.indices().exists(new GetIndexRequest(indexName.getValue()), RequestOptions.DEFAULT);
+ return client.indexExists(new ExistsRequest.Builder()
+ .index(indexName.getValue())
+ .build())
+ .map(BooleanResponse::value)
+ .block();
+ }
+
+ private IndexSettings generateSetting() {
+ return new IndexSettings.Builder()
+ .numberOfShards(Integer.toString(nbShards))
+ .numberOfReplicas(Integer.toString(nbReplica))
+ .analysis(new IndexSettingsAnalysis.Builder()
+ .normalizer(CASE_INSENSITIVE, new Normalizer.Builder()
+ .custom(generateNormalizer())
+ .build())
+ .analyzer(generateAnalyzers())
+ .tokenizer(generateTokenizers())
+ .build())
+ .build();
}
- private XContentBuilder generateSetting() throws IOException {
- return jsonBuilder()
- .startObject()
- .startObject("settings")
- .field("number_of_shards", nbShards)
- .field("number_of_replicas", nbReplica)
- .field("index.write.wait_for_active_shards", waitForActiveShards)
- .startObject("analysis")
- .startObject("normalizer")
- .startObject(CASE_INSENSITIVE)
- .field("type", "custom")
- .startArray("char_filter")
- .endArray()
- .startArray("filter")
- .value("lowercase")
- .value("asciifolding")
- .endArray()
- .endObject()
- .endObject()
- .rawField(ANALYZER, generateAnalyzers(), XContentType.JSON)
- .rawField(TOKENIZER, generateTokenizer(), XContentType.JSON)
- .endObject()
- .endObject()
- .endObject();
+ private CustomNormalizer generateNormalizer() {
+ return new CustomNormalizer.Builder()
+ .filter("lowercase", "asciifolding")
+ .build();
}
- private String analyzerDefault() throws IOException {
- XContentBuilder analyzerBuilder = jsonBuilder()
- .startObject()
- .startObject(KEEP_MAIL_AND_URL)
- .field("tokenizer", "uax_url_email")
- .startArray("filter")
- .value("lowercase")
- .value("stop")
- .endArray()
- .endObject()
- .endObject();
-
- return Strings.toString(analyzerBuilder);
+ private SnowballTokenFilter generateFilter() {
+ return new SnowballTokenFilter.Builder()
+ .language(SnowballLanguage.English)
+ .build();
}
- private InputStream generateAnalyzers() {
- return new ByteArrayInputStream(customAnalyzers.orElseGet(Throwing.supplier(() -> IndexCreationCustomElement.from(analyzerDefault())).sneakyThrow())
- .getPayload()
- .getBytes(StandardCharsets.UTF_8));
+ private Map<String, Analyzer> defaultAnalyzers() {
+ return ImmutableMap.of(
+ KEEP_MAIL_AND_URL, new Analyzer.Builder().custom(
+ new CustomAnalyzer.Builder()
+ .tokenizer("uax_url_email")
+ .filter("lowercase", "stop")
+ .build())
+ .build()
+ );
+ }
+
+ private Map<String, Analyzer> generateAnalyzers() {
+ if (customAnalyzers.isEmpty()) {
+ return defaultAnalyzers();
+ }
+ return customAnalyzers.stream()
+ .collect(Collectors.toMap(
+ IndexCreationCustomAnalyzer::getKey,
+ IndexCreationCustomAnalyzer::getAnalyzer));
}
- private InputStream generateTokenizer() {
- return new ByteArrayInputStream(customTokenizers.orElse(IndexCreationCustomElement.EMPTY)
- .getPayload()
- .getBytes(StandardCharsets.UTF_8));
+ private Map<String, Tokenizer> generateTokenizers() {
+ return customTokenizers.stream()
+ .collect(Collectors.toMap(
+ IndexCreationCustomTokenizer::getKey,
+ IndexCreationCustomTokenizer::getTokenizer
+ ));
}
}
diff --git a/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/OpenSearchHealthCheck.java b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/OpenSearchHealthCheck.java
index de26140daf..49874fc8f9 100644
--- a/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/OpenSearchHealthCheck.java
+++ b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/OpenSearchHealthCheck.java
@@ -19,7 +19,10 @@
package org.apache.james.backends.opensearch;
+import java.io.IOException;
+import java.util.List;
import java.util.Set;
+import java.util.stream.Collectors;
import javax.inject.Inject;
@@ -27,9 +30,8 @@ import org.apache.commons.lang3.NotImplementedException;
import org.apache.james.core.healthcheck.ComponentName;
import org.apache.james.core.healthcheck.HealthCheck;
import org.apache.james.core.healthcheck.Result;
-import org.opensearch.action.admin.cluster.health.ClusterHealthRequest;
-import org.opensearch.action.admin.cluster.health.ClusterHealthResponse;
-import org.opensearch.client.Requests;
+import org.opensearch.client.opensearch.cluster.HealthRequest;
+import org.opensearch.client.opensearch.cluster.HealthResponse;
import com.google.common.annotations.VisibleForTesting;
@@ -54,24 +56,30 @@ public class OpenSearchHealthCheck implements HealthCheck {
@Override
public Mono<Result> check() {
- String[] indices = indexNames.stream()
+ List<String> indices = indexNames.stream()
.map(IndexName::getValue)
- .toArray(String[]::new);
- ClusterHealthRequest request = Requests.clusterHealthRequest(indices);
+ .collect(Collectors.toList());
+ HealthRequest request = new HealthRequest.Builder()
+ .index(indices)
+ .build();
- return client.health(request)
- .map(this::toHealthCheckResult)
- .onErrorResume(e -> Mono.just(Result.unhealthy(COMPONENT_NAME, "Error while contacting cluster", e)));
+ try {
+ return client.health(request)
+ .map(this::toHealthCheckResult)
+ .onErrorResume(e -> Mono.just(Result.unhealthy(COMPONENT_NAME, "Error while contacting cluster", e)));
+ } catch (IOException e) {
+ return Mono.error(e);
+ }
}
@VisibleForTesting
- Result toHealthCheckResult(ClusterHealthResponse response) {
- switch (response.getStatus()) {
- case GREEN:
- case YELLOW:
+ Result toHealthCheckResult(HealthResponse response) {
+ switch (response.status()) {
+ case Green:
+ case Yellow:
return Result.healthy(COMPONENT_NAME);
- case RED:
- return Result.unhealthy(COMPONENT_NAME, response.getClusterName() + " status is RED");
+ case Red:
+ return Result.unhealthy(COMPONENT_NAME, response.clusterName() + " status is RED");
default:
throw new NotImplementedException("Un-handled OpenSearch cluster status");
}
diff --git a/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/OpenSearchIndexer.java b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/OpenSearchIndexer.java
index 9a0002cf91..19c46e0dac 100644
--- a/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/OpenSearchIndexer.java
+++ b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/OpenSearchIndexer.java
@@ -18,24 +18,23 @@
****************************************************************/
package org.apache.james.backends.opensearch;
+import java.io.IOException;
+import java.util.Collections;
import java.util.List;
import org.apache.commons.lang3.StringUtils;
-import org.opensearch.action.bulk.BulkRequest;
-import org.opensearch.action.bulk.BulkResponse;
-import org.opensearch.action.delete.DeleteRequest;
-import org.opensearch.action.get.GetRequest;
-import org.opensearch.action.get.GetResponse;
-import org.opensearch.action.index.IndexRequest;
-import org.opensearch.action.index.IndexResponse;
-import org.opensearch.action.update.UpdateRequest;
-import org.opensearch.client.RequestOptions;
-import org.opensearch.common.ValidationException;
-import org.opensearch.common.xcontent.XContentType;
-import org.opensearch.index.query.QueryBuilder;
+import org.opensearch.client.opensearch._types.query_dsl.Query;
+import org.opensearch.client.opensearch.core.BulkRequest;
+import org.opensearch.client.opensearch.core.BulkResponse;
+import org.opensearch.client.opensearch.core.GetRequest;
+import org.opensearch.client.opensearch.core.GetResponse;
+import org.opensearch.client.opensearch.core.IndexRequest;
+import org.opensearch.client.opensearch.core.IndexResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.fasterxml.jackson.databind.util.RawValue;
import com.google.common.base.Preconditions;
import reactor.core.publisher.Mono;
@@ -59,11 +58,17 @@ public class OpenSearchIndexer {
public Mono<IndexResponse> index(DocumentId id, String content, RoutingKey routingKey) {
checkArgument(content);
logContent(id, content);
- return client.index(new IndexRequest(aliasName.getValue())
+
+ try {
+ return client.index(new IndexRequest.Builder<>()
+ .index(aliasName.getValue())
.id(id.asString())
- .source(content, XContentType.JSON)
- .routing(routingKey.asString()),
- RequestOptions.DEFAULT);
+ .document(new RawValue(content))
+ .routing(routingKey.asString())
+ .build());
+ } catch (IOException e) {
+ return Mono.error(e);
+ }
}
private void logContent(DocumentId id, String content) {
@@ -75,50 +80,70 @@ public class OpenSearchIndexer {
public Mono<BulkResponse> update(List<UpdatedRepresentation> updatedDocumentParts, RoutingKey routingKey) {
Preconditions.checkNotNull(updatedDocumentParts);
Preconditions.checkNotNull(routingKey);
- BulkRequest request = new BulkRequest();
- updatedDocumentParts.forEach(updatedDocumentPart -> request.add(
- new UpdateRequest(aliasName.getValue(),
- updatedDocumentPart.getId().asString())
- .doc(updatedDocumentPart.getUpdatedDocumentPart(), XContentType.JSON)
- .routing(routingKey.asString())));
-
- return client.bulk(request, RequestOptions.DEFAULT)
- .onErrorResume(ValidationException.class, exception -> {
- LOGGER.warn("Error while updating index", exception);
- return Mono.empty();
- });
+
+ if (updatedDocumentParts.isEmpty()) {
+ return Mono.empty();
+ }
+
+ BulkRequest.Builder bulkBuilder = new BulkRequest.Builder();
+ updatedDocumentParts.forEach(updatedDocumentPart -> bulkBuilder.operations(
+ op -> op.update(idx -> idx
+ .index(aliasName.getValue())
+ .id(updatedDocumentPart.getId().asString())
+ .document(Collections.singletonMap("doc", new RawValue(updatedDocumentPart.getUpdatedDocumentPart())))
+ .routing(routingKey.asString())
+ )));
+
+ try {
+ return client.bulk(bulkBuilder.build());
+ } catch (IOException e) {
+ return Mono.error(e);
+ }
}
public Mono<BulkResponse> delete(List<DocumentId> ids, RoutingKey routingKey) {
- BulkRequest request = new BulkRequest();
- ids.forEach(id -> request.add(
- new DeleteRequest(aliasName.getValue())
+ if (ids.isEmpty()) {
+ return Mono.empty();
+ }
+
+ BulkRequest.Builder bulkBuilder = new BulkRequest.Builder();
+
+ ids.forEach(id -> bulkBuilder.operations(
+ op -> op.delete(idx -> idx
+ .index(aliasName.getValue())
.id(id.asString())
- .routing(routingKey.asString())));
+ .routing(routingKey.asString())
+ )));
- return client.bulk(request, RequestOptions.DEFAULT)
- .onErrorResume(ValidationException.class, exception -> {
- LOGGER.warn("Error while deleting index", exception);
- return Mono.empty();
- });
+ try {
+ return client.bulk(bulkBuilder.build());
+ } catch (IOException e) {
+ return Mono.error(e);
+ }
}
- public Mono<Void> deleteAllMatchingQuery(QueryBuilder queryBuilder, RoutingKey routingKey) {
- return deleteByQueryPerformer.perform(queryBuilder, routingKey);
+ public Mono<Void> deleteAllMatchingQuery(Query query, RoutingKey routingKey) {
+ return deleteByQueryPerformer.perform(query, routingKey);
}
private void checkArgument(String content) {
Preconditions.checkArgument(content != null, "content should be provided");
}
- public Mono<GetResponse> get(DocumentId id, RoutingKey routingKey) {
- return Mono.fromRunnable(() -> {
- Preconditions.checkNotNull(id);
- Preconditions.checkNotNull(routingKey);
- })
- .then(client.get(new GetRequest(aliasName.getValue())
- .id(id.asString())
- .routing(routingKey.asString()),
- RequestOptions.DEFAULT));
+ public Mono<GetResponse<ObjectNode>> get(DocumentId id, RoutingKey routingKey) {
+ try {
+ return Mono.fromRunnable(() -> {
+ Preconditions.checkNotNull(id);
+ Preconditions.checkNotNull(routingKey);
+ })
+ .then(client.get(
+ new GetRequest.Builder()
+ .index(aliasName.getValue())
+ .id(id.asString())
+ .routing(routingKey.asString())
+ .build()));
+ } catch (IOException e) {
+ return Mono.error(e);
+ }
}
}
diff --git a/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/ReactorOpenSearchClient.java b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/ReactorOpenSearchClient.java
index 1c739fcf37..d2818e2ad3 100644
--- a/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/ReactorOpenSearchClient.java
+++ b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/ReactorOpenSearchClient.java
@@ -20,160 +20,129 @@
package org.apache.james.backends.opensearch;
import java.io.IOException;
-import java.util.function.Consumer;
-
-import org.opensearch.action.ActionListener;
-import org.opensearch.action.admin.cluster.health.ClusterHealthRequest;
-import org.opensearch.action.admin.cluster.health.ClusterHealthResponse;
-import org.opensearch.action.admin.cluster.storedscripts.DeleteStoredScriptRequest;
-import org.opensearch.action.admin.cluster.storedscripts.GetStoredScriptRequest;
-import org.opensearch.action.admin.cluster.storedscripts.GetStoredScriptResponse;
-import org.opensearch.action.bulk.BulkRequest;
-import org.opensearch.action.bulk.BulkResponse;
-import org.opensearch.action.delete.DeleteRequest;
-import org.opensearch.action.delete.DeleteResponse;
-import org.opensearch.action.explain.ExplainRequest;
-import org.opensearch.action.explain.ExplainResponse;
-import org.opensearch.action.fieldcaps.FieldCapabilitiesRequest;
-import org.opensearch.action.fieldcaps.FieldCapabilitiesResponse;
-import org.opensearch.action.get.GetRequest;
-import org.opensearch.action.get.GetResponse;
-import org.opensearch.action.index.IndexRequest;
-import org.opensearch.action.index.IndexResponse;
-import org.opensearch.action.search.ClearScrollRequest;
-import org.opensearch.action.search.ClearScrollResponse;
-import org.opensearch.action.search.MultiSearchRequest;
-import org.opensearch.action.search.MultiSearchResponse;
-import org.opensearch.action.search.SearchRequest;
-import org.opensearch.action.search.SearchResponse;
-import org.opensearch.action.search.SearchScrollRequest;
-import org.opensearch.action.support.master.AcknowledgedResponse;
-import org.opensearch.client.IndicesClient;
-import org.opensearch.client.RequestOptions;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.BiConsumer;
+
import org.opensearch.client.RestClient;
-import org.opensearch.client.RestHighLevelClient;
-import org.opensearch.client.core.MainResponse;
-import org.opensearch.index.rankeval.RankEvalRequest;
-import org.opensearch.index.rankeval.RankEvalResponse;
-import org.opensearch.index.reindex.BulkByScrollResponse;
-import org.opensearch.index.reindex.DeleteByQueryRequest;
-import org.opensearch.script.mustache.MultiSearchTemplateRequest;
-import org.opensearch.script.mustache.MultiSearchTemplateResponse;
-import org.opensearch.script.mustache.SearchTemplateRequest;
-import org.opensearch.script.mustache.SearchTemplateResponse;
+import org.opensearch.client.opensearch.OpenSearchAsyncClient;
+import org.opensearch.client.opensearch.cluster.HealthRequest;
+import org.opensearch.client.opensearch.cluster.HealthResponse;
+import org.opensearch.client.opensearch.core.BulkRequest;
+import org.opensearch.client.opensearch.core.BulkResponse;
+import org.opensearch.client.opensearch.core.ClearScrollRequest;
+import org.opensearch.client.opensearch.core.ClearScrollResponse;
+import org.opensearch.client.opensearch.core.DeleteByQueryRequest;
+import org.opensearch.client.opensearch.core.DeleteByQueryResponse;
+import org.opensearch.client.opensearch.core.DeleteRequest;
+import org.opensearch.client.opensearch.core.DeleteResponse;
+import org.opensearch.client.opensearch.core.GetRequest;
+import org.opensearch.client.opensearch.core.GetResponse;
+import org.opensearch.client.opensearch.core.IndexRequest;
+import org.opensearch.client.opensearch.core.IndexResponse;
+import org.opensearch.client.opensearch.core.InfoResponse;
+import org.opensearch.client.opensearch.core.ScrollRequest;
+import org.opensearch.client.opensearch.core.ScrollResponse;
+import org.opensearch.client.opensearch.core.SearchRequest;
+import org.opensearch.client.opensearch.core.SearchResponse;
+import org.opensearch.client.opensearch.indices.CreateIndexRequest;
+import org.opensearch.client.opensearch.indices.CreateIndexResponse;
+import org.opensearch.client.opensearch.indices.ExistsAliasRequest;
+import org.opensearch.client.opensearch.indices.ExistsRequest;
+import org.opensearch.client.opensearch.indices.UpdateAliasesRequest;
+import org.opensearch.client.opensearch.indices.UpdateAliasesResponse;
+import org.opensearch.client.transport.endpoints.BooleanResponse;
+
+import com.fasterxml.jackson.databind.node.ObjectNode;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoSink;
import reactor.core.scheduler.Schedulers;
public class ReactorOpenSearchClient implements AutoCloseable {
- private final RestHighLevelClient client;
+ private final OpenSearchAsyncClient client;
+ private final RestClient lowLevelRestClient;
- public ReactorOpenSearchClient(RestHighLevelClient client) {
+ public ReactorOpenSearchClient(OpenSearchAsyncClient client, RestClient lowLevelRestClient) {
this.client = client;
+ this.lowLevelRestClient = lowLevelRestClient;
}
- public Mono<BulkResponse> bulk(BulkRequest bulkRequest, RequestOptions options) {
- return toReactor(listener -> client.bulkAsync(bulkRequest, options, listener));
- }
-
- public Mono<ClearScrollResponse> clearScroll(ClearScrollRequest clearScrollRequest, RequestOptions options) {
- return toReactor(listener -> client.clearScrollAsync(clearScrollRequest, options, listener));
- }
-
- public DeleteResponse delete(DeleteRequest deleteRequest, RequestOptions options) throws IOException {
- return client.delete(deleteRequest, options);
- }
-
- public Mono<BulkByScrollResponse> deleteByQuery(DeleteByQueryRequest deleteRequest, RequestOptions options) {
- return toReactor(listener -> client.deleteByQueryAsync(deleteRequest, options, listener));
+ public Mono<BulkResponse> bulk(BulkRequest bulkRequest) throws IOException {
+ return toReactor(client.bulk(bulkRequest));
}
- public Mono<AcknowledgedResponse> deleteScript(DeleteStoredScriptRequest request, RequestOptions options) {
- return toReactor(listener -> client.deleteScriptAsync(request, options, listener));
+ public Mono<ClearScrollResponse> clearScroll(ClearScrollRequest clearScrollRequest) throws IOException {
+ return toReactor(client.clearScroll(clearScrollRequest));
}
- public Mono<ExplainResponse> explain(ExplainRequest explainRequest, RequestOptions options) {
- return toReactor(listener -> client.explainAsync(explainRequest, options, listener));
+ public Mono<DeleteResponse> delete(DeleteRequest deleteRequest) throws IOException {
+ return toReactor(client.delete(deleteRequest));
}
- public Mono<FieldCapabilitiesResponse> fieldCaps(FieldCapabilitiesRequest fieldCapabilitiesRequest, RequestOptions options) {
- return toReactor(listener -> client.fieldCapsAsync(fieldCapabilitiesRequest, options, listener));
+ public Mono<DeleteByQueryResponse> deleteByQuery(DeleteByQueryRequest deleteRequest) throws IOException {
+ return toReactor(client.deleteByQuery(deleteRequest));
}
public RestClient getLowLevelClient() {
- return client.getLowLevelClient();
+ return lowLevelRestClient;
}
- public Mono<GetStoredScriptResponse> getScript(GetStoredScriptRequest request, RequestOptions options) {
- return toReactor(listener -> client.getScriptAsync(request, options, listener));
+ public <T> Mono<IndexResponse> index(IndexRequest<T> indexRequest) throws IOException {
+ return toReactor(client.index(indexRequest));
}
- public Mono<IndexResponse> index(IndexRequest indexRequest, RequestOptions options) {
- return toReactor(listener -> client.indexAsync(indexRequest, options, listener));
+ public Mono<BooleanResponse> indexExists(ExistsRequest existsRequest) throws IOException {
+ return toReactor(client.indices().exists(existsRequest));
}
- public IndicesClient indices() {
- return client.indices();
+ public Mono<BooleanResponse> aliasExists(ExistsAliasRequest existsAliasRequest) throws IOException {
+ return toReactor(client.indices().existsAlias(existsAliasRequest));
}
- public MainResponse info(RequestOptions options) throws IOException {
- return client.info(options);
+ public Mono<CreateIndexResponse> createIndex(CreateIndexRequest indexRequest) throws IOException {
+ return toReactor(client.indices().create(indexRequest));
}
- public Mono<MultiSearchResponse> msearch(MultiSearchRequest multiSearchRequest, RequestOptions options) {
- return toReactor(listener -> client.msearchAsync(multiSearchRequest, options, listener));
+ public Mono<UpdateAliasesResponse> updateAliases(UpdateAliasesRequest updateAliasesRequest) throws IOException {
+ return toReactor(client.indices().updateAliases(updateAliasesRequest));
}
- public Mono<MultiSearchTemplateResponse> msearchTemplate(MultiSearchTemplateRequest multiSearchTemplateRequest, RequestOptions options) {
- return toReactor(listener -> client.msearchTemplateAsync(multiSearchTemplateRequest, options, listener));
+ public Mono<InfoResponse> info() throws IOException {
+ return toReactor(client.info());
}
- public Mono<RankEvalResponse> rankEval(RankEvalRequest rankEvalRequest, RequestOptions options) {
- return toReactor(listener -> client.rankEvalAsync(rankEvalRequest, options, listener));
+ public Mono<ScrollResponse<ObjectNode>> scroll(ScrollRequest scrollRequest) throws IOException {
+ return toReactor(client.scroll(scrollRequest, ObjectNode.class));
}
- public Mono<SearchResponse> scroll(SearchScrollRequest searchScrollRequest, RequestOptions options) {
- return toReactor(listener -> client.scrollAsync(searchScrollRequest, options, listener));
+ public Mono<SearchResponse<ObjectNode>> search(SearchRequest searchRequest) throws IOException {
+ return toReactor(client.search(searchRequest, ObjectNode.class));
}
- public Mono<SearchResponse> search(SearchRequest searchRequest, RequestOptions options) {
- return toReactor(listener -> client.searchAsync(searchRequest, options, listener));
+ public Mono<HealthResponse> health(HealthRequest request) throws IOException {
+ return toReactor(client.cluster().health(request));
}
- public Mono<ClusterHealthResponse> health(ClusterHealthRequest request) {
- return toReactor(listener -> client.cluster()
- .healthAsync(request, RequestOptions.DEFAULT, listener));
- }
-
- public Mono<SearchTemplateResponse> searchTemplate(SearchTemplateRequest searchTemplateRequest, RequestOptions options) {
- return toReactor(listener -> client.searchTemplateAsync(searchTemplateRequest, options, listener));
- }
-
- public Mono<GetResponse> get(GetRequest getRequest, RequestOptions options) {
- return toReactor(listener -> client.getAsync(getRequest, options, listener));
+ public Mono<GetResponse<ObjectNode>> get(GetRequest getRequest) throws IOException {
+ return toReactor(client.get(getRequest, ObjectNode.class));
}
@Override
public void close() throws IOException {
- client.close();
+ lowLevelRestClient.close();
}
- private static <T> Mono<T> toReactor(Consumer<ActionListener<T>> async) {
- return Mono.<T>create(sink -> async.accept(getListener(sink)))
- .publishOn(Schedulers.boundedElastic());
+ private static <T> Mono<T> toReactor(CompletableFuture<T> async) {
+ return Mono.<T>create(sink -> async.whenComplete(getFuture(sink)))
+ .publishOn(Schedulers.elastic());
}
- private static <T> ActionListener<T> getListener(MonoSink<T> sink) {
- return new ActionListener<T>() {
- @Override
- public void onResponse(T t) {
- sink.success(t);
- }
-
- @Override
- public void onFailure(Exception e) {
- sink.error(e);
+ private static <T> BiConsumer<? super T, ? super Throwable> getFuture(MonoSink<T> sink) {
+ return (response, exception) -> {
+ if (exception != null) {
+ sink.error(exception);
+ } else {
+ sink.success(response);
}
};
}
diff --git a/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/RoutingKey.java b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/RoutingKey.java
index bb823f8e9f..4dedbc67fe 100644
--- a/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/RoutingKey.java
+++ b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/RoutingKey.java
@@ -21,9 +21,8 @@ package org.apache.james.backends.opensearch;
import java.util.Objects;
-import org.opensearch.common.Strings;
-
import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
public class RoutingKey {
public interface Factory<T> {
diff --git a/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/UpdatedRepresentation.java b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/UpdatedRepresentation.java
index f221ccd45c..76c415b9e1 100644
--- a/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/UpdatedRepresentation.java
+++ b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/UpdatedRepresentation.java
@@ -20,10 +20,9 @@ package org.apache.james.backends.opensearch;
import java.util.Objects;
-import org.opensearch.common.Strings;
-
import com.google.common.base.MoreObjects;
import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
public class UpdatedRepresentation {
private final DocumentId id;
diff --git a/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/search/ScrolledSearch.java b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/search/ScrolledSearch.java
index ecb9cf955a..3a5d22d1ce 100644
--- a/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/search/ScrolledSearch.java
+++ b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/search/ScrolledSearch.java
@@ -19,19 +19,20 @@
package org.apache.james.backends.opensearch.search;
+import java.io.IOException;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import org.apache.james.backends.opensearch.ReactorOpenSearchClient;
-import org.opensearch.action.search.ClearScrollRequest;
-import org.opensearch.action.search.SearchRequest;
-import org.opensearch.action.search.SearchResponse;
-import org.opensearch.action.search.SearchScrollRequest;
-import org.opensearch.client.RequestOptions;
-import org.opensearch.common.unit.TimeValue;
-import org.opensearch.search.SearchHit;
-
+import org.opensearch.client.opensearch._types.Time;
+import org.opensearch.client.opensearch.core.ClearScrollRequest;
+import org.opensearch.client.opensearch.core.ScrollRequest;
+import org.opensearch.client.opensearch.core.ScrollResponse;
+import org.opensearch.client.opensearch.core.SearchRequest;
+import org.opensearch.client.opensearch.core.search.Hit;
+
+import com.fasterxml.jackson.databind.node.ObjectNode;
import com.github.fge.lambdas.Throwing;
import reactor.core.publisher.Flux;
@@ -39,8 +40,9 @@ import reactor.core.publisher.FluxSink;
import reactor.core.publisher.Mono;
public class ScrolledSearch {
-
- private static final TimeValue TIMEOUT = TimeValue.timeValueMinutes(1);
+ private static final Time TIMEOUT = new Time.Builder()
+ .time("1m")
+ .build();
private final ReactorOpenSearchClient client;
private final SearchRequest searchRequest;
@@ -50,12 +52,12 @@ public class ScrolledSearch {
this.searchRequest = searchRequest;
}
- public Flux<SearchHit> searchHits() {
+ public Flux<Hit<ObjectNode>> searchHits() {
return searchResponses()
- .concatMap(searchResponse -> Flux.just(searchResponse.getHits().getHits()));
+ .concatMap(searchResponse -> Flux.fromIterable(searchResponse.hits().hits()));
}
- public Flux<SearchResponse> searchResponses() {
+ private Flux<ScrollResponse<ObjectNode>> searchResponses() {
return Flux.push(sink -> {
AtomicReference<Optional<String>> scrollId = new AtomicReference<>(Optional.empty());
sink.onRequest(numberOfRequestedElements -> next(sink, scrollId, numberOfRequestedElements));
@@ -64,17 +66,16 @@ public class ScrolledSearch {
});
}
- private void next(FluxSink<SearchResponse> sink, AtomicReference<Optional<String>> scrollId, long numberOfRequestedElements) {
+ private void next(FluxSink<ScrollResponse<ObjectNode>> sink, AtomicReference<Optional<String>> scrollId, long numberOfRequestedElements) {
if (numberOfRequestedElements <= 0) {
return;
}
- Consumer<SearchResponse> onResponse = searchResponse -> {
- scrollId.set(Optional.of(searchResponse.getScrollId()));
+ Consumer<ScrollResponse<ObjectNode>> onResponse = searchResponse -> {
+ scrollId.set(Optional.of(searchResponse.scrollId()));
sink.next(searchResponse);
- boolean noHitsLeft = searchResponse.getHits().getHits().length == 0;
- if (noHitsLeft) {
+ if (searchResponse.hits().hits().isEmpty()) {
sink.complete();
} else {
next(sink, scrollId, numberOfRequestedElements - 1);
@@ -87,22 +88,33 @@ public class ScrolledSearch {
.subscribe(onResponse, onFailure);
}
- private Mono<SearchResponse> buildRequest(Optional<String> scrollId) {
- return scrollId.map(id ->
- client.scroll(
- new SearchScrollRequest()
- .scrollId(scrollId.get())
- .scroll(TIMEOUT),
- RequestOptions.DEFAULT))
- .orElseGet(() -> client.search(searchRequest, RequestOptions.DEFAULT));
+ private Mono<ScrollResponse<ObjectNode>> buildRequest(Optional<String> scrollId) {
+ return scrollId.map(Throwing.function(id -> client.scroll(new ScrollRequest.Builder()
+ .scrollId(scrollId.get())
+ .scroll(TIMEOUT)
+ .build())).sneakyThrow())
+ .orElseGet(() -> {
+ try {
+ return client.search(searchRequest)
+ .map(response -> new ScrollResponse.Builder<ObjectNode>()
+ .scrollId(response.scrollId())
+ .hits(response.hits())
+ .took(response.took())
+ .timedOut(response.timedOut())
+ .shards(response.shards())
+ .build());
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ });
}
public void close(AtomicReference<Optional<String>> scrollId) {
- scrollId.get().map(id -> {
- ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
- clearScrollRequest.addScrollId(id);
- return clearScrollRequest;
- }).ifPresent(Throwing.<ClearScrollRequest>consumer(clearScrollRequest -> client.clearScroll(clearScrollRequest, RequestOptions.DEFAULT).subscribe()).sneakyThrow());
+ scrollId.get().map(id -> new ClearScrollRequest.Builder()
+ .scrollId(id)
+ .build())
+ .ifPresent(Throwing.<ClearScrollRequest>consumer(clearScrollRequest ->
+ client.clearScroll(clearScrollRequest).subscribe()).sneakyThrow());
}
}
diff --git a/backends-common/opensearch/src/test/java/org/apache/james/backends/opensearch/ClientProviderImplConnectionContract.java b/backends-common/opensearch/src/test/java/org/apache/james/backends/opensearch/ClientProviderImplConnectionContract.java
index 9434324172..9f89156f42 100644
--- a/backends-common/opensearch/src/test/java/org/apache/james/backends/opensearch/ClientProviderImplConnectionContract.java
+++ b/backends-common/opensearch/src/test/java/org/apache/james/backends/opensearch/ClientProviderImplConnectionContract.java
@@ -24,10 +24,8 @@ import java.util.concurrent.TimeUnit;
import org.apache.james.backends.opensearch.OpenSearchClusterExtension.OpenSearchCluster;
import org.awaitility.Awaitility;
import org.junit.jupiter.api.Test;
-import org.opensearch.action.search.SearchRequest;
-import org.opensearch.client.RequestOptions;
-import org.opensearch.index.query.QueryBuilders;
-import org.opensearch.search.builder.SearchSourceBuilder;
+import org.opensearch.client.opensearch._types.query_dsl.MatchAllQuery;
+import org.opensearch.client.opensearch.core.SearchRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -76,9 +74,9 @@ interface ClientProviderImplConnectionContract {
default boolean isConnected(ClientProvider clientProvider) {
try (ReactorOpenSearchClient client = clientProvider.get()) {
client.search(
- new SearchRequest()
- .source(new SearchSourceBuilder().query(QueryBuilders.existsQuery("any"))),
- RequestOptions.DEFAULT).block();
+ new SearchRequest.Builder()
+ .query(new MatchAllQuery.Builder().build()._toQuery())
+ .build()).block();
return true;
} catch (Exception e) {
LOGGER.info("Caught exception while trying to connect", e);
diff --git a/backends-common/opensearch/src/test/java/org/apache/james/backends/opensearch/DockerOpenSearchExtension.java b/backends-common/opensearch/src/test/java/org/apache/james/backends/opensearch/DockerOpenSearchExtension.java
index 34a105b5e6..021a60d790 100644
--- a/backends-common/opensearch/src/test/java/org/apache/james/backends/opensearch/DockerOpenSearchExtension.java
+++ b/backends-common/opensearch/src/test/java/org/apache/james/backends/opensearch/DockerOpenSearchExtension.java
@@ -19,18 +19,14 @@
package org.apache.james.backends.opensearch;
-import java.time.Duration;
-
import org.junit.jupiter.api.extension.AfterEachCallback;
import org.junit.jupiter.api.extension.BeforeEachCallback;
import org.junit.jupiter.api.extension.ExtensionContext;
import org.junit.jupiter.api.extension.ParameterContext;
import org.junit.jupiter.api.extension.ParameterResolutionException;
import org.junit.jupiter.api.extension.ParameterResolver;
-import org.opensearch.action.search.SearchRequest;
-import org.opensearch.client.RequestOptions;
-import org.opensearch.index.query.QueryBuilders;
-import org.opensearch.search.builder.SearchSourceBuilder;
+import org.opensearch.client.opensearch._types.query_dsl.MatchAllQuery;
+import org.opensearch.client.opensearch.core.SearchRequest;
import org.testcontainers.shaded.org.awaitility.Awaitility;
public class DockerOpenSearchExtension implements AfterEachCallback, BeforeEachCallback, ParameterResolver {
@@ -57,20 +53,20 @@ public class DockerOpenSearchExtension implements AfterEachCallback, BeforeEachC
}
@Override
- public void clean(DockerOpenSearch elasticSearch) {
+ public void clean(DockerOpenSearch openSearch) {
Awaitility.await()
.until(() -> {
- elasticSearch.flushIndices();
- ReactorOpenSearchClient client = elasticSearch.clientProvider().get();
+ openSearch.flushIndices();
+ ReactorOpenSearchClient client = openSearch.clientProvider().get();
new DeleteByQueryPerformer(client, aliasName)
- .perform(QueryBuilders.matchAllQuery())
+ .perform(new MatchAllQuery.Builder().build()._toQuery())
.block();
- SearchRequest searchRequest = new SearchRequest();
- searchRequest.source(new SearchSourceBuilder()
- .query(QueryBuilders.matchAllQuery()));
- elasticSearch.flushIndices();
- boolean result = client.search(new SearchRequest(searchRequest), RequestOptions.DEFAULT)
- .map(searchResponse -> searchResponse.getHits().getHits().length)
+ SearchRequest searchRequest = new SearchRequest.Builder()
+ .query(new MatchAllQuery.Builder().build()._toQuery())
+ .build();
+ openSearch.flushIndices();
+ boolean result = client.search(searchRequest)
+ .map(searchResponse -> searchResponse.hits().hits().size())
.block() == 0;
try {
@@ -83,7 +79,7 @@ public class DockerOpenSearchExtension implements AfterEachCallback, BeforeEachC
}
}
- private final DockerOpenSearch elasticSearch = DockerOpenSearchSingleton.INSTANCE;
+ private final DockerOpenSearch openSearch = DockerOpenSearchSingleton.INSTANCE;
private final CleanupStrategy cleanupStrategy;
public DockerOpenSearchExtension() {
@@ -96,13 +92,13 @@ public class DockerOpenSearchExtension implements AfterEachCallback, BeforeEachC
@Override
public void afterEach(ExtensionContext context) {
- cleanupStrategy.clean(elasticSearch);
+ cleanupStrategy.clean(openSearch);
}
@Override
public void beforeEach(ExtensionContext extensionContext) {
- if (!elasticSearch.isRunning()) {
- elasticSearch.unpause();
+ if (!openSearch.isRunning()) {
+ openSearch.unpause();
}
awaitForOpenSearch();
}
@@ -114,14 +110,14 @@ public class DockerOpenSearchExtension implements AfterEachCallback, BeforeEachC
@Override
public Object resolveParameter(ParameterContext parameterContext, ExtensionContext extensionContext) throws ParameterResolutionException {
- return elasticSearch;
+ return openSearch;
}
public void awaitForOpenSearch() {
- elasticSearch.flushIndices();
+ openSearch.flushIndices();
}
public DockerOpenSearch getDockerOpenSearch() {
- return elasticSearch;
+ return openSearch;
}
}
diff --git a/backends-common/opensearch/src/test/java/org/apache/james/backends/opensearch/IndexCreationFactoryTest.java b/backends-common/opensearch/src/test/java/org/apache/james/backends/opensearch/IndexCreationFactoryTest.java
index 8cda09676f..2b169aaaa5 100644
--- a/backends-common/opensearch/src/test/java/org/apache/james/backends/opensearch/IndexCreationFactoryTest.java
+++ b/backends-common/opensearch/src/test/java/org/apache/james/backends/opensearch/IndexCreationFactoryTest.java
@@ -19,78 +19,77 @@
package org.apache.james.backends.opensearch;
-import static org.apache.james.backends.opensearch.IndexCreationFactory.ANALYZER;
-import static org.apache.james.backends.opensearch.IndexCreationFactory.TOKENIZER;
-import static org.apache.james.backends.opensearch.IndexCreationFactory.TYPE;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
-import static org.opensearch.common.xcontent.XContentFactory.jsonBuilder;
import java.io.IOException;
import java.util.Optional;
-import org.apache.james.backends.opensearch.IndexCreationFactory.IndexCreationCustomElement;
+import org.apache.james.backends.opensearch.IndexCreationFactory.IndexCreationCustomAnalyzer;
+import org.apache.james.backends.opensearch.IndexCreationFactory.IndexCreationCustomTokenizer;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
-import org.opensearch.OpenSearchStatusException;
-import org.opensearch.common.xcontent.XContentBuilder;
+import org.opensearch.client.opensearch._types.analysis.Analyzer;
+import org.opensearch.client.opensearch._types.analysis.CustomAnalyzer;
+import org.opensearch.client.opensearch._types.analysis.NGramTokenFilter;
+import org.opensearch.client.opensearch._types.analysis.PatternTokenizer;
+import org.opensearch.client.opensearch._types.analysis.TokenFilter;
+import org.opensearch.client.opensearch._types.analysis.TokenFilterDefinition;
+import org.opensearch.client.opensearch._types.analysis.Tokenizer;
+import org.opensearch.client.opensearch._types.analysis.TokenizerDefinition;
+import org.opensearch.client.opensearch.indices.IndexSettings;
+import org.opensearch.client.opensearch.indices.IndexSettingsAnalysis;
class IndexCreationFactoryTest {
private static final IndexName INDEX_NAME = new IndexName("index");
private static final ReadAliasName ALIAS_NAME = new ReadAliasName("alias");
- public static XContentBuilder getValidIndexSetting() throws IOException {
- return jsonBuilder()
- .startObject()
- .startObject("settings")
- .startObject("index")
- .field("max_ngram_diff", 10)
- .endObject()
- .startObject("analysis")
- .startObject(ANALYZER)
- .startObject("email_ngram_filter_analyzer")
- .field(TOKENIZER, "uax_url_email")
- .startArray("filter")
- .value("ngram_filter")
- .endArray()
- .endObject()
- .endObject()
- .startObject("filter")
- .startObject("ngram_filter")
- .field(TYPE, "ngram")
- .field("min_gram", 3)
- .field("max_gram", 13)
- .endObject()
- .endObject()
- .endObject()
- .endObject()
- .endObject();
+ public static IndexSettings getValidIndexSetting() {
+ return new IndexSettings.Builder()
+ .index(new IndexSettings.Builder()
+ .maxNgramDiff(10)
+ .build())
+ .analysis(new IndexSettingsAnalysis.Builder()
+ .analyzer("email_ngram_filter_analyzer", new Analyzer.Builder()
+ .custom(generateAnalyzer())
+ .build())
+ .filter("ngram_filter", new TokenFilter.Builder()
+ .definition(new TokenFilterDefinition.Builder()
+ .ngram(generateFilter())
+ .build())
+ .build())
+ .build())
+ .build();
}
- public static XContentBuilder getInvalidIndexSetting() throws IOException {
- return jsonBuilder()
- .startObject()
- .startObject("settings")
- .startObject("analysis")
- .startObject(ANALYZER)
- .startObject("email_ngram_filter_analyzer")
- .field(TOKENIZER, "uax_url_email")
- .startArray("filter")
- .value("ngram_filter")
- .endArray()
- .endObject()
- .endObject()
- .startObject("filter")
- .startObject("ngram_filter")
- .field(TYPE, "ngram")
- .field("min_gram", 3)
- .field("max_gram", 13)
- .endObject()
- .endObject()
- .endObject()
- .endObject()
- .endObject();
+ public static IndexSettings getInvalidIndexSetting() {
+ return new IndexSettings.Builder()
+ .analysis(new IndexSettingsAnalysis.Builder()
+ .analyzer("email_ngram_filter_analyzer", new Analyzer.Builder()
+ .custom(generateAnalyzer())
+ .build())
+ .filter("ngram_filter", new TokenFilter.Builder()
+ .definition(new TokenFilterDefinition.Builder()
+ .ngram(generateFilter())
+ .build())
+ .build())
+ .build())
+ .build();
+ }
+
+ private static CustomAnalyzer generateAnalyzer() {
+ return new CustomAnalyzer.Builder()
+ .tokenizer("uax_url_email")
+ .filter("ngram_filter")
+ .build();
+ }
+
+ private static NGramTokenFilter generateFilter() {
+ return new NGramTokenFilter.Builder()
+ .minGram(3)
+ .maxGram(13)
+ .build();
}
@RegisterExtension
@@ -133,19 +132,14 @@ class IndexCreationFactoryTest {
new IndexCreationFactory(OpenSearchConfiguration.DEFAULT_CONFIGURATION)
.useIndex(INDEX_NAME)
.addAlias(ALIAS_NAME)
- .customAnalyzers(IndexCreationCustomElement.from("{" +
- " \"my_custom_analyzer\": {" +
- " \"type\": \"custom\"," +
- " \"tokenizer\": \"standard\"," +
- " \"char_filter\": [" +
- " \"html_strip\"" +
- " ]," +
- " \"filter\": [" +
- " \"lowercase\"," +
- " \"asciifolding\"" +
- " ]" +
- " }" +
- "}"))
+ .customAnalyzers(new IndexCreationCustomAnalyzer("my_custom_analyzer",
+ new Analyzer.Builder()
+ .custom(new CustomAnalyzer.Builder()
+ .tokenizer("standard")
+ .filter("lowercase", "asciifolding")
+ .charFilter("html_strip")
+ .build())
+ .build()))
.createIndexAndAliases(client);
}
@@ -155,14 +149,14 @@ class IndexCreationFactoryTest {
new IndexCreationFactory(OpenSearchConfiguration.DEFAULT_CONFIGURATION)
.useIndex(INDEX_NAME)
.addAlias(ALIAS_NAME)
- .customAnalyzers(IndexCreationCustomElement.from("{" +
- " \"my_custom_analyzer\": {" +
- " \"type\": \"invalid\"," +
- " \"tokenizer\": \"not_Found_tokenizer\"" +
- " }" +
- "}"))
+ .customAnalyzers(new IndexCreationCustomAnalyzer("my_custom_analyzer",
+ new Analyzer.Builder()
+ .custom(new CustomAnalyzer.Builder()
+ .tokenizer("not_Found_tokenizer")
+ .build())
+ .build()))
.createIndexAndAliases(client))
- .isInstanceOf(OpenSearchStatusException.class);
+ .isInstanceOf(Exception.class);
}
@Test
@@ -170,12 +164,16 @@ class IndexCreationFactoryTest {
new IndexCreationFactory(OpenSearchConfiguration.DEFAULT_CONFIGURATION)
.useIndex(INDEX_NAME)
.addAlias(ALIAS_NAME)
- .customTokenizers(IndexCreationCustomElement.from("{" +
- " \"custom_tokenizer\": { " +
- " \"type\": \"pattern\"," +
- " \"pattern\": \"[ .,!?]\"" +
- " }" +
- " }"))
+ .customTokenizers(new IndexCreationCustomTokenizer("custom_tokenizer",
+ new Tokenizer.Builder()
+ .definition(new TokenizerDefinition.Builder()
+ .pattern(new PatternTokenizer.Builder()
+ .pattern("[ .,!?]")
+ .flags("CASE_INSENSITIVE|COMMENTS")
+ .group(0)
+ .build())
+ .build())
+ .build()))
.createIndexAndAliases(client);
}
@@ -185,14 +183,16 @@ class IndexCreationFactoryTest {
new IndexCreationFactory(OpenSearchConfiguration.DEFAULT_CONFIGURATION)
.useIndex(INDEX_NAME)
.addAlias(ALIAS_NAME)
- .customTokenizers(IndexCreationCustomElement.from("{" +
- " \"custom_tokenizer\": { " +
- " \"type\": \"invalidType\"," +
- " \"pattern\": \"[ .,!?]\"" +
- " }" +
- " }"))
+ .customTokenizers(new IndexCreationCustomTokenizer("custom_tokenizer",
+ new Tokenizer.Builder()
+ .definition(new TokenizerDefinition.Builder()
+ .pattern(new PatternTokenizer.Builder()
+ .pattern("[ .,!?]")
+ .build())
+ .build())
+ .build()))
.createIndexAndAliases(client))
- .isInstanceOf(OpenSearchStatusException.class);
+ .isInstanceOf(Exception.class);
}
@Test
@@ -200,25 +200,28 @@ class IndexCreationFactoryTest {
new IndexCreationFactory(OpenSearchConfiguration.DEFAULT_CONFIGURATION)
.useIndex(INDEX_NAME)
.addAlias(ALIAS_NAME)
- .customAnalyzers(IndexCreationCustomElement.from("{" +
- " \"my_custom_analyzer\": { " +
- " \"tokenizer\": \"custom_tokenizer\"," +
- " \"filter\": [" +
- " \"lowercase\"" +
- " ]" +
- " }" +
- " }"))
- .customTokenizers(IndexCreationCustomElement.from("{" +
- " \"custom_tokenizer\": { " +
- " \"type\": \"pattern\"," +
- " \"pattern\": \"[ .,!?]\"" +
- " }" +
- " }"))
+ .customAnalyzers(new IndexCreationCustomAnalyzer("my_custom_analyzer",
+ new Analyzer.Builder()
+ .custom(new CustomAnalyzer.Builder()
+ .tokenizer("custom_tokenizer")
+ .filter("lowercase")
+ .build())
+ .build()))
+ .customTokenizers(new IndexCreationCustomTokenizer("custom_tokenizer",
+ new Tokenizer.Builder()
+ .definition(new TokenizerDefinition.Builder()
+ .pattern(new PatternTokenizer.Builder()
+ .pattern("[ .,!?]")
+ .flags("CASE_INSENSITIVE|COMMENTS")
+ .group(0)
+ .build())
+ .build())
+ .build()))
.createIndexAndAliases(client);
}
@Test
- void customIndexSettingShouldNotThrowWhenValidSetting() throws IOException {
+ void customIndexSettingShouldNotThrowWhenValidSetting() {
new IndexCreationFactory(OpenSearchConfiguration.DEFAULT_CONFIGURATION)
.useIndex(INDEX_NAME)
.addAlias(ALIAS_NAME)
@@ -232,7 +235,7 @@ class IndexCreationFactoryTest {
.useIndex(INDEX_NAME)
.addAlias(ALIAS_NAME)
.createIndexAndAliases(client, Optional.of(getInvalidIndexSetting()), Optional.empty()))
- .isInstanceOf(OpenSearchStatusException.class);
+ .isInstanceOf(Exception.class);
}
}
\ No newline at end of file
diff --git a/backends-common/opensearch/src/test/java/org/apache/james/backends/opensearch/OpenSearchHealthCheckTest.java b/backends-common/opensearch/src/test/java/org/apache/james/backends/opensearch/OpenSearchHealthCheckTest.java
index 5bacd05967..1e40074c7f 100644
--- a/backends-common/opensearch/src/test/java/org/apache/james/backends/opensearch/OpenSearchHealthCheckTest.java
+++ b/backends-common/opensearch/src/test/java/org/apache/james/backends/opensearch/OpenSearchHealthCheckTest.java
@@ -22,17 +22,32 @@ import static org.assertj.core.api.Assertions.assertThat;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-import org.opensearch.action.admin.cluster.health.ClusterHealthResponse;
-import org.opensearch.cluster.ClusterName;
-import org.opensearch.cluster.ClusterState;
-import org.opensearch.cluster.block.ClusterBlocks;
-import org.opensearch.cluster.health.ClusterHealthStatus;
-import org.opensearch.cluster.node.DiscoveryNodes;
-import org.opensearch.cluster.routing.RoutingTable;
+import org.opensearch.client.opensearch._types.HealthStatus;
+import org.opensearch.client.opensearch.cluster.HealthResponse;
import com.google.common.collect.ImmutableSet;
class OpenSearchHealthCheckTest {
+ private static HealthResponse fakeHealthResponse(HealthStatus status) {
+ return new HealthResponse.Builder()
+ .clusterName("fake-cluster")
+ .activePrimaryShards(0)
+ .activeShards(0)
+ .activeShardsPercentAsNumber("0")
+ .delayedUnassignedShards(0)
+ .initializingShards(0)
+ .numberOfDataNodes(0)
+ .numberOfInFlightFetch(0)
+ .numberOfNodes(0)
+ .numberOfPendingTasks(0)
+ .relocatingShards(0)
+ .taskMaxWaitingInQueueMillis(String.valueOf(System.currentTimeMillis()))
+ .timedOut(false)
+ .unassignedShards(0)
+ .status(status)
+ .build();
+ }
+
private OpenSearchHealthCheck healthCheck;
@BeforeEach
@@ -42,39 +57,22 @@ class OpenSearchHealthCheckTest {
@Test
void checkShouldReturnHealthyWhenOpenSearchClusterHealthStatusIsGreen() {
- FakeClusterHealthResponse response = new FakeClusterHealthResponse(ClusterHealthStatus.GREEN);
+ HealthResponse response = fakeHealthResponse(HealthStatus.Green);
assertThat(healthCheck.toHealthCheckResult(response).isHealthy()).isTrue();
}
@Test
void checkShouldReturnUnHealthyWhenOpenSearchClusterHealthStatusIsRed() {
- FakeClusterHealthResponse response = new FakeClusterHealthResponse(ClusterHealthStatus.RED);
+ HealthResponse response = fakeHealthResponse(HealthStatus.Red);
assertThat(healthCheck.toHealthCheckResult(response).isUnHealthy()).isTrue();
}
@Test
void checkShouldReturnHealthyWhenOpenSearchClusterHealthStatusIsYellow() {
- FakeClusterHealthResponse response = new FakeClusterHealthResponse(ClusterHealthStatus.YELLOW);
+ HealthResponse response = fakeHealthResponse(HealthStatus.Yellow);
assertThat(healthCheck.toHealthCheckResult(response).isHealthy()).isTrue();
}
-
- private static class FakeClusterHealthResponse extends ClusterHealthResponse {
- private final ClusterHealthStatus status;
-
- private FakeClusterHealthResponse(ClusterHealthStatus clusterHealthStatus) {
- super("fake-cluster", new String[0],
- new ClusterState(new ClusterName("fake-cluster"), 0, null, null, RoutingTable.builder().build(),
- DiscoveryNodes.builder().build(),
- ClusterBlocks.builder().build(), null, 0, false));
- this.status = clusterHealthStatus;
- }
-
- @Override
- public ClusterHealthStatus getStatus() {
- return this.status;
- }
- }
}
diff --git a/backends-common/opensearch/src/test/java/org/apache/james/backends/opensearch/OpenSearchIndexerTest.java b/backends-common/opensearch/src/test/java/org/apache/james/backends/opensearch/OpenSearchIndexerTest.java
index 2c8ba75c6e..9af1bd3806 100644
--- a/backends-common/opensearch/src/test/java/org/apache/james/backends/opensearch/OpenSearchIndexerTest.java
+++ b/backends-common/opensearch/src/test/java/org/apache/james/backends/opensearch/OpenSearchIndexerTest.java
@@ -23,7 +23,6 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.awaitility.Durations.ONE_HUNDRED_MILLISECONDS;
-import static org.opensearch.index.query.QueryBuilders.termQuery;
import java.io.IOException;
@@ -34,12 +33,13 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
-import org.opensearch.action.get.GetResponse;
-import org.opensearch.action.search.SearchRequest;
-import org.opensearch.client.RequestOptions;
-import org.opensearch.index.query.QueryBuilder;
-import org.opensearch.index.query.QueryBuilders;
-import org.opensearch.search.builder.SearchSourceBuilder;
+import org.opensearch.client.opensearch._types.FieldValue;
+import org.opensearch.client.opensearch._types.query_dsl.MatchAllQuery;
+import org.opensearch.client.opensearch._types.query_dsl.MatchQuery;
+import org.opensearch.client.opensearch._types.query_dsl.Query;
+import org.opensearch.client.opensearch._types.query_dsl.TermQuery;
+import org.opensearch.client.opensearch.core.GetResponse;
+import org.opensearch.client.opensearch.core.SearchRequest;
import com.google.common.collect.ImmutableList;
@@ -79,32 +79,36 @@ class OpenSearchIndexerTest {
}
@Test
- void indexMessageShouldWork() {
+ void indexMessageShouldWork() throws IOException {
DocumentId documentId = DocumentId.fromString("1");
String content = "{\"message\": \"trying out Elasticsearch\"}";
testee.index(documentId, content, useDocumentId(documentId)).block();
- awaitForOpenSearch(QueryBuilders.matchQuery("message", "trying"), 1L);
+ awaitForOpenSearch(new MatchQuery.Builder()
+ .field("message")
+ .query(new FieldValue.Builder().stringValue("trying").build())
+ .build()
+ ._toQuery(), 1L);
}
@Test
void indexMessageShouldThrowWhenJsonIsNull() {
- assertThatThrownBy(() -> testee.index(DOCUMENT_ID, null, ROUTING))
+ assertThatThrownBy(() -> testee.index(DOCUMENT_ID, null, ROUTING).block())
.isInstanceOf(IllegalArgumentException.class);
}
@Test
- void updateMessages() {
+ void updateMessages() throws IOException {
String content = "{\"message\": \"trying out Elasticsearch\",\"field\":\"Should be unchanged\"}";
testee.index(DOCUMENT_ID, content, useDocumentId(DOCUMENT_ID)).block();
- awaitForOpenSearch(QueryBuilders.matchAllQuery(), 1L);
+ awaitForOpenSearch(new MatchAllQuery.Builder().build()._toQuery(), 1L);
testee.update(ImmutableList.of(new UpdatedRepresentation(DOCUMENT_ID, "{\"message\": \"mastering out Elasticsearch\"}")), useDocumentId(DOCUMENT_ID)).block();
- awaitForOpenSearch(QueryBuilders.matchQuery("message", "mastering"), 1L);
+ awaitForOpenSearch(new MatchQuery.Builder().field("message").query(new FieldValue.Builder().stringValue("mastering").build()).build()._toQuery(), 1L);
- awaitForOpenSearch(QueryBuilders.matchQuery("field", "unchanged"), 1L);
+ awaitForOpenSearch(new MatchQuery.Builder().field("field").query(new FieldValue.Builder().stringValue("unchanged").build()).build()._toQuery(), 1L);
}
@Test
@@ -136,21 +140,21 @@ class OpenSearchIndexerTest {
}
@Test
- void deleteByQueryShouldWorkOnSingleMessage() {
+ void deleteByQueryShouldWorkOnSingleMessage() throws IOException {
DocumentId documentId = DocumentId.fromString("1:2");
String content = "{\"message\": \"trying out Elasticsearch\", \"property\":\"1\"}";
RoutingKey routingKey = useDocumentId(documentId);
testee.index(documentId, content, routingKey).block();
- awaitForOpenSearch(QueryBuilders.matchAllQuery(), 1L);
+ awaitForOpenSearch(new MatchAllQuery.Builder().build()._toQuery(), 1L);
- testee.deleteAllMatchingQuery(termQuery("property", "1"), routingKey).block();
+ testee.deleteAllMatchingQuery(new TermQuery.Builder().field("property").value(new FieldValue.Builder().stringValue("1").build()).build()._toQuery(), routingKey).block();
- awaitForOpenSearch(QueryBuilders.matchAllQuery(), 0L);
+ awaitForOpenSearch(new MatchAllQuery.Builder().build()._toQuery(), 0L);
}
@Test
- void deleteByQueryShouldWorkWhenMultipleMessages() {
+ void deleteByQueryShouldWorkWhenMultipleMessages() throws IOException {
DocumentId documentId = DocumentId.fromString("1:1");
String content = "{\"message\": \"trying out Elasticsearch\", \"property\":\"1\"}";
@@ -165,28 +169,28 @@ class OpenSearchIndexerTest {
String content3 = "{\"message\": \"trying out Elasticsearch 3\", \"property\":\"2\"}";
testee.index(documentId3, content3, ROUTING).block();
- awaitForOpenSearch(QueryBuilders.matchAllQuery(), 3L);
+ awaitForOpenSearch(new MatchAllQuery.Builder().build()._toQuery(), 3L);
- testee.deleteAllMatchingQuery(termQuery("property", "1"), ROUTING).block();
+ testee.deleteAllMatchingQuery(new TermQuery.Builder().field("property").value(new FieldValue.Builder().stringValue("1").build()).build()._toQuery(), ROUTING).block();
- awaitForOpenSearch(QueryBuilders.matchAllQuery(), 1L);
+ awaitForOpenSearch(new MatchAllQuery.Builder().build()._toQuery(), 1L);
}
@Test
- void deleteMessage() {
+ void deleteMessage() throws IOException {
DocumentId documentId = DocumentId.fromString("1:2");
String content = "{\"message\": \"trying out Elasticsearch\"}";
testee.index(documentId, content, useDocumentId(documentId)).block();
- awaitForOpenSearch(QueryBuilders.matchAllQuery(), 1L);
+ awaitForOpenSearch(new MatchAllQuery.Builder().build()._toQuery(), 1L);
testee.delete(ImmutableList.of(documentId), useDocumentId(documentId)).block();
- awaitForOpenSearch(QueryBuilders.matchAllQuery(), 0L);
+ awaitForOpenSearch(new MatchAllQuery.Builder().build()._toQuery(), 0L);
}
@Test
- void deleteShouldWorkWhenMultipleMessages() {
+ void deleteShouldWorkWhenMultipleMessages() throws IOException {
DocumentId documentId = DocumentId.fromString("1:1");
String content = "{\"message\": \"trying out Elasticsearch\", \"mailboxId\":\"1\"}";
testee.index(documentId, content, ROUTING).block();
@@ -199,11 +203,11 @@ class OpenSearchIndexerTest {
String content3 = "{\"message\": \"trying out Elasticsearch 3\", \"mailboxId\":\"2\"}";
testee.index(documentId3, content3, ROUTING).block();
- awaitForOpenSearch(QueryBuilders.matchAllQuery(), 3L);
+ awaitForOpenSearch(new MatchAllQuery.Builder().build()._toQuery(), 3L);
testee.delete(ImmutableList.of(documentId, documentId3), ROUTING).block();
- awaitForOpenSearch(QueryBuilders.matchAllQuery(), 1L);
+ awaitForOpenSearch(new MatchAllQuery.Builder().build()._toQuery(), 1L);
}
@Test
@@ -219,16 +223,16 @@ class OpenSearchIndexerTest {
}
@Test
- void getShouldWork() {
+ void getShouldWork() throws IOException {
DocumentId documentId = DocumentId.fromString("1");
String content = "{\"message\":\"trying out Elasticsearch\"}";
testee.index(documentId, content, useDocumentId(documentId)).block();
- awaitForOpenSearch(QueryBuilders.matchAllQuery(), 1L);
+ awaitForOpenSearch(new MatchAllQuery.Builder().build()._toQuery(), 1L);
GetResponse getResponse = testee.get(documentId, useDocumentId(documentId)).block();
- assertThat(getResponse.getSourceAsString()).isEqualTo(content);
+ assertThat(getResponse.source().toString()).isEqualTo(content);
}
@Test
@@ -243,13 +247,13 @@ class OpenSearchIndexerTest {
.isInstanceOf(NullPointerException.class);
}
- private void awaitForOpenSearch(QueryBuilder query, long totalHits) {
+ private void awaitForOpenSearch(Query query, long totalHits) {
CALMLY_AWAIT.atMost(Durations.TEN_SECONDS)
.untilAsserted(() -> assertThat(client.search(
- new SearchRequest(INDEX_NAME.getValue())
- .source(new SearchSourceBuilder().query(query)),
- RequestOptions.DEFAULT)
+ new SearchRequest.Builder()
+ .query(query)
+ .build())
.block()
- .getHits().getTotalHits().value).isEqualTo(totalHits));
+ .hits().total().value()).isEqualTo(totalHits));
}
}
diff --git a/backends-common/opensearch/src/test/java/org/apache/james/backends/opensearch/search/ScrolledSearchTest.java b/backends-common/opensearch/src/test/java/org/apache/james/backends/opensearch/search/ScrolledSearchTest.java
index dcd10da200..9e3e32455c 100644
--- a/backends-common/opensearch/src/test/java/org/apache/james/backends/opensearch/search/ScrolledSearchTest.java
+++ b/backends-common/opensearch/src/test/java/org/apache/james/backends/opensearch/search/ScrolledSearchTest.java
@@ -23,6 +23,7 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;
import java.io.IOException;
+import java.util.List;
import org.apache.james.backends.opensearch.DockerOpenSearchExtension;
import org.apache.james.backends.opensearch.IndexCreationFactory;
@@ -36,18 +37,20 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
-import org.opensearch.action.index.IndexRequest;
-import org.opensearch.action.search.SearchRequest;
-import org.opensearch.client.RequestOptions;
-import org.opensearch.common.unit.TimeValue;
-import org.opensearch.index.query.QueryBuilders;
-import org.opensearch.search.SearchHit;
-import org.opensearch.search.builder.SearchSourceBuilder;
+import org.opensearch.client.opensearch._types.Time;
+import org.opensearch.client.opensearch._types.query_dsl.MatchAllQuery;
+import org.opensearch.client.opensearch.core.IndexRequest;
+import org.opensearch.client.opensearch.core.SearchRequest;
+import org.opensearch.client.opensearch.core.search.Hit;
+
+import com.fasterxml.jackson.databind.util.RawValue;
class ScrolledSearchTest {
- private static final TimeValue TIMEOUT = TimeValue.timeValueMinutes(1);
+ private static final Time TIMEOUT = new Time.Builder()
+ .time("1m")
+ .build();
private static final int SIZE = 2;
- private static final String MESSAGE = "message";
+ private static final String CONTENT = "{\"message\": \"Sample message\"}";
private static final IndexName INDEX_NAME = new IndexName("index");
private static final ReadAliasName ALIAS_NAME = new ReadAliasName("alias");
@@ -74,119 +77,129 @@ class ScrolledSearchTest {
@Test
void scrollIterableShouldWorkWhenEmpty() {
- SearchRequest searchRequest = new SearchRequest(INDEX_NAME.getValue())
+ SearchRequest searchRequest = new SearchRequest.Builder()
+ .index(INDEX_NAME.getValue())
.scroll(TIMEOUT)
- .source(new SearchSourceBuilder()
- .query(QueryBuilders.matchAllQuery())
- .size(SIZE));
+ .query(new MatchAllQuery.Builder().build()._toQuery())
+ .size(SIZE)
+ .build();
assertThat(new ScrolledSearch(client, searchRequest).searchHits().collectList().block())
.isEmpty();
}
@Test
- void scrollIterableShouldWorkWhenOneElement() throws Exception {
+ void scrollIterableShouldWorkWhenOneElement() throws IOException {
String id = "1";
- client.index(new IndexRequest(INDEX_NAME.getValue())
+ client.index(new IndexRequest.Builder<>()
+ .index(INDEX_NAME.getValue())
.id(id)
- .source(MESSAGE, "Sample message"),
- RequestOptions.DEFAULT)
- .block();
+ .document(new RawValue(CONTENT))
+ .build())
+ .block();
openSearch.awaitForOpenSearch();
WAIT_CONDITION.untilAsserted(() -> hasIdsInIndex(client, id));
- SearchRequest searchRequest = new SearchRequest(INDEX_NAME.getValue())
+ SearchRequest searchRequest = new SearchRequest.Builder()
+ .index(INDEX_NAME.getValue())
.scroll(TIMEOUT)
- .source(new SearchSourceBuilder()
- .query(QueryBuilders.matchAllQuery())
- .size(SIZE));
+ .query(new MatchAllQuery.Builder().build()._toQuery())
+ .size(SIZE)
+ .build();
assertThat(new ScrolledSearch(client, searchRequest).searchHits().collectList().block())
- .extracting(SearchHit::getId)
+ .extracting(Hit::id)
.containsOnly(id);
}
@Test
- void scrollIterableShouldWorkWhenSizeElement() {
+ void scrollIterableShouldWorkWhenSizeElement() throws IOException {
String id1 = "1";
- client.index(new IndexRequest(INDEX_NAME.getValue())
+ client.index(new IndexRequest.Builder<>()
+ .index(INDEX_NAME.getValue())
.id(id1)
- .source(MESSAGE, "Sample message"),
- RequestOptions.DEFAULT)
+ .document(new RawValue(CONTENT))
+ .build())
.block();
String id2 = "2";
- client.index(new IndexRequest(INDEX_NAME.getValue())
+ client.index(new IndexRequest.Builder<>()
+ .index(INDEX_NAME.getValue())
.id(id2)
- .source(MESSAGE, "Sample message"),
- RequestOptions.DEFAULT)
+ .document(new RawValue(CONTENT))
+ .build())
.block();
openSearch.awaitForOpenSearch();
WAIT_CONDITION.untilAsserted(() -> hasIdsInIndex(client, id1, id2));
- SearchRequest searchRequest = new SearchRequest(INDEX_NAME.getValue())
+ SearchRequest searchRequest = new SearchRequest.Builder()
+ .index(INDEX_NAME.getValue())
.scroll(TIMEOUT)
- .source(new SearchSourceBuilder()
- .query(QueryBuilders.matchAllQuery())
- .size(SIZE));
+ .query(new MatchAllQuery.Builder().build()._toQuery())
+ .size(SIZE)
+ .build();
assertThat(new ScrolledSearch(client, searchRequest).searchHits().collectList().block())
- .extracting(SearchHit::getId)
+ .extracting(Hit::id)
.containsOnly(id1, id2);
}
@Test
- void scrollIterableShouldWorkWhenMoreThanSizeElement() {
+ void scrollIterableShouldWorkWhenMoreThanSizeElement() throws IOException {
String id1 = "1";
- client.index(new IndexRequest(INDEX_NAME.getValue())
+ client.index(new IndexRequest.Builder<>()
+ .index(INDEX_NAME.getValue())
.id(id1)
- .source(MESSAGE, "Sample message"),
- RequestOptions.DEFAULT)
+ .document(new RawValue(CONTENT))
+ .build())
.block();
String id2 = "2";
- client.index(new IndexRequest(INDEX_NAME.getValue())
+ client.index(new IndexRequest.Builder<>()
+ .index(INDEX_NAME.getValue())
.id(id2)
- .source(MESSAGE, "Sample message"),
- RequestOptions.DEFAULT)
+ .document(new RawValue(CONTENT))
+ .build())
.block();
String id3 = "3";
- client.index(new IndexRequest(INDEX_NAME.getValue())
+ client.index(new IndexRequest.Builder<>()
+ .index(INDEX_NAME.getValue())
.id(id3)
- .source(MESSAGE, "Sample message"),
- RequestOptions.DEFAULT)
+ .document(new RawValue(CONTENT))
+ .build())
.block();
openSearch.awaitForOpenSearch();
WAIT_CONDITION.untilAsserted(() -> hasIdsInIndex(client, id1, id2, id3));
- SearchRequest searchRequest = new SearchRequest(INDEX_NAME.getValue())
+ SearchRequest searchRequest = new SearchRequest.Builder()
+ .index(INDEX_NAME.getValue())
.scroll(TIMEOUT)
- .source(new SearchSourceBuilder()
- .query(QueryBuilders.matchAllQuery())
- .size(SIZE));
+ .query(new MatchAllQuery.Builder().build()._toQuery())
+ .size(SIZE)
+ .build();
assertThat(new ScrolledSearch(client, searchRequest).searchHits().collectList().block())
- .extracting(SearchHit::getId)
+ .extracting(Hit::id)
.containsOnly(id1, id2, id3);
}
- private void hasIdsInIndex(ReactorOpenSearchClient client, String... ids) {
- SearchRequest searchRequest = new SearchRequest(INDEX_NAME.getValue())
- .scroll(TIMEOUT)
- .source(new SearchSourceBuilder()
- .query(QueryBuilders.matchAllQuery()));
+ private void hasIdsInIndex(ReactorOpenSearchClient client, String... ids) throws IOException {
+ SearchRequest searchRequest = new SearchRequest.Builder()
+ .index(INDEX_NAME.getValue())
+ .query(new MatchAllQuery.Builder().build()._toQuery())
+ .build();
- SearchHit[] hits = client.search(searchRequest, RequestOptions.DEFAULT)
- .block()
- .getHits()
- .getHits();
+ List<String> hitIds = client.search(searchRequest)
+ .flatMapIterable(response -> response.hits().hits())
+ .map(Hit::id)
+ .collectList()
+ .block();
- assertThat(hits)
- .extracting(SearchHit::getId)
+ assertThat(hitIds)
.contains(ids);
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 10/12: JAMES-3771 Prevent SPI lookups on the HTTP eventloop
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 49c358a39dad08f3c238716595d7795447152aa5
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Thu Dec 15 13:59:24 2022 +0700
JAMES-3771 Prevent SPI lookups on the HTTP eventloop
---
.../james/backends/opensearch/ClientProvider.java | 2 +-
.../opensearch/json/JsonpDeserializerBase.java | 397 +++++++++++++++++++++
.../backends/opensearch/json/JsonpMapperBase.java | 102 ++++++
.../james/backends/opensearch/json/JsonpUtils.java | 207 +++++++++++
.../json/UnexpectedJsonEventException.java | 40 +++
.../json/jackson/JacksonJsonProvider.java | 298 ++++++++++++++++
.../json/jackson/JacksonJsonpGenerator.java | 374 +++++++++++++++++++
.../json/jackson/JacksonJsonpLocation.java | 54 +++
.../json/jackson/JacksonJsonpMapper.java | 123 +++++++
.../json/jackson/JacksonJsonpParser.java | 313 ++++++++++++++++
.../opensearch/json/jackson/JacksonUtils.java | 43 +++
.../opensearch/json/jackson/JsonValueParser.java | 108 ++++++
.../backends/opensearch/json/package-info.java | 29 ++
13 files changed, 2089 insertions(+), 1 deletion(-)
diff --git a/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/ClientProvider.java b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/ClientProvider.java
index 89a104bb8b..abd0fc9945 100644
--- a/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/ClientProvider.java
+++ b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/ClientProvider.java
@@ -44,9 +44,9 @@ import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.apache.http.ssl.SSLContextBuilder;
import org.apache.http.ssl.TrustStrategy;
+import org.apache.james.backends.opensearch.json.jackson.JacksonJsonpMapper;
import org.apache.james.util.concurrent.NamedThreadFactory;
import org.opensearch.client.RestClient;
-import org.opensearch.client.json.jackson.JacksonJsonpMapper;
import org.opensearch.client.opensearch.OpenSearchAsyncClient;
import org.opensearch.client.transport.rest_client.RestClientTransport;
import org.slf4j.Logger;
diff --git a/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/JsonpDeserializerBase.java b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/JsonpDeserializerBase.java
new file mode 100644
index 0000000000..c790e4ab2c
--- /dev/null
+++ b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/JsonpDeserializerBase.java
@@ -0,0 +1,397 @@
+/****************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one *
+ * or more contributor license agreements. See the NOTICE file *
+ * distributed with this work for additional information *
+ * regarding copyright ownership. The ASF licenses this file *
+ * to you 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.apache.james.backends.opensearch.json;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.EnumSet;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.opensearch.client.json.JsonpDeserializer;
+import org.opensearch.client.json.JsonpMapper;
+
+import jakarta.json.JsonNumber;
+import jakarta.json.JsonValue;
+import jakarta.json.stream.JsonParser;
+import jakarta.json.stream.JsonParser.Event;
+import jakarta.json.stream.JsonParsingException;
+
+/**
+ * Base class for {@link JsonpDeserializer} implementations that accept a set of JSON events known at instanciation time.
+ */
+public abstract class JsonpDeserializerBase<V> implements JsonpDeserializer<V> {
+
+ private final EnumSet<Event> acceptedEvents;
+ private final EnumSet<Event> nativeEvents;
+
+ protected JsonpDeserializerBase(EnumSet<Event> acceptedEvents) {
+ this(acceptedEvents, acceptedEvents);
+ }
+
+ protected JsonpDeserializerBase(EnumSet<Event> acceptedEvents, EnumSet<Event> nativeEvents) {
+ this.acceptedEvents = acceptedEvents;
+ this.nativeEvents = nativeEvents;
+ }
+
+ /** Combines accepted events from a number of deserializers */
+ protected static EnumSet<Event> allAcceptedEvents(JsonpDeserializer<?>... deserializers) {
+ EnumSet<Event> result = EnumSet.noneOf(Event.class);
+ for (JsonpDeserializer<?> deserializer: deserializers) {
+
+ EnumSet<Event> set = deserializer.acceptedEvents();
+ // Disabled for now. Only happens with the experimental Union2 and is caused by string and number
+ // parsers leniency. Need to be replaced with a check on a preferred event type.
+ //if (!Collections.disjoint(result, set)) {
+ // throw new IllegalArgumentException("Deserializer accepted events are not disjoint");
+ //}
+
+ result.addAll(set);
+ }
+ return result;
+ }
+
+ @Override
+ public EnumSet<Event> nativeEvents() {
+ return nativeEvents;
+ }
+
+ /**
+ * The JSON events this deserializer accepts as a starting point
+ */
+ public final EnumSet<Event> acceptedEvents() {
+ return acceptedEvents;
+ }
+
+ /**
+ * Convenience method for {@code acceptedEvents.contains(event)}
+ */
+ public final boolean accepts(Event event) {
+ return acceptedEvents.contains(event);
+ }
+
+ //---------------------------------------------------------------------------------------------
+
+ //----- Builtin types
+
+ static final JsonpDeserializer<String> STRING =
+ // String parsing is lenient and accepts any other primitive type
+ new JsonpDeserializerBase<String>(EnumSet.of(
+ Event.KEY_NAME, Event.VALUE_STRING, Event.VALUE_NUMBER,
+ Event.VALUE_FALSE, Event.VALUE_TRUE
+ ),
+ EnumSet.of(Event.VALUE_STRING)
+ ) {
+ @Override
+ public String deserialize(JsonParser parser, JsonpMapper mapper, Event event) {
+ if (event == Event.VALUE_TRUE) {
+ return "true";
+ }
+ if (event == Event.VALUE_FALSE) {
+ return "false";
+ }
+ return parser.getString(); // also accepts numbers
+ }
+ };
+
+ static final JsonpDeserializer<Integer> INTEGER =
+ new JsonpDeserializerBase<Integer>(
+ EnumSet.of(Event.VALUE_NUMBER, Event.VALUE_STRING),
+ EnumSet.of(Event.VALUE_NUMBER)
+ ) {
+ @Override
+ public Integer deserialize(JsonParser parser, JsonpMapper mapper, Event event) {
+ if (event == Event.VALUE_STRING) {
+ return Integer.valueOf(parser.getString());
+ }
+ return parser.getInt();
+ }
+ };
+
+ static final JsonpDeserializer<Boolean> BOOLEAN =
+ new JsonpDeserializerBase<Boolean>(
+ EnumSet.of(Event.VALUE_FALSE, Event.VALUE_TRUE, Event.VALUE_STRING),
+ EnumSet.of(Event.VALUE_FALSE, Event.VALUE_TRUE)
+ ) {
+ @Override
+ public Boolean deserialize(JsonParser parser, JsonpMapper mapper, Event event) {
+ if (event == Event.VALUE_STRING) {
+ return Boolean.parseBoolean(parser.getString());
+ } else {
+ return event == Event.VALUE_TRUE;
+ }
+ }
+ };
+
+ static final JsonpDeserializer<Long> LONG =
+ new JsonpDeserializerBase<Long>(
+ EnumSet.of(Event.VALUE_NUMBER, Event.VALUE_STRING),
+ EnumSet.of(Event.VALUE_NUMBER)
+ ) {
+ @Override
+ public Long deserialize(JsonParser parser, JsonpMapper mapper, Event event) {
+ if (event == Event.VALUE_STRING) {
+ return Long.valueOf(parser.getString());
+ }
+ return parser.getLong();
+ }
+ };
+
+ static final JsonpDeserializer<Float> FLOAT =
+ new JsonpDeserializerBase<Float>(
+ EnumSet.of(Event.VALUE_NUMBER, Event.VALUE_STRING),
+ EnumSet.of(Event.VALUE_NUMBER)
+
+ ) {
+ @Override
+ public Float deserialize(JsonParser parser, JsonpMapper mapper, Event event) {
+ if (event == Event.VALUE_STRING) {
+ return Float.valueOf(parser.getString());
+ }
+ return parser.getBigDecimal().floatValue();
+ }
+ };
+
+ static final JsonpDeserializer<Double> DOUBLE =
+ new JsonpDeserializerBase<Double>(
+ EnumSet.of(Event.VALUE_NUMBER, Event.VALUE_STRING),
+ EnumSet.of(Event.VALUE_NUMBER)
+ ) {
+ @Override
+ public Double deserialize(JsonParser parser, JsonpMapper mapper, Event event) {
+ if (event == Event.VALUE_STRING) {
+ return Double.valueOf(parser.getString());
+ }
+ return parser.getBigDecimal().doubleValue();
+ }
+ };
+
+ static final class DoubleOrNullDeserializer extends JsonpDeserializerBase<Double> {
+ static final EnumSet<Event> nativeEvents = EnumSet.of(Event.VALUE_NUMBER, Event.VALUE_NULL);
+ static final EnumSet<Event> acceptedEvents = EnumSet.of(Event.VALUE_STRING, Event.VALUE_NUMBER, Event.VALUE_NULL);
+ private final double defaultValue;
+
+ DoubleOrNullDeserializer(double defaultValue) {
+ super(acceptedEvents, nativeEvents);
+ this.defaultValue = defaultValue;
+ }
+
+ @Override
+ public Double deserialize(JsonParser parser, JsonpMapper mapper, Event event) {
+ if (event == Event.VALUE_NULL) {
+ return defaultValue;
+ }
+ if (event == Event.VALUE_STRING) {
+ return Double.valueOf(parser.getString());
+ }
+ return parser.getBigDecimal().doubleValue();
+ }
+ }
+
+ static final class IntOrNullDeserializer extends JsonpDeserializerBase<Integer> {
+ static final EnumSet<Event> nativeEvents = EnumSet.of(Event.VALUE_NUMBER, Event.VALUE_NULL);
+ static final EnumSet<Event> acceptedEvents = EnumSet.of(Event.VALUE_STRING, Event.VALUE_NUMBER, Event.VALUE_NULL);
+ private final int defaultValue;
+
+ IntOrNullDeserializer(int defaultValue) {
+ super(acceptedEvents, nativeEvents);
+ this.defaultValue = defaultValue;
+ }
+
+ @Override
+ public Integer deserialize(JsonParser parser, JsonpMapper mapper, Event event) {
+ if (event == Event.VALUE_NULL) {
+ return defaultValue;
+ }
+ if (event == Event.VALUE_STRING) {
+ return Integer.valueOf(parser.getString());
+ }
+ return parser.getInt();
+ }
+ }
+
+ static final class StringOrNullDeserializer extends JsonpDeserializerBase<String> {
+ static final EnumSet<Event> nativeEvents = EnumSet.of(Event.VALUE_STRING, Event.VALUE_NULL);
+ static final EnumSet<Event> acceptedEvents = EnumSet.of(Event.KEY_NAME, Event.VALUE_STRING,
+ Event.VALUE_NUMBER, Event.VALUE_FALSE, Event.VALUE_TRUE, Event.VALUE_NULL);
+
+ StringOrNullDeserializer() {
+ super(acceptedEvents, nativeEvents);
+ }
+
+ @Override
+ public String deserialize(JsonParser parser, JsonpMapper mapper, Event event) {
+ if (event == Event.VALUE_NULL) {
+ return null;
+ }
+ if (event == Event.VALUE_TRUE) {
+ return "true";
+ }
+ if (event == Event.VALUE_FALSE) {
+ return "false";
+ }
+ return parser.getString();
+ }
+ }
+
+ static final JsonpDeserializer<Double> DOUBLE_OR_NAN =
+ new JsonpDeserializerBase<Double>(
+ EnumSet.of(Event.VALUE_NUMBER, Event.VALUE_STRING, Event.VALUE_NULL),
+ EnumSet.of(Event.VALUE_NUMBER, Event.VALUE_NULL)
+ ) {
+ @Override
+ public Double deserialize(JsonParser parser, JsonpMapper mapper, Event event) {
+ if (event == Event.VALUE_NULL) {
+ return Double.NaN;
+ }
+ if (event == Event.VALUE_STRING) {
+ return Double.valueOf(parser.getString());
+ }
+ return parser.getBigDecimal().doubleValue();
+ }
+ };
+
+ static final JsonpDeserializer<Number> NUMBER =
+ new JsonpDeserializerBase<Number>(
+ EnumSet.of(Event.VALUE_NUMBER, Event.VALUE_STRING),
+ EnumSet.of(Event.VALUE_NUMBER)
+ ) {
+ @Override
+ public Number deserialize(JsonParser parser, JsonpMapper mapper, Event event) {
+ if (event == Event.VALUE_STRING) {
+ return Double.valueOf(parser.getString());
+ }
+ return ((JsonNumber)parser.getValue()).numberValue();
+ }
+ };
+
+ static final JsonpDeserializer<JsonValue> JSON_VALUE =
+ new JsonpDeserializerBase<JsonValue>(
+ EnumSet.allOf(Event.class)
+ ) {
+ @Override
+ public JsonValue deserialize(JsonParser parser, JsonpMapper mapper, Event event) {
+ return parser.getValue();
+ }
+ };
+
+ static final JsonpDeserializer<Void> VOID = new JsonpDeserializerBase<Void>(
+ EnumSet.noneOf(Event.class)
+ ) {
+ @Override
+ public Void deserialize(JsonParser parser, JsonpMapper mapper) {
+ throw new JsonParsingException("Void types should not have any value", parser.getLocation());
+ }
+
+ @Override
+ public Void deserialize(JsonParser parser, JsonpMapper mapper, Event event) {
+ return deserialize(parser, mapper);
+ }
+ };
+
+ //----- Collections
+
+ static class ArrayDeserializer<T> implements JsonpDeserializer<List<T>> {
+ private final JsonpDeserializer<T> itemDeserializer;
+ private EnumSet<Event> acceptedEvents;
+ private static final EnumSet<Event> nativeEvents = EnumSet.of(Event.START_ARRAY);
+
+ protected ArrayDeserializer(JsonpDeserializer<T> itemDeserializer) {
+ this.itemDeserializer = itemDeserializer;
+ }
+
+ @Override
+ public EnumSet<Event> nativeEvents() {
+ return nativeEvents;
+ }
+
+ @Override
+ public EnumSet<Event> acceptedEvents() {
+ // Accepted events is computed lazily
+ // no need for double-checked lock, we don't care about computing it several times
+ if (acceptedEvents == null) {
+ acceptedEvents = EnumSet.of(Event.START_ARRAY);
+ acceptedEvents.addAll(itemDeserializer.acceptedEvents());
+ }
+ return acceptedEvents;
+ }
+
+ @Override
+ public List<T> deserialize(JsonParser parser, JsonpMapper mapper, Event event) {
+ if (event == Event.START_ARRAY) {
+ List<T> result = new ArrayList<>();
+ while ((event = parser.next()) != Event.END_ARRAY) {
+ JsonpUtils.ensureAccepts(itemDeserializer, parser, event);
+ result.add(itemDeserializer.deserialize(parser, mapper, event));
+ }
+ return result;
+ } else {
+ // Single-value mode
+ JsonpUtils.ensureAccepts(itemDeserializer, parser, event);
+ return Collections.singletonList(itemDeserializer.deserialize(parser, mapper, event));
+ }
+ }
+ }
+
+ static class StringMapDeserializer<T> extends JsonpDeserializerBase<Map<String, T>> {
+ private final JsonpDeserializer<T> itemDeserializer;
+
+ protected StringMapDeserializer(JsonpDeserializer<T> itemDeserializer) {
+ super(EnumSet.of(Event.START_OBJECT));
+ this.itemDeserializer = itemDeserializer;
+ }
+
+ @Override
+ public Map<String, T> deserialize(JsonParser parser, JsonpMapper mapper, Event event) {
+ Map<String, T> result = new HashMap<>();
+ while ((event = parser.next()) != Event.END_OBJECT) {
+ JsonpUtils.expectEvent(parser, Event.KEY_NAME, event);
+ String key = parser.getString();
+ T value = itemDeserializer.deserialize(parser, mapper);
+ result.put(key, value);
+ }
+ return result;
+ }
+ }
+
+ static class EnumMapDeserializer<K, V> extends JsonpDeserializerBase<Map<K, V>> {
+ private final JsonpDeserializer<K> keyDeserializer;
+ private final JsonpDeserializer<V> valueDeserializer;
+
+ protected EnumMapDeserializer(JsonpDeserializer<K> keyDeserializer, JsonpDeserializer<V> valueDeserializer) {
+ super(EnumSet.of(Event.START_OBJECT));
+ this.keyDeserializer = keyDeserializer;
+ this.valueDeserializer = valueDeserializer;
+ }
+
+ @Override
+ public Map<K, V> deserialize(JsonParser parser, JsonpMapper mapper, Event event) {
+ Map<K, V> result = new HashMap<>();
+ while ((event = parser.next()) != Event.END_OBJECT) {
+ JsonpUtils.expectEvent(parser, Event.KEY_NAME, event);
+ K key = keyDeserializer.deserialize(parser, mapper, event);
+ V value = valueDeserializer.deserialize(parser, mapper);
+ result.put(key, value);
+ }
+ return result;
+ }
+ }
+}
diff --git a/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/JsonpMapperBase.java b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/JsonpMapperBase.java
new file mode 100644
index 0000000000..5a710795b8
--- /dev/null
+++ b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/JsonpMapperBase.java
@@ -0,0 +1,102 @@
+/****************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one *
+ * or more contributor license agreements. See the NOTICE file *
+ * distributed with this work for additional information *
+ * regarding copyright ownership. The ASF licenses this file *
+ * to you 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.apache.james.backends.opensearch.json;
+
+import java.lang.reflect.Field;
+
+import javax.annotation.Nullable;
+
+import org.opensearch.client.json.JsonpDeserializable;
+import org.opensearch.client.json.JsonpDeserializer;
+import org.opensearch.client.json.JsonpMapper;
+import org.opensearch.client.json.JsonpSerializable;
+import org.opensearch.client.json.JsonpSerializer;
+
+import jakarta.json.JsonValue;
+import jakarta.json.stream.JsonGenerator;
+import jakarta.json.stream.JsonParser;
+
+public abstract class JsonpMapperBase implements JsonpMapper {
+
+ /** Get a serializer when none of the builtin ones are applicable */
+ protected abstract <T> JsonpDeserializer<T> getDefaultDeserializer(Class<T> clazz);
+
+ @Override
+ public <T> T deserialize(JsonParser parser, Class<T> clazz) {
+ JsonpDeserializer<T> deserializer = findDeserializer(clazz);
+ if (deserializer != null) {
+ return deserializer.deserialize(parser, this);
+ }
+
+ return getDefaultDeserializer(clazz).deserialize(parser, this);
+ }
+
+ @Nullable
+ @SuppressWarnings("unchecked")
+ public static <T> JsonpDeserializer<T> findDeserializer(Class<T> clazz) {
+ JsonpDeserializable annotation = clazz.getAnnotation(JsonpDeserializable.class);
+ if (annotation != null) {
+ try {
+ Field field = clazz.getDeclaredField(annotation.field());
+ return (JsonpDeserializer<T>)field.get(null);
+ } catch (Exception e) {
+ throw new RuntimeException("No deserializer found in '" + clazz.getName() + "." + annotation.field() + "'");
+ }
+ }
+
+ return null;
+ }
+
+ @Nullable
+ @SuppressWarnings("unchecked")
+ public static <T> JsonpSerializer<T> findSerializer(T value) {
+ Class<?> valueClass = value.getClass();
+ if (JsonpSerializable.class.isAssignableFrom(valueClass)) {
+ return (JsonpSerializer<T>) JsonpSerializableSerializer.INSTANCE;
+ }
+
+ if (JsonValue.class.isAssignableFrom(valueClass)) {
+ return (JsonpSerializer<T>) JsonpValueSerializer.INSTANCE;
+ }
+
+ return null;
+ }
+
+ protected static class JsonpSerializableSerializer<T extends JsonpSerializable> implements JsonpSerializer<T> {
+ @Override
+ public void serialize(T value, JsonGenerator generator, JsonpMapper mapper) {
+ value.serialize(generator, mapper);
+ }
+
+ protected static final JsonpSerializer<?> INSTANCE = new JsonpSerializableSerializer<>();
+
+ }
+
+ protected static class JsonpValueSerializer implements JsonpSerializer<JsonValue> {
+ @Override
+ public void serialize(JsonValue value, JsonGenerator generator, JsonpMapper mapper) {
+ generator.write(value);
+ }
+
+ protected static final JsonpSerializer<?> INSTANCE = new JsonpValueSerializer();
+
+ }
+
+}
diff --git a/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/JsonpUtils.java b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/JsonpUtils.java
new file mode 100644
index 0000000000..5694a5cbee
--- /dev/null
+++ b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/JsonpUtils.java
@@ -0,0 +1,207 @@
+/****************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one *
+ * or more contributor license agreements. See the NOTICE file *
+ * distributed with this work for additional information *
+ * regarding copyright ownership. The ASF licenses this file *
+ * to you 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.apache.james.backends.opensearch.json;
+
+import java.io.StringReader;
+import java.util.AbstractMap;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import javax.annotation.Nullable;
+
+import org.opensearch.client.json.JsonpDeserializer;
+import org.opensearch.client.json.JsonpMapper;
+import org.opensearch.client.json.JsonpSerializable;
+import org.opensearch.client.json.JsonpSerializer;
+import org.opensearch.client.util.ObjectBuilder;
+
+import jakarta.json.JsonObject;
+import jakarta.json.JsonString;
+import jakarta.json.JsonValue;
+import jakarta.json.stream.JsonGenerator;
+import jakarta.json.stream.JsonParser;
+import jakarta.json.stream.JsonParser.Event;
+import jakarta.json.stream.JsonParsingException;
+
+public class JsonpUtils {
+
+ /**
+ * Advances the parser to the next event and checks that this even is the expected one.
+ *
+ * @return the expected event
+ *
+ * @throws jakarta.json.JsonException if an i/o error occurs (IOException would be cause of JsonException)
+ * @throws JsonParsingException if the event is not the expected one, or if the parser encounters invalid
+ * JSON when advancing to next state.
+ * @throws java.util.NoSuchElementException if there are no more parsing states.
+ */
+ public static Event expectNextEvent(JsonParser parser, Event expected) {
+ Event event = parser.next();
+ expectEvent(parser, expected, event);
+ return event;
+ }
+
+ public static void expectEvent(JsonParser parser, Event expected, Event event) {
+ if (event != expected) {
+ throw new UnexpectedJsonEventException(parser, event, expected);
+ }
+ }
+
+ public static String expectKeyName(JsonParser parser, Event event) {
+ JsonpUtils.expectEvent(parser, Event.KEY_NAME, event);
+ return parser.getString();
+ }
+
+ public static void ensureAccepts(JsonpDeserializer<?> deserializer, JsonParser parser, Event event) {
+ if (!deserializer.acceptedEvents().contains(event)) {
+ throw new UnexpectedJsonEventException(parser, event, deserializer.acceptedEvents());
+ }
+ }
+
+ /**
+ * Skip the value at the next position of the parser.
+ */
+ public static void skipValue(JsonParser parser) {
+ skipValue(parser, parser.next());
+ }
+
+ /**
+ * Skip the value at the current position of the parser.
+ */
+ public static void skipValue(JsonParser parser, Event event) {
+ switch (event) {
+ case START_OBJECT:
+ parser.skipObject();
+ break;
+
+ case START_ARRAY:
+ parser.skipArray();
+ break;
+
+ default:
+ // Not a structure, no additional skipping needed
+ break;
+ }
+ }
+
+ public static <T> T buildVariant(JsonParser parser, ObjectBuilder<T> builder) {
+ if (builder == null) {
+ throw new JsonParsingException("No variant found", parser.getLocation());
+ }
+ return builder.build();
+ }
+
+ public static <T> void serialize(T value, JsonGenerator generator, @Nullable JsonpSerializer<T> serializer, JsonpMapper mapper) {
+ if (serializer != null) {
+ serializer.serialize(value, generator, mapper);
+ } else if (value instanceof JsonpSerializable) {
+ ((JsonpSerializable) value).serialize(generator, mapper);
+ } else {
+ mapper.serialize(value, generator);
+ }
+ }
+
+ /**
+ * Looks ahead a field value in the Json object from the upcoming object in a parser, which should be on the
+ * START_OBJECT event.
+ *
+ * Returns a pair containing that value and a parser that should be used to actually parse the object
+ * (the object has been consumed from the original one).
+ */
+ public static Map.Entry<String, JsonParser> lookAheadFieldValue(
+ String name, String defaultValue, JsonParser parser, JsonpMapper mapper
+ ) {
+ // FIXME: need a buffering parser wrapper so that we don't roundtrip through a JsonObject and a String
+ // FIXME: resulting parser should return locations that are offset with the original parser's location
+ JsonObject object = parser.getObject();
+ String result = object.getString(name, null);
+
+ if (result == null) {
+ result = defaultValue;
+ }
+
+ if (result == null) {
+ throw new JsonParsingException("Property '" + name + "' not found", parser.getLocation());
+ }
+
+ return new AbstractMap.SimpleImmutableEntry<>(result, objectParser(object, mapper));
+ }
+
+ /**
+ * Create a parser that traverses a JSON object
+ */
+ public static JsonParser objectParser(JsonObject object, JsonpMapper mapper) {
+ // FIXME: we should have used createParser(object), but this doesn't work as it creates a
+ // org.glassfish.json.JsonStructureParser that doesn't implement the JsonP 1.0.1 features, in particular
+ // parser.getObject(). So deserializing recursive internally-tagged union would fail with UnsupportedOperationException
+ // While glassfish has this issue or until we write our own, we roundtrip through a string.
+
+ String strObject = object.toString();
+ return mapper.jsonProvider().createParser(new StringReader(strObject));
+ }
+
+ public static String toString(JsonValue value) {
+ switch (value.getValueType()) {
+ case OBJECT:
+ throw new IllegalArgumentException("Json objects cannot be used as string");
+
+ case ARRAY:
+ return value.asJsonArray().stream()
+ .map(JsonpUtils::toString)
+ .collect(Collectors.joining(","));
+
+ case STRING:
+ return ((JsonString)value).getString();
+
+ case TRUE:
+ return "true";
+
+ case FALSE:
+ return "false";
+
+ case NULL:
+ return "null";
+
+ case NUMBER:
+ return value.toString();
+
+ default:
+ throw new IllegalArgumentException("Unknown JSON value type: '" + value + "'");
+ }
+ }
+
+ public static void serializeDoubleOrNull(JsonGenerator generator, double value, double defaultValue) {
+ // Only output null if the default value isn't finite, which cannot be represented as JSON
+ if (value == defaultValue && !Double.isFinite(defaultValue)) {
+ generator.writeNull();
+ } else {
+ generator.write(value);
+ }
+ }
+
+ public static void serializeIntOrNull(JsonGenerator generator, int value, int defaultValue) {
+ // Only output null if the default value isn't finite, which cannot be represented as JSON
+ if (value == defaultValue && defaultValue == Integer.MAX_VALUE || defaultValue == Integer.MIN_VALUE) {
+ generator.writeNull();
+ } else {
+ generator.write(value);
+ }
+ }
+}
diff --git a/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/UnexpectedJsonEventException.java b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/UnexpectedJsonEventException.java
new file mode 100644
index 0000000000..faf7d5fe85
--- /dev/null
+++ b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/UnexpectedJsonEventException.java
@@ -0,0 +1,40 @@
+/****************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one *
+ * or more contributor license agreements. See the NOTICE file *
+ * distributed with this work for additional information *
+ * regarding copyright ownership. The ASF licenses this file *
+ * to you 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.apache.james.backends.opensearch.json;
+
+import java.util.EnumSet;
+
+import jakarta.json.stream.JsonParser;
+import jakarta.json.stream.JsonParser.Event;
+import jakarta.json.stream.JsonParsingException;
+
+public class UnexpectedJsonEventException extends JsonParsingException {
+ public UnexpectedJsonEventException(JsonParser parser, Event event) {
+ super("Unexpected JSON event '" + event + "'", parser.getLocation());
+ }
+
+ public UnexpectedJsonEventException(JsonParser parser, Event event, Event expected) {
+ super("Unexpected JSON event '" + event + "' instead of '" + expected + "'", parser.getLocation());
+ }
+
+ public UnexpectedJsonEventException(JsonParser parser, Event event, EnumSet<Event> expected) {
+ super("Unexpected JSON event '" + event + "' instead of '" + expected + "'", parser.getLocation());
+ }
+}
diff --git a/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/jackson/JacksonJsonProvider.java b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/jackson/JacksonJsonProvider.java
new file mode 100644
index 0000000000..e7d128977f
--- /dev/null
+++ b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/jackson/JacksonJsonProvider.java
@@ -0,0 +1,298 @@
+/****************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one *
+ * or more contributor license agreements. See the NOTICE file *
+ * distributed with this work for additional information *
+ * regarding copyright ownership. The ASF licenses this file *
+ * to you 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.apache.james.backends.opensearch.json.jackson;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.io.OutputStreamWriter;
+import java.io.Reader;
+import java.io.Writer;
+import java.nio.charset.Charset;
+import java.util.Collections;
+import java.util.Map;
+
+import com.fasterxml.jackson.core.JsonFactory;
+
+import jakarta.json.JsonArray;
+import jakarta.json.JsonArrayBuilder;
+import jakarta.json.JsonBuilderFactory;
+import jakarta.json.JsonObject;
+import jakarta.json.JsonObjectBuilder;
+import jakarta.json.JsonReader;
+import jakarta.json.JsonReaderFactory;
+import jakarta.json.JsonWriter;
+import jakarta.json.JsonWriterFactory;
+import jakarta.json.spi.JsonProvider;
+import jakarta.json.stream.JsonGenerator;
+import jakarta.json.stream.JsonGeneratorFactory;
+import jakarta.json.stream.JsonParser;
+import jakarta.json.stream.JsonParserFactory;
+
+/**
+ * A partial implementation of JSONP's SPI on top of Jackson.
+ */
+public class JacksonJsonProvider extends JsonProvider {
+
+ private final JsonFactory jsonFactory;
+
+ public JacksonJsonProvider(JsonFactory jsonFactory) {
+ this.jsonFactory = jsonFactory;
+ }
+
+ public JacksonJsonProvider() {
+ this(new JsonFactory());
+ }
+
+ /**
+ * Return the underlying Jackson {@link JsonFactory}.
+ */
+ public JsonFactory jacksonJsonFactory() {
+ return this.jsonFactory;
+ }
+
+ //---------------------------------------------------------------------------------------------
+ // Parser
+
+ private final ParserFactory defaultParserFactory = new ParserFactory(null);
+
+ @Override
+ public JsonParserFactory createParserFactory(Map<String, ?> config) {
+ if (config == null || config.isEmpty()) {
+ return defaultParserFactory;
+ } else {
+ // TODO: handle specific configuration
+ return defaultParserFactory;
+ }
+ }
+
+ @Override
+ public JsonParser createParser(Reader reader) {
+ return defaultParserFactory.createParser(reader);
+ }
+
+ @Override
+ public JsonParser createParser(InputStream in) {
+ return defaultParserFactory.createParser(in);
+ }
+
+ private class ParserFactory implements JsonParserFactory {
+
+ private final Map<String, ?> config;
+
+ ParserFactory(Map<String, ?> config) {
+ this.config = config == null ? Collections.emptyMap() : config;
+ }
+
+ @Override
+ public JsonParser createParser(Reader reader) {
+ try {
+ return new JacksonJsonpParser(jsonFactory.createParser(reader));
+ } catch (IOException ioe) {
+ throw JacksonUtils.convertException(ioe);
+ }
+ }
+
+ @Override
+ public JsonParser createParser(InputStream in) {
+ try {
+ return new JacksonJsonpParser(jsonFactory.createParser(in));
+ } catch (IOException ioe) {
+ throw JacksonUtils.convertException(ioe);
+ }
+ }
+
+ @Override
+ public JsonParser createParser(InputStream in, Charset charset) {
+ try {
+ return new JacksonJsonpParser(jsonFactory.createParser(new InputStreamReader(in, charset)));
+ } catch (IOException ioe) {
+ throw JacksonUtils.convertException(ioe);
+ }
+ }
+
+ /**
+ * Not implemented.
+ */
+ @Override
+ public JsonParser createParser(JsonObject obj) {
+ return JsonProvider.provider().createParserFactory(null).createParser(obj);
+ }
+
+ /**
+ * Not implemented.
+ */
+ @Override
+ public JsonParser createParser(JsonArray array) {
+ return JsonProvider.provider().createParserFactory(null).createParser(array);
+ }
+
+ /**
+ * Not implemented.
+ */
+ @Override
+ public Map<String, ?> getConfigInUse() {
+ return config;
+ }
+ }
+
+ //---------------------------------------------------------------------------------------------
+ // Generator
+
+ private final JsonGeneratorFactory defaultGeneratorFactory = new GeneratorFactory(null);
+
+ @Override
+ public JsonGeneratorFactory createGeneratorFactory(Map<String, ?> config) {
+ if (config == null || config.isEmpty()) {
+ return defaultGeneratorFactory;
+ } else {
+ // TODO: handle specific configuration
+ return defaultGeneratorFactory;
+ }
+ }
+
+ @Override
+ public JsonGenerator createGenerator(Writer writer) {
+ return defaultGeneratorFactory.createGenerator(writer);
+ }
+
+ @Override
+ public JsonGenerator createGenerator(OutputStream out) {
+ return defaultGeneratorFactory.createGenerator(out);
+ }
+
+ private class GeneratorFactory implements JsonGeneratorFactory {
+
+ private final Map<String, ?> config;
+
+ GeneratorFactory(Map<String, ?> config) {
+ this.config = config == null ? Collections.emptyMap() : config;
+ }
+
+ @Override
+ public JsonGenerator createGenerator(Writer writer) {
+ try {
+ return new JacksonJsonpGenerator(jsonFactory.createGenerator(writer));
+ } catch (IOException ioe) {
+ throw JacksonUtils.convertException(ioe);
+ }
+ }
+
+ @Override
+ public JsonGenerator createGenerator(OutputStream out) {
+ try {
+ return new JacksonJsonpGenerator(jsonFactory.createGenerator(out));
+ } catch (IOException ioe) {
+ throw JacksonUtils.convertException(ioe);
+ }
+ }
+
+ @Override
+ public JsonGenerator createGenerator(OutputStream out, Charset charset) {
+ try {
+ return new JacksonJsonpGenerator(jsonFactory.createGenerator(new OutputStreamWriter(out, charset)));
+ } catch (IOException ioe) {
+ throw JacksonUtils.convertException(ioe);
+ }
+
+ }
+
+ @Override
+ public Map<String, ?> getConfigInUse() {
+ return config;
+ }
+ }
+
+ //---------------------------------------------------------------------------------------------
+ // Unsupported operations
+
+ /**
+ * Not implemented.
+ */
+ @Override
+ public JsonReader createReader(Reader reader) {
+ throw new UnsupportedOperationException();
+ }
+
+ /**
+ * Not implemented.
+ */
+ @Override
+ public JsonReader createReader(InputStream in) {
+ throw new UnsupportedOperationException();
+ }
+
+ /**
+ * Not implemented.
+ */
+ @Override
+ public JsonWriter createWriter(Writer writer) {
+ throw new UnsupportedOperationException();
+ }
+
+ /**
+ * Not implemented.
+ */
+ @Override
+ public JsonWriter createWriter(OutputStream out) {
+ throw new UnsupportedOperationException();
+ }
+
+ /**
+ * Not implemented.
+ */
+ @Override
+ public JsonWriterFactory createWriterFactory(Map<String, ?> config) {
+ throw new UnsupportedOperationException();
+ }
+
+ /**
+ * Not implemented.
+ */
+ @Override
+ public JsonReaderFactory createReaderFactory(Map<String, ?> config) {
+ throw new UnsupportedOperationException();
+ }
+
+ /**
+ * Not implemented.
+ */
+ @Override
+ public JsonObjectBuilder createObjectBuilder() {
+ throw new UnsupportedOperationException();
+ }
+
+ /**
+ * Not implemented.
+ */
+ @Override
+ public JsonArrayBuilder createArrayBuilder() {
+ throw new UnsupportedOperationException();
+ }
+
+ /**
+ * Not implemented.
+ */
+ @Override
+ public JsonBuilderFactory createBuilderFactory(Map<String, ?> config) {
+ throw new UnsupportedOperationException();
+ }
+}
diff --git a/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/jackson/JacksonJsonpGenerator.java b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/jackson/JacksonJsonpGenerator.java
new file mode 100644
index 0000000000..c8758d11d8
--- /dev/null
+++ b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/jackson/JacksonJsonpGenerator.java
@@ -0,0 +1,374 @@
+/****************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one *
+ * or more contributor license agreements. See the NOTICE file *
+ * distributed with this work for additional information *
+ * regarding copyright ownership. The ASF licenses this file *
+ * to you 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.apache.james.backends.opensearch.json.jackson;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.util.Map;
+
+import com.fasterxml.jackson.core.JsonStreamContext;
+
+import jakarta.json.JsonNumber;
+import jakarta.json.JsonString;
+import jakarta.json.JsonValue;
+import jakarta.json.stream.JsonGenerationException;
+import jakarta.json.stream.JsonGenerator;
+
+/**
+ * A JSONP generator implementation on top of Jackson.
+ */
+public class JacksonJsonpGenerator implements JsonGenerator {
+
+ private final com.fasterxml.jackson.core.JsonGenerator generator;
+
+ public JacksonJsonpGenerator(com.fasterxml.jackson.core.JsonGenerator generator) {
+ this.generator = generator;
+ }
+
+ /**
+ * Returns the underlying Jackson generator.
+ */
+ public com.fasterxml.jackson.core.JsonGenerator jacksonGenerator() {
+ return generator;
+ }
+
+ @Override
+ public JsonGenerator writeStartObject() {
+ try {
+ generator.writeStartObject();
+ } catch (IOException e) {
+ throw JacksonUtils.convertException(e);
+ }
+ return this;
+ }
+
+ @Override
+ public JsonGenerator writeStartObject(String name) {
+ try {
+ generator.writeFieldName(name);
+ generator.writeStartObject();
+ } catch (IOException e) {
+ throw JacksonUtils.convertException(e);
+ }
+ return this;
+ }
+
+ @Override
+ public JsonGenerator writeStartArray() {
+ try {
+ generator.writeStartArray();
+ } catch (IOException e) {
+ throw JacksonUtils.convertException(e);
+ }
+ return this;
+ }
+
+ @Override
+ public JsonGenerator writeStartArray(String name) {
+ try {
+ generator.writeFieldName(name);
+ generator.writeStartArray();
+ } catch (IOException e) {
+ throw JacksonUtils.convertException(e);
+ }
+ return this;
+ }
+
+ @Override
+ public JsonGenerator writeKey(String name) {
+ try {
+ generator.writeFieldName(name);
+ } catch (IOException e) {
+ throw JacksonUtils.convertException(e);
+ }
+ return this;
+ }
+
+ @Override
+ public JsonGenerator write(String name, JsonValue value) {
+ try {
+ generator.writeFieldName(name);
+ writeValue(value);
+ } catch (IOException e) {
+ throw JacksonUtils.convertException(e);
+ }
+ return this;
+ }
+
+ @Override
+ public JsonGenerator write(String name, String value) {
+ try {
+ generator.writeFieldName(name);
+ generator.writeString(value);
+ } catch (IOException e) {
+ throw JacksonUtils.convertException(e);
+ }
+ return this;
+ }
+
+ @Override
+ public JsonGenerator write(String name, BigInteger value) {
+ try {
+ generator.writeFieldName(name);
+ generator.writeNumber(value);
+ } catch (IOException e) {
+ throw JacksonUtils.convertException(e);
+ }
+ return this;
+ }
+
+ @Override
+ public JsonGenerator write(String name, BigDecimal value) {
+ try {
+ generator.writeFieldName(name);
+ generator.writeNumber(value);
+ } catch (IOException e) {
+ throw JacksonUtils.convertException(e);
+ }
+ return this;
+ }
+
+ @Override
+ public JsonGenerator write(String name, int value) {
+ try {
+ generator.writeFieldName(name);
+ generator.writeNumber(value);
+ } catch (IOException e) {
+ throw JacksonUtils.convertException(e);
+ }
+ return this;
+ }
+
+ @Override
+ public JsonGenerator write(String name, long value) {
+ try {
+ generator.writeFieldName(name);
+ generator.writeNumber(value);
+ } catch (IOException e) {
+ throw JacksonUtils.convertException(e);
+ }
+ return this;
+ }
+
+ @Override
+ public JsonGenerator write(String name, double value) {
+ try {
+ generator.writeFieldName(name);
+ generator.writeNumber(value);
+ } catch (IOException e) {
+ throw JacksonUtils.convertException(e);
+ }
+ return this;
+ }
+
+ @Override
+ public JsonGenerator write(String name, boolean value) {
+ try {
+ generator.writeFieldName(name);
+ generator.writeBooleanField(name, value);
+ } catch (IOException e) {
+ throw JacksonUtils.convertException(e);
+ }
+ return this;
+ }
+
+ @Override
+ public JsonGenerator writeNull(String name) {
+ try {
+ generator.writeFieldName(name);
+ generator.writeNull();
+ } catch (IOException e) {
+ throw JacksonUtils.convertException(e);
+ }
+ return this;
+ }
+
+ @Override
+ public JsonGenerator writeEnd() {
+ try {
+ JsonStreamContext ctx = generator.getOutputContext();
+ if (ctx.inObject()) {
+ generator.writeEndObject();
+ } else if (ctx.inArray()) {
+ generator.writeEndArray();
+ } else {
+ throw new JsonGenerationException("Unexpected context: '" + ctx.typeDesc() + "'");
+ }
+ } catch (IOException e) {
+ throw JacksonUtils.convertException(e);
+ }
+ return this;
+ }
+
+ @Override
+ public JsonGenerator write(JsonValue value) {
+ try {
+ writeValue(value);
+ } catch (IOException e) {
+ throw JacksonUtils.convertException(e);
+ }
+ return this;
+ }
+
+ @Override
+ public JsonGenerator write(String value) {
+ try {
+ generator.writeString(value);
+ } catch (IOException e) {
+ throw JacksonUtils.convertException(e);
+ }
+ return this;
+ }
+
+ @Override
+ public JsonGenerator write(BigDecimal value) {
+ try {
+ generator.writeNumber(value);
+ } catch (IOException e) {
+ throw JacksonUtils.convertException(e);
+ }
+ return this;
+ }
+
+ @Override
+ public JsonGenerator write(BigInteger value) {
+ try {
+ generator.writeNumber(value);
+ } catch (IOException e) {
+ throw JacksonUtils.convertException(e);
+ }
+ return this;
+ }
+
+ @Override
+ public JsonGenerator write(int value) {
+ try {
+ generator.writeNumber(value);
+ } catch (IOException e) {
+ throw JacksonUtils.convertException(e);
+ }
+ return this;
+ }
+
+ @Override
+ public JsonGenerator write(long value) {
+ try {
+ generator.writeNumber(value);
+ } catch (IOException e) {
+ throw JacksonUtils.convertException(e);
+ }
+ return this;
+ }
+
+ @Override
+ public JsonGenerator write(double value) {
+ try {
+ generator.writeNumber(value);
+ } catch (IOException e) {
+ throw JacksonUtils.convertException(e);
+ }
+ return this;
+ }
+
+ @Override
+ public JsonGenerator write(boolean value) {
+ try {
+ generator.writeBoolean(value);
+ } catch (IOException e) {
+ throw JacksonUtils.convertException(e);
+ }
+ return this;
+ }
+
+ @Override
+ public JsonGenerator writeNull() {
+ try {
+ generator.writeNull();
+ } catch (IOException e) {
+ throw JacksonUtils.convertException(e);
+ }
+ return this;
+ }
+
+ @Override
+ public void close() {
+ try {
+ generator.close();
+ } catch (IOException e) {
+ throw JacksonUtils.convertException(e);
+ }
+ }
+
+ @Override
+ public void flush() {
+ try {
+ generator.flush();
+ } catch (IOException e) {
+ throw JacksonUtils.convertException(e);
+ }
+ }
+
+ private void writeValue(JsonValue value) throws IOException {
+ switch (value.getValueType()) {
+ case OBJECT:
+ generator.writeStartObject();
+ for (Map.Entry<String, JsonValue> entry: value.asJsonObject().entrySet()) {
+ generator.writeFieldName(entry.getKey());
+ writeValue(entry.getValue());
+ }
+ generator.writeEndObject();
+ break;
+
+ case ARRAY:
+ generator.writeStartArray();
+ for (JsonValue item: value.asJsonArray()) {
+ writeValue(item);
+ }
+ generator.writeEndArray();
+ break;
+
+ case STRING:
+ generator.writeString(((JsonString)value).getString());
+ break;
+
+ case FALSE:
+ generator.writeBoolean(false);
+ break;
+
+ case TRUE:
+ generator.writeBoolean(true);
+ break;
+
+ case NULL:
+ generator.writeNull();
+ break;
+
+ case NUMBER:
+ JsonNumber n = (JsonNumber) value;
+ if (n.isIntegral()) {
+ generator.writeNumber(n.longValue());
+ } else {
+ generator.writeNumber(n.doubleValue());
+ }
+ break;
+ }
+ }
+}
diff --git a/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/jackson/JacksonJsonpLocation.java b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/jackson/JacksonJsonpLocation.java
new file mode 100644
index 0000000000..132f40365e
--- /dev/null
+++ b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/jackson/JacksonJsonpLocation.java
@@ -0,0 +1,54 @@
+/****************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one *
+ * or more contributor license agreements. See the NOTICE file *
+ * distributed with this work for additional information *
+ * regarding copyright ownership. The ASF licenses this file *
+ * to you 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.apache.james.backends.opensearch.json.jackson;
+
+import jakarta.json.stream.JsonLocation;
+
+/**
+ * Translate a Jackson location to a JSONP location.
+ */
+public class JacksonJsonpLocation implements JsonLocation {
+
+ private final com.fasterxml.jackson.core.JsonLocation location;
+
+ JacksonJsonpLocation(com.fasterxml.jackson.core.JsonLocation location) {
+ this.location = location;
+ }
+
+ JacksonJsonpLocation(com.fasterxml.jackson.core.JsonParser parser) {
+ this(parser.getTokenLocation());
+ }
+
+ @Override
+ public long getLineNumber() {
+ return location.getLineNr();
+ }
+
+ @Override
+ public long getColumnNumber() {
+ return location.getColumnNr();
+ }
+
+ @Override
+ public long getStreamOffset() {
+ long charOffset = location.getCharOffset();
+ return charOffset == -1 ? location.getByteOffset() : charOffset;
+ }
+}
diff --git a/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/jackson/JacksonJsonpMapper.java b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/jackson/JacksonJsonpMapper.java
new file mode 100644
index 0000000000..692b0e937c
--- /dev/null
+++ b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/jackson/JacksonJsonpMapper.java
@@ -0,0 +1,123 @@
+/****************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one *
+ * or more contributor license agreements. See the NOTICE file *
+ * distributed with this work for additional information *
+ * regarding copyright ownership. The ASF licenses this file *
+ * to you 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.apache.james.backends.opensearch.json.jackson;
+
+import java.io.IOException;
+import java.util.EnumSet;
+
+import org.apache.james.backends.opensearch.json.JsonpDeserializerBase;
+import org.apache.james.backends.opensearch.json.JsonpMapperBase;
+import org.opensearch.client.json.JsonpDeserializer;
+import org.opensearch.client.json.JsonpMapper;
+import org.opensearch.client.json.JsonpSerializer;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.core.JsonFactory;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
+
+import jakarta.json.spi.JsonProvider;
+import jakarta.json.stream.JsonGenerator;
+import jakarta.json.stream.JsonParser;
+
+public class JacksonJsonpMapper extends JsonpMapperBase {
+
+ private final JacksonJsonProvider provider;
+ private final ObjectMapper objectMapper;
+
+ public JacksonJsonpMapper(ObjectMapper objectMapper) {
+ this(objectMapper, new JsonFactory());
+ }
+
+ public JacksonJsonpMapper(ObjectMapper objectMapper, JsonFactory jsonFactory) {
+ this.provider = new JacksonJsonProvider(jsonFactory);
+ this.objectMapper = objectMapper
+ .configure(SerializationFeature.INDENT_OUTPUT, false)
+ .setSerializationInclusion(JsonInclude.Include.NON_NULL);
+ }
+
+ public JacksonJsonpMapper() {
+ this(new ObjectMapper());
+ }
+
+ /**
+ * Returns the underlying Jackson mapper.
+ */
+ public ObjectMapper objectMapper() {
+ return this.objectMapper;
+ }
+
+ @Override
+ public JsonProvider jsonProvider() {
+ return provider;
+ }
+
+ @Override
+ protected <T> JsonpDeserializer<T> getDefaultDeserializer(Class<T> clazz) {
+ return new JacksonValueParser<>(clazz);
+ }
+
+ @Override
+ public <T> void serialize(T value, JsonGenerator generator) {
+
+ if (!(generator instanceof JacksonJsonpGenerator)) {
+ throw new IllegalArgumentException("Jackson's ObjectMapper can only be used with the JacksonJsonpProvider");
+ }
+
+ JsonpSerializer<T> serializer = findSerializer(value);
+ if (serializer != null) {
+ serializer.serialize(value, generator, this);
+ return;
+ }
+
+ com.fasterxml.jackson.core.JsonGenerator jkGenerator = ((JacksonJsonpGenerator)generator).jacksonGenerator();
+ try {
+ objectMapper.writeValue(jkGenerator, value);
+ } catch (IOException ioe) {
+ throw JacksonUtils.convertException(ioe);
+ }
+ }
+
+ private class JacksonValueParser<T> extends JsonpDeserializerBase<T> {
+
+ private final Class<T> clazz;
+
+ protected JacksonValueParser(Class<T> clazz) {
+ super(EnumSet.allOf(JsonParser.Event.class));
+ this.clazz = clazz;
+ }
+
+ @Override
+ public T deserialize(JsonParser parser, JsonpMapper mapper, JsonParser.Event event) {
+
+ if (!(parser instanceof JacksonJsonpParser)) {
+ throw new IllegalArgumentException("Jackson's ObjectMapper can only be used with the JacksonJsonpProvider");
+ }
+
+ com.fasterxml.jackson.core.JsonParser jkParser = ((JacksonJsonpParser)parser).jacksonParser();
+
+ try {
+ return objectMapper.readValue(jkParser, clazz);
+ } catch (IOException ioe) {
+ throw JacksonUtils.convertException(ioe);
+ }
+ }
+ }
+}
diff --git a/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/jackson/JacksonJsonpParser.java b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/jackson/JacksonJsonpParser.java
new file mode 100644
index 0000000000..b1dcacccb2
--- /dev/null
+++ b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/jackson/JacksonJsonpParser.java
@@ -0,0 +1,313 @@
+/****************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one *
+ * or more contributor license agreements. See the NOTICE file *
+ * distributed with this work for additional information *
+ * regarding copyright ownership. The ASF licenses this file *
+ * to you 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.apache.james.backends.opensearch.json.jackson;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.util.EnumMap;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.stream.Stream;
+
+import com.fasterxml.jackson.core.JsonToken;
+
+import jakarta.json.JsonArray;
+import jakarta.json.JsonObject;
+import jakarta.json.JsonValue;
+import jakarta.json.stream.JsonLocation;
+import jakarta.json.stream.JsonParser;
+import jakarta.json.stream.JsonParsingException;
+
+/**
+ * A JSONP parser implementation on top of Jackson.
+ * <p>
+ * <b>Warning:</b> this implementation isn't fully compliant with the JSONP specification: calling {@link #hasNext()}
+ * moves forward the underlying Jackson parser as Jackson doesn't provide an equivalent method. This means no value
+ * getter method (e.g. {@link #getInt()} or {@link #getString()} should be called until the next call to {@link #next()}.
+ * Such calls will throw an {@code IllegalStateException}.
+ */
+public class JacksonJsonpParser implements JsonParser {
+
+ private final com.fasterxml.jackson.core.JsonParser parser;
+
+ private boolean hasNextWasCalled = false;
+
+ private static final EnumMap<JsonToken, Event> tokenToEvent;
+
+ static {
+ tokenToEvent = new EnumMap<>(JsonToken.class);
+ tokenToEvent.put(JsonToken.END_ARRAY, Event.END_ARRAY);
+ tokenToEvent.put(JsonToken.END_OBJECT, Event.END_OBJECT);
+ tokenToEvent.put(JsonToken.FIELD_NAME, Event.KEY_NAME);
+ tokenToEvent.put(JsonToken.START_ARRAY, Event.START_ARRAY);
+ tokenToEvent.put(JsonToken.START_OBJECT, Event.START_OBJECT);
+ tokenToEvent.put(JsonToken.VALUE_FALSE, Event.VALUE_FALSE);
+ tokenToEvent.put(JsonToken.VALUE_NULL, Event.VALUE_NULL);
+ tokenToEvent.put(JsonToken.VALUE_NUMBER_FLOAT, Event.VALUE_NUMBER);
+ tokenToEvent.put(JsonToken.VALUE_NUMBER_INT, Event.VALUE_NUMBER);
+ tokenToEvent.put(JsonToken.VALUE_STRING, Event.VALUE_STRING);
+ tokenToEvent.put(JsonToken.VALUE_TRUE, Event.VALUE_TRUE);
+
+ // No equivalent for
+ // - VALUE_EMBEDDED_OBJECT
+ // - NOT_AVAILABLE
+ }
+
+ public JacksonJsonpParser(com.fasterxml.jackson.core.JsonParser parser) {
+ this.parser = parser;
+ }
+
+ /**
+ * Returns the underlying Jackson parser.
+ */
+ public com.fasterxml.jackson.core.JsonParser jacksonParser() {
+ return this.parser;
+ }
+
+ private JsonParsingException convertException(IOException ioe) {
+ return new JsonParsingException("Jackson exception: " + ioe.getMessage(), ioe, getLocation());
+ }
+
+ private JsonToken fetchNextToken() {
+ try {
+ return parser.nextToken();
+ } catch (IOException e) {
+ throw convertException(e);
+ }
+ }
+
+ private void ensureTokenIsCurrent() {
+ if (hasNextWasCalled) {
+ throw new IllegalStateException("Cannot get event data as parser as already been moved to the next event");
+ }
+ }
+
+ @Override
+ public boolean hasNext() {
+ if (hasNextWasCalled) {
+ return parser.currentToken() != null;
+ } else {
+ hasNextWasCalled = true;
+ return fetchNextToken() != null;
+ }
+ }
+
+ @Override
+ public Event next() {
+ JsonToken token;
+ if (hasNextWasCalled) {
+ token = parser.getCurrentToken();
+ hasNextWasCalled = false;
+ } else {
+ token = fetchNextToken();
+ }
+
+ if (token == null) {
+ throw new NoSuchElementException();
+ }
+
+ Event result = tokenToEvent.get(token);
+ if (result == null) {
+ throw new JsonParsingException("Unsupported Jackson event type '" + token + "'", getLocation());
+ }
+
+ return result;
+ }
+
+ @Override
+ public String getString() {
+ ensureTokenIsCurrent();
+ try {
+ return parser.getValueAsString();
+ } catch (IOException e) {
+ throw convertException(e);
+ }
+ }
+
+ @Override
+ public boolean isIntegralNumber() {
+ ensureTokenIsCurrent();
+ return parser.isExpectedNumberIntToken();
+ }
+
+ @Override
+ public int getInt() {
+ ensureTokenIsCurrent();
+ try {
+ return parser.getIntValue();
+ } catch (IOException e) {
+ throw convertException(e);
+ }
+ }
+
+ @Override
+ public long getLong() {
+ ensureTokenIsCurrent();
+ try {
+ return parser.getLongValue();
+ } catch (IOException e) {
+ throw convertException(e);
+ }
+ }
+
+ @Override
+ public BigDecimal getBigDecimal() {
+ ensureTokenIsCurrent();
+ try {
+ return parser.getDecimalValue();
+ } catch (IOException e) {
+ throw convertException(e);
+ }
+ }
+
+ @Override
+ public JsonLocation getLocation() {
+ return new JacksonJsonpLocation(parser.getCurrentLocation());
+ }
+
+ @Override
+ public void close() {
+ try {
+ parser.close();
+ } catch (IOException e) {
+ throw convertException(e);
+ }
+ }
+
+ private JsonValueParser valueParser;
+
+ @Override
+ public JsonObject getObject() {
+ ensureTokenIsCurrent();
+ if (parser.currentToken() != JsonToken.START_OBJECT) {
+ throw new IllegalStateException("Unexpected event '" + parser.currentToken() +
+ "' at " + parser.getTokenLocation());
+ }
+ if (valueParser == null) {
+ valueParser = new JsonValueParser();
+ }
+ try {
+ return valueParser.parseObject(parser);
+ } catch (IOException e) {
+ throw convertException(e);
+ }
+ }
+
+ @Override
+ public JsonArray getArray() {
+ ensureTokenIsCurrent();
+ if (valueParser == null) {
+ valueParser = new JsonValueParser();
+ }
+ if (parser.currentToken() != JsonToken.START_ARRAY) {
+ throw new IllegalStateException("Unexpected event '" + parser.currentToken() +
+ "' at " + parser.getTokenLocation());
+ }
+ try {
+ return valueParser.parseArray(parser);
+ } catch (IOException e) {
+ throw convertException(e);
+ }
+ }
+
+ @Override
+ public JsonValue getValue() {
+ ensureTokenIsCurrent();
+ if (valueParser == null) {
+ valueParser = new JsonValueParser();
+ }
+ try {
+ return valueParser.parseValue(parser);
+ } catch (IOException e) {
+ throw convertException(e);
+ }
+ }
+
+ @Override
+ public void skipObject() {
+ ensureTokenIsCurrent();
+ if (parser.currentToken() != JsonToken.START_OBJECT) {
+ return;
+ }
+
+ try {
+ int depth = 1;
+ JsonToken token;
+ do {
+ token = parser.nextToken();
+ switch (token) {
+ case START_OBJECT:
+ depth++;
+ break;
+ case END_OBJECT:
+ depth--;
+ break;
+ }
+ } while (!(token == JsonToken.END_OBJECT && depth == 0));
+ } catch (IOException e) {
+ throw convertException(e);
+ }
+ }
+
+ @Override
+ public void skipArray() {
+ ensureTokenIsCurrent();
+ if (parser.currentToken() != JsonToken.START_ARRAY) {
+ return;
+ }
+
+ try {
+ int depth = 1;
+ JsonToken token;
+ do {
+ token = parser.nextToken();
+ switch (token) {
+ case START_ARRAY:
+ depth++;
+ break;
+ case END_ARRAY:
+ depth--;
+ break;
+ }
+ } while (!(token == JsonToken.END_ARRAY && depth == 0));
+ } catch (IOException e) {
+ throw convertException(e);
+ }
+ }
+
+ @Override
+ public Stream<Map.Entry<String, JsonValue>> getObjectStream() {
+ return getObject().entrySet().stream();
+ }
+
+ @Override
+ public Stream<JsonValue> getArrayStream() {
+ return getArray().stream();
+ }
+
+ /**
+ * Not implemented.
+ */
+ @Override
+ public Stream<JsonValue> getValueStream() {
+ return JsonParser.super.getValueStream();
+ }
+}
+
diff --git a/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/jackson/JacksonUtils.java b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/jackson/JacksonUtils.java
new file mode 100644
index 0000000000..d5e7782395
--- /dev/null
+++ b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/jackson/JacksonUtils.java
@@ -0,0 +1,43 @@
+/****************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one *
+ * or more contributor license agreements. See the NOTICE file *
+ * distributed with this work for additional information *
+ * regarding copyright ownership. The ASF licenses this file *
+ * to you 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.apache.james.backends.opensearch.json.jackson;
+
+import java.io.IOException;
+
+import com.fasterxml.jackson.core.JsonParseException;
+
+import jakarta.json.JsonException;
+import jakarta.json.stream.JsonGenerationException;
+import jakarta.json.stream.JsonParsingException;
+
+class JacksonUtils {
+ public static JsonException convertException(IOException ioe) {
+ if (ioe instanceof com.fasterxml.jackson.core.JsonGenerationException) {
+ return new JsonGenerationException(ioe.getMessage(), ioe);
+
+ } else if (ioe instanceof JsonParseException) {
+ JsonParseException jpe = (JsonParseException) ioe;
+ return new JsonParsingException(ioe.getMessage(), jpe, new JacksonJsonpLocation(jpe.getLocation()));
+
+ } else {
+ return new JsonException("Jackson exception", ioe);
+ }
+ }
+}
diff --git a/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/jackson/JsonValueParser.java b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/jackson/JsonValueParser.java
new file mode 100644
index 0000000000..a66bbd3edb
--- /dev/null
+++ b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/jackson/JsonValueParser.java
@@ -0,0 +1,108 @@
+/****************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one *
+ * or more contributor license agreements. See the NOTICE file *
+ * distributed with this work for additional information *
+ * regarding copyright ownership. The ASF licenses this file *
+ * to you 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.apache.james.backends.opensearch.json.jackson;
+
+import java.io.IOException;
+
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.core.JsonToken;
+
+import jakarta.json.JsonArray;
+import jakarta.json.JsonArrayBuilder;
+import jakarta.json.JsonObject;
+import jakarta.json.JsonObjectBuilder;
+import jakarta.json.JsonValue;
+import jakarta.json.spi.JsonProvider;
+import jakarta.json.stream.JsonParsingException;
+
+/**
+ * Reads a Jsonp value/object/array from a Jackson parser. The parser's current token should be the start of the
+ * object (e.g. START_OBJECT, VALUE_NUMBER, etc).
+ */
+class JsonValueParser {
+ private static final JsonProvider provider = JsonProvider.provider();
+
+ public JsonObject parseObject(JsonParser parser) throws IOException {
+
+ JsonObjectBuilder ob = provider.createObjectBuilder();
+
+ JsonToken token;
+ while ((token = parser.nextToken()) != JsonToken.END_OBJECT) {
+ if (token != JsonToken.FIELD_NAME) {
+ throw new JsonParsingException("Expected a property name", new JacksonJsonpLocation(parser));
+ }
+ String name = parser.getCurrentName();
+ parser.nextToken();
+ ob.add(name, parseValue(parser));
+ }
+ return ob.build();
+ }
+
+ public JsonArray parseArray(JsonParser parser) throws IOException {
+ JsonArrayBuilder ab = provider.createArrayBuilder();
+
+ while (parser.nextToken() != JsonToken.END_ARRAY) {
+ ab.add(parseValue(parser));
+ }
+ return ab.build();
+ }
+
+ public JsonValue parseValue(JsonParser parser) throws IOException {
+ switch (parser.currentToken()) {
+ case START_OBJECT:
+ return parseObject(parser);
+
+ case START_ARRAY:
+ return parseArray(parser);
+
+ case VALUE_TRUE:
+ return JsonValue.TRUE;
+
+ case VALUE_FALSE:
+ return JsonValue.FALSE;
+
+ case VALUE_NULL:
+ return JsonValue.NULL;
+
+ case VALUE_STRING:
+ return provider.createValue(parser.getText());
+
+ case VALUE_NUMBER_FLOAT:
+ case VALUE_NUMBER_INT:
+ switch (parser.getNumberType()) {
+ case INT:
+ return provider.createValue(parser.getIntValue());
+ case LONG:
+ return provider.createValue(parser.getLongValue());
+ case FLOAT:
+ case DOUBLE:
+ return provider.createValue(parser.getDoubleValue());
+ case BIG_DECIMAL:
+ return provider.createValue(parser.getDecimalValue());
+ case BIG_INTEGER:
+ return provider.createValue(parser.getBigIntegerValue());
+ }
+
+ default:
+ throw new JsonParsingException("Unexpected token '" + parser.currentToken() + "'", new JacksonJsonpLocation(parser));
+
+ }
+ }
+}
diff --git a/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/package-info.java b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/package-info.java
new file mode 100644
index 0000000000..db5d3f4c13
--- /dev/null
+++ b/backends-common/opensearch/src/main/java/org/apache/james/backends/opensearch/json/package-info.java
@@ -0,0 +1,29 @@
+/****************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one *
+ * or more contributor license agreements. See the NOTICE file *
+ * distributed with this work for additional information *
+ * regarding copyright ownership. The ASF licenses this file *
+ * to you 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.apache.james.backends.opensearch.json;
+
+/**
+ * See {@link https://github.com/opensearch-project/opensearch-java/issues/292}
+ *
+ * The Jackson related code was copied from {@link https://github.com/opensearch-project/opensearch-java} in order to get rid of a SPI lookup for each deserialized request,
+ * that was happening on the event loop.
+ *
+ * The only modified class is {@link jackson.JsonValueParser} where the SPI lookup was made static.
+ */
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 12/12: JAMES-3771 Date searching should align IMAP4rev2 specifications
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 19984d6765d6caf539858025d954e18284df98c0
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Tue Jan 3 17:35:44 2023 +0700
JAMES-3771 Date searching should align IMAP4rev2 specifications
---
.../org/apache/james/mailbox/opensearch/query/CriterionConverter.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/query/CriterionConverter.java b/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/query/CriterionConverter.java
index d7b019c8ce..9efe66925a 100644
--- a/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/query/CriterionConverter.java
+++ b/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/query/CriterionConverter.java
@@ -470,13 +470,13 @@ public class CriterionConverter {
case BEFORE:
return new RangeQuery.Builder()
.field(field)
- .lt(JsonData.of(upDateString))
+ .lt(JsonData.of(lowDateString))
.build()
._toQuery();
case AFTER:
return new RangeQuery.Builder()
.field(field)
- .gte(JsonData.of(lowDateString))
+ .gte(JsonData.of(upDateString))
.build()
._toQuery();
case ON:
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 11/12: JAMES-3771 Fix thread leak issue within DeleteAllIndexDocumentsCleanupStrategy
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 250a79dcf0aef85badda0dfb5f9bc76f2c07a516
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Tue Jan 3 17:33:15 2023 +0700
JAMES-3771 Fix thread leak issue within DeleteAllIndexDocumentsCleanupStrategy
---
.../opensearch/DockerOpenSearchExtension.java | 19 ++++++++++---------
1 file changed, 10 insertions(+), 9 deletions(-)
diff --git a/backends-common/opensearch/src/test/java/org/apache/james/backends/opensearch/DockerOpenSearchExtension.java b/backends-common/opensearch/src/test/java/org/apache/james/backends/opensearch/DockerOpenSearchExtension.java
index aee8790003..0f27bdd978 100644
--- a/backends-common/opensearch/src/test/java/org/apache/james/backends/opensearch/DockerOpenSearchExtension.java
+++ b/backends-common/opensearch/src/test/java/org/apache/james/backends/opensearch/DockerOpenSearchExtension.java
@@ -47,6 +47,7 @@ public class DockerOpenSearchExtension implements AfterEachCallback, BeforeEachC
public static class DeleteAllIndexDocumentsCleanupStrategy implements CleanupStrategy {
private final WriteAliasName aliasName;
+ private ReactorOpenSearchClient client;
public DeleteAllIndexDocumentsCleanupStrategy(WriteAliasName aliasName) {
this.aliasName = aliasName;
@@ -58,7 +59,7 @@ public class DockerOpenSearchExtension implements AfterEachCallback, BeforeEachC
.ignoreExceptions()
.until(() -> {
openSearch.flushIndices();
- ReactorOpenSearchClient client = openSearch.clientProvider().get();
+ ReactorOpenSearchClient client = client(openSearch);
new DeleteByQueryPerformer(client, aliasName)
.perform(new MatchAllQuery.Builder().build()._toQuery())
.block();
@@ -66,18 +67,18 @@ public class DockerOpenSearchExtension implements AfterEachCallback, BeforeEachC
.query(new MatchAllQuery.Builder().build()._toQuery())
.build();
openSearch.flushIndices();
- boolean result = client.search(searchRequest)
+ return client.search(searchRequest)
.map(searchResponse -> searchResponse.hits().hits().size())
.block() == 0;
-
- try {
- client.close();
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- return result;
});
}
+
+ private ReactorOpenSearchClient client(DockerOpenSearch openSearch) {
+ if (client == null) {
+ client = openSearch.clientProvider().get();
+ }
+ return client;
+ }
}
private final DockerOpenSearch openSearch = DockerOpenSearchSingleton.INSTANCE;
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 06/12: JAMES-3771 Upgrade to opensearch java client 2.1.0
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 7e1c35fa64d7a1e0dcfcd18a6d6880497f933fa6
Author: Rene Cordier <rc...@linagora.com>
AuthorDate: Tue Aug 2 10:55:52 2022 +0700
JAMES-3771 Upgrade to opensearch java client 2.1.0
With the fix on null sort fields
---
backends-common/opensearch/pom.xml | 2 +-
.../apache/james/mailbox/opensearch/OpenSearchIntegrationTest.java | 6 ------
2 files changed, 1 insertion(+), 7 deletions(-)
diff --git a/backends-common/opensearch/pom.xml b/backends-common/opensearch/pom.xml
index a53014a685..86237bd857 100644
--- a/backends-common/opensearch/pom.xml
+++ b/backends-common/opensearch/pom.xml
@@ -83,7 +83,7 @@
<dependency>
<groupId>org.opensearch.client</groupId>
<artifactId>opensearch-java</artifactId>
- <version>2.0.0</version>
+ <version>2.1.0</version>
</dependency>
<dependency>
<groupId>org.opensearch.client</groupId>
diff --git a/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/OpenSearchIntegrationTest.java b/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/OpenSearchIntegrationTest.java
index c87203b756..4d6c23f7a1 100644
--- a/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/OpenSearchIntegrationTest.java
+++ b/mailbox/opensearch/src/test/java/org/apache/james/mailbox/opensearch/OpenSearchIntegrationTest.java
@@ -367,12 +367,6 @@ class OpenSearchIntegrationTest extends AbstractMessageSearchIndexTest {
.containsOnly(messageId2.getUid());
}
- @Disabled("JAMES-3771 Waiting for a fix on opensearch client to be merged: https://github.com/opensearch-project/opensearch-java/pull/169")
- @Test
- public void sortOnCcShouldWork() throws Exception {
-
- }
-
@Disabled("MAILBOX-403 Relaxed the matching constraints for email addresses in text bodies to reduce OpenSearch disk space usage")
@Test
public void textShouldNotMatchOtherAddressesOfTheSameDomain() {
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 09/12: JAMES-3771 Refactor extract operations from hit in OpenSearchListeningMessageSearchIndex
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 5934ecaa4d543b2fd8ff1fb41ad853d12aacbb67
Author: Rene Cordier <rc...@linagora.com>
AuthorDate: Wed Dec 14 17:13:46 2022 +0700
JAMES-3771 Refactor extract operations from hit in OpenSearchListeningMessageSearchIndex
JsonpDeserializer.arrayDeserializer seems to do some blocking action in its code, resulting in severe issues when running perf tests on James with it.
---
.../events/OpenSearchListeningMessageSearchIndex.java | 15 ++++++++++-----
1 file changed, 10 insertions(+), 5 deletions(-)
diff --git a/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/events/OpenSearchListeningMessageSearchIndex.java b/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/events/OpenSearchListeningMessageSearchIndex.java
index 420d734d31..5284ec9d2e 100644
--- a/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/events/OpenSearchListeningMessageSearchIndex.java
+++ b/mailbox/opensearch/src/main/java/org/apache/james/mailbox/opensearch/events/OpenSearchListeningMessageSearchIndex.java
@@ -62,7 +62,6 @@ import org.apache.james.mailbox.store.MailboxSessionMapperFactory;
import org.apache.james.mailbox.store.mail.model.MailboxMessage;
import org.apache.james.mailbox.store.search.ListeningMessageSearchIndex;
import org.opensearch.client.json.JsonData;
-import org.opensearch.client.json.JsonpDeserializer;
import org.opensearch.client.opensearch._types.FieldValue;
import org.opensearch.client.opensearch._types.query_dsl.Query;
import org.opensearch.client.opensearch._types.query_dsl.TermQuery;
@@ -273,8 +272,11 @@ public class OpenSearchListeningMessageSearchIndex extends ListeningMessageSearc
private void extractMessageIdFromHit(Hit<ObjectNode> hit, SynchronousSink<MessageId> sink) {
JsonData messageId = hit.fields().get(MESSAGE_ID);
if (messageId != null) {
- List<String> extractMessageId = messageId.deserialize(JsonpDeserializer.arrayDeserializer(JsonpDeserializer.stringDeserializer()));
- sink.next(messageIdFactory.fromString(extractMessageId.get(0)));
+ String messageIdAsString = messageId.toJson()
+ .asJsonArray()
+ .getString(0);
+
+ sink.next(messageIdFactory.fromString(messageIdAsString));
} else {
LOGGER.warn("Can not extract UID, MessageID and/or MailboxId for search result {}", hit.id());
}
@@ -283,8 +285,11 @@ public class OpenSearchListeningMessageSearchIndex extends ListeningMessageSearc
private void extractUidFromHit(Hit<ObjectNode> hit, SynchronousSink<MessageUid> sink) {
JsonData uid = hit.fields().get(UID);
if (uid != null) {
- List<Number> uidAsNumber = uid.deserialize(JsonpDeserializer.arrayDeserializer(JsonpDeserializer.numberDeserializer()));
- sink.next(MessageUid.of(uidAsNumber.get(0).longValue()));
+ int uidAsInt = uid.toJson()
+ .asJsonArray()
+ .getInt(0);
+
+ sink.next(MessageUid.of(uidAsInt));
} else {
LOGGER.warn("Can not extract UID, MessageID and/or MailboxId for search result {}", hit.id());
}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org