You are viewing a plain text version of this content. The canonical link for it is here.
Posted to server-dev@james.apache.org by bt...@apache.org on 2020/04/08 02:25:45 UTC
[james-project] branch master updated (e2596a7 -> b5ed9c2)
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 e2596a7 [Enhencement] Add glowroot pugins
new 1469c4a JAMES-2997 Drop Cassandra schema version prior version 5
new 32bea1d [Release] Post 3.5.0 release removal
new 97e06e9 JAMES-3144 Use ElasticSearch reactively
new a55a1a6 JAMES-3144 Ensure ElasticSearch Docker is running before each test
new a8dcf21 JAMES-3144 Reactify ScrollSearch
new b5ed9c2 JAMES-3144 Increase ElasticSearch timeout in RabbitMQReindexingWithEventDeadLettersTest
The 6 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:
.../versions/CassandraSchemaVersionManager.java | 2 +-
.../apache/james/backends/es/ClientProvider.java | 12 +-
.../james/backends/es/DeleteByQueryPerformer.java | 17 +-
.../backends/es/ElasticSearchHealthCheck.java | 5 +-
.../james/backends/es/ElasticSearchIndexer.java | 67 ++-
.../james/backends/es/IndexCreationFactory.java | 11 +-
.../apache/james/backends/es/ListenerToFuture.java | 51 ---
.../james/backends/es/NodeMappingFactory.java | 7 +-
.../backends/es/ReactorElasticSearchClient.java | 171 ++++++++
.../james/backends/es/search/ScrolledSearch.java | 122 +++---
.../es/ClientProviderImplConnectionContract.java | 5 +-
.../backends/es/DockerElasticSearchExtension.java | 11 +-
.../es/ElasticSearchHealthCheckConnectionTest.java | 3 +-
.../backends/es/ElasticSearchIndexerTest.java | 72 ++--
.../backends/es/IndexCreationFactoryTest.java | 3 +-
.../backends/es/NodeMappingFactoryAuthTest.java | 3 +-
.../james/backends/es/NodeMappingFactoryTest.java | 3 +-
.../backends/es/search/ScrolledSearchTest.java | 37 +-
.../CassandraMailboxSessionMapperFactory.java | 7 +-
.../cassandra/mail/CassandraAttachmentMapper.java | 11 +-
.../migration/AttachmentMessageIdCreation.java | 62 ---
.../mail/migration/AttachmentV2Migration.java | 65 ---
.../CassandraSubscriptionManagerTest.java | 4 -
.../cassandra/mail/CassandraAttachmentDAOTest.java | 118 ------
.../mail/CassandraAttachmentFallbackTest.java | 180 --------
.../migration/AttachmentMessageIdCreationTest.java | 198 ---------
.../mail/migration/AttachmentV2MigrationTest.java | 221 ----------
.../SolveMailboxInconsistenciesServiceTest.java | 2 +-
.../elasticsearch/MailboxIndexCreationUtil.java | 14 +-
.../ElasticSearchListeningMessageSearchIndex.java | 54 +--
.../search/ElasticSearchSearcher.java | 23 +-
.../ElasticSearchIntegrationTest.java | 4 +-
...asticSearchListeningMessageSearchIndexTest.java | 10 +-
.../search/ElasticSearchSearcherTest.java | 4 +-
.../elasticsearch/ElasticSearchQuotaSearcher.java | 35 +-
.../QuotaSearchIndexCreationUtil.java | 6 +-
.../events/ElasticSearchQuotaMailboxListener.java | 8 +-
...lasticSearchQuotaSearchTestSystemExtension.java | 4 +-
.../ElasticSearchQuotaMailboxListenerTest.java | 13 +-
mpt/antlib/LICENSE.txt | 176 --------
mpt/antlib/NOTICE.txt | 11 -
mpt/antlib/pom.xml | 165 --------
.../apache/james/mpt/ant/MailProtocolTestTask.java | 426 -------------------
.../resources/org/apache/james/mpt/ant/antlib.xml | 26 --
mpt/antlib/src/reporting-site/site.xml | 29 --
.../src/site/resources/images/asf-logo-reduced.gif | Bin 6636 -> 0 bytes
.../src/site/resources/images/james-mpt-logo.gif | Bin 5179 -> 0 bytes
mpt/antlib/src/site/site.xml | 37 --
mpt/antlib/src/site/xdoc/index.xml | 183 --------
.../java/org/apache/james/mpt/ant/TestAddUser.java | 68 ---
.../org/apache/james/mpt/ant/TestRunScripts.java | 102 -----
mpt/antlib/src/test/resources/build.xml | 140 -------
.../host/ElasticSearchHostSystem.java | 4 +-
mpt/pom.xml | 1 -
.../modules/mailbox/CassandraMailboxModule.java | 2 -
.../modules/mailbox/ElasticSearchClientModule.java | 4 +-
.../mailbox/ElasticSearchMailboxModule.java | 10 +-
.../mailbox/ElasticSearchQuotaSearcherModule.java | 10 +-
.../modules/mailbox/ElasticSearchStartUpCheck.java | 6 +-
.../modules/webadmin/CassandraRoutesModule.java | 8 -
.../apache/james/DockerElasticSearchExtension.java | 10 +-
.../test/java/org/apache/james/ESReporterTest.java | 5 +-
.../metric/es/DockerElasticSearchExtension.java | 3 +
.../james/transport/mailets/BayesianAnalysis.java | 460 ---------------------
.../transport/mailets/BayesianAnalysisFeeder.java | 336 ---------------
.../james/transport/mailets/ToRecipientFolder.java | 99 -----
.../mailets/delivery/ToRecipientFolderTest.java | 198 ---------
...RabbitMQReindexingWithEventDeadLettersTest.java | 4 +-
.../routes/ElasticSearchQuotaSearchExtension.java | 4 +-
upgrade-instructions.md | 12 +
70 files changed, 510 insertions(+), 3674 deletions(-)
delete mode 100644 backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ListenerToFuture.java
create mode 100644 backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ReactorElasticSearchClient.java
delete mode 100644 mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/migration/AttachmentMessageIdCreation.java
delete mode 100644 mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/migration/AttachmentV2Migration.java
delete mode 100644 mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraAttachmentDAOTest.java
delete mode 100644 mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraAttachmentFallbackTest.java
delete mode 100644 mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/migration/AttachmentMessageIdCreationTest.java
delete mode 100644 mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/migration/AttachmentV2MigrationTest.java
delete mode 100644 mpt/antlib/LICENSE.txt
delete mode 100644 mpt/antlib/NOTICE.txt
delete mode 100644 mpt/antlib/pom.xml
delete mode 100644 mpt/antlib/src/main/java/org/apache/james/mpt/ant/MailProtocolTestTask.java
delete mode 100644 mpt/antlib/src/main/resources/org/apache/james/mpt/ant/antlib.xml
delete mode 100644 mpt/antlib/src/reporting-site/site.xml
delete mode 100644 mpt/antlib/src/site/resources/images/asf-logo-reduced.gif
delete mode 100644 mpt/antlib/src/site/resources/images/james-mpt-logo.gif
delete mode 100644 mpt/antlib/src/site/site.xml
delete mode 100644 mpt/antlib/src/site/xdoc/index.xml
delete mode 100644 mpt/antlib/src/test/java/org/apache/james/mpt/ant/TestAddUser.java
delete mode 100644 mpt/antlib/src/test/java/org/apache/james/mpt/ant/TestRunScripts.java
delete mode 100644 mpt/antlib/src/test/resources/build.xml
delete mode 100644 server/mailet/mailets/src/main/java/org/apache/james/transport/mailets/BayesianAnalysis.java
delete mode 100644 server/mailet/mailets/src/main/java/org/apache/james/transport/mailets/BayesianAnalysisFeeder.java
delete mode 100644 server/mailet/mailets/src/main/java/org/apache/james/transport/mailets/ToRecipientFolder.java
delete mode 100644 server/mailet/mailets/src/test/java/org/apache/james/transport/mailets/delivery/ToRecipientFolderTest.java
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 04/06: JAMES-3144 Ensure ElasticSearch Docker is
running before 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 a55a1a6bc88e2eae99d332bae288fdbc1eb432fe
Author: Gautier DI FOLCO <gd...@linagora.com>
AuthorDate: Tue Mar 17 09:11:21 2020 +0100
JAMES-3144 Ensure ElasticSearch Docker is running before each test
---
.../backends/es/DockerElasticSearchExtension.java | 11 ++++++++++-
.../ElasticSearchListeningMessageSearchIndex.java | 20 ++++++++++----------
.../apache/james/DockerElasticSearchExtension.java | 10 +++++++++-
.../metric/es/DockerElasticSearchExtension.java | 3 +++
4 files changed, 32 insertions(+), 12 deletions(-)
diff --git a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/DockerElasticSearchExtension.java b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/DockerElasticSearchExtension.java
index 6583954..438517a 100644
--- a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/DockerElasticSearchExtension.java
+++ b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/DockerElasticSearchExtension.java
@@ -20,12 +20,13 @@
package org.apache.james.backends.es;
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;
-public class DockerElasticSearchExtension implements AfterEachCallback, ParameterResolver {
+public class DockerElasticSearchExtension implements AfterEachCallback, BeforeEachCallback, ParameterResolver {
private final DockerElasticSearch elasticSearch = DockerElasticSearchSingleton.INSTANCE;
@@ -35,6 +36,14 @@ public class DockerElasticSearchExtension implements AfterEachCallback, Paramete
}
@Override
+ public void beforeEach(ExtensionContext extensionContext) {
+ if (!elasticSearch.isRunning()) {
+ elasticSearch.unpause();
+ }
+ awaitForElasticSearch();
+ }
+
+ @Override
public boolean supportsParameter(ParameterContext parameterContext, ExtensionContext extensionContext) throws ParameterResolutionException {
return (parameterContext.getParameter().getType() == DockerElasticSearch.class);
}
diff --git a/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/events/ElasticSearchListeningMessageSearchIndex.java b/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/events/ElasticSearchListeningMessageSearchIndex.java
index 1430111..b64ecb3 100644
--- a/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/events/ElasticSearchListeningMessageSearchIndex.java
+++ b/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/events/ElasticSearchListeningMessageSearchIndex.java
@@ -166,12 +166,12 @@ public class ElasticSearchListeningMessageSearchIndex extends ListeningMessageSe
@Override
public void delete(MailboxSession session, Mailbox mailbox, Collection<MessageUid> expungedUids) {
- elasticSearchIndexer
- .delete(expungedUids.stream()
- .map(uid -> indexIdFor(mailbox, uid))
- .collect(Guavate.toImmutableList()),
- routingKeyFactory.from(mailbox.getMailboxId()))
- .block();
+ elasticSearchIndexer
+ .delete(expungedUids.stream()
+ .map(uid -> indexIdFor(mailbox, uid))
+ .collect(Guavate.toImmutableList()),
+ routingKeyFactory.from(mailbox.getMailboxId()))
+ .block();
}
@Override
@@ -198,10 +198,10 @@ public class ElasticSearchListeningMessageSearchIndex extends ListeningMessageSe
}
private UpdatedRepresentation createUpdatedDocumentPartFromUpdatedFlags(Mailbox mailbox, UpdatedFlags updatedFlags) throws JsonProcessingException {
- return new UpdatedRepresentation(
- indexIdFor(mailbox, updatedFlags.getUid()),
- messageToElasticSearchJson
- .getUpdatedJsonMessagePart(updatedFlags.getNewFlags(), updatedFlags.getModSeq()));
+ return new UpdatedRepresentation(
+ indexIdFor(mailbox, updatedFlags.getUid()),
+ messageToElasticSearchJson
+ .getUpdatedJsonMessagePart(updatedFlags.getNewFlags(), updatedFlags.getModSeq()));
}
private DocumentId indexIdFor(Mailbox mailbox, MessageUid uid) {
diff --git a/server/container/guice/cassandra-guice/src/test/java/org/apache/james/DockerElasticSearchExtension.java b/server/container/guice/cassandra-guice/src/test/java/org/apache/james/DockerElasticSearchExtension.java
index 7ff07f4..7f76c79 100644
--- a/server/container/guice/cassandra-guice/src/test/java/org/apache/james/DockerElasticSearchExtension.java
+++ b/server/container/guice/cassandra-guice/src/test/java/org/apache/james/DockerElasticSearchExtension.java
@@ -49,11 +49,19 @@ public class DockerElasticSearchExtension implements GuiceModuleTestExtension {
}
@Override
- public void beforeEach(ExtensionContext extensionContext) {
+ public void beforeAll(ExtensionContext extensionContext) {
getDockerES().start();
}
@Override
+ public void beforeEach(ExtensionContext extensionContext) {
+ if (!getDockerES().isRunning()) {
+ getDockerES().unpause();
+ }
+ await();
+ }
+
+ @Override
public void afterEach(ExtensionContext extensionContext) {
}
diff --git a/server/container/metrics/metrics-es-reporter/src/test/java/org/apache/james/metric/es/DockerElasticSearchExtension.java b/server/container/metrics/metrics-es-reporter/src/test/java/org/apache/james/metric/es/DockerElasticSearchExtension.java
index cfedbc0..459ac2a 100644
--- a/server/container/metrics/metrics-es-reporter/src/test/java/org/apache/james/metric/es/DockerElasticSearchExtension.java
+++ b/server/container/metrics/metrics-es-reporter/src/test/java/org/apache/james/metric/es/DockerElasticSearchExtension.java
@@ -45,6 +45,9 @@ public class DockerElasticSearchExtension implements AfterAllCallback, BeforeAll
@Override
public void beforeEach(ExtensionContext context) throws Exception {
+ if (!elasticSearch.isRunning()) {
+ elasticSearch.unpause();
+ }
elasticSearch.flushIndices();
}
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 05/06: JAMES-3144 Reactify ScrollSearch
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 a8dcf2192885b4c523cca85d69b06194a98301fe
Author: Matthieu Baechler <ma...@apache.org>
AuthorDate: Fri Mar 13 11:43:33 2020 +0100
JAMES-3144 Reactify ScrollSearch
---
.../james/backends/es/search/ScrolledSearch.java | 75 +++++++++++++---------
1 file changed, 43 insertions(+), 32 deletions(-)
diff --git a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/search/ScrolledSearch.java b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/search/ScrolledSearch.java
index 10146ea..cf8a83b 100644
--- a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/search/ScrolledSearch.java
+++ b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/search/ScrolledSearch.java
@@ -21,6 +21,7 @@ package org.apache.james.backends.es.search;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;
import org.apache.james.backends.es.ReactorElasticSearchClient;
import org.elasticsearch.action.search.ClearScrollRequest;
@@ -31,11 +32,14 @@ import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.search.SearchHit;
-import reactor.core.Disposable;
+import com.github.fge.lambdas.Throwing;
+
import reactor.core.publisher.Flux;
+import reactor.core.publisher.FluxSink;
import reactor.core.publisher.Mono;
public class ScrolledSearch {
+
private static final TimeValue TIMEOUT = TimeValue.timeValueMinutes(1);
private final ReactorElasticSearchClient client;
@@ -46,52 +50,59 @@ public class ScrolledSearch {
this.searchRequest = searchRequest;
}
-
public Flux<SearchHit> searchHits() {
return searchResponses()
- .flatMap(searchResponse -> Flux.fromArray(searchResponse.getHits().getHits()));
+ .concatMap(searchResponse -> Flux.just(searchResponse.getHits().getHits()));
}
public Flux<SearchResponse> searchResponses() {
- return ensureClosing(Flux.from(startScrolling(searchRequest))
- .expand(this::nextResponse));
- }
+ return Flux.push(sink -> {
+ AtomicReference<Optional<String>> scrollId = new AtomicReference<>(Optional.empty());
+ sink.onRequest(numberOfRequestedElements -> next(sink, scrollId, numberOfRequestedElements));
- private Mono<SearchResponse> startScrolling(SearchRequest searchRequest) {
- return client.search(searchRequest, RequestOptions.DEFAULT);
+ sink.onDispose(() -> close(scrollId));
+ });
}
- public Mono<SearchResponse> nextResponse(SearchResponse previous) {
- if (allSearchResponsesConsumed(previous)) {
- return Mono.empty();
+ private void next(FluxSink<SearchResponse> sink, AtomicReference<Optional<String>> scrollId, long numberOfRequestedElements) {
+ if (numberOfRequestedElements <= 0) {
+ return;
}
- return client.scroll(
- new SearchScrollRequest()
- .scrollId(previous.getScrollId())
- .scroll(TIMEOUT),
- RequestOptions.DEFAULT);
- }
+ Consumer<SearchResponse> onResponse = searchResponse -> {
+ scrollId.set(Optional.of(searchResponse.getScrollId()));
+ sink.next(searchResponse);
- private boolean allSearchResponsesConsumed(SearchResponse searchResponse) {
- return searchResponse.getHits().getHits().length == 0;
- }
+ boolean noHitsLeft = searchResponse.getHits().getHits().length == 0;
+ if (noHitsLeft) {
+ sink.complete();
+ } else {
+ next(sink, scrollId, numberOfRequestedElements - 1);
+ }
+ };
- private Flux<SearchResponse> ensureClosing(Flux<SearchResponse> origin) {
- AtomicReference<SearchResponse> latest = new AtomicReference<>();
- return origin
- .doOnNext(latest::set)
- .doOnTerminate(close(latest));
- }
+ Consumer<Throwable> onFailure = sink::error;
- public Runnable close(AtomicReference<SearchResponse> latest) {
- return () -> Optional.ofNullable(latest.getAndSet(null)).map(this::clearScroll);
+ buildRequest(scrollId.get())
+ .subscribe(onResponse, onFailure);
}
- private Disposable clearScroll(SearchResponse current) {
- ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
- clearScrollRequest.addScrollId(current.getScrollId());
+ 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));
+ }
- return client.clearScroll(clearScrollRequest, RequestOptions.DEFAULT).subscribe();
+ 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());
}
+
}
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 03/06: JAMES-3144 Use ElasticSearch reactively
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 97e06e9ccafa1d5a8869fdcc2b5ead42564a3c90
Author: Gautier DI FOLCO <gd...@linagora.com>
AuthorDate: Fri Mar 13 16:05:21 2020 +0100
JAMES-3144 Use ElasticSearch reactively
---
.../apache/james/backends/es/ClientProvider.java | 12 +-
.../james/backends/es/DeleteByQueryPerformer.java | 17 +-
.../backends/es/ElasticSearchHealthCheck.java | 5 +-
.../james/backends/es/ElasticSearchIndexer.java | 67 ++++----
.../james/backends/es/IndexCreationFactory.java | 11 +-
.../apache/james/backends/es/ListenerToFuture.java | 51 ------
.../james/backends/es/NodeMappingFactory.java | 7 +-
.../backends/es/ReactorElasticSearchClient.java | 171 +++++++++++++++++++++
.../james/backends/es/search/ScrolledSearch.java | 113 ++++++--------
.../es/ClientProviderImplConnectionContract.java | 5 +-
.../es/ElasticSearchHealthCheckConnectionTest.java | 3 +-
.../backends/es/ElasticSearchIndexerTest.java | 72 +++++----
.../backends/es/IndexCreationFactoryTest.java | 3 +-
.../backends/es/NodeMappingFactoryAuthTest.java | 3 +-
.../james/backends/es/NodeMappingFactoryTest.java | 3 +-
.../backends/es/search/ScrolledSearchTest.java | 37 +++--
.../elasticsearch/MailboxIndexCreationUtil.java | 14 +-
.../ElasticSearchListeningMessageSearchIndex.java | 38 +++--
.../search/ElasticSearchSearcher.java | 23 +--
.../ElasticSearchIntegrationTest.java | 4 +-
...asticSearchListeningMessageSearchIndexTest.java | 10 +-
.../search/ElasticSearchSearcherTest.java | 4 +-
.../elasticsearch/ElasticSearchQuotaSearcher.java | 35 ++---
.../QuotaSearchIndexCreationUtil.java | 6 +-
.../events/ElasticSearchQuotaMailboxListener.java | 8 +-
...lasticSearchQuotaSearchTestSystemExtension.java | 4 +-
.../ElasticSearchQuotaMailboxListenerTest.java | 13 +-
.../host/ElasticSearchHostSystem.java | 4 +-
.../modules/mailbox/ElasticSearchClientModule.java | 4 +-
.../mailbox/ElasticSearchMailboxModule.java | 10 +-
.../mailbox/ElasticSearchQuotaSearcherModule.java | 10 +-
.../modules/mailbox/ElasticSearchStartUpCheck.java | 6 +-
.../test/java/org/apache/james/ESReporterTest.java | 5 +-
.../routes/ElasticSearchQuotaSearchExtension.java | 4 +-
34 files changed, 450 insertions(+), 332 deletions(-)
diff --git a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ClientProvider.java b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ClientProvider.java
index b879a88..247ed50 100644
--- a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ClientProvider.java
+++ b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ClientProvider.java
@@ -57,7 +57,7 @@ import com.google.common.annotations.VisibleForTesting;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
-public class ClientProvider implements Provider<RestHighLevelClient> {
+public class ClientProvider implements Provider<ReactorElasticSearchClient> {
private static class HttpAsyncClientConfigurer {
@@ -165,15 +165,17 @@ public class ClientProvider implements Provider<RestHighLevelClient> {
private static final Logger LOGGER = LoggerFactory.getLogger(ClientProvider.class);
private final ElasticSearchConfiguration configuration;
- private final RestHighLevelClient client;
+ private final RestHighLevelClient elasticSearchRestHighLevelClient;
private final HttpAsyncClientConfigurer httpAsyncClientConfigurer;
+ private final ReactorElasticSearchClient client;
@Inject
@VisibleForTesting
ClientProvider(ElasticSearchConfiguration configuration) {
this.httpAsyncClientConfigurer = new HttpAsyncClientConfigurer(configuration);
this.configuration = configuration;
- this.client = connect(configuration);
+ this.elasticSearchRestHighLevelClient = connect(configuration);
+ this.client = new ReactorElasticSearchClient(this.elasticSearchRestHighLevelClient);
}
private RestHighLevelClient connect(ElasticSearchConfiguration configuration) {
@@ -206,12 +208,12 @@ public class ClientProvider implements Provider<RestHighLevelClient> {
}
@Override
- public RestHighLevelClient get() {
+ public ReactorElasticSearchClient get() {
return client;
}
@PreDestroy
public void close() throws IOException {
- client.close();
+ elasticSearchRestHighLevelClient.close();
}
}
diff --git a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/DeleteByQueryPerformer.java b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/DeleteByQueryPerformer.java
index 66ad682..5277694 100644
--- a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/DeleteByQueryPerformer.java
+++ b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/DeleteByQueryPerformer.java
@@ -26,7 +26,6 @@ import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
-import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.search.SearchHit;
@@ -34,18 +33,17 @@ import org.elasticsearch.search.builder.SearchSourceBuilder;
import com.google.common.annotations.VisibleForTesting;
-import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
public class DeleteByQueryPerformer {
private static final TimeValue TIMEOUT = new TimeValue(60000);
- private final RestHighLevelClient client;
+ private final ReactorElasticSearchClient client;
private final int batchSize;
private final WriteAliasName aliasName;
@VisibleForTesting
- DeleteByQueryPerformer(RestHighLevelClient client, int batchSize, WriteAliasName aliasName) {
+ DeleteByQueryPerformer(ReactorElasticSearchClient client, int batchSize, WriteAliasName aliasName) {
this.client = client;
this.batchSize = batchSize;
this.aliasName = aliasName;
@@ -54,9 +52,10 @@ public class DeleteByQueryPerformer {
public Mono<Void> perform(QueryBuilder queryBuilder, RoutingKey routingKey) {
SearchRequest searchRequest = prepareSearch(queryBuilder, routingKey);
- return Flux.fromStream(new ScrolledSearch(client, searchRequest).searchResponses())
- .flatMap(searchResponse -> deleteRetrievedIds(client, searchResponse, routingKey))
- .thenEmpty(Mono.empty());
+ return new ScrolledSearch(client, searchRequest).searchResponses()
+ .filter(searchResponse -> searchResponse.getHits().getHits().length > 0)
+ .flatMap(searchResponse -> deleteRetrievedIds(searchResponse, routingKey))
+ .then();
}
private SearchRequest prepareSearch(QueryBuilder queryBuilder, RoutingKey routingKey) {
@@ -73,7 +72,7 @@ public class DeleteByQueryPerformer {
.size(batchSize);
}
- private Mono<BulkResponse> deleteRetrievedIds(RestHighLevelClient client, SearchResponse searchResponse, RoutingKey routingKey) {
+ private Mono<BulkResponse> deleteRetrievedIds(SearchResponse searchResponse, RoutingKey routingKey) {
BulkRequest request = new BulkRequest();
for (SearchHit hit : searchResponse.getHits()) {
@@ -84,6 +83,6 @@ public class DeleteByQueryPerformer {
.routing(routingKey.asString()));
}
- return Mono.fromCallable(() -> client.bulk(request, RequestOptions.DEFAULT));
+ return client.bulk(request, RequestOptions.DEFAULT);
}
}
diff --git a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ElasticSearchHealthCheck.java b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ElasticSearchHealthCheck.java
index 1305420..58cd813 100644
--- a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ElasticSearchHealthCheck.java
+++ b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ElasticSearchHealthCheck.java
@@ -32,7 +32,6 @@ import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.Requests;
-import org.elasticsearch.client.RestHighLevelClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -44,10 +43,10 @@ public class ElasticSearchHealthCheck implements HealthCheck {
private static final ComponentName COMPONENT_NAME = new ComponentName("ElasticSearch Backend");
private final Set<IndexName> indexNames;
- private final RestHighLevelClient client;
+ private final ReactorElasticSearchClient client;
@Inject
- ElasticSearchHealthCheck(RestHighLevelClient client, Set<IndexName> indexNames) {
+ ElasticSearchHealthCheck(ReactorElasticSearchClient client, Set<IndexName> indexNames) {
this.client = client;
this.indexNames = indexNames;
}
diff --git a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ElasticSearchIndexer.java b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ElasticSearchIndexer.java
index 99a9f80..0fcbc11 100644
--- a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ElasticSearchIndexer.java
+++ b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ElasticSearchIndexer.java
@@ -18,9 +18,7 @@
****************************************************************/
package org.apache.james.backends.es;
-import java.io.IOException;
import java.util.List;
-import java.util.Optional;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.action.bulk.BulkRequest;
@@ -30,7 +28,6 @@ import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.RequestOptions;
-import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.ValidationException;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.QueryBuilder;
@@ -40,23 +37,25 @@ import org.slf4j.LoggerFactory;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
+import reactor.core.publisher.Mono;
+
public class ElasticSearchIndexer {
private static final int DEBUG_MAX_LENGTH_CONTENT = 1000;
private static final int DEFAULT_BATCH_SIZE = 100;
private static final Logger LOGGER = LoggerFactory.getLogger(ElasticSearchIndexer.class);
- private final RestHighLevelClient client;
+ private final ReactorElasticSearchClient client;
private final AliasName aliasName;
private final DeleteByQueryPerformer deleteByQueryPerformer;
- public ElasticSearchIndexer(RestHighLevelClient client,
+ public ElasticSearchIndexer(ReactorElasticSearchClient client,
WriteAliasName aliasName) {
this(client, aliasName, DEFAULT_BATCH_SIZE);
}
@VisibleForTesting
- public ElasticSearchIndexer(RestHighLevelClient client,
+ public ElasticSearchIndexer(ReactorElasticSearchClient client,
WriteAliasName aliasName,
int batchSize) {
this.client = client;
@@ -64,7 +63,7 @@ public class ElasticSearchIndexer {
this.aliasName = aliasName;
}
- public IndexResponse index(DocumentId id, String content, RoutingKey routingKey) throws IOException {
+ public Mono<IndexResponse> index(DocumentId id, String content, RoutingKey routingKey) {
checkArgument(content);
logContent(id, content);
return client.index(new IndexRequest(aliasName.getValue())
@@ -81,37 +80,37 @@ public class ElasticSearchIndexer {
}
}
- public Optional<BulkResponse> update(List<UpdatedRepresentation> updatedDocumentParts, RoutingKey routingKey) throws IOException {
- try {
- Preconditions.checkNotNull(updatedDocumentParts);
- Preconditions.checkNotNull(routingKey);
- BulkRequest request = new BulkRequest();
- updatedDocumentParts.forEach(updatedDocumentPart -> request.add(
- new UpdateRequest(aliasName.getValue(),
- NodeMappingFactory.DEFAULT_MAPPING_NAME,
- updatedDocumentPart.getId().asString())
+ 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(),
+ NodeMappingFactory.DEFAULT_MAPPING_NAME,
+ updatedDocumentPart.getId().asString())
.doc(updatedDocumentPart.getUpdatedDocumentPart(), XContentType.JSON)
.routing(routingKey.asString())));
- return Optional.of(client.bulk(request, RequestOptions.DEFAULT));
- } catch (ValidationException e) {
- LOGGER.warn("Error while updating index", e);
- return Optional.empty();
- }
+
+ return client.bulk(request, RequestOptions.DEFAULT)
+ .onErrorResume(ValidationException.class, exception -> {
+ LOGGER.warn("Error while updating index", exception);
+ return Mono.empty();
+ });
}
- public Optional<BulkResponse> delete(List<DocumentId> ids, RoutingKey routingKey) throws IOException {
- try {
- BulkRequest request = new BulkRequest();
- ids.forEach(id -> request.add(
- new DeleteRequest(aliasName.getValue())
- .type(NodeMappingFactory.DEFAULT_MAPPING_NAME)
- .id(id.asString())
- .routing(routingKey.asString())));
- return Optional.of(client.bulk(request, RequestOptions.DEFAULT));
- } catch (ValidationException e) {
- LOGGER.warn("Error while deleting index", e);
- return Optional.empty();
- }
+ public Mono<BulkResponse> delete(List<DocumentId> ids, RoutingKey routingKey) {
+ BulkRequest request = new BulkRequest();
+ ids.forEach(id -> request.add(
+ new DeleteRequest(aliasName.getValue())
+ .type(NodeMappingFactory.DEFAULT_MAPPING_NAME)
+ .id(id.asString())
+ .routing(routingKey.asString())));
+
+ return client.bulk(request, RequestOptions.DEFAULT)
+ .onErrorResume(ValidationException.class, exception -> {
+ LOGGER.warn("Error while deleting index", exception);
+ return Mono.empty();
+ });
}
public void deleteAllMatchingQuery(QueryBuilder queryBuilder, RoutingKey routingKey) {
diff --git a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/IndexCreationFactory.java b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/IndexCreationFactory.java
index 9d6f834..07e41e8 100644
--- a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/IndexCreationFactory.java
+++ b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/IndexCreationFactory.java
@@ -31,7 +31,6 @@ import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequest;
import org.elasticsearch.action.admin.indices.alias.get.GetAliasesRequest;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
import org.elasticsearch.client.RequestOptions;
-import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -63,7 +62,7 @@ public class IndexCreationFactory {
return this;
}
- public RestHighLevelClient createIndexAndAliases(RestHighLevelClient client) {
+ public ReactorElasticSearchClient createIndexAndAliases(ReactorElasticSearchClient client) {
return new IndexCreationPerformer(nbShards, nbReplica, waitForActiveShards, indexName, aliases.build()).createIndexAndAliases(client);
}
}
@@ -83,7 +82,7 @@ public class IndexCreationFactory {
this.aliases = aliases;
}
- public RestHighLevelClient createIndexAndAliases(RestHighLevelClient client) {
+ public ReactorElasticSearchClient createIndexAndAliases(ReactorElasticSearchClient client) {
Preconditions.checkNotNull(indexName);
try {
createIndexIfNeeded(client, indexName, generateSetting(nbShards, nbReplica, waitForActiveShards));
@@ -95,7 +94,7 @@ public class IndexCreationFactory {
return client;
}
- private void createAliasIfNeeded(RestHighLevelClient client, IndexName indexName, AliasName aliasName) throws IOException {
+ private void createAliasIfNeeded(ReactorElasticSearchClient client, IndexName indexName, AliasName aliasName) throws IOException {
if (!aliasExist(client, aliasName)) {
client.indices()
.updateAliases(
@@ -107,12 +106,12 @@ public class IndexCreationFactory {
}
}
- private boolean aliasExist(RestHighLevelClient client, AliasName aliasName) throws IOException {
+ private boolean aliasExist(ReactorElasticSearchClient client, AliasName aliasName) throws IOException {
return client.indices()
.existsAlias(new GetAliasesRequest().aliases(aliasName.getValue()), RequestOptions.DEFAULT);
}
- private void createIndexIfNeeded(RestHighLevelClient client, IndexName indexName, XContentBuilder settings) throws IOException {
+ private void createIndexIfNeeded(ReactorElasticSearchClient client, IndexName indexName, XContentBuilder settings) throws IOException {
try {
client.indices()
.create(
diff --git a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ListenerToFuture.java b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ListenerToFuture.java
deleted file mode 100644
index 1df57b0..0000000
--- a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ListenerToFuture.java
+++ /dev/null
@@ -1,51 +0,0 @@
-/****************************************************************
- * 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.es;
-
-import java.util.concurrent.CompletableFuture;
-
-import org.elasticsearch.action.ActionListener;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class ListenerToFuture<T> implements ActionListener<T> {
- private static final Logger LOGGER = LoggerFactory.getLogger(ListenerToFuture.class);
-
- private CompletableFuture<T> future;
-
- public ListenerToFuture() {
- this.future = new CompletableFuture<>();
- }
-
- @Override
- public void onResponse(T t) {
- future.complete(t);
- }
-
- @Override
- public void onFailure(Exception e) {
- LOGGER.warn("Error while waiting ElasticSearch query execution: ", e);
- future.completeExceptionally(e);
- }
-
- public CompletableFuture<T> getFuture() {
- return future;
- }
-}
diff --git a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/NodeMappingFactory.java b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/NodeMappingFactory.java
index f4359f4..25b82e7 100644
--- a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/NodeMappingFactory.java
+++ b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/NodeMappingFactory.java
@@ -25,7 +25,6 @@ import org.apache.http.HttpStatus;
import org.elasticsearch.action.admin.indices.mapping.put.PutMappingRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.ResponseException;
-import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentBuilder;
public class NodeMappingFactory {
@@ -55,7 +54,7 @@ public class NodeMappingFactory {
public static final String SNOWBALL = "snowball";
public static final String IGNORE_ABOVE = "ignore_above";
- public static RestHighLevelClient applyMapping(RestHighLevelClient client, IndexName indexName, XContentBuilder mappingsSources) throws IOException {
+ public static ReactorElasticSearchClient applyMapping(ReactorElasticSearchClient client, IndexName indexName, XContentBuilder mappingsSources) throws IOException {
if (!mappingAlreadyExist(client, indexName)) {
createMapping(client, indexName, mappingsSources);
}
@@ -64,7 +63,7 @@ public class NodeMappingFactory {
// ElasticSearch 6.3.2 does not support field master_timeout that is set up by 6.4.3 REST client when relying on getMapping
@SuppressWarnings("deprecation")
- public static boolean mappingAlreadyExist(RestHighLevelClient client, IndexName indexName) throws IOException {
+ public static boolean mappingAlreadyExist(ReactorElasticSearchClient client, IndexName indexName) throws IOException {
try {
client.getLowLevelClient().performRequest("GET", "/" + indexName.getValue() + "/_mapping/" + NodeMappingFactory.DEFAULT_MAPPING_NAME);
return true;
@@ -76,7 +75,7 @@ public class NodeMappingFactory {
return false;
}
- public static void createMapping(RestHighLevelClient client, IndexName indexName, XContentBuilder mappingsSources) throws IOException {
+ public static void createMapping(ReactorElasticSearchClient client, IndexName indexName, XContentBuilder mappingsSources) throws IOException {
client.indices().putMapping(
new PutMappingRequest(indexName.getValue())
.type(NodeMappingFactory.DEFAULT_MAPPING_NAME)
diff --git a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ReactorElasticSearchClient.java b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ReactorElasticSearchClient.java
new file mode 100644
index 0000000..3b0c18b
--- /dev/null
+++ b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ReactorElasticSearchClient.java
@@ -0,0 +1,171 @@
+/****************************************************************
+ * 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.es;
+
+import java.io.IOException;
+import java.util.function.Consumer;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.admin.cluster.storedscripts.DeleteStoredScriptRequest;
+import org.elasticsearch.action.admin.cluster.storedscripts.DeleteStoredScriptResponse;
+import org.elasticsearch.action.admin.cluster.storedscripts.GetStoredScriptRequest;
+import org.elasticsearch.action.admin.cluster.storedscripts.GetStoredScriptResponse;
+import org.elasticsearch.action.bulk.BulkRequest;
+import org.elasticsearch.action.bulk.BulkResponse;
+import org.elasticsearch.action.delete.DeleteRequest;
+import org.elasticsearch.action.delete.DeleteResponse;
+import org.elasticsearch.action.explain.ExplainRequest;
+import org.elasticsearch.action.explain.ExplainResponse;
+import org.elasticsearch.action.fieldcaps.FieldCapabilitiesRequest;
+import org.elasticsearch.action.fieldcaps.FieldCapabilitiesResponse;
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.action.index.IndexResponse;
+import org.elasticsearch.action.main.MainResponse;
+import org.elasticsearch.action.search.ClearScrollRequest;
+import org.elasticsearch.action.search.ClearScrollResponse;
+import org.elasticsearch.action.search.MultiSearchRequest;
+import org.elasticsearch.action.search.MultiSearchResponse;
+import org.elasticsearch.action.search.SearchRequest;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.action.search.SearchScrollRequest;
+import org.elasticsearch.client.ClusterClient;
+import org.elasticsearch.client.IndicesClient;
+import org.elasticsearch.client.RequestOptions;
+import org.elasticsearch.client.RestClient;
+import org.elasticsearch.client.RestHighLevelClient;
+import org.elasticsearch.index.rankeval.RankEvalRequest;
+import org.elasticsearch.index.rankeval.RankEvalResponse;
+import org.elasticsearch.script.mustache.MultiSearchTemplateRequest;
+import org.elasticsearch.script.mustache.MultiSearchTemplateResponse;
+import org.elasticsearch.script.mustache.SearchTemplateRequest;
+import org.elasticsearch.script.mustache.SearchTemplateResponse;
+
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.MonoSink;
+import reactor.core.scheduler.Schedulers;
+
+public class ReactorElasticSearchClient implements AutoCloseable {
+ private final RestHighLevelClient client;
+
+ public ReactorElasticSearchClient(RestHighLevelClient client) {
+ this.client = client;
+ }
+
+ 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 ClusterClient cluster() {
+ return client.cluster();
+ }
+
+ public DeleteResponse delete(DeleteRequest deleteRequest, RequestOptions options) throws IOException {
+ return client.delete(deleteRequest, options);
+ }
+
+ public Mono<DeleteStoredScriptResponse> deleteScript(DeleteStoredScriptRequest request, RequestOptions options) {
+ return toReactor(listener -> client.deleteScriptAsync(request, options, listener));
+ }
+
+ public Mono<ExplainResponse> explain(ExplainRequest explainRequest, RequestOptions options) {
+ return toReactor(listener -> client.explainAsync(explainRequest, options, listener));
+ }
+
+ public Mono<FieldCapabilitiesResponse> fieldCaps(FieldCapabilitiesRequest fieldCapabilitiesRequest, RequestOptions options) {
+ return toReactor(listener -> client.fieldCapsAsync(fieldCapabilitiesRequest, options, listener));
+ }
+
+ public RestClient getLowLevelClient() {
+ return client.getLowLevelClient();
+ }
+
+ public Mono<GetStoredScriptResponse> getScript(GetStoredScriptRequest request, RequestOptions options) {
+ return toReactor(listener -> client.getScriptAsync(request, options, listener));
+ }
+
+ public Mono<IndexResponse> index(IndexRequest indexRequest, RequestOptions options) {
+ return toReactor(listener -> client.indexAsync(indexRequest, options, listener));
+ }
+
+ public IndicesClient indices() {
+ return client.indices();
+ }
+
+ public MainResponse info(RequestOptions options) throws IOException {
+ return client.info(options);
+ }
+
+ public Mono<MultiSearchResponse> msearch(MultiSearchRequest multiSearchRequest, RequestOptions options) {
+ return toReactor(listener -> client.msearchAsync(multiSearchRequest, options, listener));
+ }
+
+ public Mono<MultiSearchTemplateResponse> msearchTemplate(MultiSearchTemplateRequest multiSearchTemplateRequest, RequestOptions options) {
+ return toReactor(listener -> client.msearchTemplateAsync(multiSearchTemplateRequest, options, listener));
+ }
+
+ public Mono<RankEvalResponse> rankEval(RankEvalRequest rankEvalRequest, RequestOptions options) {
+ return toReactor(listener -> client.rankEvalAsync(rankEvalRequest, options, listener));
+ }
+
+ public Mono<SearchResponse> scroll(SearchScrollRequest searchScrollRequest, RequestOptions options) {
+ return toReactor(listener -> client.scrollAsync(searchScrollRequest, options, listener));
+ }
+
+ @Deprecated
+ public Mono<SearchResponse> search(SearchRequest searchRequest) {
+ return toReactor(listener -> client.searchAsync(searchRequest, listener));
+ }
+
+ public Mono<SearchResponse> search(SearchRequest searchRequest, RequestOptions options) {
+ return toReactor(listener -> client.searchAsync(searchRequest, options, listener));
+ }
+
+ public Mono<SearchTemplateResponse> searchTemplate(SearchTemplateRequest searchTemplateRequest, RequestOptions options) {
+ return toReactor(listener -> client.searchTemplateAsync(searchTemplateRequest, options, listener));
+ }
+
+ @Override
+ public void close() throws IOException {
+ client.close();
+ }
+
+ private static <T> Mono<T> toReactor(Consumer<ActionListener<T>> async) {
+ return Mono.<T>create(sink -> async.accept(getListener(sink)))
+ .subscribeOn(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);
+ }
+ };
+ }
+}
diff --git a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/search/ScrolledSearch.java b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/search/ScrolledSearch.java
index 834f801..10146ea 100644
--- a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/search/ScrolledSearch.java
+++ b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/search/ScrolledSearch.java
@@ -19,94 +19,79 @@
package org.apache.james.backends.es.search;
-import java.io.Closeable;
-import java.io.IOException;
-import java.util.Arrays;
-import java.util.Iterator;
-import java.util.concurrent.CompletableFuture;
-import java.util.stream.Stream;
-
-import org.apache.james.backends.es.ListenerToFuture;
-import org.apache.james.util.streams.Iterators;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.james.backends.es.ReactorElasticSearchClient;
import org.elasticsearch.action.search.ClearScrollRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchScrollRequest;
import org.elasticsearch.client.RequestOptions;
-import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.search.SearchHit;
-import com.github.fge.lambdas.Throwing;
+import reactor.core.Disposable;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
public class ScrolledSearch {
- private static class ScrollIterator implements Iterator<SearchResponse>, Closeable {
- private final RestHighLevelClient client;
- private CompletableFuture<SearchResponse> searchResponseFuture;
+ private static final TimeValue TIMEOUT = TimeValue.timeValueMinutes(1);
- ScrollIterator(RestHighLevelClient client, SearchRequest searchRequest) {
- this.client = client;
- ListenerToFuture<SearchResponse> listener = new ListenerToFuture<>();
- client.searchAsync(searchRequest, RequestOptions.DEFAULT, listener);
+ private final ReactorElasticSearchClient client;
+ private final SearchRequest searchRequest;
- this.searchResponseFuture = listener.getFuture();
- }
+ public ScrolledSearch(ReactorElasticSearchClient client, SearchRequest searchRequest) {
+ this.client = client;
+ this.searchRequest = searchRequest;
+ }
- @Override
- public void close() throws IOException {
- ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
- clearScrollRequest.addScrollId(searchResponseFuture.join().getScrollId());
- client.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);
- }
- @Override
- public boolean hasNext() {
- SearchResponse join = searchResponseFuture.join();
- return !allSearchResponsesConsumed(join);
- }
+ public Flux<SearchHit> searchHits() {
+ return searchResponses()
+ .flatMap(searchResponse -> Flux.fromArray(searchResponse.getHits().getHits()));
+ }
- @Override
- public SearchResponse next() {
- SearchResponse result = searchResponseFuture.join();
- ListenerToFuture<SearchResponse> listener = new ListenerToFuture<>();
- client.scrollAsync(
- new SearchScrollRequest()
- .scrollId(result.getScrollId())
- .scroll(TIMEOUT),
- RequestOptions.DEFAULT,
- listener);
- searchResponseFuture = listener.getFuture();
- return result;
- }
+ public Flux<SearchResponse> searchResponses() {
+ return ensureClosing(Flux.from(startScrolling(searchRequest))
+ .expand(this::nextResponse));
+ }
- public Stream<SearchResponse> stream() {
- return Iterators.toStream(this)
- .onClose(Throwing.runnable(this::close));
- }
+ private Mono<SearchResponse> startScrolling(SearchRequest searchRequest) {
+ return client.search(searchRequest, RequestOptions.DEFAULT);
+ }
- private boolean allSearchResponsesConsumed(SearchResponse searchResponse) {
- return searchResponse.getHits().getHits().length == 0;
+ public Mono<SearchResponse> nextResponse(SearchResponse previous) {
+ if (allSearchResponsesConsumed(previous)) {
+ return Mono.empty();
}
- }
- private static final TimeValue TIMEOUT = TimeValue.timeValueMinutes(1);
+ return client.scroll(
+ new SearchScrollRequest()
+ .scrollId(previous.getScrollId())
+ .scroll(TIMEOUT),
+ RequestOptions.DEFAULT);
+ }
- private final RestHighLevelClient client;
- private final SearchRequest searchRequest;
+ private boolean allSearchResponsesConsumed(SearchResponse searchResponse) {
+ return searchResponse.getHits().getHits().length == 0;
+ }
- public ScrolledSearch(RestHighLevelClient client, SearchRequest searchRequest) {
- this.client = client;
- this.searchRequest = searchRequest;
+ private Flux<SearchResponse> ensureClosing(Flux<SearchResponse> origin) {
+ AtomicReference<SearchResponse> latest = new AtomicReference<>();
+ return origin
+ .doOnNext(latest::set)
+ .doOnTerminate(close(latest));
}
- public Stream<SearchHit> searchHits() {
- return searchResponses()
- .flatMap(searchResponse -> Arrays.stream(searchResponse.getHits().getHits()));
+ public Runnable close(AtomicReference<SearchResponse> latest) {
+ return () -> Optional.ofNullable(latest.getAndSet(null)).map(this::clearScroll);
}
- @SuppressWarnings("resource")
- public Stream<SearchResponse> searchResponses() {
- return new ScrollIterator(client, searchRequest)
- .stream();
+ private Disposable clearScroll(SearchResponse current) {
+ ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
+ clearScrollRequest.addScrollId(current.getScrollId());
+
+ return client.clearScroll(clearScrollRequest, RequestOptions.DEFAULT).subscribe();
}
}
diff --git a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ClientProviderImplConnectionContract.java b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ClientProviderImplConnectionContract.java
index 05a7ec4..4a4328c 100644
--- a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ClientProviderImplConnectionContract.java
+++ b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ClientProviderImplConnectionContract.java
@@ -25,7 +25,6 @@ import org.apache.james.backends.es.ElasticSearchClusterExtension.ElasticSearchC
import org.awaitility.Awaitility;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.client.RequestOptions;
-import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.junit.jupiter.api.Test;
@@ -75,11 +74,11 @@ interface ClientProviderImplConnectionContract {
}
default boolean isConnected(ClientProvider clientProvider) {
- try (RestHighLevelClient client = clientProvider.get()) {
+ try (ReactorElasticSearchClient client = clientProvider.get()) {
client.search(
new SearchRequest()
.source(new SearchSourceBuilder().query(QueryBuilders.existsQuery("any"))),
- RequestOptions.DEFAULT);
+ RequestOptions.DEFAULT).block();
return true;
} catch (Exception e) {
LOGGER.info("Caught exception while trying to connect", e);
diff --git a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ElasticSearchHealthCheckConnectionTest.java b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ElasticSearchHealthCheckConnectionTest.java
index 51e1b2e..0f5ddd6 100644
--- a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ElasticSearchHealthCheckConnectionTest.java
+++ b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ElasticSearchHealthCheckConnectionTest.java
@@ -22,7 +22,6 @@ import static org.assertj.core.api.Assertions.assertThat;
import java.time.Duration;
-import org.elasticsearch.client.RestHighLevelClient;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
@@ -38,7 +37,7 @@ class ElasticSearchHealthCheckConnectionTest {
@BeforeEach
void setUp() {
- RestHighLevelClient client = elasticSearch.getDockerElasticSearch().clientProvider(REQUEST_TIMEOUT).get();
+ ReactorElasticSearchClient client = elasticSearch.getDockerElasticSearch().clientProvider(REQUEST_TIMEOUT).get();
elasticSearchHealthCheck = new ElasticSearchHealthCheck(client, ImmutableSet.of());
}
diff --git a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ElasticSearchIndexerTest.java b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ElasticSearchIndexerTest.java
index c7372a5..f5f8a40 100644
--- a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ElasticSearchIndexerTest.java
+++ b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ElasticSearchIndexerTest.java
@@ -33,7 +33,6 @@ import org.awaitility.core.ConditionFactory;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
-import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.junit.jupiter.api.AfterEach;
@@ -62,7 +61,7 @@ class ElasticSearchIndexerTest {
@RegisterExtension
public DockerElasticSearchExtension elasticSearch = new DockerElasticSearchExtension();
private ElasticSearchIndexer testee;
- private RestHighLevelClient client;
+ private ReactorElasticSearchClient client;
@BeforeEach
void setup() {
@@ -80,17 +79,18 @@ class ElasticSearchIndexerTest {
}
@Test
- void indexMessageShouldWork() throws Exception {
+ void indexMessageShouldWork() {
DocumentId documentId = DocumentId.fromString("1");
String content = "{\"message\": \"trying out Elasticsearch\"}";
- testee.index(documentId, content, useDocumentId(documentId));
+ testee.index(documentId, content, useDocumentId(documentId)).block();
elasticSearch.awaitForElasticSearch();
-
+
SearchResponse searchResponse = client.search(
new SearchRequest(INDEX_NAME.getValue())
.source(new SearchSourceBuilder().query(QueryBuilders.matchQuery("message", "trying"))),
- RequestOptions.DEFAULT);
+ RequestOptions.DEFAULT)
+ .block();
assertThat(searchResponse.getHits().getTotalHits()).isEqualTo(1);
}
@@ -101,64 +101,66 @@ class ElasticSearchIndexerTest {
}
@Test
- void updateMessages() throws Exception {
+ void updateMessages() {
String content = "{\"message\": \"trying out Elasticsearch\",\"field\":\"Should be unchanged\"}";
- testee.index(DOCUMENT_ID, content, useDocumentId(DOCUMENT_ID));
+ testee.index(DOCUMENT_ID, content, useDocumentId(DOCUMENT_ID)).block();
elasticSearch.awaitForElasticSearch();
- testee.update(ImmutableList.of(new UpdatedRepresentation(DOCUMENT_ID, "{\"message\": \"mastering out Elasticsearch\"}")), useDocumentId(DOCUMENT_ID));
+ testee.update(ImmutableList.of(new UpdatedRepresentation(DOCUMENT_ID, "{\"message\": \"mastering out Elasticsearch\"}")), useDocumentId(DOCUMENT_ID)).block();
elasticSearch.awaitForElasticSearch();
SearchResponse searchResponse = client.search(
new SearchRequest(INDEX_NAME.getValue())
.source(new SearchSourceBuilder().query(QueryBuilders.matchQuery("message", "mastering"))),
- RequestOptions.DEFAULT);
+ RequestOptions.DEFAULT)
+ .block();
assertThat(searchResponse.getHits().getTotalHits()).isEqualTo(1);
SearchResponse searchResponse2 = client.search(
new SearchRequest(INDEX_NAME.getValue())
.source(new SearchSourceBuilder().query(QueryBuilders.matchQuery("field", "unchanged"))),
- RequestOptions.DEFAULT);
+ RequestOptions.DEFAULT)
+ .block();
assertThat(searchResponse2.getHits().getTotalHits()).isEqualTo(1);
}
@Test
void updateMessageShouldThrowWhenJsonIsNull() {
assertThatThrownBy(() -> testee.update(ImmutableList.of(
- new UpdatedRepresentation(DOCUMENT_ID, null)), ROUTING))
+ new UpdatedRepresentation(DOCUMENT_ID, null)), ROUTING).block())
.isInstanceOf(IllegalArgumentException.class);
}
@Test
void updateMessageShouldThrowWhenIdIsNull() {
assertThatThrownBy(() -> testee.update(ImmutableList.of(
- new UpdatedRepresentation(null, "{\"message\": \"mastering out Elasticsearch\"}")), ROUTING))
+ new UpdatedRepresentation(null, "{\"message\": \"mastering out Elasticsearch\"}")), ROUTING).block())
.isInstanceOf(NullPointerException.class);
}
@Test
void updateMessageShouldThrowWhenJsonIsEmpty() {
assertThatThrownBy(() -> testee.update(ImmutableList.of(
- new UpdatedRepresentation(DOCUMENT_ID, "")), ROUTING))
+ new UpdatedRepresentation(DOCUMENT_ID, "")), ROUTING).block())
.isInstanceOf(IllegalArgumentException.class);
}
@Test
void updateMessageShouldThrowWhenRoutingKeyIsNull() {
assertThatThrownBy(() -> testee.update(ImmutableList.of(
- new UpdatedRepresentation(DOCUMENT_ID, "{\"message\": \"mastering out Elasticsearch\"}")), null))
+ new UpdatedRepresentation(DOCUMENT_ID, "{\"message\": \"mastering out Elasticsearch\"}")), null).block())
.isInstanceOf(NullPointerException.class);
}
@Test
- void deleteByQueryShouldWorkOnSingleMessage() throws Exception {
+ void deleteByQueryShouldWorkOnSingleMessage() {
DocumentId documentId = DocumentId.fromString("1:2");
String content = "{\"message\": \"trying out Elasticsearch\", \"property\":\"1\"}";
RoutingKey routingKey = useDocumentId(documentId);
- testee.index(documentId, content, routingKey);
+ testee.index(documentId, content, routingKey).block();
elasticSearch.awaitForElasticSearch();
testee.deleteAllMatchingQuery(termQuery("property", "1"), routingKey);
@@ -169,25 +171,26 @@ class ElasticSearchIndexerTest {
new SearchRequest(INDEX_NAME.getValue())
.source(new SearchSourceBuilder().query(QueryBuilders.matchAllQuery())),
RequestOptions.DEFAULT)
+ .block()
.getHits().getTotalHits() == 0);
}
@Test
- void deleteByQueryShouldWorkWhenMultipleMessages() throws Exception {
+ void deleteByQueryShouldWorkWhenMultipleMessages() {
DocumentId documentId = DocumentId.fromString("1:1");
String content = "{\"message\": \"trying out Elasticsearch\", \"property\":\"1\"}";
- testee.index(documentId, content, ROUTING);
+ testee.index(documentId, content, ROUTING).block();
DocumentId documentId2 = DocumentId.fromString("1:2");
String content2 = "{\"message\": \"trying out Elasticsearch 2\", \"property\":\"1\"}";
- testee.index(documentId2, content2, ROUTING);
+ testee.index(documentId2, content2, ROUTING).block();
DocumentId documentId3 = DocumentId.fromString("2:3");
String content3 = "{\"message\": \"trying out Elasticsearch 3\", \"property\":\"2\"}";
- testee.index(documentId3, content3, ROUTING);
+ testee.index(documentId3, content3, ROUTING).block();
elasticSearch.awaitForElasticSearch();
testee.deleteAllMatchingQuery(termQuery("property", "1"), ROUTING);
@@ -198,64 +201,67 @@ class ElasticSearchIndexerTest {
new SearchRequest(INDEX_NAME.getValue())
.source(new SearchSourceBuilder().query(QueryBuilders.matchAllQuery())),
RequestOptions.DEFAULT)
+ .block()
.getHits().getTotalHits() == 1);
}
@Test
- void deleteMessage() throws Exception {
+ void deleteMessage() {
DocumentId documentId = DocumentId.fromString("1:2");
String content = "{\"message\": \"trying out Elasticsearch\"}";
- testee.index(documentId, content, useDocumentId(documentId));
+ testee.index(documentId, content, useDocumentId(documentId)).block();
elasticSearch.awaitForElasticSearch();
- testee.delete(ImmutableList.of(documentId), useDocumentId(documentId));
+ testee.delete(ImmutableList.of(documentId), useDocumentId(documentId)).block();
elasticSearch.awaitForElasticSearch();
SearchResponse searchResponse = client.search(
new SearchRequest(INDEX_NAME.getValue())
.source(new SearchSourceBuilder().query(QueryBuilders.matchAllQuery())),
- RequestOptions.DEFAULT);
+ RequestOptions.DEFAULT)
+ .block();
assertThat(searchResponse.getHits().getTotalHits()).isEqualTo(0);
}
@Test
- void deleteShouldWorkWhenMultipleMessages() throws Exception {
+ void deleteShouldWorkWhenMultipleMessages() {
DocumentId documentId = DocumentId.fromString("1:1");
String content = "{\"message\": \"trying out Elasticsearch\", \"mailboxId\":\"1\"}";
- testee.index(documentId, content, ROUTING);
+ testee.index(documentId, content, ROUTING).block();
DocumentId documentId2 = DocumentId.fromString("1:2");
String content2 = "{\"message\": \"trying out Elasticsearch 2\", \"mailboxId\":\"1\"}";
- testee.index(documentId2, content2, ROUTING);
+ testee.index(documentId2, content2, ROUTING).block();
DocumentId documentId3 = DocumentId.fromString("2:3");
String content3 = "{\"message\": \"trying out Elasticsearch 3\", \"mailboxId\":\"2\"}";
- testee.index(documentId3, content3, ROUTING);
+ testee.index(documentId3, content3, ROUTING).block();
elasticSearch.awaitForElasticSearch();
- testee.delete(ImmutableList.of(documentId, documentId3), ROUTING);
+ testee.delete(ImmutableList.of(documentId, documentId3), ROUTING).block();
elasticSearch.awaitForElasticSearch();
SearchResponse searchResponse = client.search(
new SearchRequest(INDEX_NAME.getValue())
.source(new SearchSourceBuilder().query(QueryBuilders.matchAllQuery())),
- RequestOptions.DEFAULT);
+ RequestOptions.DEFAULT)
+ .block();
assertThat(searchResponse.getHits().getTotalHits()).isEqualTo(1);
}
@Test
void updateMessagesShouldNotThrowWhenEmptyList() {
- assertThatCode(() -> testee.update(ImmutableList.of(), ROUTING))
+ assertThatCode(() -> testee.update(ImmutableList.of(), ROUTING).block())
.doesNotThrowAnyException();
}
@Test
void deleteMessagesShouldNotThrowWhenEmptyList() {
- assertThatCode(() -> testee.delete(ImmutableList.of(), ROUTING))
+ assertThatCode(() -> testee.delete(ImmutableList.of(), ROUTING).block())
.doesNotThrowAnyException();
}
}
diff --git a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/IndexCreationFactoryTest.java b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/IndexCreationFactoryTest.java
index 0862711..6ff2f7e 100644
--- a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/IndexCreationFactoryTest.java
+++ b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/IndexCreationFactoryTest.java
@@ -23,7 +23,6 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
import java.io.IOException;
-import org.elasticsearch.client.RestHighLevelClient;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -35,7 +34,7 @@ class IndexCreationFactoryTest {
@RegisterExtension
public DockerElasticSearchExtension elasticSearch = new DockerElasticSearchExtension();
- private RestHighLevelClient client;
+ private ReactorElasticSearchClient client;
@BeforeEach
void setUp() {
diff --git a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/NodeMappingFactoryAuthTest.java b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/NodeMappingFactoryAuthTest.java
index dddd34a..8c29fca 100644
--- a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/NodeMappingFactoryAuthTest.java
+++ b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/NodeMappingFactoryAuthTest.java
@@ -25,7 +25,6 @@ import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
import java.io.IOException;
import java.util.Optional;
-import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
@@ -42,7 +41,7 @@ class NodeMappingFactoryAuthTest {
DockerAuthElasticSearchSingleton.INSTANCE,
new DockerElasticSearch.WithAuth()));
- private RestHighLevelClient client;
+ private ReactorElasticSearchClient client;
@BeforeEach
void setUp(ElasticSearchClusterExtension.ElasticSearchCluster esCluster) throws Exception {
diff --git a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/NodeMappingFactoryTest.java b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/NodeMappingFactoryTest.java
index 7d2f52e..e686473 100644
--- a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/NodeMappingFactoryTest.java
+++ b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/NodeMappingFactoryTest.java
@@ -24,7 +24,6 @@ import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
import java.io.IOException;
-import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
@@ -38,7 +37,7 @@ class NodeMappingFactoryTest {
@RegisterExtension
public DockerElasticSearchExtension elasticSearch = new DockerElasticSearchExtension();
- private RestHighLevelClient client;
+ private ReactorElasticSearchClient client;
@BeforeEach
void setUp() throws Exception {
diff --git a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/search/ScrolledSearchTest.java b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/search/ScrolledSearchTest.java
index c4e9e2b..e07368e 100644
--- a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/search/ScrolledSearchTest.java
+++ b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/search/ScrolledSearchTest.java
@@ -29,13 +29,13 @@ import org.apache.james.backends.es.ElasticSearchConfiguration;
import org.apache.james.backends.es.IndexCreationFactory;
import org.apache.james.backends.es.IndexName;
import org.apache.james.backends.es.NodeMappingFactory;
+import org.apache.james.backends.es.ReactorElasticSearchClient;
import org.apache.james.backends.es.ReadAliasName;
import org.awaitility.Duration;
import org.awaitility.core.ConditionFactory;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.client.RequestOptions;
-import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
@@ -56,7 +56,7 @@ class ScrolledSearchTest {
@RegisterExtension
public DockerElasticSearchExtension elasticSearch = new DockerElasticSearchExtension();
- private RestHighLevelClient client;
+ private ReactorElasticSearchClient client;
@BeforeEach
void setUp() {
@@ -81,7 +81,7 @@ class ScrolledSearchTest {
.query(QueryBuilders.matchAllQuery())
.size(SIZE));
- assertThat(new ScrolledSearch(client, searchRequest).searchHits())
+ assertThat(new ScrolledSearch(client, searchRequest).searchHits().collectList().block())
.isEmpty();
}
@@ -92,7 +92,8 @@ class ScrolledSearchTest {
.type(NodeMappingFactory.DEFAULT_MAPPING_NAME)
.id(id)
.source(MESSAGE, "Sample message"),
- RequestOptions.DEFAULT);
+ RequestOptions.DEFAULT)
+ .block();
elasticSearch.awaitForElasticSearch();
WAIT_CONDITION.untilAsserted(() -> hasIdsInIndex(client, id));
@@ -103,26 +104,28 @@ class ScrolledSearchTest {
.query(QueryBuilders.matchAllQuery())
.size(SIZE));
- assertThat(new ScrolledSearch(client, searchRequest).searchHits())
+ assertThat(new ScrolledSearch(client, searchRequest).searchHits().collectList().block())
.extracting(SearchHit::getId)
.containsOnly(id);
}
@Test
- void scrollIterableShouldWorkWhenSizeElement() throws Exception {
+ void scrollIterableShouldWorkWhenSizeElement() {
String id1 = "1";
client.index(new IndexRequest(INDEX_NAME.getValue())
.type(NodeMappingFactory.DEFAULT_MAPPING_NAME)
.id(id1)
.source(MESSAGE, "Sample message"),
- RequestOptions.DEFAULT);
+ RequestOptions.DEFAULT)
+ .block();
String id2 = "2";
client.index(new IndexRequest(INDEX_NAME.getValue())
.type(NodeMappingFactory.DEFAULT_MAPPING_NAME)
.id(id2)
.source(MESSAGE, "Sample message"),
- RequestOptions.DEFAULT);
+ RequestOptions.DEFAULT)
+ .block();
elasticSearch.awaitForElasticSearch();
WAIT_CONDITION.untilAsserted(() -> hasIdsInIndex(client, id1, id2));
@@ -133,33 +136,36 @@ class ScrolledSearchTest {
.query(QueryBuilders.matchAllQuery())
.size(SIZE));
- assertThat(new ScrolledSearch(client, searchRequest).searchHits())
+ assertThat(new ScrolledSearch(client, searchRequest).searchHits().collectList().block())
.extracting(SearchHit::getId)
.containsOnly(id1, id2);
}
@Test
- void scrollIterableShouldWorkWhenMoreThanSizeElement() throws Exception {
+ void scrollIterableShouldWorkWhenMoreThanSizeElement() {
String id1 = "1";
client.index(new IndexRequest(INDEX_NAME.getValue())
.type(NodeMappingFactory.DEFAULT_MAPPING_NAME)
.id(id1)
.source(MESSAGE, "Sample message"),
- RequestOptions.DEFAULT);
+ RequestOptions.DEFAULT)
+ .block();
String id2 = "2";
client.index(new IndexRequest(INDEX_NAME.getValue())
.type(NodeMappingFactory.DEFAULT_MAPPING_NAME)
.id(id2)
.source(MESSAGE, "Sample message"),
- RequestOptions.DEFAULT);
+ RequestOptions.DEFAULT)
+ .block();
String id3 = "3";
client.index(new IndexRequest(INDEX_NAME.getValue())
.type(NodeMappingFactory.DEFAULT_MAPPING_NAME)
.id(id3)
.source(MESSAGE, "Sample message"),
- RequestOptions.DEFAULT);
+ RequestOptions.DEFAULT)
+ .block();
elasticSearch.awaitForElasticSearch();
WAIT_CONDITION.untilAsserted(() -> hasIdsInIndex(client, id1, id2, id3));
@@ -170,18 +176,19 @@ class ScrolledSearchTest {
.query(QueryBuilders.matchAllQuery())
.size(SIZE));
- assertThat(new ScrolledSearch(client, searchRequest).searchHits())
+ assertThat(new ScrolledSearch(client, searchRequest).searchHits().collectList().block())
.extracting(SearchHit::getId)
.containsOnly(id1, id2, id3);
}
- private void hasIdsInIndex(RestHighLevelClient client, String... ids) throws IOException {
+ private void hasIdsInIndex(ReactorElasticSearchClient client, String... ids) {
SearchRequest searchRequest = new SearchRequest(INDEX_NAME.getValue())
.scroll(TIMEOUT)
.source(new SearchSourceBuilder()
.query(QueryBuilders.matchAllQuery()));
SearchHit[] hits = client.search(searchRequest, RequestOptions.DEFAULT)
+ .block()
.getHits()
.getHits();
diff --git a/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/MailboxIndexCreationUtil.java b/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/MailboxIndexCreationUtil.java
index 4b76dad..9871600 100644
--- a/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/MailboxIndexCreationUtil.java
+++ b/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/MailboxIndexCreationUtil.java
@@ -25,17 +25,17 @@ import org.apache.james.backends.es.ElasticSearchConfiguration;
import org.apache.james.backends.es.IndexCreationFactory;
import org.apache.james.backends.es.IndexName;
import org.apache.james.backends.es.NodeMappingFactory;
+import org.apache.james.backends.es.ReactorElasticSearchClient;
import org.apache.james.backends.es.ReadAliasName;
import org.apache.james.backends.es.WriteAliasName;
-import org.elasticsearch.client.RestHighLevelClient;
public class MailboxIndexCreationUtil {
- public static RestHighLevelClient prepareClient(RestHighLevelClient client,
- ReadAliasName readAlias,
- WriteAliasName writeAlias,
- IndexName indexName,
- ElasticSearchConfiguration configuration) throws IOException {
+ public static ReactorElasticSearchClient prepareClient(ReactorElasticSearchClient client,
+ ReadAliasName readAlias,
+ WriteAliasName writeAlias,
+ IndexName indexName,
+ ElasticSearchConfiguration configuration) throws IOException {
return NodeMappingFactory.applyMapping(
new IndexCreationFactory(configuration)
.useIndex(indexName)
@@ -46,7 +46,7 @@ public class MailboxIndexCreationUtil {
MailboxMappingFactory.getMappingContent());
}
- public static RestHighLevelClient prepareDefaultClient(RestHighLevelClient client, ElasticSearchConfiguration configuration) throws IOException {
+ public static ReactorElasticSearchClient prepareDefaultClient(ReactorElasticSearchClient client, ElasticSearchConfiguration configuration) throws IOException {
return prepareClient(client,
MailboxElasticSearchConstants.DEFAULT_MAILBOX_READ_ALIAS,
MailboxElasticSearchConstants.DEFAULT_MAILBOX_WRITE_ALIAS,
diff --git a/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/events/ElasticSearchListeningMessageSearchIndex.java b/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/events/ElasticSearchListeningMessageSearchIndex.java
index c3d8595..1430111 100644
--- a/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/events/ElasticSearchListeningMessageSearchIndex.java
+++ b/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/events/ElasticSearchListeningMessageSearchIndex.java
@@ -52,7 +52,6 @@ import org.apache.james.mailbox.model.UpdatedFlags;
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.apache.james.util.OptionalUtils;
import org.elasticsearch.index.query.TermQueryBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -63,6 +62,8 @@ import com.github.steveash.guavate.Guavate;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
+import reactor.core.publisher.Mono;
+
public class ElasticSearchListeningMessageSearchIndex extends ListeningMessageSearchIndex {
public static class ElasticSearchListeningMessageSearchIndexGroup extends Group {
@@ -112,7 +113,8 @@ public class ElasticSearchListeningMessageSearchIndex extends ListeningMessageSe
return searcher
.search(ImmutableList.of(mailbox.getMailboxId()), searchQuery, noLimit)
- .map(SearchResult::getMessageUid);
+ .map(SearchResult::getMessageUid)
+ .toStream();
}
@Override
@@ -123,15 +125,14 @@ public class ElasticSearchListeningMessageSearchIndex extends ListeningMessageSe
return ImmutableList.of();
}
- try (Stream<SearchResult> searchResults = searcher.search(mailboxIds, searchQuery, Optional.empty())) {
- return searchResults
- .peek(this::logIfNoMessageId)
- .map(SearchResult::getMessageId)
- .flatMap(OptionalUtils::toStream)
- .distinct()
- .limit(limit)
- .collect(Guavate.toImmutableList());
- }
+ return searcher.search(mailboxIds, searchQuery, Optional.empty())
+ .doOnNext(this::logIfNoMessageId)
+ .map(SearchResult::getMessageId)
+ .flatMap(Mono::justOrEmpty)
+ .distinct()
+ .take(limit)
+ .collect(Guavate.toImmutableList())
+ .block();
}
@Override
@@ -144,7 +145,9 @@ public class ElasticSearchListeningMessageSearchIndex extends ListeningMessageSe
String jsonContent = generateIndexedJson(mailbox, message, session);
- elasticSearchIndexer.index(indexIdFor(mailbox, message.getUid()), jsonContent, routingKeyFactory.from(mailbox.getMailboxId()));
+ elasticSearchIndexer
+ .index(indexIdFor(mailbox, message.getUid()), jsonContent, routingKeyFactory.from(mailbox.getMailboxId()))
+ .block();
}
private String generateIndexedJson(Mailbox mailbox, MailboxMessage message, MailboxSession session) throws JsonProcessingException {
@@ -162,12 +165,13 @@ public class ElasticSearchListeningMessageSearchIndex extends ListeningMessageSe
}
@Override
- public void delete(MailboxSession session, Mailbox mailbox, Collection<MessageUid> expungedUids) throws IOException {
+ public void delete(MailboxSession session, Mailbox mailbox, Collection<MessageUid> expungedUids) {
elasticSearchIndexer
.delete(expungedUids.stream()
.map(uid -> indexIdFor(mailbox, uid))
.collect(Guavate.toImmutableList()),
- routingKeyFactory.from(mailbox.getMailboxId()));
+ routingKeyFactory.from(mailbox.getMailboxId()))
+ .block();
}
@Override
@@ -181,14 +185,16 @@ public class ElasticSearchListeningMessageSearchIndex extends ListeningMessageSe
}
@Override
- public void update(MailboxSession session, Mailbox mailbox, List<UpdatedFlags> updatedFlagsList) throws IOException {
+ public void update(MailboxSession session, Mailbox mailbox, List<UpdatedFlags> updatedFlagsList) {
ImmutableList<UpdatedRepresentation> updates = updatedFlagsList.stream()
.map(Throwing.<UpdatedFlags, UpdatedRepresentation>function(
updatedFlags -> createUpdatedDocumentPartFromUpdatedFlags(mailbox, updatedFlags))
.sneakyThrow())
.collect(Guavate.toImmutableList());
- elasticSearchIndexer.update(updates, routingKeyFactory.from(mailbox.getMailboxId()));
+ elasticSearchIndexer
+ .update(updates, routingKeyFactory.from(mailbox.getMailboxId()))
+ .block();
}
private UpdatedRepresentation createUpdatedDocumentPartFromUpdatedFlags(Mailbox mailbox, UpdatedFlags updatedFlags) throws JsonProcessingException {
diff --git a/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/search/ElasticSearchSearcher.java b/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/search/ElasticSearchSearcher.java
index 85cceb1..7dd6509 100644
--- a/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/search/ElasticSearchSearcher.java
+++ b/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/search/ElasticSearchSearcher.java
@@ -21,10 +21,10 @@ package org.apache.james.mailbox.elasticsearch.search;
import java.util.Collection;
import java.util.Optional;
-import java.util.stream.Stream;
import org.apache.james.backends.es.AliasName;
import org.apache.james.backends.es.NodeMappingFactory;
+import org.apache.james.backends.es.ReactorElasticSearchClient;
import org.apache.james.backends.es.ReadAliasName;
import org.apache.james.backends.es.RoutingKey;
import org.apache.james.backends.es.search.ScrolledSearch;
@@ -37,7 +37,6 @@ import org.apache.james.mailbox.model.MessageId;
import org.apache.james.mailbox.model.SearchQuery;
import org.apache.james.mailbox.store.search.MessageSearchIndex;
import org.elasticsearch.action.search.SearchRequest;
-import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.document.DocumentField;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.search.SearchHit;
@@ -47,6 +46,8 @@ import org.slf4j.LoggerFactory;
import com.google.common.collect.ImmutableList;
+import reactor.core.publisher.Flux;
+
public class ElasticSearchSearcher {
public static final int DEFAULT_SEARCH_SIZE = 100;
private static final Logger LOGGER = LoggerFactory.getLogger(ElasticSearchSearcher.class);
@@ -55,7 +56,7 @@ public class ElasticSearchSearcher {
JsonMessageConstants.UID, JsonMessageConstants.MESSAGE_ID);
private static final int MAX_ROUTING_KEY = 5;
- private final RestHighLevelClient client;
+ private final ReactorElasticSearchClient client;
private final QueryConverter queryConverter;
private final int size;
private final MailboxId.Factory mailboxIdFactory;
@@ -63,7 +64,7 @@ public class ElasticSearchSearcher {
private final AliasName aliasName;
private final RoutingKey.Factory<MailboxId> routingKeyFactory;
- public ElasticSearchSearcher(RestHighLevelClient client, QueryConverter queryConverter, int size,
+ public ElasticSearchSearcher(ReactorElasticSearchClient client, QueryConverter queryConverter, int size,
MailboxId.Factory mailboxIdFactory, MessageId.Factory messageIdFactory,
ReadAliasName aliasName, RoutingKey.Factory<MailboxId> routingKeyFactory) {
this.client = client;
@@ -75,14 +76,14 @@ public class ElasticSearchSearcher {
this.routingKeyFactory = routingKeyFactory;
}
- public Stream<MessageSearchIndex.SearchResult> search(Collection<MailboxId> mailboxIds, SearchQuery query,
- Optional<Integer> limit) {
+ public Flux<MessageSearchIndex.SearchResult> search(Collection<MailboxId> mailboxIds, SearchQuery query,
+ Optional<Integer> limit) {
SearchRequest searchRequest = prepareSearch(mailboxIds, query, limit);
- Stream<MessageSearchIndex.SearchResult> pairStream = new ScrolledSearch(client, searchRequest)
+ Flux<MessageSearchIndex.SearchResult> pairStream = new ScrolledSearch(client, searchRequest)
.searchHits()
.flatMap(this::extractContentFromHit);
- return limit.map(pairStream::limit)
+ return limit.map(pairStream::take)
.orElse(pairStream);
}
@@ -122,20 +123,20 @@ public class ElasticSearchSearcher {
.orElse(size);
}
- private Stream<MessageSearchIndex.SearchResult> extractContentFromHit(SearchHit hit) {
+ private Flux<MessageSearchIndex.SearchResult> extractContentFromHit(SearchHit hit) {
DocumentField mailboxId = hit.field(JsonMessageConstants.MAILBOX_ID);
DocumentField uid = hit.field(JsonMessageConstants.UID);
Optional<DocumentField> id = retrieveMessageIdField(hit);
if (mailboxId != null && uid != null) {
Number uidAsNumber = uid.getValue();
- return Stream.of(
+ return Flux.just(
new MessageSearchIndex.SearchResult(
id.map(field -> messageIdFactory.fromString(field.getValue())),
mailboxIdFactory.fromString(mailboxId.getValue()),
MessageUid.of(uidAsNumber.longValue())));
} else {
LOGGER.warn("Can not extract UID, MessageID and/or MailboxId for search result {}", hit.getId());
- return Stream.empty();
+ return Flux.empty();
}
}
diff --git a/mailbox/elasticsearch/src/test/java/org/apache/james/mailbox/elasticsearch/ElasticSearchIntegrationTest.java b/mailbox/elasticsearch/src/test/java/org/apache/james/mailbox/elasticsearch/ElasticSearchIntegrationTest.java
index 04bc2d6..1813ef3 100644
--- a/mailbox/elasticsearch/src/test/java/org/apache/james/mailbox/elasticsearch/ElasticSearchIntegrationTest.java
+++ b/mailbox/elasticsearch/src/test/java/org/apache/james/mailbox/elasticsearch/ElasticSearchIntegrationTest.java
@@ -27,6 +27,7 @@ import java.time.ZoneId;
import org.apache.james.backends.es.DockerElasticSearchExtension;
import org.apache.james.backends.es.ElasticSearchIndexer;
+import org.apache.james.backends.es.ReactorElasticSearchClient;
import org.apache.james.mailbox.MailboxSession;
import org.apache.james.mailbox.MailboxSessionUtil;
import org.apache.james.mailbox.MessageManager;
@@ -48,7 +49,6 @@ import org.apache.james.mailbox.tika.TikaHttpClientImpl;
import org.apache.james.mailbox.tika.TikaTextExtractor;
import org.apache.james.metrics.tests.RecordingMetricFactory;
import org.apache.james.mime4j.dom.Message;
-import org.elasticsearch.client.RestHighLevelClient;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
@@ -67,7 +67,7 @@ class ElasticSearchIntegrationTest extends AbstractMessageSearchIndexTest {
DockerElasticSearchExtension elasticSearch = new DockerElasticSearchExtension();
TikaTextExtractor textExtractor;
- RestHighLevelClient client;
+ ReactorElasticSearchClient client;
@AfterEach
void tearDown() throws IOException {
diff --git a/mailbox/elasticsearch/src/test/java/org/apache/james/mailbox/elasticsearch/events/ElasticSearchListeningMessageSearchIndexTest.java b/mailbox/elasticsearch/src/test/java/org/apache/james/mailbox/elasticsearch/events/ElasticSearchListeningMessageSearchIndexTest.java
index f6d1391..95b67f8 100644
--- a/mailbox/elasticsearch/src/test/java/org/apache/james/mailbox/elasticsearch/events/ElasticSearchListeningMessageSearchIndexTest.java
+++ b/mailbox/elasticsearch/src/test/java/org/apache/james/mailbox/elasticsearch/events/ElasticSearchListeningMessageSearchIndexTest.java
@@ -33,6 +33,7 @@ import javax.mail.util.SharedByteArrayInputStream;
import org.apache.james.backends.es.DockerElasticSearchExtension;
import org.apache.james.backends.es.ElasticSearchIndexer;
+import org.apache.james.backends.es.ReactorElasticSearchClient;
import org.apache.james.core.Username;
import org.apache.james.mailbox.DefaultMailboxes;
import org.apache.james.mailbox.MailboxSession;
@@ -73,7 +74,6 @@ import org.apache.james.mailbox.store.extractor.DefaultTextExtractor;
import org.apache.james.mailbox.store.mail.model.impl.PropertyBuilder;
import org.apache.james.mailbox.store.mail.model.impl.SimpleMailboxMessage;
import org.awaitility.Duration;
-import org.elasticsearch.client.RestHighLevelClient;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
@@ -155,7 +155,7 @@ class ElasticSearchListeningMessageSearchIndexTest {
InMemoryMessageId.Factory messageIdFactory = new InMemoryMessageId.Factory();
- RestHighLevelClient client = MailboxIndexCreationUtil.prepareDefaultClient(
+ ReactorElasticSearchClient client = MailboxIndexCreationUtil.prepareDefaultClient(
elasticSearch.getDockerElasticSearch().clientProvider().get(),
elasticSearch.getDockerElasticSearch().configuration());
@@ -256,7 +256,7 @@ class ElasticSearchListeningMessageSearchIndexTest {
Thread.sleep(Duration.FIVE_SECONDS.getValueInMS()); // Docker pause is asynchronous and we found no way to poll for it
assertThatThrownBy(() -> testee.add(session, mailbox, MESSAGE_1))
- .isInstanceOf(IOException.class);
+ .hasCauseInstanceOf(IOException.class);
elasticSearch.getDockerElasticSearch().unpause();
}
@@ -330,7 +330,7 @@ class ElasticSearchListeningMessageSearchIndexTest {
Thread.sleep(Duration.FIVE_SECONDS.getValueInMS()); // Docker pause is asynchronous and we found no way to poll for it
assertThatThrownBy(() -> testee.delete(session, mailbox, Lists.newArrayList(MESSAGE_UID_1)))
- .isInstanceOf(IOException.class);
+ .hasCauseInstanceOf(IOException.class);
elasticSearch.getDockerElasticSearch().unpause();
}
@@ -413,7 +413,7 @@ class ElasticSearchListeningMessageSearchIndexTest {
.build();
assertThatThrownBy(() -> testee.update(session, mailbox, Lists.newArrayList(updatedFlags)))
- .isInstanceOf(IOException.class);
+ .hasCauseInstanceOf(IOException.class);
elasticSearch.getDockerElasticSearch().unpause();
}
diff --git a/mailbox/elasticsearch/src/test/java/org/apache/james/mailbox/elasticsearch/search/ElasticSearchSearcherTest.java b/mailbox/elasticsearch/src/test/java/org/apache/james/mailbox/elasticsearch/search/ElasticSearchSearcherTest.java
index f8be9a0..f7d6564 100644
--- a/mailbox/elasticsearch/src/test/java/org/apache/james/mailbox/elasticsearch/search/ElasticSearchSearcherTest.java
+++ b/mailbox/elasticsearch/src/test/java/org/apache/james/mailbox/elasticsearch/search/ElasticSearchSearcherTest.java
@@ -30,6 +30,7 @@ import java.util.stream.IntStream;
import org.apache.james.backends.es.DockerElasticSearchExtension;
import org.apache.james.backends.es.ElasticSearchIndexer;
+import org.apache.james.backends.es.ReactorElasticSearchClient;
import org.apache.james.core.Username;
import org.apache.james.mailbox.MailboxSession;
import org.apache.james.mailbox.MailboxSessionUtil;
@@ -58,7 +59,6 @@ import org.apache.james.mailbox.tika.TikaHttpClientImpl;
import org.apache.james.mailbox.tika.TikaTextExtractor;
import org.apache.james.metrics.tests.RecordingMetricFactory;
import org.apache.james.mime4j.dom.Message;
-import org.elasticsearch.client.RestHighLevelClient;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -80,7 +80,7 @@ class ElasticSearchSearcherTest {
DockerElasticSearchExtension elasticSearch = new DockerElasticSearchExtension();
TikaTextExtractor textExtractor;
- RestHighLevelClient client;
+ ReactorElasticSearchClient client;
private InMemoryMailboxManager storeMailboxManager;
@BeforeEach
diff --git a/mailbox/plugin/quota-search-elasticsearch/src/main/java/org/apache/james/quota/search/elasticsearch/ElasticSearchQuotaSearcher.java b/mailbox/plugin/quota-search-elasticsearch/src/main/java/org/apache/james/quota/search/elasticsearch/ElasticSearchQuotaSearcher.java
index 0a1e082..8765345 100644
--- a/mailbox/plugin/quota-search-elasticsearch/src/main/java/org/apache/james/quota/search/elasticsearch/ElasticSearchQuotaSearcher.java
+++ b/mailbox/plugin/quota-search-elasticsearch/src/main/java/org/apache/james/quota/search/elasticsearch/ElasticSearchQuotaSearcher.java
@@ -21,13 +21,11 @@ package org.apache.james.quota.search.elasticsearch;
import static org.apache.james.quota.search.elasticsearch.json.JsonMessageConstants.USER;
-import java.io.IOException;
-import java.util.Arrays;
import java.util.List;
-import java.util.stream.Stream;
import org.apache.james.backends.es.AliasName;
import org.apache.james.backends.es.NodeMappingFactory;
+import org.apache.james.backends.es.ReactorElasticSearchClient;
import org.apache.james.backends.es.ReadAliasName;
import org.apache.james.backends.es.search.ScrolledSearch;
import org.apache.james.core.Username;
@@ -35,7 +33,6 @@ import org.apache.james.quota.search.QuotaQuery;
import org.apache.james.quota.search.QuotaSearcher;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.client.RequestOptions;
-import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
@@ -44,14 +41,16 @@ import org.elasticsearch.search.sort.SortOrder;
import com.github.steveash.guavate.Guavate;
+import reactor.core.publisher.Flux;
+
public class ElasticSearchQuotaSearcher implements QuotaSearcher {
private static final TimeValue TIMEOUT = TimeValue.timeValueMinutes(1);
- private final RestHighLevelClient client;
+ private final ReactorElasticSearchClient client;
private final AliasName readAlias;
private final QuotaQueryConverter quotaQueryConverter;
- public ElasticSearchQuotaSearcher(RestHighLevelClient client, ReadAliasName readAlias) {
+ public ElasticSearchQuotaSearcher(ReactorElasticSearchClient client, ReadAliasName readAlias) {
this.client = client;
this.readAlias = readAlias;
this.quotaQueryConverter = new QuotaQueryConverter();
@@ -60,18 +59,17 @@ public class ElasticSearchQuotaSearcher implements QuotaSearcher {
@Override
public List<Username> search(QuotaQuery query) {
try {
- try (Stream<SearchHit> searchHits = searchHits(query)) {
- return searchHits
- .map(SearchHit::getId)
- .map(Username::of)
- .collect(Guavate.toImmutableList());
- }
- } catch (IOException e) {
+ return searchHits(query)
+ .map(SearchHit::getId)
+ .map(Username::of)
+ .collect(Guavate.toImmutableList())
+ .block();
+ } catch (Exception e) {
throw new RuntimeException("Unexpected exception while executing " + query, e);
}
}
- private Stream<SearchHit> searchHits(QuotaQuery query) throws IOException {
+ private Flux<SearchHit> searchHits(QuotaQuery query) {
if (query.getLimit().isLimited()) {
return executeSingleSearch(query);
} else {
@@ -79,7 +77,7 @@ public class ElasticSearchQuotaSearcher implements QuotaSearcher {
}
}
- private Stream<SearchHit> executeSingleSearch(QuotaQuery query) throws IOException {
+ private Flux<SearchHit> executeSingleSearch(QuotaQuery query) {
SearchSourceBuilder searchSourceBuilder = searchSourceBuilder(query)
.from(query.getOffset().getValue());
query.getLimit().getValue()
@@ -89,12 +87,11 @@ public class ElasticSearchQuotaSearcher implements QuotaSearcher {
.types(NodeMappingFactory.DEFAULT_MAPPING_NAME)
.source(searchSourceBuilder);
- return Arrays.stream(client.search(searchRequest, RequestOptions.DEFAULT)
- .getHits()
- .getHits());
+ return client.search(searchRequest, RequestOptions.DEFAULT)
+ .flatMapMany(searchResponse -> Flux.fromArray(searchResponse.getHits().getHits()));
}
- private Stream<SearchHit> executeScrolledSearch(QuotaQuery query) {
+ private Flux<SearchHit> executeScrolledSearch(QuotaQuery query) {
return new ScrolledSearch(client,
new SearchRequest(readAlias.getValue())
.types(NodeMappingFactory.DEFAULT_MAPPING_NAME)
diff --git a/mailbox/plugin/quota-search-elasticsearch/src/main/java/org/apache/james/quota/search/elasticsearch/QuotaSearchIndexCreationUtil.java b/mailbox/plugin/quota-search-elasticsearch/src/main/java/org/apache/james/quota/search/elasticsearch/QuotaSearchIndexCreationUtil.java
index c1918a9..cf1aad6 100644
--- a/mailbox/plugin/quota-search-elasticsearch/src/main/java/org/apache/james/quota/search/elasticsearch/QuotaSearchIndexCreationUtil.java
+++ b/mailbox/plugin/quota-search-elasticsearch/src/main/java/org/apache/james/quota/search/elasticsearch/QuotaSearchIndexCreationUtil.java
@@ -26,11 +26,11 @@ import org.apache.james.backends.es.ElasticSearchConfiguration;
import org.apache.james.backends.es.IndexCreationFactory;
import org.apache.james.backends.es.IndexName;
import org.apache.james.backends.es.NodeMappingFactory;
-import org.elasticsearch.client.RestHighLevelClient;
+import org.apache.james.backends.es.ReactorElasticSearchClient;
public class QuotaSearchIndexCreationUtil {
- public static RestHighLevelClient prepareClient(RestHighLevelClient client,
+ public static ReactorElasticSearchClient prepareClient(ReactorElasticSearchClient client,
AliasName readAlias,
AliasName writeAlias,
IndexName indexName,
@@ -46,7 +46,7 @@ public class QuotaSearchIndexCreationUtil {
QuotaRatioMappingFactory.getMappingContent());
}
- public static RestHighLevelClient prepareDefaultClient(RestHighLevelClient client, ElasticSearchConfiguration configuration) throws IOException {
+ public static ReactorElasticSearchClient prepareDefaultClient(ReactorElasticSearchClient client, ElasticSearchConfiguration configuration) throws IOException {
return prepareClient(client,
QuotaRatioElasticSearchConstants.DEFAULT_QUOTA_RATIO_READ_ALIAS,
QuotaRatioElasticSearchConstants.DEFAULT_QUOTA_RATIO_WRITE_ALIAS,
diff --git a/mailbox/plugin/quota-search-elasticsearch/src/main/java/org/apache/james/quota/search/elasticsearch/events/ElasticSearchQuotaMailboxListener.java b/mailbox/plugin/quota-search-elasticsearch/src/main/java/org/apache/james/quota/search/elasticsearch/events/ElasticSearchQuotaMailboxListener.java
index b124c18..0a80a56 100644
--- a/mailbox/plugin/quota-search-elasticsearch/src/main/java/org/apache/james/quota/search/elasticsearch/events/ElasticSearchQuotaMailboxListener.java
+++ b/mailbox/plugin/quota-search-elasticsearch/src/main/java/org/apache/james/quota/search/elasticsearch/events/ElasticSearchQuotaMailboxListener.java
@@ -70,9 +70,11 @@ public class ElasticSearchQuotaMailboxListener implements MailboxListener.GroupM
private void handleEvent(QuotaUsageUpdatedEvent event) throws IOException {
Username user = event.getUsername();
- indexer.index(toDocumentId(user),
- quotaRatioToElasticSearchJson.convertToJson(event),
- routingKeyFactory.from(user));
+ indexer
+ .index(toDocumentId(user),
+ quotaRatioToElasticSearchJson.convertToJson(event),
+ routingKeyFactory.from(user))
+ .block();
}
private DocumentId toDocumentId(Username user) {
diff --git a/mailbox/plugin/quota-search-elasticsearch/src/test/java/org/apache/james/quota/search/elasticsearch/ElasticSearchQuotaSearchTestSystemExtension.java b/mailbox/plugin/quota-search-elasticsearch/src/test/java/org/apache/james/quota/search/elasticsearch/ElasticSearchQuotaSearchTestSystemExtension.java
index cc7d8b2..991c5d2 100644
--- a/mailbox/plugin/quota-search-elasticsearch/src/test/java/org/apache/james/quota/search/elasticsearch/ElasticSearchQuotaSearchTestSystemExtension.java
+++ b/mailbox/plugin/quota-search-elasticsearch/src/test/java/org/apache/james/quota/search/elasticsearch/ElasticSearchQuotaSearchTestSystemExtension.java
@@ -26,6 +26,7 @@ import java.io.IOException;
import org.apache.james.backends.es.DockerElasticSearch;
import org.apache.james.backends.es.DockerElasticSearchSingleton;
import org.apache.james.backends.es.ElasticSearchIndexer;
+import org.apache.james.backends.es.ReactorElasticSearchClient;
import org.apache.james.dnsservice.api.DNSService;
import org.apache.james.domainlist.memory.MemoryDomainList;
import org.apache.james.mailbox.inmemory.manager.InMemoryIntegrationResources;
@@ -34,7 +35,6 @@ import org.apache.james.quota.search.QuotaSearchTestSystem;
import org.apache.james.quota.search.elasticsearch.events.ElasticSearchQuotaMailboxListener;
import org.apache.james.quota.search.elasticsearch.json.QuotaRatioToElasticSearchJson;
import org.apache.james.user.memory.MemoryUsersRepository;
-import org.elasticsearch.client.RestHighLevelClient;
import org.junit.jupiter.api.extension.AfterEachCallback;
import org.junit.jupiter.api.extension.BeforeEachCallback;
import org.junit.jupiter.api.extension.ExtensionContext;
@@ -45,7 +45,7 @@ import org.junit.jupiter.api.extension.ParameterResolver;
public class ElasticSearchQuotaSearchTestSystemExtension implements ParameterResolver, BeforeEachCallback, AfterEachCallback {
private final DockerElasticSearch elasticSearch = DockerElasticSearchSingleton.INSTANCE;
- private RestHighLevelClient client;
+ private ReactorElasticSearchClient client;
@Override
public boolean supportsParameter(ParameterContext parameterContext, ExtensionContext extensionContext) throws ParameterResolutionException {
diff --git a/mailbox/plugin/quota-search-elasticsearch/src/test/java/org/apache/james/quota/search/elasticsearch/events/ElasticSearchQuotaMailboxListenerTest.java b/mailbox/plugin/quota-search-elasticsearch/src/test/java/org/apache/james/quota/search/elasticsearch/events/ElasticSearchQuotaMailboxListenerTest.java
index ba482aa..9e5671b 100644
--- a/mailbox/plugin/quota-search-elasticsearch/src/test/java/org/apache/james/quota/search/elasticsearch/events/ElasticSearchQuotaMailboxListenerTest.java
+++ b/mailbox/plugin/quota-search-elasticsearch/src/test/java/org/apache/james/quota/search/elasticsearch/events/ElasticSearchQuotaMailboxListenerTest.java
@@ -30,6 +30,7 @@ import org.apache.james.backends.es.DockerElasticSearchExtension;
import org.apache.james.backends.es.ElasticSearchConfiguration;
import org.apache.james.backends.es.ElasticSearchIndexer;
import org.apache.james.backends.es.NodeMappingFactory;
+import org.apache.james.backends.es.ReactorElasticSearchClient;
import org.apache.james.mailbox.events.Event;
import org.apache.james.mailbox.events.Group;
import org.apache.james.mailbox.quota.QuotaFixture.Counts;
@@ -41,7 +42,6 @@ import org.apache.james.quota.search.elasticsearch.UserRoutingKeyFactory;
import org.apache.james.quota.search.elasticsearch.json.QuotaRatioToElasticSearchJson;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
-import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.junit.jupiter.api.AfterEach;
@@ -58,7 +58,7 @@ class ElasticSearchQuotaMailboxListenerTest {
DockerElasticSearchExtension elasticSearch = new DockerElasticSearchExtension();
ElasticSearchQuotaMailboxListener quotaMailboxListener;
- RestHighLevelClient client;
+ ReactorElasticSearchClient client;
@BeforeEach
void setUp() throws IOException {
@@ -100,10 +100,11 @@ class ElasticSearchQuotaMailboxListenerTest {
elasticSearch.awaitForElasticSearch();
- SearchResponse searchResponse = client.search(new SearchRequest(QuotaRatioElasticSearchConstants.DEFAULT_QUOTA_RATIO_READ_ALIAS.getValue())
- .types(NodeMappingFactory.DEFAULT_MAPPING_NAME)
- .source(new SearchSourceBuilder()
- .query(QueryBuilders.matchAllQuery())));
+ SearchRequest searchRequest = new SearchRequest(QuotaRatioElasticSearchConstants.DEFAULT_QUOTA_RATIO_READ_ALIAS.getValue())
+ .types(NodeMappingFactory.DEFAULT_MAPPING_NAME)
+ .source(new SearchSourceBuilder()
+ .query(QueryBuilders.matchAllQuery()));
+ SearchResponse searchResponse = client.search(searchRequest).block();
assertThat(searchResponse.getHits().getTotalHits()).isEqualTo(1);
}
diff --git a/mpt/impl/imap-mailbox/elasticsearch/src/test/java/org/apache/james/mpt/imapmailbox/elasticsearch/host/ElasticSearchHostSystem.java b/mpt/impl/imap-mailbox/elasticsearch/src/test/java/org/apache/james/mpt/imapmailbox/elasticsearch/host/ElasticSearchHostSystem.java
index 38e1d5c..233fb26 100644
--- a/mpt/impl/imap-mailbox/elasticsearch/src/test/java/org/apache/james/mpt/imapmailbox/elasticsearch/host/ElasticSearchHostSystem.java
+++ b/mpt/impl/imap-mailbox/elasticsearch/src/test/java/org/apache/james/mpt/imapmailbox/elasticsearch/host/ElasticSearchHostSystem.java
@@ -27,6 +27,7 @@ import org.apache.james.backends.es.DockerElasticSearch;
import org.apache.james.backends.es.DockerElasticSearchSingleton;
import org.apache.james.backends.es.ElasticSearchConfiguration;
import org.apache.james.backends.es.ElasticSearchIndexer;
+import org.apache.james.backends.es.ReactorElasticSearchClient;
import org.apache.james.core.quota.QuotaCountLimit;
import org.apache.james.core.quota.QuotaSizeLimit;
import org.apache.james.imap.api.process.ImapProcessor;
@@ -54,7 +55,6 @@ import org.apache.james.metrics.logger.DefaultMetricFactory;
import org.apache.james.mpt.api.ImapFeatures;
import org.apache.james.mpt.api.ImapFeatures.Feature;
import org.apache.james.mpt.host.JamesImapHostSystem;
-import org.elasticsearch.client.RestHighLevelClient;
public class ElasticSearchHostSystem extends JamesImapHostSystem {
@@ -63,7 +63,7 @@ public class ElasticSearchHostSystem extends JamesImapHostSystem {
private DockerElasticSearch dockerElasticSearch;
private StoreMailboxManager mailboxManager;
- private RestHighLevelClient client;
+ private ReactorElasticSearchClient client;
@Override
public void beforeTest() throws Exception {
diff --git a/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchClientModule.java b/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchClientModule.java
index 7aab691..ed9b27e 100644
--- a/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchClientModule.java
+++ b/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchClientModule.java
@@ -24,10 +24,10 @@ import java.util.Set;
import org.apache.james.backends.es.ClientProvider;
import org.apache.james.backends.es.ElasticSearchHealthCheck;
import org.apache.james.backends.es.IndexName;
+import org.apache.james.backends.es.ReactorElasticSearchClient;
import org.apache.james.core.healthcheck.HealthCheck;
import org.apache.james.mailbox.elasticsearch.ElasticSearchMailboxConfiguration;
import org.apache.james.quota.search.elasticsearch.ElasticSearchQuotaConfiguration;
-import org.elasticsearch.client.RestHighLevelClient;
import com.google.common.collect.ImmutableSet;
import com.google.inject.AbstractModule;
@@ -41,7 +41,7 @@ public class ElasticSearchClientModule extends AbstractModule {
@Override
protected void configure() {
bind(ClientProvider.class).in(Scopes.SINGLETON);
- bind(RestHighLevelClient.class).toProvider(ClientProvider.class);
+ bind(ReactorElasticSearchClient.class).toProvider(ClientProvider.class);
Multibinder.newSetBinder(binder(), HealthCheck.class)
.addBinding()
diff --git a/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchMailboxModule.java b/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchMailboxModule.java
index bcc8148..b02c837 100644
--- a/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchMailboxModule.java
+++ b/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchMailboxModule.java
@@ -32,6 +32,7 @@ import org.apache.commons.configuration2.Configuration;
import org.apache.commons.configuration2.ex.ConfigurationException;
import org.apache.james.backends.es.ElasticSearchConfiguration;
import org.apache.james.backends.es.ElasticSearchIndexer;
+import org.apache.james.backends.es.ReactorElasticSearchClient;
import org.apache.james.backends.es.RoutingKey;
import org.apache.james.lifecycle.api.StartUpCheck;
import org.apache.james.lifecycle.api.Startable;
@@ -51,7 +52,6 @@ import org.apache.james.mailbox.store.search.MessageSearchIndex;
import org.apache.james.utils.InitializationOperation;
import org.apache.james.utils.InitilizationOperationBuilder;
import org.apache.james.utils.PropertiesProvider;
-import org.elasticsearch.client.RestHighLevelClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -68,12 +68,12 @@ public class ElasticSearchMailboxModule extends AbstractModule {
private final ElasticSearchConfiguration configuration;
private final ElasticSearchMailboxConfiguration mailboxConfiguration;
- private final RestHighLevelClient client;
+ private final ReactorElasticSearchClient client;
@Inject
MailboxIndexCreator(ElasticSearchConfiguration configuration,
ElasticSearchMailboxConfiguration mailboxConfiguration,
- RestHighLevelClient client) {
+ ReactorElasticSearchClient client) {
this.configuration = configuration;
this.mailboxConfiguration = mailboxConfiguration;
this.client = client;
@@ -114,7 +114,7 @@ public class ElasticSearchMailboxModule extends AbstractModule {
@Provides
@Singleton
@Named(MailboxElasticSearchConstants.InjectionNames.MAILBOX)
- private ElasticSearchIndexer createMailboxElasticSearchIndexer(RestHighLevelClient client,
+ private ElasticSearchIndexer createMailboxElasticSearchIndexer(ReactorElasticSearchClient client,
ElasticSearchMailboxConfiguration configuration) {
return new ElasticSearchIndexer(
client,
@@ -123,7 +123,7 @@ public class ElasticSearchMailboxModule extends AbstractModule {
@Provides
@Singleton
- private ElasticSearchSearcher createMailboxElasticSearchSearcher(RestHighLevelClient client,
+ private ElasticSearchSearcher createMailboxElasticSearchSearcher(ReactorElasticSearchClient client,
QueryConverter queryConverter,
MailboxId.Factory mailboxIdFactory,
MessageId.Factory messageIdFactory,
diff --git a/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchQuotaSearcherModule.java b/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchQuotaSearcherModule.java
index eb4f9b4..123b02f 100644
--- a/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchQuotaSearcherModule.java
+++ b/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchQuotaSearcherModule.java
@@ -30,6 +30,7 @@ import org.apache.commons.configuration2.Configuration;
import org.apache.commons.configuration2.ex.ConfigurationException;
import org.apache.james.backends.es.ElasticSearchConfiguration;
import org.apache.james.backends.es.ElasticSearchIndexer;
+import org.apache.james.backends.es.ReactorElasticSearchClient;
import org.apache.james.lifecycle.api.Startable;
import org.apache.james.mailbox.events.MailboxListener;
import org.apache.james.quota.search.QuotaSearcher;
@@ -42,7 +43,6 @@ import org.apache.james.quota.search.elasticsearch.json.QuotaRatioToElasticSearc
import org.apache.james.utils.InitializationOperation;
import org.apache.james.utils.InitilizationOperationBuilder;
import org.apache.james.utils.PropertiesProvider;
-import org.elasticsearch.client.RestHighLevelClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -57,12 +57,12 @@ public class ElasticSearchQuotaSearcherModule extends AbstractModule {
static class ElasticSearchQuotaIndexCreator implements Startable {
private final ElasticSearchConfiguration configuration;
private final ElasticSearchQuotaConfiguration quotaConfiguration;
- private final RestHighLevelClient client;
+ private final ReactorElasticSearchClient client;
@Inject
ElasticSearchQuotaIndexCreator(ElasticSearchConfiguration configuration,
ElasticSearchQuotaConfiguration quotaConfiguration,
- RestHighLevelClient client) {
+ ReactorElasticSearchClient client) {
this.configuration = configuration;
this.quotaConfiguration = quotaConfiguration;
this.client = client;
@@ -88,7 +88,7 @@ public class ElasticSearchQuotaSearcherModule extends AbstractModule {
@Provides
@Singleton
- public QuotaSearcher provideSearcher(RestHighLevelClient client, ElasticSearchQuotaConfiguration configuration) {
+ public QuotaSearcher provideSearcher(ReactorElasticSearchClient client, ElasticSearchQuotaConfiguration configuration) {
return new ElasticSearchQuotaSearcher(client,
configuration.getReadAliasQuotaRatioName());
}
@@ -107,7 +107,7 @@ public class ElasticSearchQuotaSearcherModule extends AbstractModule {
@Provides
@Singleton
- public ElasticSearchQuotaMailboxListener provideListener(RestHighLevelClient client,
+ public ElasticSearchQuotaMailboxListener provideListener(ReactorElasticSearchClient client,
ElasticSearchQuotaConfiguration configuration) {
return new ElasticSearchQuotaMailboxListener(
new ElasticSearchIndexer(client,
diff --git a/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchStartUpCheck.java b/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchStartUpCheck.java
index 4bf89d7..3565b38 100644
--- a/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchStartUpCheck.java
+++ b/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchStartUpCheck.java
@@ -24,10 +24,10 @@ import java.io.IOException;
import javax.inject.Inject;
import org.apache.james.backends.es.ElasticSearchConfiguration;
+import org.apache.james.backends.es.ReactorElasticSearchClient;
import org.apache.james.lifecycle.api.StartUpCheck;
import org.elasticsearch.Version;
import org.elasticsearch.client.RequestOptions;
-import org.elasticsearch.client.RestHighLevelClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -40,10 +40,10 @@ public class ElasticSearchStartUpCheck implements StartUpCheck {
public static final String CHECK_NAME = "ElasticSearchStartUpCheck";
- private final RestHighLevelClient client;
+ private final ReactorElasticSearchClient client;
@Inject
- private ElasticSearchStartUpCheck(RestHighLevelClient client) {
+ private ElasticSearchStartUpCheck(ReactorElasticSearchClient client) {
this.client = client;
}
diff --git a/server/container/guice/cassandra-guice/src/test/java/org/apache/james/ESReporterTest.java b/server/container/guice/cassandra-guice/src/test/java/org/apache/james/ESReporterTest.java
index 23e7cff..81aa4c0 100644
--- a/server/container/guice/cassandra-guice/src/test/java/org/apache/james/ESReporterTest.java
+++ b/server/container/guice/cassandra-guice/src/test/java/org/apache/james/ESReporterTest.java
@@ -34,6 +34,7 @@ import java.util.TimerTask;
import java.util.stream.Collectors;
import org.apache.commons.net.imap.IMAPClient;
+import org.apache.james.backends.es.ReactorElasticSearchClient;
import org.apache.james.core.Username;
import org.apache.james.jmap.AccessToken;
import org.apache.james.jmap.draft.JmapGuiceProbe;
@@ -45,7 +46,6 @@ import org.apache.james.modules.protocols.ImapGuiceProbe;
import org.apache.james.utils.DataProbeImpl;
import org.awaitility.Duration;
import org.elasticsearch.action.search.SearchRequest;
-import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.junit.jupiter.api.AfterEach;
@@ -152,12 +152,13 @@ class ESReporterTest {
}
private boolean checkMetricRecordedInElasticSearch() {
- try (RestHighLevelClient client = elasticSearchExtension.getDockerES().clientProvider().get()) {
+ try (ReactorElasticSearchClient client = elasticSearchExtension.getDockerES().clientProvider().get()) {
SearchRequest searchRequest = new SearchRequest()
.source(new SearchSourceBuilder()
.query(QueryBuilders.matchAllQuery()));
return !Arrays.stream(client
.search(searchRequest)
+ .block()
.getHits()
.getHits())
.filter(searchHit -> searchHit.getIndex().startsWith(TestDockerESMetricReporterModule.METRICS_INDEX))
diff --git a/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/ElasticSearchQuotaSearchExtension.java b/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/ElasticSearchQuotaSearchExtension.java
index e57771f..392b859 100644
--- a/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/ElasticSearchQuotaSearchExtension.java
+++ b/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/ElasticSearchQuotaSearchExtension.java
@@ -27,6 +27,7 @@ import org.apache.james.backends.es.DockerElasticSearch;
import org.apache.james.backends.es.DockerElasticSearchSingleton;
import org.apache.james.backends.es.ElasticSearchConfiguration;
import org.apache.james.backends.es.ElasticSearchIndexer;
+import org.apache.james.backends.es.ReactorElasticSearchClient;
import org.apache.james.dnsservice.api.DNSService;
import org.apache.james.domainlist.memory.MemoryDomainList;
import org.apache.james.mailbox.inmemory.manager.InMemoryIntegrationResources;
@@ -39,7 +40,6 @@ import org.apache.james.quota.search.elasticsearch.UserRoutingKeyFactory;
import org.apache.james.quota.search.elasticsearch.events.ElasticSearchQuotaMailboxListener;
import org.apache.james.quota.search.elasticsearch.json.QuotaRatioToElasticSearchJson;
import org.apache.james.user.memory.MemoryUsersRepository;
-import org.elasticsearch.client.RestHighLevelClient;
import org.junit.jupiter.api.extension.AfterEachCallback;
import org.junit.jupiter.api.extension.BeforeEachCallback;
import org.junit.jupiter.api.extension.ExtensionContext;
@@ -53,7 +53,7 @@ public class ElasticSearchQuotaSearchExtension implements ParameterResolver, Bef
private final DockerElasticSearch elasticSearch = DockerElasticSearchSingleton.INSTANCE;
private WebAdminQuotaSearchTestSystem restQuotaSearchTestSystem;
private TemporaryFolder temporaryFolder = new TemporaryFolder();
- private RestHighLevelClient client;
+ private ReactorElasticSearchClient client;
@Override
public void beforeEach(ExtensionContext context) {
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 01/06: JAMES-2997 Drop Cassandra schema version
prior version 5
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 1469c4a41cd7f083317d6d0efe44d6a278057f04
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Mon Apr 6 19:41:25 2020 +0700
JAMES-2997 Drop Cassandra schema version prior version 5
---
.../versions/CassandraSchemaVersionManager.java | 2 +-
.../CassandraMailboxSessionMapperFactory.java | 7 +-
.../cassandra/mail/CassandraAttachmentMapper.java | 11 +-
.../migration/AttachmentMessageIdCreation.java | 62 ------
.../mail/migration/AttachmentV2Migration.java | 65 ------
.../CassandraSubscriptionManagerTest.java | 4 -
.../cassandra/mail/CassandraAttachmentDAOTest.java | 118 -----------
.../mail/CassandraAttachmentFallbackTest.java | 180 -----------------
.../migration/AttachmentMessageIdCreationTest.java | 198 ------------------
.../mail/migration/AttachmentV2MigrationTest.java | 221 ---------------------
.../SolveMailboxInconsistenciesServiceTest.java | 2 +-
.../modules/mailbox/CassandraMailboxModule.java | 2 -
.../modules/webadmin/CassandraRoutesModule.java | 8 -
upgrade-instructions.md | 12 ++
14 files changed, 18 insertions(+), 874 deletions(-)
diff --git a/backends-common/cassandra/src/main/java/org/apache/james/backends/cassandra/versions/CassandraSchemaVersionManager.java b/backends-common/cassandra/src/main/java/org/apache/james/backends/cassandra/versions/CassandraSchemaVersionManager.java
index ef56bdb..9b47f39 100644
--- a/backends-common/cassandra/src/main/java/org/apache/james/backends/cassandra/versions/CassandraSchemaVersionManager.java
+++ b/backends-common/cassandra/src/main/java/org/apache/james/backends/cassandra/versions/CassandraSchemaVersionManager.java
@@ -33,7 +33,7 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
public class CassandraSchemaVersionManager {
- public static final SchemaVersion MIN_VERSION = new SchemaVersion(2);
+ public static final SchemaVersion MIN_VERSION = new SchemaVersion(5);
public static final SchemaVersion MAX_VERSION = new SchemaVersion(7);
public static final SchemaVersion DEFAULT_VERSION = MIN_VERSION;
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/CassandraMailboxSessionMapperFactory.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/CassandraMailboxSessionMapperFactory.java
index 301e28a..d6f506e 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/CassandraMailboxSessionMapperFactory.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/CassandraMailboxSessionMapperFactory.java
@@ -29,7 +29,6 @@ import org.apache.james.mailbox.MailboxSession;
import org.apache.james.mailbox.cassandra.mail.CassandraACLMapper;
import org.apache.james.mailbox.cassandra.mail.CassandraAnnotationMapper;
import org.apache.james.mailbox.cassandra.mail.CassandraApplicableFlagDAO;
-import org.apache.james.mailbox.cassandra.mail.CassandraAttachmentDAO;
import org.apache.james.mailbox.cassandra.mail.CassandraAttachmentDAOV2;
import org.apache.james.mailbox.cassandra.mail.CassandraAttachmentMapper;
import org.apache.james.mailbox.cassandra.mail.CassandraAttachmentMessageIdDAO;
@@ -85,7 +84,6 @@ public class CassandraMailboxSessionMapperFactory extends MailboxSessionMapperFa
private final CassandraMailboxPathV2DAO mailboxPathV2DAO;
private final CassandraFirstUnseenDAO firstUnseenDAO;
private final CassandraApplicableFlagDAO applicableFlagDAO;
- private final CassandraAttachmentDAO attachmentDAO;
private final CassandraAttachmentDAOV2 attachmentDAOV2;
private final CassandraDeletedMessageDAO deletedMessageDAO;
private final BlobStore blobStore;
@@ -103,7 +101,7 @@ public class CassandraMailboxSessionMapperFactory extends MailboxSessionMapperFa
CassandraMessageIdDAO messageIdDAO, CassandraMessageIdToImapUidDAO imapUidDAO,
CassandraMailboxCounterDAO mailboxCounterDAO, CassandraMailboxRecentsDAO mailboxRecentsDAO, CassandraMailboxDAO mailboxDAO,
CassandraMailboxPathDAOImpl mailboxPathDAO, CassandraMailboxPathV2DAO mailboxPathV2DAO, CassandraFirstUnseenDAO firstUnseenDAO, CassandraApplicableFlagDAO applicableFlagDAO,
- CassandraAttachmentDAO attachmentDAO, CassandraAttachmentDAOV2 attachmentDAOV2, CassandraDeletedMessageDAO deletedMessageDAO,
+ CassandraAttachmentDAOV2 attachmentDAOV2, CassandraDeletedMessageDAO deletedMessageDAO,
BlobStore blobStore, CassandraAttachmentMessageIdDAO attachmentMessageIdDAO,
CassandraAttachmentOwnerDAO ownerDAO, CassandraACLMapper aclMapper,
CassandraUserMailboxRightsDAO userMailboxRightsDAO,
@@ -121,7 +119,6 @@ public class CassandraMailboxSessionMapperFactory extends MailboxSessionMapperFa
this.mailboxPathDAO = mailboxPathDAO;
this.mailboxPathV2DAO = mailboxPathV2DAO;
this.firstUnseenDAO = firstUnseenDAO;
- this.attachmentDAO = attachmentDAO;
this.attachmentDAOV2 = attachmentDAOV2;
this.deletedMessageDAO = deletedMessageDAO;
this.applicableFlagDAO = applicableFlagDAO;
@@ -174,7 +171,7 @@ public class CassandraMailboxSessionMapperFactory extends MailboxSessionMapperFa
@Override
public CassandraAttachmentMapper createAttachmentMapper(MailboxSession mailboxSession) {
- return new CassandraAttachmentMapper(attachmentDAO, attachmentDAOV2, blobStore, attachmentMessageIdDAO, ownerDAO);
+ return new CassandraAttachmentMapper(attachmentDAOV2, blobStore, attachmentMessageIdDAO, ownerDAO);
}
@Override
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraAttachmentMapper.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraAttachmentMapper.java
index a2c4bad..9c7f0b2 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraAttachmentMapper.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraAttachmentMapper.java
@@ -48,15 +48,13 @@ import reactor.core.publisher.Mono;
public class CassandraAttachmentMapper implements AttachmentMapper {
private static final Logger LOGGER = LoggerFactory.getLogger(CassandraAttachmentMapper.class);
- private final CassandraAttachmentDAO attachmentDAO;
private final CassandraAttachmentDAOV2 attachmentDAOV2;
private final BlobStore blobStore;
private final CassandraAttachmentMessageIdDAO attachmentMessageIdDAO;
private final CassandraAttachmentOwnerDAO ownerDAO;
@Inject
- public CassandraAttachmentMapper(CassandraAttachmentDAO attachmentDAO, CassandraAttachmentDAOV2 attachmentDAOV2, BlobStore blobStore, CassandraAttachmentMessageIdDAO attachmentMessageIdDAO, CassandraAttachmentOwnerDAO ownerDAO) {
- this.attachmentDAO = attachmentDAO;
+ public CassandraAttachmentMapper(CassandraAttachmentDAOV2 attachmentDAOV2, BlobStore blobStore, CassandraAttachmentMessageIdDAO attachmentMessageIdDAO, CassandraAttachmentOwnerDAO ownerDAO) {
this.attachmentDAOV2 = attachmentDAOV2;
this.blobStore = blobStore;
this.attachmentMessageIdDAO = attachmentMessageIdDAO;
@@ -101,12 +99,7 @@ public class CassandraAttachmentMapper implements AttachmentMapper {
private Mono<Attachment> getAttachmentInternal(AttachmentId id) {
return attachmentDAOV2.getAttachment(id)
- .flatMap(this::retrievePayload)
- .switchIfEmpty(fallbackToV1(id));
- }
-
- private Mono<Attachment> fallbackToV1(AttachmentId attachmentId) {
- return attachmentDAO.getAttachment(attachmentId);
+ .flatMap(this::retrievePayload);
}
@Override
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/migration/AttachmentMessageIdCreation.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/migration/AttachmentMessageIdCreation.java
deleted file mode 100644
index b70b8db..0000000
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/migration/AttachmentMessageIdCreation.java
+++ /dev/null
@@ -1,62 +0,0 @@
-/****************************************************************
- * 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.mailbox.cassandra.mail.migration;
-
-import javax.inject.Inject;
-
-import org.apache.james.backends.cassandra.migration.Migration;
-import org.apache.james.mailbox.cassandra.mail.CassandraAttachmentMessageIdDAO;
-import org.apache.james.mailbox.cassandra.mail.CassandraMessageDAO;
-import org.apache.james.mailbox.cassandra.mail.CassandraMessageDAO.MessageIdAttachmentIds;
-import org.apache.james.mailbox.model.MessageId;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-
-public class AttachmentMessageIdCreation implements Migration {
- private static final Logger LOGGER = LoggerFactory.getLogger(AttachmentMessageIdCreation.class);
- private final CassandraMessageDAO cassandraMessageDAO;
- private final CassandraAttachmentMessageIdDAO attachmentMessageIdDAO;
-
- @Inject
- public AttachmentMessageIdCreation(CassandraMessageDAO cassandraMessageDAO,
- CassandraAttachmentMessageIdDAO attachmentMessageIdDAO) {
- this.cassandraMessageDAO = cassandraMessageDAO;
- this.attachmentMessageIdDAO = attachmentMessageIdDAO;
- }
-
- @Override
- public void apply() {
- cassandraMessageDAO.retrieveAllMessageIdAttachmentIds()
- .flatMap(this::createIndex)
- .doOnError(e -> LOGGER.error("Error while creation attachmentId -> messageIds index", e))
- .blockLast();
- }
-
- private Mono<Void> createIndex(MessageIdAttachmentIds message) {
- MessageId messageId = message.getMessageId();
- return Flux.fromIterable(message.getAttachmentId())
- .flatMap(attachmentId -> attachmentMessageIdDAO.storeAttachmentForMessageId(attachmentId, messageId))
- .then();
- }
-
-}
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/migration/AttachmentV2Migration.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/migration/AttachmentV2Migration.java
deleted file mode 100644
index 020a8d5..0000000
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/migration/AttachmentV2Migration.java
+++ /dev/null
@@ -1,65 +0,0 @@
-/****************************************************************
- * 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.mailbox.cassandra.mail.migration;
-
-import static org.apache.james.blob.api.BlobStore.StoragePolicy.LOW_COST;
-
-import javax.inject.Inject;
-
-import org.apache.james.backends.cassandra.migration.Migration;
-import org.apache.james.blob.api.BlobStore;
-import org.apache.james.mailbox.cassandra.mail.CassandraAttachmentDAO;
-import org.apache.james.mailbox.cassandra.mail.CassandraAttachmentDAOV2;
-import org.apache.james.mailbox.model.Attachment;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import reactor.core.publisher.Mono;
-
-public class AttachmentV2Migration implements Migration {
- private static final Logger LOGGER = LoggerFactory.getLogger(AttachmentV2Migration.class);
- private final CassandraAttachmentDAO attachmentDAOV1;
- private final CassandraAttachmentDAOV2 attachmentDAOV2;
- private final BlobStore blobStore;
-
- @Inject
- public AttachmentV2Migration(CassandraAttachmentDAO attachmentDAOV1,
- CassandraAttachmentDAOV2 attachmentDAOV2,
- BlobStore blobStore) {
- this.attachmentDAOV1 = attachmentDAOV1;
- this.attachmentDAOV2 = attachmentDAOV2;
- this.blobStore = blobStore;
- }
-
- @Override
- public void apply() {
- attachmentDAOV1.retrieveAll()
- .flatMap(this::migrateAttachment)
- .doOnError(t -> LOGGER.error("Error while performing attachmentDAO V2 migration", t))
- .blockLast();
- }
-
- private Mono<Void> migrateAttachment(Attachment attachment) {
- return Mono.from(blobStore.save(blobStore.getDefaultBucketName(), attachment.getBytes(), LOW_COST))
- .map(blobId -> CassandraAttachmentDAOV2.from(attachment, blobId))
- .flatMap(attachmentDAOV2::storeAttachment)
- .then(attachmentDAOV1.deleteAttachment(attachment.getAttachmentId()));
- }
-}
diff --git a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/CassandraSubscriptionManagerTest.java b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/CassandraSubscriptionManagerTest.java
index a230e11..f555a7f 100644
--- a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/CassandraSubscriptionManagerTest.java
+++ b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/CassandraSubscriptionManagerTest.java
@@ -22,14 +22,12 @@ package org.apache.james.mailbox.cassandra;
import org.apache.james.backends.cassandra.CassandraClusterExtension;
import org.apache.james.backends.cassandra.init.configuration.CassandraConfiguration;
import org.apache.james.backends.cassandra.utils.CassandraUtils;
-import org.apache.james.backends.cassandra.versions.CassandraSchemaVersionDAO;
import org.apache.james.backends.cassandra.versions.CassandraSchemaVersionManager;
import org.apache.james.blob.api.BlobStore;
import org.apache.james.mailbox.SubscriptionManager;
import org.apache.james.mailbox.SubscriptionManagerContract;
import org.apache.james.mailbox.cassandra.mail.CassandraACLMapper;
import org.apache.james.mailbox.cassandra.mail.CassandraApplicableFlagDAO;
-import org.apache.james.mailbox.cassandra.mail.CassandraAttachmentDAO;
import org.apache.james.mailbox.cassandra.mail.CassandraAttachmentDAOV2;
import org.apache.james.mailbox.cassandra.mail.CassandraAttachmentMessageIdDAO;
import org.apache.james.mailbox.cassandra.mail.CassandraAttachmentOwnerDAO;
@@ -78,7 +76,6 @@ class CassandraSubscriptionManagerTest implements SubscriptionManagerContract {
CassandraMailboxPathV2DAO mailboxPathV2DAO = null;
CassandraFirstUnseenDAO firstUnseenDAO = null;
CassandraApplicableFlagDAO applicableFlagDAO = null;
- CassandraAttachmentDAO attachmentDAO = null;
CassandraDeletedMessageDAO deletedMessageDAO = null;
CassandraAttachmentDAOV2 attachmentDAOV2 = null;
CassandraAttachmentMessageIdDAO attachmentMessageIdDAO = null;
@@ -105,7 +102,6 @@ class CassandraSubscriptionManagerTest implements SubscriptionManagerContract {
mailboxPathV2DAO,
firstUnseenDAO,
applicableFlagDAO,
- attachmentDAO,
attachmentDAOV2,
deletedMessageDAO,
blobStore,
diff --git a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraAttachmentDAOTest.java b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraAttachmentDAOTest.java
deleted file mode 100644
index ec41c87..0000000
--- a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraAttachmentDAOTest.java
+++ /dev/null
@@ -1,118 +0,0 @@
-/****************************************************************
- * 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.mailbox.cassandra.mail;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
-import java.nio.charset.StandardCharsets;
-import java.util.Optional;
-
-import org.apache.james.backends.cassandra.CassandraCluster;
-import org.apache.james.backends.cassandra.CassandraClusterExtension;
-import org.apache.james.backends.cassandra.init.configuration.CassandraConfiguration;
-import org.apache.james.mailbox.cassandra.modules.CassandraAttachmentModule;
-import org.apache.james.mailbox.model.Attachment;
-import org.apache.james.mailbox.model.AttachmentId;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.RegisterExtension;
-
-class CassandraAttachmentDAOTest {
- private static final AttachmentId ATTACHMENT_ID = AttachmentId.from("id1");
- private static final AttachmentId ATTACHMENT_ID_2 = AttachmentId.from("id2");
-
- @RegisterExtension
- static CassandraClusterExtension cassandraCluster = new CassandraClusterExtension(CassandraAttachmentModule.MODULE);
-
- private CassandraAttachmentDAO testee;
-
- @BeforeEach
- void setUp(CassandraCluster cassandra) {
- testee = new CassandraAttachmentDAO(cassandra.getConf(),
- CassandraConfiguration.DEFAULT_CONFIGURATION);
- }
-
- @Test
- void getAttachmentShouldReturnEmptyWhenAbsent() {
- Optional<Attachment> attachment = testee.getAttachment(ATTACHMENT_ID).blockOptional();
-
- assertThat(attachment).isEmpty();
- }
-
- @Test
- void retrieveAllShouldReturnEmptyByDefault() {
- assertThat(
- testee.retrieveAll()
- .collectList()
- .block())
- .isEmpty();
- }
-
- @Test
- void retrieveAllShouldReturnStoredAttachments() throws Exception {
- Attachment attachment1 = Attachment.builder()
- .attachmentId(ATTACHMENT_ID)
- .type("application/json")
- .bytes("{\"property\":`\"value1\"}".getBytes(StandardCharsets.UTF_8))
- .build();
- Attachment attachment2 = Attachment.builder()
- .attachmentId(ATTACHMENT_ID_2)
- .type("application/json")
- .bytes("{\"property\":`\"value2\"}".getBytes(StandardCharsets.UTF_8))
- .build();
- testee.storeAttachment(attachment1).block();
- testee.storeAttachment(attachment2).block();
-
- assertThat(
- testee.retrieveAll()
- .collectList()
- .block())
- .containsOnly(attachment1, attachment2);
- }
-
- @Test
- void getAttachmentShouldReturnAttachmentWhenStored() throws Exception {
- Attachment attachment = Attachment.builder()
- .attachmentId(ATTACHMENT_ID)
- .type("application/json")
- .bytes("{\"property\":`\"value\"}".getBytes(StandardCharsets.UTF_8))
- .build();
- testee.storeAttachment(attachment).block();
-
- Optional<Attachment> actual = testee.getAttachment(ATTACHMENT_ID).blockOptional();
-
- assertThat(actual).contains(attachment);
- }
-
- @Test
- void deleteAttachmentShouldRemoveAttachment() throws Exception {
- Attachment attachment = Attachment.builder()
- .attachmentId(ATTACHMENT_ID)
- .type("application/json")
- .bytes("{\"property\":`\"value\"}".getBytes(StandardCharsets.UTF_8))
- .build();
- testee.storeAttachment(attachment).block();
-
- testee.deleteAttachment(attachment.getAttachmentId()).block();
-
- assertThat(testee.getAttachment(attachment.getAttachmentId()).blockOptional())
- .isEmpty();
- }
-}
diff --git a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraAttachmentFallbackTest.java b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraAttachmentFallbackTest.java
deleted file mode 100644
index 98aacf0..0000000
--- a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraAttachmentFallbackTest.java
+++ /dev/null
@@ -1,180 +0,0 @@
-/****************************************************************
- * 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.mailbox.cassandra.mail;
-
-import static org.apache.james.blob.api.BlobStore.StoragePolicy.LOW_COST;
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
-
-import java.nio.charset.StandardCharsets;
-import java.util.List;
-
-import org.apache.james.backends.cassandra.CassandraCluster;
-import org.apache.james.backends.cassandra.CassandraClusterExtension;
-import org.apache.james.backends.cassandra.components.CassandraModule;
-import org.apache.james.backends.cassandra.init.configuration.CassandraConfiguration;
-import org.apache.james.blob.api.BlobId;
-import org.apache.james.blob.api.HashBlobId;
-import org.apache.james.blob.cassandra.CassandraBlobModule;
-import org.apache.james.blob.cassandra.CassandraBlobStore;
-import org.apache.james.mailbox.cassandra.ids.CassandraMessageId;
-import org.apache.james.mailbox.cassandra.modules.CassandraAttachmentModule;
-import org.apache.james.mailbox.exception.AttachmentNotFoundException;
-import org.apache.james.mailbox.model.Attachment;
-import org.apache.james.mailbox.model.AttachmentId;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.RegisterExtension;
-
-import com.google.common.collect.ImmutableList;
-
-class CassandraAttachmentFallbackTest {
- private static final AttachmentId ATTACHMENT_ID_1 = AttachmentId.from("id1");
- private static final AttachmentId ATTACHMENT_ID_2 = AttachmentId.from("id2");
- private static final HashBlobId.Factory BLOB_ID_FACTORY = new HashBlobId.Factory();
-
- @RegisterExtension
- static CassandraClusterExtension cassandraCluster = new CassandraClusterExtension(
- CassandraModule.aggregateModules(
- CassandraAttachmentModule.MODULE,
- CassandraBlobModule.MODULE));
-
- private CassandraAttachmentDAOV2 attachmentDAOV2;
- private CassandraAttachmentDAO attachmentDAO;
- private CassandraAttachmentMapper attachmentMapper;
- private CassandraBlobStore blobStore;
- private CassandraAttachmentMessageIdDAO attachmentMessageIdDAO;
-
-
- @BeforeEach
- void setUp(CassandraCluster cassandra) {
- attachmentDAOV2 = new CassandraAttachmentDAOV2(BLOB_ID_FACTORY, cassandra.getConf());
- attachmentDAO = new CassandraAttachmentDAO(cassandra.getConf(),
- CassandraConfiguration.DEFAULT_CONFIGURATION);
- blobStore = CassandraBlobStore.forTesting(cassandra.getConf());
- attachmentMessageIdDAO = new CassandraAttachmentMessageIdDAO(cassandra.getConf(), new CassandraMessageId.Factory());
- CassandraAttachmentOwnerDAO ownerDAO = new CassandraAttachmentOwnerDAO(cassandra.getConf());
- attachmentMapper = new CassandraAttachmentMapper(attachmentDAO, attachmentDAOV2, blobStore, attachmentMessageIdDAO, ownerDAO);
- }
-
- @Test
- void getAttachmentShouldThrowWhenAbsentFromV1AndV2() {
- assertThatThrownBy(() -> attachmentMapper.getAttachment(ATTACHMENT_ID_1))
- .isInstanceOf(AttachmentNotFoundException.class);
- }
-
- @Test
- void getAttachmentsShouldReturnEmptyWhenAbsentFromV1AndV2() {
- assertThat(attachmentMapper.getAttachments(ImmutableList.of(ATTACHMENT_ID_1)))
- .isEmpty();
- }
-
- @Test
- void getAttachmentShouldReturnV2WhenPresentInV1AndV2() throws Exception {
- Attachment attachment = Attachment.builder()
- .attachmentId(ATTACHMENT_ID_1)
- .type("application/json")
- .bytes("{\"property\":`\"value\"}".getBytes(StandardCharsets.UTF_8))
- .build();
- Attachment otherAttachment = Attachment.builder()
- .attachmentId(ATTACHMENT_ID_1)
- .type("application/json")
- .bytes("{\"property\":`\"different\"}".getBytes(StandardCharsets.UTF_8))
- .build();
-
- BlobId blobId = blobStore.save(blobStore.getDefaultBucketName(), attachment.getBytes(), LOW_COST).block();
- attachmentDAOV2.storeAttachment(CassandraAttachmentDAOV2.from(attachment, blobId)).block();
- attachmentDAO.storeAttachment(otherAttachment).block();
-
- assertThat(attachmentMapper.getAttachment(ATTACHMENT_ID_1))
- .isEqualTo(attachment);
- }
-
- @Test
- void getAttachmentShouldReturnV1WhenV2Absent() throws Exception {
- Attachment attachment = Attachment.builder()
- .attachmentId(ATTACHMENT_ID_1)
- .type("application/json")
- .bytes("{\"property\":`\"value\"}".getBytes(StandardCharsets.UTF_8))
- .build();
-
- attachmentDAO.storeAttachment(attachment).block();
-
- assertThat(attachmentMapper.getAttachment(ATTACHMENT_ID_1))
- .isEqualTo(attachment);
- }
-
- @Test
- void getAttachmentsShouldReturnV2WhenV2AndV1() throws Exception {
- Attachment attachment = Attachment.builder()
- .attachmentId(ATTACHMENT_ID_1)
- .type("application/json")
- .bytes("{\"property\":`\"value\"}".getBytes(StandardCharsets.UTF_8))
- .build();
- Attachment otherAttachment = Attachment.builder()
- .attachmentId(ATTACHMENT_ID_1)
- .type("application/json")
- .bytes("{\"property\":`\"different\"}".getBytes(StandardCharsets.UTF_8))
- .build();
-
- BlobId blobId = blobStore.save(blobStore.getDefaultBucketName(), attachment.getBytes(), LOW_COST).block();
- attachmentDAOV2.storeAttachment(CassandraAttachmentDAOV2.from(attachment, blobId)).block();
- attachmentDAO.storeAttachment(otherAttachment).block();
-
- assertThat(attachmentMapper.getAttachments(ImmutableList.of(ATTACHMENT_ID_1)))
- .containsExactly(attachment);
- }
-
- @Test
- void getAttachmentsShouldReturnV1WhenV2Absent() throws Exception {
- Attachment attachment = Attachment.builder()
- .attachmentId(ATTACHMENT_ID_1)
- .type("application/json")
- .bytes("{\"property\":`\"value\"}".getBytes(StandardCharsets.UTF_8))
- .build();
-
- attachmentDAO.storeAttachment(attachment).block();
-
- assertThat(attachmentMapper.getAttachments(ImmutableList.of(ATTACHMENT_ID_1)))
- .containsExactly(attachment);
- }
-
- @Test
- void getAttachmentsShouldCombineElementsFromV1AndV2() throws Exception {
- Attachment attachment = Attachment.builder()
- .attachmentId(ATTACHMENT_ID_1)
- .type("application/json")
- .bytes("{\"property\":`\"value\"}".getBytes(StandardCharsets.UTF_8))
- .build();
- Attachment otherAttachment = Attachment.builder()
- .attachmentId(ATTACHMENT_ID_2)
- .type("application/json")
- .bytes("{\"property\":`\"different\"}".getBytes(StandardCharsets.UTF_8))
- .build();
-
- BlobId blobId = blobStore.save(blobStore.getDefaultBucketName(), attachment.getBytes(), LOW_COST).block();
- attachmentDAOV2.storeAttachment(CassandraAttachmentDAOV2.from(attachment, blobId)).block();
- attachmentDAO.storeAttachment(otherAttachment).block();
-
- List<Attachment> attachments = attachmentMapper.getAttachments(ImmutableList.of(ATTACHMENT_ID_1, ATTACHMENT_ID_2));
- assertThat(attachments)
- .containsExactlyInAnyOrder(attachment, otherAttachment);
- }
-}
diff --git a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/migration/AttachmentMessageIdCreationTest.java b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/migration/AttachmentMessageIdCreationTest.java
deleted file mode 100644
index 531c719..0000000
--- a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/migration/AttachmentMessageIdCreationTest.java
+++ /dev/null
@@ -1,198 +0,0 @@
-/****************************************************************
- * 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.mailbox.cassandra.mail.migration;
-
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
-import java.nio.charset.StandardCharsets;
-import java.util.Collection;
-import java.util.Date;
-import java.util.List;
-
-import javax.mail.Flags;
-import javax.mail.util.SharedByteArrayInputStream;
-
-import org.apache.james.backends.cassandra.CassandraCluster;
-import org.apache.james.backends.cassandra.CassandraClusterExtension;
-import org.apache.james.backends.cassandra.components.CassandraModule;
-import org.apache.james.backends.cassandra.versions.CassandraSchemaVersionModule;
-import org.apache.james.blob.api.HashBlobId;
-import org.apache.james.blob.cassandra.CassandraBlobModule;
-import org.apache.james.blob.cassandra.CassandraBlobStore;
-import org.apache.james.mailbox.MessageUid;
-import org.apache.james.mailbox.cassandra.ids.CassandraId;
-import org.apache.james.mailbox.cassandra.ids.CassandraMessageId;
-import org.apache.james.mailbox.cassandra.mail.CassandraAttachmentMessageIdDAO;
-import org.apache.james.mailbox.cassandra.mail.CassandraMessageDAO;
-import org.apache.james.mailbox.cassandra.modules.CassandraAttachmentModule;
-import org.apache.james.mailbox.cassandra.modules.CassandraMessageModule;
-import org.apache.james.mailbox.model.Attachment;
-import org.apache.james.mailbox.model.AttachmentId;
-import org.apache.james.mailbox.model.MessageAttachment;
-import org.apache.james.mailbox.model.MessageId;
-import org.apache.james.mailbox.store.mail.model.impl.PropertyBuilder;
-import org.apache.james.mailbox.store.mail.model.impl.SimpleMailboxMessage;
-import org.apache.james.task.Task;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.RegisterExtension;
-
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableSet;
-
-import reactor.core.publisher.Flux;
-
-class AttachmentMessageIdCreationTest {
- public static final CassandraModule MODULES = CassandraModule.aggregateModules(
- CassandraMessageModule.MODULE,
- CassandraAttachmentModule.MODULE,
- CassandraBlobModule.MODULE,
- CassandraSchemaVersionModule.MODULE);
-
- @RegisterExtension
- static CassandraClusterExtension cassandraCluster = new CassandraClusterExtension(
- MODULES);
-
- private CassandraBlobStore blobStore;
- private CassandraMessageDAO cassandraMessageDAO;
- private CassandraAttachmentMessageIdDAO attachmentMessageIdDAO;
-
- private AttachmentMessageIdCreation migration;
-
- private SimpleMailboxMessage message;
- private CassandraMessageId messageId;
-
- @BeforeEach
- void setUp(CassandraCluster cassandra) {
- CassandraMessageId.Factory messageIdFactory = new CassandraMessageId.Factory();
-
- blobStore = CassandraBlobStore.forTesting(cassandra.getConf());
- cassandraMessageDAO = new CassandraMessageDAO(cassandra.getConf(), cassandra.getTypesProvider(),
- blobStore, new HashBlobId.Factory(), messageIdFactory);
-
- attachmentMessageIdDAO = new CassandraAttachmentMessageIdDAO(cassandra.getConf(),
- new CassandraMessageId.Factory());
-
- migration = new AttachmentMessageIdCreation(cassandraMessageDAO, attachmentMessageIdDAO);
-
- messageId = messageIdFactory.generate();
- }
-
- @Test
- void emptyMigrationShouldSucceed() throws InterruptedException {
- assertThat(migration.asTask().run())
- .isEqualTo(Task.Result.COMPLETED);
- }
-
- @Test
- void migrationShouldSucceedWhenNoAttachment() throws Exception {
- List<MessageAttachment> noAttachment = ImmutableList.of();
- message = createMessage(messageId, noAttachment);
-
- cassandraMessageDAO.save(message).block();
-
- assertThat(migration.asTask().run())
- .isEqualTo(Task.Result.COMPLETED);
- }
-
- @Test
- void migrationShouldSucceedWhenAttachment() throws Exception {
- MessageAttachment attachment = createAttachment();
- message = createMessage(messageId, ImmutableList.of(attachment));
-
- cassandraMessageDAO.save(message).block();
-
- assertThat(migration.asTask().run())
- .isEqualTo(Task.Result.COMPLETED);
- }
-
- @Test
- void migrationShouldCreateAttachmentIdOnAttachmentMessageIdTableFromMessage() throws Exception {
- MessageAttachment attachment = createAttachment();
- message = createMessage(messageId, ImmutableList.of(attachment));
-
- cassandraMessageDAO.save(message).block();
-
- migration.apply();
-
- assertThat(attachmentMessageIdDAO.getOwnerMessageIds(attachment.getAttachmentId()).toIterable())
- .containsExactly(messageId);
- }
-
- @Test
- void migrationShouldReturnPartialWhenRetrieveAllAttachmentIdFromMessageFail() throws Exception {
- CassandraMessageDAO cassandraMessageDAO = mock(CassandraMessageDAO.class);
- CassandraAttachmentMessageIdDAO attachmentMessageIdDAO = mock(CassandraAttachmentMessageIdDAO.class);
- migration = new AttachmentMessageIdCreation(cassandraMessageDAO, attachmentMessageIdDAO);
-
- when(cassandraMessageDAO.retrieveAllMessageIdAttachmentIds()).thenReturn(Flux.error(new RuntimeException("Mocked exception")));
-
- assertThat(migration.asTask().run()).isEqualTo(Task.Result.PARTIAL);
- }
-
- @Test
- void migrationShouldReturnPartialWhenSavingAttachmentIdForMessageIdFail() throws Exception {
- CassandraMessageDAO cassandraMessageDAO = mock(CassandraMessageDAO.class);
- CassandraAttachmentMessageIdDAO attachmentMessageIdDAO = mock(CassandraAttachmentMessageIdDAO.class);
- CassandraMessageDAO.MessageIdAttachmentIds messageIdAttachmentIds = mock(CassandraMessageDAO.MessageIdAttachmentIds.class);
-
- migration = new AttachmentMessageIdCreation(cassandraMessageDAO, attachmentMessageIdDAO);
-
- when(messageIdAttachmentIds.getAttachmentId()).thenReturn(ImmutableSet.of(AttachmentId.from("any")));
- when(cassandraMessageDAO.retrieveAllMessageIdAttachmentIds())
- .thenReturn(Flux.just(messageIdAttachmentIds));
- when(attachmentMessageIdDAO.storeAttachmentForMessageId(any(AttachmentId.class), any(MessageId.class)))
- .thenThrow(new RuntimeException());
-
- assertThat(migration.asTask().run()).isEqualTo(Task.Result.PARTIAL);
- }
-
- private SimpleMailboxMessage createMessage(MessageId messageId, Collection<MessageAttachment> attachments) {
- MessageUid messageUid = MessageUid.of(1);
- CassandraId mailboxId = CassandraId.timeBased();
- String content = "Subject: Any subject \n\nThis is the body\n.\n";
- int bodyStart = 22;
-
- return SimpleMailboxMessage.builder()
- .messageId(messageId)
- .mailboxId(mailboxId)
- .uid(messageUid)
- .internalDate(new Date())
- .bodyStartOctet(bodyStart)
- .size(content.length())
- .content(new SharedByteArrayInputStream(content.getBytes(StandardCharsets.UTF_8)))
- .flags(new Flags())
- .propertyBuilder(new PropertyBuilder())
- .addAttachments(attachments)
- .build();
- }
-
- private MessageAttachment createAttachment() {
- return MessageAttachment.builder()
- .attachment(Attachment.builder()
- .bytes("content".getBytes(StandardCharsets.UTF_8))
- .type("type")
- .build())
- .build();
- }
-}
\ No newline at end of file
diff --git a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/migration/AttachmentV2MigrationTest.java b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/migration/AttachmentV2MigrationTest.java
deleted file mode 100644
index 4e40a47..0000000
--- a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/migration/AttachmentV2MigrationTest.java
+++ /dev/null
@@ -1,221 +0,0 @@
-/****************************************************************
- * 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.mailbox.cassandra.mail.migration;
-
-import static org.apache.james.blob.api.BlobStore.StoragePolicy.LOW_COST;
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.eq;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
-import java.nio.charset.StandardCharsets;
-
-import org.apache.james.backends.cassandra.CassandraCluster;
-import org.apache.james.backends.cassandra.CassandraClusterExtension;
-import org.apache.james.backends.cassandra.components.CassandraModule;
-import org.apache.james.backends.cassandra.init.configuration.CassandraConfiguration;
-import org.apache.james.blob.api.BucketName;
-import org.apache.james.blob.api.HashBlobId;
-import org.apache.james.blob.cassandra.CassandraBlobModule;
-import org.apache.james.blob.cassandra.CassandraBlobStore;
-import org.apache.james.mailbox.cassandra.mail.CassandraAttachmentDAO;
-import org.apache.james.mailbox.cassandra.mail.CassandraAttachmentDAOV2;
-import org.apache.james.mailbox.cassandra.modules.CassandraAttachmentModule;
-import org.apache.james.mailbox.model.Attachment;
-import org.apache.james.mailbox.model.AttachmentId;
-import org.apache.james.task.Task;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.RegisterExtension;
-
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-
-class AttachmentV2MigrationTest {
- private static final AttachmentId ATTACHMENT_ID = AttachmentId.from("id1");
- private static final AttachmentId ATTACHMENT_ID_2 = AttachmentId.from("id2");
- private static final HashBlobId.Factory BLOB_ID_FACTORY = new HashBlobId.Factory();
-
- @RegisterExtension
- static CassandraClusterExtension cassandraCluster = new CassandraClusterExtension(
- CassandraModule.aggregateModules(
- CassandraAttachmentModule.MODULE,
- CassandraBlobModule.MODULE));
-
- private CassandraAttachmentDAO attachmentDAO;
- private CassandraAttachmentDAOV2 attachmentDAOV2;
- private CassandraBlobStore blobsStore;
- private AttachmentV2Migration migration;
- private Attachment attachment1;
- private Attachment attachment2;
-
- @BeforeEach
- void setUp(CassandraCluster cassandra) {
- attachmentDAO = new CassandraAttachmentDAO(cassandra.getConf(),
- CassandraConfiguration.DEFAULT_CONFIGURATION);
- attachmentDAOV2 = new CassandraAttachmentDAOV2(BLOB_ID_FACTORY, cassandra.getConf());
- blobsStore = CassandraBlobStore.forTesting(cassandra.getConf());
- migration = new AttachmentV2Migration(attachmentDAO, attachmentDAOV2, blobsStore);
-
- attachment1 = Attachment.builder()
- .attachmentId(ATTACHMENT_ID)
- .type("application/json")
- .bytes("{\"property\":`\"value1\"}".getBytes(StandardCharsets.UTF_8))
- .build();
- attachment2 = Attachment.builder()
- .attachmentId(ATTACHMENT_ID_2)
- .type("application/json")
- .bytes("{\"property\":`\"value2\"}".getBytes(StandardCharsets.UTF_8))
- .build();
- }
-
- @Test
- void emptyMigrationShouldSucceed() throws InterruptedException {
- assertThat(migration.asTask().run())
- .isEqualTo(Task.Result.COMPLETED);
- }
-
- @Test
- void migrationShouldSucceed() throws Exception {
- attachmentDAO.storeAttachment(attachment1).block();
- attachmentDAO.storeAttachment(attachment2).block();
-
- assertThat(migration.asTask().run())
- .isEqualTo(Task.Result.COMPLETED);
- }
-
- @Test
- void migrationShouldMoveAttachmentsToV2() throws Exception {
- attachmentDAO.storeAttachment(attachment1).block();
- attachmentDAO.storeAttachment(attachment2).block();
-
- migration.apply();
-
- assertThat(attachmentDAOV2.getAttachment(ATTACHMENT_ID).blockOptional())
- .contains(CassandraAttachmentDAOV2.from(attachment1, BLOB_ID_FACTORY.forPayload(attachment1.getBytes())));
- assertThat(attachmentDAOV2.getAttachment(ATTACHMENT_ID_2).blockOptional())
- .contains(CassandraAttachmentDAOV2.from(attachment2, BLOB_ID_FACTORY.forPayload(attachment2.getBytes())));
- assertThat(blobsStore.readBytes(blobsStore.getDefaultBucketName(), BLOB_ID_FACTORY.forPayload(attachment1.getBytes())).block())
- .isEqualTo(attachment1.getBytes());
- assertThat(blobsStore.readBytes(blobsStore.getDefaultBucketName(), BLOB_ID_FACTORY.forPayload(attachment2.getBytes())).block())
- .isEqualTo(attachment2.getBytes());
- }
-
- @Test
- void migrationShouldRemoveAttachmentsFromV1() throws Exception {
- attachmentDAO.storeAttachment(attachment1).block();
- attachmentDAO.storeAttachment(attachment2).block();
-
- migration.apply();
-
- assertThat(attachmentDAO.getAttachment(ATTACHMENT_ID).blockOptional())
- .isEmpty();
- assertThat(attachmentDAO.getAttachment(ATTACHMENT_ID_2).blockOptional())
- .isEmpty();
- }
-
- @Test
- void runShouldReturnPartialWhenInitialReadFail() throws InterruptedException {
- CassandraAttachmentDAO attachmentDAO = mock(CassandraAttachmentDAO.class);
- CassandraAttachmentDAOV2 attachmentDAOV2 = mock(CassandraAttachmentDAOV2.class);
- CassandraBlobStore blobsStore = mock(CassandraBlobStore.class);
- migration = new AttachmentV2Migration(attachmentDAO, attachmentDAOV2, blobsStore);
-
- when(attachmentDAO.retrieveAll()).thenReturn(Flux.error(new RuntimeException()));
-
- assertThat(migration.asTask().run()).isEqualTo(Task.Result.PARTIAL);
- }
-
- @Test
- void runShouldReturnPartialWhenSavingBlobsFails() throws InterruptedException {
- CassandraAttachmentDAO attachmentDAO = mock(CassandraAttachmentDAO.class);
- CassandraAttachmentDAOV2 attachmentDAOV2 = mock(CassandraAttachmentDAOV2.class);
- CassandraBlobStore blobsStore = mock(CassandraBlobStore.class);
- migration = new AttachmentV2Migration(attachmentDAO, attachmentDAOV2, blobsStore);
-
- when(attachmentDAO.retrieveAll()).thenReturn(Flux.just(
- attachment1,
- attachment2));
- when(blobsStore.save(any(BucketName.class), any(byte[].class), eq(LOW_COST))).thenThrow(new RuntimeException());
-
- assertThat(migration.asTask().run()).isEqualTo(Task.Result.PARTIAL);
- }
-
- @Test
- void runShouldReturnPartialWhenSavingAttachmentV2Fail() throws InterruptedException {
- CassandraAttachmentDAO attachmentDAO = mock(CassandraAttachmentDAO.class);
- CassandraAttachmentDAOV2 attachmentDAOV2 = mock(CassandraAttachmentDAOV2.class);
- CassandraBlobStore blobsStore = mock(CassandraBlobStore.class);
- migration = new AttachmentV2Migration(attachmentDAO, attachmentDAOV2, blobsStore);
-
- when(attachmentDAO.retrieveAll()).thenReturn(Flux.just(
- attachment1,
- attachment2));
- when(blobsStore.save(blobsStore.getDefaultBucketName(), attachment1.getBytes(), LOW_COST))
- .thenReturn(Mono.just(BLOB_ID_FACTORY.forPayload(attachment1.getBytes())));
- when(blobsStore.save(blobsStore.getDefaultBucketName(), attachment2.getBytes(), LOW_COST))
- .thenReturn(Mono.just(BLOB_ID_FACTORY.forPayload(attachment2.getBytes())));
- when(attachmentDAOV2.storeAttachment(any())).thenThrow(new RuntimeException());
-
- assertThat(migration.asTask().run()).isEqualTo(Task.Result.PARTIAL);
- }
-
- @Test
- void runShouldReturnPartialWhenDeleteV1AttachmentFail() throws InterruptedException {
- CassandraAttachmentDAO attachmentDAO = mock(CassandraAttachmentDAO.class);
- CassandraAttachmentDAOV2 attachmentDAOV2 = mock(CassandraAttachmentDAOV2.class);
- CassandraBlobStore blobsStore = mock(CassandraBlobStore.class);
- migration = new AttachmentV2Migration(attachmentDAO, attachmentDAOV2, blobsStore);
-
- when(attachmentDAO.retrieveAll()).thenReturn(Flux.just(
- attachment1,
- attachment2));
- when(blobsStore.save(blobsStore.getDefaultBucketName(), attachment1.getBytes(), LOW_COST))
- .thenReturn(Mono.just(BLOB_ID_FACTORY.forPayload(attachment1.getBytes())));
- when(blobsStore.save(blobsStore.getDefaultBucketName(), attachment2.getBytes(), LOW_COST))
- .thenReturn(Mono.just(BLOB_ID_FACTORY.forPayload(attachment2.getBytes())));
- when(attachmentDAOV2.storeAttachment(any())).thenReturn(Mono.empty());
- when(attachmentDAO.deleteAttachment(any())).thenThrow(new RuntimeException());
-
- assertThat(migration.asTask().run()).isEqualTo(Task.Result.PARTIAL);
- }
-
- @Test
- void runShouldReturnPartialWhenAtLeastOneAttachmentMigrationFails() throws InterruptedException {
- CassandraAttachmentDAO attachmentDAO = mock(CassandraAttachmentDAO.class);
- CassandraAttachmentDAOV2 attachmentDAOV2 = mock(CassandraAttachmentDAOV2.class);
- CassandraBlobStore blobsStore = mock(CassandraBlobStore.class);
- migration = new AttachmentV2Migration(attachmentDAO, attachmentDAOV2, blobsStore);
-
- when(attachmentDAO.retrieveAll()).thenReturn(Flux.just(
- attachment1,
- attachment2));
- when(blobsStore.save(blobsStore.getDefaultBucketName(), attachment1.getBytes(), LOW_COST))
- .thenReturn(Mono.just(BLOB_ID_FACTORY.forPayload(attachment1.getBytes())));
- when(blobsStore.save(blobsStore.getDefaultBucketName(), attachment2.getBytes(), LOW_COST))
- .thenThrow(new RuntimeException());
- when(attachmentDAOV2.storeAttachment(any())).thenReturn(Mono.empty());
- when(attachmentDAO.deleteAttachment(any())).thenReturn(Mono.empty());
-
- assertThat(migration.asTask().run()).isEqualTo(Task.Result.PARTIAL);
- }
-
-}
diff --git a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/task/SolveMailboxInconsistenciesServiceTest.java b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/task/SolveMailboxInconsistenciesServiceTest.java
index b99367a..5b9b1ec 100644
--- a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/task/SolveMailboxInconsistenciesServiceTest.java
+++ b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/task/SolveMailboxInconsistenciesServiceTest.java
@@ -97,7 +97,7 @@ class SolveMailboxInconsistenciesServiceTest {
assertThatThrownBy(() -> testee.fixMailboxInconsistencies(new Context()).block())
.isInstanceOf(IllegalStateException.class)
- .hasMessage("Schema version 6 is required in order to ensure mailboxPathV2DAO to be correctly populated, got 2");
+ .hasMessage("Schema version 6 is required in order to ensure mailboxPathV2DAO to be correctly populated, got 5");
}
@Test
diff --git a/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/CassandraMailboxModule.java b/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/CassandraMailboxModule.java
index 2733ac6..80d5579 100644
--- a/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/CassandraMailboxModule.java
+++ b/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/CassandraMailboxModule.java
@@ -39,7 +39,6 @@ import org.apache.james.mailbox.cassandra.ids.CassandraId;
import org.apache.james.mailbox.cassandra.ids.CassandraMessageId;
import org.apache.james.mailbox.cassandra.mail.CassandraACLMapper;
import org.apache.james.mailbox.cassandra.mail.CassandraApplicableFlagDAO;
-import org.apache.james.mailbox.cassandra.mail.CassandraAttachmentDAO;
import org.apache.james.mailbox.cassandra.mail.CassandraAttachmentDAOV2;
import org.apache.james.mailbox.cassandra.mail.CassandraAttachmentMessageIdDAO;
import org.apache.james.mailbox.cassandra.mail.CassandraAttachmentOwnerDAO;
@@ -111,7 +110,6 @@ public class CassandraMailboxModule extends AbstractModule {
install(new CassandraDeadLetterModule());
bind(CassandraApplicableFlagDAO.class).in(Scopes.SINGLETON);
- bind(CassandraAttachmentDAO.class).in(Scopes.SINGLETON);
bind(CassandraAttachmentDAOV2.class).in(Scopes.SINGLETON);
bind(CassandraAttachmentMessageIdDAO.class).in(Scopes.SINGLETON);
bind(CassandraAttachmentOwnerDAO.class).in(Scopes.SINGLETON);
diff --git a/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/webadmin/CassandraRoutesModule.java b/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/webadmin/CassandraRoutesModule.java
index 8812df0..23a4f0a 100644
--- a/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/webadmin/CassandraRoutesModule.java
+++ b/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/webadmin/CassandraRoutesModule.java
@@ -25,8 +25,6 @@ import org.apache.james.backends.cassandra.migration.MigrationTask;
import org.apache.james.backends.cassandra.versions.CassandraSchemaVersionManager;
import org.apache.james.backends.cassandra.versions.SchemaTransition;
import org.apache.james.backends.cassandra.versions.SchemaVersion;
-import org.apache.james.mailbox.cassandra.mail.migration.AttachmentMessageIdCreation;
-import org.apache.james.mailbox.cassandra.mail.migration.AttachmentV2Migration;
import org.apache.james.mailbox.cassandra.mail.migration.MailboxPathV2Migration;
import org.apache.james.rrt.cassandra.migration.MappingsSourcesMigration;
import org.apache.james.webadmin.Routes;
@@ -40,9 +38,6 @@ import com.google.inject.multibindings.Multibinder;
import com.google.inject.name.Names;
public class CassandraRoutesModule extends AbstractModule {
- private static final SchemaTransition FROM_V2_TO_V3 = SchemaTransition.to(new SchemaVersion(3));
- private static final SchemaTransition FROM_V3_TO_V4 = SchemaTransition.to(new SchemaVersion(4));
- private static final SchemaTransition FROM_V4_TO_V5 = SchemaTransition.to(new SchemaVersion(5));
private static final SchemaTransition FROM_V5_TO_V6 = SchemaTransition.to(new SchemaVersion(6));
private static final SchemaTransition FROM_V6_TO_V7 = SchemaTransition.to(new SchemaVersion(7));
@@ -60,9 +55,6 @@ public class CassandraRoutesModule extends AbstractModule {
routesMultibinder.addBinding().to(CassandraMailboxMergingRoutes.class);
MapBinder<SchemaTransition, Migration> allMigrationClazzBinder = MapBinder.newMapBinder(binder(), SchemaTransition.class, Migration.class);
- allMigrationClazzBinder.addBinding(FROM_V2_TO_V3).toInstance(() -> { });
- allMigrationClazzBinder.addBinding(FROM_V3_TO_V4).to(AttachmentV2Migration.class);
- allMigrationClazzBinder.addBinding(FROM_V4_TO_V5).to(AttachmentMessageIdCreation.class);
allMigrationClazzBinder.addBinding(FROM_V5_TO_V6).to(MailboxPathV2Migration.class);
allMigrationClazzBinder.addBinding(FROM_V6_TO_V7).to(MappingsSourcesMigration.class);
diff --git a/upgrade-instructions.md b/upgrade-instructions.md
index 399451a..4f7c11d 100644
--- a/upgrade-instructions.md
+++ b/upgrade-instructions.md
@@ -33,6 +33,18 @@ Change list:
- [ProtocolSession storng typing](#protocolsession-storng-typing)
- [Tune Cassandra time serie tables options](#tune-cassandra-time-serie-tables-options)
- [Log4J2 Adoption](#log4j2-adoption)
+ - [Drop Cassandra schema version prior version 5](#drop-cassandra-schema-version-prior-version-5)
+
+### Drop Cassandra schema version prior version 5
+
+Date 06/04/2020
+
+JIRA: https://issues.apache.org/jira/browse/JAMES-2997
+
+In an effort to simplify the code base, we dropped support for Cassandra schema version prior version 5.
+
+Installation running older schema version than version 5 needs to rely on release 3.5.0 to upgrade to schema version 7,
+before upgrading to an eventual newer version.
### Log4J2 Adoption
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 06/06: JAMES-3144 Increase ElasticSearch timeout in
RabbitMQReindexingWithEventDeadLettersTest
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 b5ed9c258faff04798c633149a85cc4994ceee3b
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Sun Apr 5 22:23:42 2020 +0700
JAMES-3144 Increase ElasticSearch timeout in RabbitMQReindexingWithEventDeadLettersTest
If not this results in continuous socket timeouts.
Waiting long enough is required.
To be noted the interupted exception occuring before is due to the test
timeout causing a stop while the elastic executor is still blocked.
---
.../rabbitmq/RabbitMQReindexingWithEventDeadLettersTest.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
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 b9a95b8..bf5f7b4 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
@@ -73,11 +73,11 @@ class RabbitMQReindexingWithEventDeadLettersTest {
private static final ConditionFactory CALMLY_AWAIT = Awaitility
.with().pollInterval(ONE_HUNDRED_MILLISECONDS)
.and().pollDelay(ONE_HUNDRED_MILLISECONDS)
- .atMost(Duration.ONE_MINUTE)
+ .atMost(Duration.FIVE_MINUTES)
.await();
private static final DockerElasticSearchExtension dockerElasticSearch =
- new DockerElasticSearchExtension().withRequestTimeout(java.time.Duration.ofSeconds(1));
+ new DockerElasticSearchExtension().withRequestTimeout(java.time.Duration.ofSeconds(5));
private static final JamesServerBuilder.ServerProvider CONFIGURATION_BUILDER = configuration -> GuiceJamesServer
.forConfiguration(configuration)
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 02/06: [Release] Post 3.5.0 release removal
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 32bea1de906dabacc3f344701b5426a6ea34065a
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Mon Apr 6 18:39:10 2020 +0700
[Release] Post 3.5.0 release removal
- To be deprecated in upcoming 3.5.0 and removed in 3.6.0:
- james-server-mailet BayesianAnalysis + BayesianAnalysisFeeder +
JDBCBayesianAnalyser
mailet/ai mailets to be used instead.
- ToRecipientFolder.
WithStorageDirective + LocalDelivery to be used instead.
This will limit user misunderstandings
- mpt/antlib + mpt/mpt-maven-plugin + mpt/app
Not aware of usages, not exposed to users
---
mpt/antlib/LICENSE.txt | 176 --------
mpt/antlib/NOTICE.txt | 11 -
mpt/antlib/pom.xml | 165 --------
.../apache/james/mpt/ant/MailProtocolTestTask.java | 426 -------------------
.../resources/org/apache/james/mpt/ant/antlib.xml | 26 --
mpt/antlib/src/reporting-site/site.xml | 29 --
.../src/site/resources/images/asf-logo-reduced.gif | Bin 6636 -> 0 bytes
.../src/site/resources/images/james-mpt-logo.gif | Bin 5179 -> 0 bytes
mpt/antlib/src/site/site.xml | 37 --
mpt/antlib/src/site/xdoc/index.xml | 183 --------
.../java/org/apache/james/mpt/ant/TestAddUser.java | 68 ---
.../org/apache/james/mpt/ant/TestRunScripts.java | 102 -----
mpt/antlib/src/test/resources/build.xml | 140 -------
mpt/pom.xml | 1 -
.../james/transport/mailets/BayesianAnalysis.java | 460 ---------------------
.../transport/mailets/BayesianAnalysisFeeder.java | 336 ---------------
.../james/transport/mailets/ToRecipientFolder.java | 99 -----
.../mailets/delivery/ToRecipientFolderTest.java | 198 ---------
18 files changed, 2457 deletions(-)
diff --git a/mpt/antlib/LICENSE.txt b/mpt/antlib/LICENSE.txt
deleted file mode 100644
index 2bb9ad2..0000000
--- a/mpt/antlib/LICENSE.txt
+++ /dev/null
@@ -1,176 +0,0 @@
- Apache License
- Version 2.0, January 2004
- http://www.apache.org/licenses/
-
- TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
-
- 1. Definitions.
-
- "License" shall mean the terms and conditions for use, reproduction,
- and distribution as defined by Sections 1 through 9 of this document.
-
- "Licensor" shall mean the copyright owner or entity authorized by
- the copyright owner that is granting the License.
-
- "Legal Entity" shall mean the union of the acting entity and all
- other entities that control, are controlled by, or are under common
- control with that entity. For the purposes of this definition,
- "control" means (i) the power, direct or indirect, to cause the
- direction or management of such entity, whether by contract or
- otherwise, or (ii) ownership of fifty percent (50%) or more of the
- outstanding shares, or (iii) beneficial ownership of such entity.
-
- "You" (or "Your") shall mean an individual or Legal Entity
- exercising permissions granted by this License.
-
- "Source" form shall mean the preferred form for making modifications,
- including but not limited to software source code, documentation
- source, and configuration files.
-
- "Object" form shall mean any form resulting from mechanical
- transformation or translation of a Source form, including but
- not limited to compiled object code, generated documentation,
- and conversions to other media types.
-
- "Work" shall mean the work of authorship, whether in Source or
- Object form, made available under the License, as indicated by a
- copyright notice that is included in or attached to the work
- (an example is provided in the Appendix below).
-
- "Derivative Works" shall mean any work, whether in Source or Object
- form, that is based on (or derived from) the Work and for which the
- editorial revisions, annotations, elaborations, or other modifications
- represent, as a whole, an original work of authorship. For the purposes
- of this License, Derivative Works shall not include works that remain
- separable from, or merely link (or bind by name) to the interfaces of,
- the Work and Derivative Works thereof.
-
- "Contribution" shall mean any work of authorship, including
- the original version of the Work and any modifications or additions
- to that Work or Derivative Works thereof, that is intentionally
- submitted to Licensor for inclusion in the Work by the copyright owner
- or by an individual or Legal Entity authorized to submit on behalf of
- the copyright owner. For the purposes of this definition, "submitted"
- means any form of electronic, verbal, or written communication sent
- to the Licensor or its representatives, including but not limited to
- communication on electronic mailing lists, source code control systems,
- and issue tracking systems that are managed by, or on behalf of, the
- Licensor for the purpose of discussing and improving the Work, but
- excluding communication that is conspicuously marked or otherwise
- designated in writing by the copyright owner as "Not a Contribution."
-
- "Contributor" shall mean Licensor and any individual or Legal Entity
- on behalf of whom a Contribution has been received by Licensor and
- subsequently incorporated within the Work.
-
- 2. Grant of Copyright License. Subject to the terms and conditions of
- this License, each Contributor hereby grants to You a perpetual,
- worldwide, non-exclusive, no-charge, royalty-free, irrevocable
- copyright license to reproduce, prepare Derivative Works of,
- publicly display, publicly perform, sublicense, and distribute the
- Work and such Derivative Works in Source or Object form.
-
- 3. Grant of Patent License. Subject to the terms and conditions of
- this License, each Contributor hereby grants to You a perpetual,
- worldwide, non-exclusive, no-charge, royalty-free, irrevocable
- (except as stated in this section) patent license to make, have made,
- use, offer to sell, sell, import, and otherwise transfer the Work,
- where such license applies only to those patent claims licensable
- by such Contributor that are necessarily infringed by their
- Contribution(s) alone or by combination of their Contribution(s)
- with the Work to which such Contribution(s) was submitted. If You
- institute patent litigation against any entity (including a
- cross-claim or counterclaim in a lawsuit) alleging that the Work
- or a Contribution incorporated within the Work constitutes direct
- or contributory patent infringement, then any patent licenses
- granted to You under this License for that Work shall terminate
- as of the date such litigation is filed.
-
- 4. Redistribution. You may reproduce and distribute copies of the
- Work or Derivative Works thereof in any medium, with or without
- modifications, and in Source or Object form, provided that You
- meet the following conditions:
-
- (a) You must give any other recipients of the Work or
- Derivative Works a copy of this License; and
-
- (b) You must cause any modified files to carry prominent notices
- stating that You changed the files; and
-
- (c) You must retain, in the Source form of any Derivative Works
- that You distribute, all copyright, patent, trademark, and
- attribution notices from the Source form of the Work,
- excluding those notices that do not pertain to any part of
- the Derivative Works; and
-
- (d) If the Work includes a "NOTICE" text file as part of its
- distribution, then any Derivative Works that You distribute must
- include a readable copy of the attribution notices contained
- within such NOTICE file, excluding those notices that do not
- pertain to any part of the Derivative Works, in at least one
- of the following places: within a NOTICE text file distributed
- as part of the Derivative Works; within the Source form or
- documentation, if provided along with the Derivative Works; or,
- within a display generated by the Derivative Works, if and
- wherever such third-party notices normally appear. The contents
- of the NOTICE file are for informational purposes only and
- do not modify the License. You may add Your own attribution
- notices within Derivative Works that You distribute, alongside
- or as an addendum to the NOTICE text from the Work, provided
- that such additional attribution notices cannot be construed
- as modifying the License.
-
- You may add Your own copyright statement to Your modifications and
- may provide additional or different license terms and conditions
- for use, reproduction, or distribution of Your modifications, or
- for any such Derivative Works as a whole, provided Your use,
- reproduction, and distribution of the Work otherwise complies with
- the conditions stated in this License.
-
- 5. Submission of Contributions. Unless You explicitly state otherwise,
- any Contribution intentionally submitted for inclusion in the Work
- by You to the Licensor shall be under the terms and conditions of
- this License, without any additional terms or conditions.
- Notwithstanding the above, nothing herein shall supersede or modify
- the terms of any separate license agreement you may have executed
- with Licensor regarding such Contributions.
-
- 6. Trademarks. This License does not grant permission to use the trade
- names, trademarks, service marks, or product names of the Licensor,
- except as required for reasonable and customary use in describing the
- origin of the Work and reproducing the content of the NOTICE file.
-
- 7. Disclaimer of Warranty. Unless required by applicable law or
- agreed to in writing, Licensor provides the Work (and each
- Contributor provides its Contributions) on an "AS IS" BASIS,
- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
- implied, including, without limitation, any warranties or conditions
- of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
- PARTICULAR PURPOSE. You are solely responsible for determining the
- appropriateness of using or redistributing the Work and assume any
- risks associated with Your exercise of permissions under this License.
-
- 8. Limitation of Liability. In no event and under no legal theory,
- whether in tort (including negligence), contract, or otherwise,
- unless required by applicable law (such as deliberate and grossly
- negligent acts) or agreed to in writing, shall any Contributor be
- liable to You for damages, including any direct, indirect, special,
- incidental, or consequential damages of any character arising as a
- result of this License or out of the use or inability to use the
- Work (including but not limited to damages for loss of goodwill,
- work stoppage, computer failure or malfunction, or any and all
- other commercial damages or losses), even if such Contributor
- has been advised of the possibility of such damages.
-
- 9. Accepting Warranty or Additional Liability. While redistributing
- the Work or Derivative Works thereof, You may choose to offer,
- and charge a fee for, acceptance of support, warranty, indemnity,
- or other liability obligations and/or rights consistent with this
- License. However, in accepting such obligations, You may act only
- on Your own behalf and on Your sole responsibility, not on behalf
- of any other Contributor, and only if You agree to indemnify,
- defend, and hold each Contributor harmless for any liability
- incurred by, or claims asserted against, such Contributor by reason
- of your accepting any such warranty or additional liability.
-
- END OF TERMS AND CONDITIONS
\ No newline at end of file
diff --git a/mpt/antlib/NOTICE.txt b/mpt/antlib/NOTICE.txt
deleted file mode 100644
index 198c869..0000000
--- a/mpt/antlib/NOTICE.txt
+++ /dev/null
@@ -1,11 +0,0 @@
-
-=========================================================================
-== NOTICE file for use with the Apache License, Version 2.0, ==
-=========================================================================
-
-Apache JAMES
-Copyright 2007-2011 The Apache Software Foundation
-
-This product includes software developed at
-The Apache Software Foundation (http://www.apache.org/).
-
diff --git a/mpt/antlib/pom.xml b/mpt/antlib/pom.xml
deleted file mode 100644
index 40231a1..0000000
--- a/mpt/antlib/pom.xml
+++ /dev/null
@@ -1,165 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
- 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.
--->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
- <modelVersion>4.0.0</modelVersion>
-
- <parent>
- <groupId>org.apache.james</groupId>
- <artifactId>apache-james-mpt</artifactId>
- <version>3.6.0-SNAPSHOT</version>
- <relativePath>../pom.xml</relativePath>
- </parent>
-
- <artifactId>apache-james-mpt-antlib</artifactId>
-
- <name>Apache James MPT AntLib</name>
- <description>Apache James Mail Protocol Tester (MPT) is a library providing a framework for the
-scritable functional testing of ASCII based line protocols. This AntLib provides an easy interface
-to the library requiring no extra coding.</description>
- <url>http://james.apache.org/mpt/antlib</url>
-
- <dependencies>
- <dependency>
- <groupId>${james.groupId}</groupId>
- <artifactId>apache-james-mpt-core</artifactId>
- </dependency>
- <dependency>
- <groupId>${james.groupId}</groupId>
- <artifactId>apache-james-mpt-core</artifactId>
- <type>test-jar</type>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>${james.groupId}</groupId>
- <artifactId>testing-base</artifactId>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.ant</groupId>
- <artifactId>ant</artifactId>
- </dependency>
- <dependency>
- <groupId>org.apache.ant</groupId>
- <artifactId>ant-antunit</artifactId>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>org.slf4j</groupId>
- <artifactId>jcl-over-slf4j</artifactId>
- </dependency>
- </dependencies>
-
- <build>
- <plugins>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-antrun-plugin</artifactId>
- <dependencies>
- <!-- AntUnit require Ant 1.7+ -->
- <dependency>
- <groupId>org.apache.ant</groupId>
- <artifactId>ant</artifactId>
- <version>1.7.1</version>
- </dependency>
- </dependencies>
- <executions>
- <execution>
- <goals>
- <goal>run</goal>
- </goals>
- <phase>test</phase>
- <configuration>
- <tasks>
- <ant antfile="src/test/resources/build.xml" inheritAll="false" inheritRefs="false">
- <property location="${project.build.outputDirectory}" name="jar.name" />
- <property location="${basedir}" name="base.dir" />
- <property name="test.skipped" value="${maven.test.skip}" />
- <property name="test.classpath" refid="maven.test.classpath" />
- </ant>
- </tasks>
- </configuration>
- </execution>
- </executions>
- </plugin>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-jar-plugin</artifactId>
- <configuration>
- <archive>
- <manifestFile>${project.build.outputDirectory}/META-INF/MANIFEST.MF</manifestFile>
- <manifestEntries>
- <Specification-Title>Apache James MPT AntLib</Specification-Title>
- <Specification-Version>${project.version}</Specification-Version>
- <Specification-Vendor>The Apache Software Foundation</Specification-Vendor>
- <Implementation-Title>Apache James MPT AntLib</Implementation-Title>
- <Implementation-Version>${project.version}</Implementation-Version>
- <Implementation-Vendor>The Apache Software Foundation</Implementation-Vendor>
- <Implementation-Vendor-Id>org.apache</Implementation-Vendor-Id>
- <url>${project.url}</url>
- </manifestEntries>
- </archive>
- </configuration>
- <executions>
- <execution>
- <goals>
- <goal>test-jar</goal>
- </goals>
- </execution>
- </executions>
- </plugin>
-
- <plugin>
- <groupId>org.apache.felix</groupId>
- <artifactId>maven-bundle-plugin</artifactId>
- <version>${felix.plugin.version}</version>
- <extensions>true</extensions>
- <configuration>
- <instructions>
- <Export-Package>org.apache.james.mpt.ant</Export-Package>
- <Embed-Dependency>*;scope=runtime</Embed-Dependency>
- </instructions>
- </configuration>
- <executions>
- <execution>
- <id>bundle-manifest</id>
- <goals>
- <goal>manifest</goal>
- </goals>
- <phase>process-classes</phase>
- </execution>
- </executions>
- </plugin>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-javadoc-plugin</artifactId>
- <executions>
- <execution>
- <id>create-javadocs</id>
- <goals>
- <goal>javadoc</goal>
- <goal>jar</goal>
- </goals>
- <phase>package</phase>
- </execution>
- </executions>
- </plugin>
- </plugins>
- </build>
-</project>
diff --git a/mpt/antlib/src/main/java/org/apache/james/mpt/ant/MailProtocolTestTask.java b/mpt/antlib/src/main/java/org/apache/james/mpt/ant/MailProtocolTestTask.java
deleted file mode 100644
index 24c8570..0000000
--- a/mpt/antlib/src/main/java/org/apache/james/mpt/ant/MailProtocolTestTask.java
+++ /dev/null
@@ -1,426 +0,0 @@
-/****************************************************************
- * 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.mpt.ant;
-
-import java.io.File;
-import java.io.FileNotFoundException;
-import java.io.FileReader;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.Reader;
-import java.io.StringReader;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Optional;
-
-import org.apache.james.core.Username;
-import org.apache.james.mpt.Runner;
-import org.apache.james.mpt.api.ImapFeatures;
-import org.apache.james.mpt.api.ImapFeatures.Feature;
-import org.apache.james.mpt.api.Monitor;
-import org.apache.james.mpt.host.ExternalHostSystem;
-import org.apache.james.mpt.protocol.ProtocolSessionBuilder;
-import org.apache.james.mpt.user.ScriptedUserAdder;
-import org.apache.james.util.Port;
-import org.apache.tools.ant.BuildException;
-import org.apache.tools.ant.Project;
-import org.apache.tools.ant.Task;
-import org.apache.tools.ant.types.Resource;
-import org.apache.tools.ant.types.ResourceCollection;
-import org.apache.tools.ant.types.resources.FileResource;
-import org.apache.tools.ant.types.resources.Union;
-
-/**
- * Task executes MPT scripts against a server
- * running on a given port and host.
- */
-public class MailProtocolTestTask extends Task implements Monitor {
-
- private static final ImapFeatures SUPPORTED_FEATURES = ImapFeatures.of(Feature.NAMESPACE_SUPPORT);
-
- private boolean quiet = false;
- private File script;
- private Union scripts;
- private Optional<Port> port = Optional.empty();
- private String host = "127.0.0.1";
- private boolean skip = false;
- private String shabang = null;
- private final Collection<AddUser> users = new ArrayList<>();
- private String errorProperty;
-
- /**
- * Gets the error property.
- *
- * @return name of the ant property to be set on error,
- * null if the script should terminate on error
- */
- public String getErrorProperty() {
- return errorProperty;
- }
-
- /**
- * Sets the error property.
- * @param errorProperty name of the ant property to be set on error,
- * nul if the script should terminate on error
- */
- public void setErrorProperty(String errorProperty) {
- this.errorProperty = errorProperty;
- }
-
- /**
- * Should progress output be suppressed?
- * @return true if progress information should be suppressed,
- * false otherwise
- */
- public boolean isQuiet() {
- return quiet;
- }
-
- /**
- * Sets whether progress output should be suppressed/
- * @param quiet true if progress information should be suppressed,
- * false otherwise
- */
- public void setQuiet(boolean quiet) {
- this.quiet = quiet;
- }
-
- /**
- * Should the execution be skipped?
- * @return true if exection should be skipped,
- * otherwise false
- */
- public boolean isSkip() {
- return skip;
- }
-
- /**
- * Sets execution skipping.
- * @param skip true to skip excution
- */
- public void setSkip(boolean skip) {
- this.skip = skip;
- }
-
- /**
- * Gets the host (either name or number) against which this
- * test will run.
- * @return host, not null
- */
- public String getHost() {
- return host;
- }
-
- /**
- * Sets the host (either name or number) against which this
- * test will run.
- * @param host not null
- */
- public void setHost(String host) {
- this.host = host;
- }
-
- /**
- * Gets the port against which this test will run.
- * @return port number
- */
- public int getPort() {
- return port
- .map(Port::getValue)
- .orElseThrow(() -> new RuntimeException("Port must be set"));
- }
-
- /**
- * Sets the port aginst which this test will run.
- * @param port port number
- */
- public void setPort(int port) {
- this.port = Optional.of(new Port(port));
- }
-
- /**
- * Gets the script to execute.
- * @return file containing test script
- */
- public File getScript() {
- return script;
- }
-
- /**
- * Sets the script to execute.
- * @param script not null
- */
- public void setScript(File script) {
- this.script = script;
- }
-
- /**
- * Gets script shabang.
- * This will be substituted for the first server response.
- * @return script shabang,
- * or null for no shabang
- */
- public String getShabang() {
- return shabang;
- }
-
- /**
- * Sets the script shabang.
- * When not null, this value will be used to be substituted for the
- * first server response.
- * @param shabang script shabang,
- * or null for no shabang.
- */
- public void setShabang(String shabang) {
- this.shabang = shabang;
- }
-
- @Override
- public void execute() throws BuildException {
- if (! port.isPresent()) {
- throw new BuildException("Port must be set");
- }
-
- if (scripts == null && script == null) {
- throw new BuildException("Scripts must be specified as an embedded resource collection");
- }
-
- if (scripts != null && script != null) {
- throw new BuildException("Scripts can be specified either by the script attribute or as resource collections but not both.");
- }
-
- for (AddUser user: users) {
- user.validate();
- }
-
- if (skip) {
- log("Skipping excution");
- } else if (errorProperty == null) {
- doExecute();
- } else {
- try {
- doExecute();
- } catch (BuildException e) {
- final Project project = getProject();
- project.setProperty(errorProperty, e.getMessage());
- log(e, Project.MSG_DEBUG);
- }
- }
- }
-
- public void add(ResourceCollection resources) {
- if (scripts == null) {
- scripts = new Union();
- }
- scripts.add(resources);
- }
-
- private void doExecute() throws BuildException {
- for (AddUser userAdder: users) {
- userAdder.execute();
- }
-
- final ExternalHostSystem host = new ExternalHostSystem(SUPPORTED_FEATURES, getHost(), port.get(), this, getShabang(), null);
- final ProtocolSessionBuilder builder = new ProtocolSessionBuilder();
-
- if (scripts == null) {
- scripts = new Union();
- scripts.add(new FileResource(script));
- }
-
- for (final Resource resource : scripts) {
- try {
- final Runner runner = new Runner();
-
- try {
-
- final InputStream inputStream = resource.getInputStream();
- final String name = resource.getName();
- builder.addProtocolLines(name == null ? "[Unknown]" : name, inputStream, runner.getTestElements());
- runner.runSessions(host);
-
- } catch (UnsupportedOperationException e) {
- log("Resource cannot be read: " + resource.getName(), Project.MSG_WARN);
- }
- } catch (IOException e) {
- throw new BuildException("Cannot load script " + resource.getName(), e);
- } catch (Exception e) {
- log(e.getMessage(), Project.MSG_ERR);
- throw new BuildException("[FAILURE] in script " + resource.getName() + "\n" + e.getMessage(), e);
- }
-
- }
-
- }
-
- public AddUser createAddUser() {
- final AddUser result = new AddUser();
- users.add(result);
- return result;
- }
-
- /**
- * Adds a user.
- */
- public class AddUser {
-
- private Port port;
- private String user;
- private String passwd;
- private File script;
- private String scriptText;
-
- /**
- * Gets the port against which the user addition
- * script should be executed.
- * @return port number
- */
- public int getPort() {
- return port.getValue();
- }
-
- /**
- * Sets the port against which the user addition
- * script should be executed.
- * @param port port number
- */
- public void setPort(int port) {
- this.port = new Port(port);
- }
-
- /**
- * Gets the password for the user.
- * @return password not null
- */
- public String getPasswd() {
- return passwd;
- }
-
- /**
- * Sets the password for the user.
- * This will be passed in the user creation script.
- * @param passwd not null
- */
- public void setPasswd(String passwd) {
- this.passwd = passwd;
- }
-
- /**
- * Gets the name of the user to be created.
- * @return user name, not null
- */
- public String getUser() {
- return user;
- }
-
- /**
- * Sets the name of the user to be created.
- * @param user not null
- */
- public void setUser(String user) {
- this.user = user;
- }
-
- /**
- * Sets user addition script.
- * @param scriptText not null
- */
- public void addText(String scriptText) {
- this.scriptText = getProject().replaceProperties(scriptText);
- }
-
- /**
- * Gets the file containing the user creation script.
- * @return not null
- */
- public File getScript() {
- return script;
- }
-
- /**
- * Sets the file containing the user creation script.
- * @param script not null
- */
- public void setScript(File script) {
- this.script = script;
- }
-
- /**
- * Validates mandatory fields have been filled.
- */
- void validate() throws BuildException {
- if (script == null && scriptText == null) {
- throw new BuildException("Either the 'script' attribute must be set, or the body must contain the text of the script");
- }
-
- if (script != null && scriptText != null) {
- throw new BuildException("Choose either script text or script attribute but not both.");
- }
-
- if (port == null) {
- throw new BuildException("'port' attribute must be set on AddUser to the port against which the script should run.");
- }
- }
-
- /**
- * Creates a user.
- * @throws BuildException
- */
- void execute() throws BuildException {
- validate();
- try {
- final File scriptFile = getScript();
- final ScriptedUserAdder adder = new ScriptedUserAdder(getHost(), port, MailProtocolTestTask.this);
- try (Reader reader = newReader(scriptFile)) {
- adder.addUser(Username.of(getUser()), getPasswd(), reader);
- }
- } catch (Exception e) {
- log(e.getMessage(), Project.MSG_ERR);
- throw new BuildException("User addition failed: \n" + e.getMessage(), e);
- }
- }
-
- private Reader newReader(File scriptFile) throws FileNotFoundException {
- if (scriptFile == null) {
- return new StringReader(scriptText);
- }
- return new FileReader(scriptFile);
- }
- }
-
- @Override
- public void note(String message) {
- if (quiet) {
- log(message, Project.MSG_DEBUG);
- } else {
- log(message, Project.MSG_INFO);
- }
- }
-
- @Override
- public void debug(char character) {
- log("'" + character + "'", Project.MSG_DEBUG);
- }
-
- @Override
- public void debug(String message) {
- log(message, Project.MSG_DEBUG);
- }
-}
diff --git a/mpt/antlib/src/main/resources/org/apache/james/mpt/ant/antlib.xml b/mpt/antlib/src/main/resources/org/apache/james/mpt/ant/antlib.xml
deleted file mode 100644
index eed1393..0000000
--- a/mpt/antlib/src/main/resources/org/apache/james/mpt/ant/antlib.xml
+++ /dev/null
@@ -1,26 +0,0 @@
-<?xml version='1.0'?>
-<!--
-
- 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. *
-
--->
-
-<antlib>
- <typedef name="mpt"
- classname="org.apache.james.mpt.ant.MailProtocolTestTask"/>
-</antlib>
\ No newline at end of file
diff --git a/mpt/antlib/src/reporting-site/site.xml b/mpt/antlib/src/reporting-site/site.xml
deleted file mode 100644
index d919164..0000000
--- a/mpt/antlib/src/reporting-site/site.xml
+++ /dev/null
@@ -1,29 +0,0 @@
-<?xml version="1.0" encoding="ISO-8859-1"?>
-<!--
- 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.
--->
-<project name="${project.name}">
-
- <body>
-
- <menu ref="parent" />
- <menu ref="reports" />
-
- </body>
-
-</project>
diff --git a/mpt/antlib/src/site/resources/images/asf-logo-reduced.gif b/mpt/antlib/src/site/resources/images/asf-logo-reduced.gif
deleted file mode 100644
index 93cc102..0000000
Binary files a/mpt/antlib/src/site/resources/images/asf-logo-reduced.gif and /dev/null differ
diff --git a/mpt/antlib/src/site/resources/images/james-mpt-logo.gif b/mpt/antlib/src/site/resources/images/james-mpt-logo.gif
deleted file mode 100644
index 9a75b4f..0000000
Binary files a/mpt/antlib/src/site/resources/images/james-mpt-logo.gif and /dev/null differ
diff --git a/mpt/antlib/src/site/site.xml b/mpt/antlib/src/site/site.xml
deleted file mode 100644
index 5437b74..0000000
--- a/mpt/antlib/src/site/site.xml
+++ /dev/null
@@ -1,37 +0,0 @@
-<?xml version="1.0" encoding="ISO-8859-1"?>
-<!--
- 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.
--->
-<project name="JAMES Mail Protocol Tester (MPT)">
-
- <body>
-
- <menu ref="parent" />
-
- <menu name="MPT AntLib">
- <item name="Overview" href="index.html"/>
- <item
- name="DOAP"
- href="doap_apache-james-mpt-antlib.rdf"
- img='http://www.w3.org/RDF/icons/rdf_metadata_button.32'/>
- </menu>
-
- <menu ref="reports" />
-
- </body>
-</project>
diff --git a/mpt/antlib/src/site/xdoc/index.xml b/mpt/antlib/src/site/xdoc/index.xml
deleted file mode 100644
index d95095b..0000000
--- a/mpt/antlib/src/site/xdoc/index.xml
+++ /dev/null
@@ -1,183 +0,0 @@
-<?xml version="1.0"?>
-<!--
- 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.
--->
-
-<document>
-
- <properties>
- <title>Overview</title>
- <author email="server-dev@james.apache.org">Mail Protocol Tester AntLib</author>
- </properties>
-
-<body>
-<section name="What is the Apache James MPT?">
-<p>
-Apache James Mail Protocol Tester (MPT) is a library providing a framework for the
-scritable functional testing of ASCII based line protocols.</p>
-<p>
-An antlib defines a suite of extensions for <a href='http://ant.apache.org'>Apache Ant</a>.
-This antlib defines a suite of functional testing tools particularly suitable for
-use with ASCII based line protocols (as are common in mail).
-</p>
-</section>
-<section name='Usage'>
-<p>
-See:
-</p>
-<ul>
-<li>Apache Ant <a href='http://ant.apache.org/manual/CoreTypes/antlib.html'>manual</a></li>
-<li><a href='http://www.onjava.com/pub/a/onjava/2006/08/09/ant-1-7-using-antlibs.html'>Using AntLibs</a> (article)</li>
-<li><a href='http://www.oracle.com/technology/pub/articles/bodewig_taskwriters.html'>Ant 1.6 for Task Writers</a> (article)</li>
-</ul>
-</section>
-<section name='Tasks'>
-<subsection name='mpt'>
-<p>
-Executes <a href='../main'>MPT</a> scripts against a running independent server.
-If these scripts require users then these can be added by script by using the <code>AddUser</code>
-element.
-</p>
-<subsection name='Parameters'>
-<table>
-<tr>
- <td><strong>Attribute</strong></td> <td><strong>Description</strong></td> <td><strong>Required</strong></td>
-</tr>
-<tr>
- <td>port</td>
- <td>Port number that the scripts should be executed against</td>
- <td>Yes</td>
-</tr>
-<tr>
- <td>host</td>
- <td>Host name or IP address of the server against which the scripts should be executed</td>
- <td>No (Defaults to <code>127.0.0.1</code>)</td>
-</tr>
-<tr>
- <td>script</td>
- <td>File containing a script to be executed. To execute multiple scripts,
- use a <a href='#Any Resource Collection'>Resource Collection</a></td>
- <td>When no resource collections elements exist</td>
-</tr>
-<tr>
- <td>skip</td>
- <td>Skips execution of scripts. For simple conditional execution.</td>
- <td>No</td>
-</tr>
-<tr>
- <td>shabang</td>
- <td>Optional server shabang will be substituted for first server response.</td>
- <td>No</td>
-</tr>
-<tr>
- <td>quiet</td>
- <td>Optional boolean setting when true will suppress progress output.</td>
- <td>No (defaults to false)</td>
-</tr>
-<tr>
- <td>errorproperty</td>
- <td>Optionally names a property which will be set (with error message) when test fails.
- When unset, the build will terminate when test fails.</td>
- <td>No</td>
-</tr>
-</table>
-</subsection>
-
-<subsection name='Nested Elements'>
-<subsection name='Any Resource Collection'>
-<p>
-<a href='http://ant.apache.org/manual/CoreTypes/resources.html#collection'>Resource Collection</a>s
-are used to specify a group of scripts to execute. When resource collections are set, attribute <code>
-</code>
-</p>
-</subsection>
-<subsection name='AddUser'>
-<p>
-Allows scripted addition of any users required by the tests.
-The port can vary but the host is inherited.
-</p>
-<table>
-<tr>
- <td><strong>Attribute</strong></td> <td><strong>Description</strong></td> <td><strong>Required</strong></td>
-</tr>
-<tr>
- <td>port</td>
- <td>Port number that the user addition script should be executed against</td>
- <td>Yes</td>
-</tr>
-
-<tr>
- <td>passwd</td>
- <td>Password to be set for user created. Will be passed to the script.</td>
- <td>No</td>
-</tr>
-
-<tr>
- <td>user</td>
- <td>Name of the user to be created. Will be passed to the script.</td>
- <td>No</td>
-</tr>
-
-<tr>
- <td>script</td>
- <td>File containing the user creation script</td>
- <td>Either this attribute must be set, or the user addition script </td>
-</tr>
-</table>
-<p>
-User addition scripts are typically short. As an alternative to the script attribute
-(which specifies the file containing the script), the script text can be added directly
-as text to the body of this element.
-</p>
-</subsection>
-</subsection>
-<subsection name='Examples'>
-<source>
-<![CDATA[
-<project
- name="tests"
- default="main"
- xmlns:mpt="antlib:org.apache.james.mpt.ant">
-
- <!-- Connect to port 10000 on localhost and run all scripts in src/test/resources/ -->
- <target name='ConnectToServerPort10000'>
- <mpt:mpt port='10000'>
- <fileset dir='src/test/resources/>
- <include name='**/*.mpt'/>
- </fileset>
- </mpt:mpt>
- </target>
-
- <!--
- Connects to port 4001 on example.org then runs user addition script
- Then connects to port 143 on example.org and runs test script 'test.mpt'
- -->
- <target name='ConnectToExampleDotOrgPort143'>
- <mpt:mpt port='143' host='example.org' script='test.mpt'>
- <!-- Add users by running 'Add User "${user}" "${passwd}"' -->
- <addUser port='4001' user='user' passwd='passwd'>C: Add User "${user}" "${passwd}"</addUser>
- </mpt:mpt>
- </target>
-</project>
-]]>
-</source>
-</subsection>
-</subsection>
-</section>
-</body>
-</document>
\ No newline at end of file
diff --git a/mpt/antlib/src/test/java/org/apache/james/mpt/ant/TestAddUser.java b/mpt/antlib/src/test/java/org/apache/james/mpt/ant/TestAddUser.java
deleted file mode 100644
index 2b1ec72..0000000
--- a/mpt/antlib/src/test/java/org/apache/james/mpt/ant/TestAddUser.java
+++ /dev/null
@@ -1,68 +0,0 @@
-/****************************************************************
- * 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.mpt.ant;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
-import org.apache.james.mpt.DiscardProtocol;
-import org.apache.tools.ant.Project;
-import org.apache.tools.ant.types.resources.Union;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
-
-public class TestAddUser {
-
- DiscardProtocol fakeServer;
-
- DiscardProtocol.Record record;
-
- MailProtocolTestTask subject;
-
- @Before
- public void setUp() throws Exception {
- fakeServer = new DiscardProtocol();
- fakeServer.start();
- record = fakeServer.recordNext();
-
- subject = new MailProtocolTestTask();
- subject.setHost("127.0.0.1");
- subject.setPort(fakeServer.getPort().getValue());
- subject.add(new Union());
- subject.setProject(new Project());
- }
-
- @After
- public void tearDown() throws Exception {
- fakeServer.stop();
- }
-
- @Test
- public void testShouldExecuteScriptAgainstPort() throws Exception {
- MailProtocolTestTask.AddUser user = subject.createAddUser();
- user.setPort(fakeServer.getPort().getValue());
- user.setPasswd("PASSWORD");
- user.setUser("USER");
- final String script = "This script adds a user";
- user.addText("C: " + script);
- subject.execute();
- assertThat(record.complete()).isEqualTo(script + "\r\n");
- }
-}
diff --git a/mpt/antlib/src/test/java/org/apache/james/mpt/ant/TestRunScripts.java b/mpt/antlib/src/test/java/org/apache/james/mpt/ant/TestRunScripts.java
deleted file mode 100644
index 04aef79..0000000
--- a/mpt/antlib/src/test/java/org/apache/james/mpt/ant/TestRunScripts.java
+++ /dev/null
@@ -1,102 +0,0 @@
-/****************************************************************
- * 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.mpt.ant;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
-import java.io.File;
-import java.io.FileWriter;
-import java.io.InputStream;
-
-import org.apache.james.mpt.DiscardProtocol;
-import org.apache.tools.ant.Project;
-import org.apache.tools.ant.types.Resource;
-import org.apache.tools.ant.types.resources.StringResource;
-import org.apache.tools.ant.types.resources.Union;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
-
-public class TestRunScripts {
-
- private static final String SCRIPT = "A script";
-
- Union stubResourceCollection;
- Resource stubResource;
-
- DiscardProtocol fakeServer;
- DiscardProtocol.Record record;
-
- MailProtocolTestTask subject;
-
- @Before
- public void setUp() throws Exception {
- fakeServer = new DiscardProtocol();
- fakeServer.start();
- record = fakeServer.recordNext();
-
- stubResourceCollection = new Union();
- stubResource = new StringResource("C: " + SCRIPT);
- stubResourceCollection.add(stubResource);
-
- subject = new MailProtocolTestTask();
- subject.setHost("127.0.0.1");
- subject.setPort(fakeServer.getPort().getValue());
- subject.setProject(new Project());
- }
-
- @After
- public void tearDown() throws Exception {
- fakeServer.stop();
- }
-
- @Test
- public void testIgnoreUnsupportedResource() throws Exception {
- final Resource unsupportedResource = new StringResource() {
- @Override
- public InputStream getInputStream() {
- throw new UnsupportedOperationException();
- }
- };
- stubResourceCollection.add(unsupportedResource);
- subject.add(stubResourceCollection);
- subject.execute();
- assertThat(record.complete()).isEqualTo(SCRIPT + "\r\n");
- }
-
- @Test
- public void testRunOneScriptFromCollection() throws Exception {
- subject.add(stubResourceCollection);
- subject.execute();
- assertThat(record.complete()).isEqualTo(SCRIPT + "\r\n");
- }
-
- @Test
- public void testRunOneScriptFromAttribute() throws Exception {
- final File file = File.createTempFile("Test", "mpt");
- file.deleteOnExit();
- final FileWriter writer = new FileWriter(file);
- writer.write("C: " + SCRIPT);
- writer.close();
- subject.setScript(file);
- subject.execute();
- assertThat(record.complete()).isEqualTo(SCRIPT + "\r\n");
- }
-}
diff --git a/mpt/antlib/src/test/resources/build.xml b/mpt/antlib/src/test/resources/build.xml
deleted file mode 100644
index f3946a7..0000000
--- a/mpt/antlib/src/test/resources/build.xml
+++ /dev/null
@@ -1,140 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
- 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.
--->
-<project name="tests" default="all" xmlns:au="antlib:org.apache.ant.antunit" xmlns:mpt="antlib:org.apache.james.mpt.ant">
-
- <taskdef uri="antlib:org.apache.ant.antunit" resource="org/apache/ant/antunit/antlib.xml" classpath="${test.classpath}" />
-
- <taskdef uri="antlib:org.apache.james.mpt.ant" resource="org/apache/james/mpt/ant/antlib.xml">
- <classpath>
- <pathelement location="${jar.name}" />
- <pathelement path='${test.classpath}' />
- </classpath>
- </taskdef>
-
- <target name='testMustNotAllowScriptsAndScript'>
- <au:expectfailure>
- <mpt:mpt skip='true' port='10000' script='pom.xml'>
- <fileset dir='${basedir}' />
- </mpt:mpt>
- </au:expectfailure>
- </target>
-
- <target name='testMustSetScriptsOrScript'>
- <au:expectfailure>
- <mpt:mpt skip='true' port='10000' />
- </au:expectfailure>
- <mpt:mpt port='10000' skip='true'>
- <fileset dir='${basedir}' />
- </mpt:mpt>
- <mpt:mpt port='10000' skip='true' script='pom.xml' />
- </target>
-
- <target name='testMustSetPort'>
- <au:expectfailure>
- <mpt:mpt skip='true'>
- <fileset dir='${basedir}' />
- </mpt:mpt>
- </au:expectfailure>
- <mpt:mpt port='10000' skip='true'>
- <fileset dir='${basedir}' />
- </mpt:mpt>
- </target>
-
- <target name='testMaySetHost'>
- <mpt:mpt port='10000' skip='true' host='example.org'>
- <fileset dir='${basedir}' />
- </mpt:mpt>
- <mpt:mpt port='10000' skip='true' host='10.0.0.66'>
- <fileset dir='${basedir}' />
- </mpt:mpt>
- </target>
-
- <target name='testMaySetShabang'>
- <mpt:mpt port='10000' skip='true' shabang='Whatever'>
- <fileset dir='${basedir}' />
- </mpt:mpt>
- </target>
-
- <target name='testMaySetQuiet'>
- <mpt:mpt port='10000' skip='true' quiet='true'>
- <fileset dir='${basedir}' />
- </mpt:mpt>
- </target>
-
- <target name='testMaySetErrorProperty'>
- <mpt:mpt port='10000' skip='true' errorproperty='mpt.failed'>
- <fileset dir='${basedir}' />
- </mpt:mpt>
- </target>
-
- <target name='testShouldAllowAddUserToBeSet'>
- <mpt:mpt port='10000' skip='true' host='example.org'>
- <fileset dir='${basedir}' />
- <addUser port='10001' user='user' passwd='passwd' script='pom.xml' />
- </mpt:mpt>
- </target>
-
- <target name='testAddUserMaySetNamePassword'>
- <mpt:mpt port='10000' skip='true' host='example.org'>
- <fileset dir='${basedir}' />
- <addUser port='10001' user='user' passwd='passwd' script='pom.xml'/>
- </mpt:mpt>
- <mpt:mpt port='10000' skip='true' host='example.org'>
- <fileset dir='${basedir}' />
- <addUser port='10001' script='pom.xml'/>
- </mpt:mpt>
- </target>
-
- <target name='testAddUserMustSetPortScript'>
- <au:expectfailure>
- <mpt:mpt port='10000' skip='true' host='example.org'>
- <fileset dir='${basedir}' />
- <addUser script='pom.xml' />
- </mpt:mpt>
- </au:expectfailure>
- <au:expectfailure>
- <mpt:mpt port='10000' skip='true' host='example.org'>
- <fileset dir='${basedir}' />
- <addUser port='10001'/>
- </mpt:mpt>
- </au:expectfailure>
- </target>
-
- <target name='testAddUserMayIncludeScriptText'>
- <mpt:mpt port='10000' skip='true' host='example.org'>
- <fileset dir='${basedir}'/>
- <addUser port='10000' user='user' passwd='passwd'>C: Foo Bar</addUser>
- </mpt:mpt>
- </target>
-
- <target name='testAddUserBothScriptTextAndScriptAttributeMayNotBeSet'>
- <au:expectfailure>
- <mpt:mpt port='10000' skip='true' host='example.org'>
- <fileset dir='${basedir}'/>
- <addUser port='10000' user='user' passwd='passwd' script='pom.xml'>C: Foo Bar</addUser>
- </mpt:mpt>
- </au:expectfailure>
- </target>
-
- <target name='testMptAttributes' depends='testMustSetPort, testMaySetHost,testMustSetScriptsOrScript, testMustNotAllowScriptsAndScript, testMaySetShabang, testMaySetQuiet, testMaySetErrorProperty'/>
- <target name='testAddUser' depends='testShouldAllowAddUserToBeSet, testAddUserMustSetPortScript, testAddUserMayIncludeScriptText, testAddUserBothScriptTextAndScriptAttributeMayNotBeSet, testAddUserMaySetNamePassword'/>
-
- <target name='all' depends='testMptAttributes, testAddUser'/>
-</project>
diff --git a/mpt/pom.xml b/mpt/pom.xml
index f51691d..9064939 100644
--- a/mpt/pom.xml
+++ b/mpt/pom.xml
@@ -37,7 +37,6 @@
<modules>
<module>all</module>
- <module>antlib</module>
<module>app</module>
<module>core</module>
<module>impl/imap-mailbox</module>
diff --git a/server/mailet/mailets/src/main/java/org/apache/james/transport/mailets/BayesianAnalysis.java b/server/mailet/mailets/src/main/java/org/apache/james/transport/mailets/BayesianAnalysis.java
deleted file mode 100644
index 2d07e75..0000000
--- a/server/mailet/mailets/src/main/java/org/apache/james/transport/mailets/BayesianAnalysis.java
+++ /dev/null
@@ -1,460 +0,0 @@
-/****************************************************************
- * 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.transport.mailets;
-
-import java.io.BufferedReader;
-import java.io.ByteArrayOutputStream;
-import java.io.StringReader;
-import java.sql.Connection;
-import java.text.DecimalFormat;
-import java.util.Collection;
-
-import javax.inject.Inject;
-import javax.mail.MessagingException;
-import javax.mail.internet.MimeMessage;
-import javax.sql.DataSource;
-
-import org.apache.james.core.MailAddress;
-import org.apache.james.filesystem.api.FileSystem;
-import org.apache.james.util.bayesian.JDBCBayesianAnalyzer;
-import org.apache.james.util.sql.JDBCUtil;
-import org.apache.mailet.Attribute;
-import org.apache.mailet.AttributeName;
-import org.apache.mailet.AttributeValue;
-import org.apache.mailet.Experimental;
-import org.apache.mailet.Mail;
-import org.apache.mailet.base.GenericMailet;
-import org.apache.mailet.base.RFC2822Headers;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.base.Joiner;
-
-/**
- * <p>
- * Spam detection mailet using bayesian analysis techniques.
- * </p>
- *
- * <p>
- * Sets an email message header indicating the probability that an email message
- * is SPAM.
- * </p>
- *
- * <p>
- * Based upon the principals described in: <a
- * href="http://www.paulgraham.com/spam.html">A Plan For Spam</a> by Paul
- * Graham. Extended to Paul Grahams' <a
- * href="http://paulgraham.com/better.html">Better Bayesian Filtering</a>.
- * </p>
- *
- * <p>
- * The analysis capabilities are based on token frequencies (the <i>Corpus</i>)
- * learned through a training process (see {@link BayesianAnalysisFeeder}) and
- * stored in a JDBC database. After a training session, the Corpus must be
- * rebuilt from the database in order to acquire the new frequencies. Every 10
- * minutes a special thread in this mailet will check if any change was made to
- * the database by the feeder, and rebuild the corpus if necessary.
- * </p>
- *
- * <p>
- * A <code>org.apache.james.spam.probability</code> mail attribute will be
- * created containing the computed spam probability as a
- * {@link java.lang.Double}. The <code>headerName</code> message header string
- * will be created containing such probability in floating point representation.
- * </p>
- *
- * <p>
- * Sample configuration:
- * </p>
- *
- * <pre>
- * <code>
- * <mailet match="All" class="BayesianAnalysis">
- * <repositoryPath>db://maildb</repositoryPath>
- * <!--
- * Set this to the header name to add with the spam probability
- * (default is "X-MessageIsSpamProbability").
- * -->
- * <headerName>X-MessageIsSpamProbability</headerName>
- * <!--
- * Set this to true if you want to ignore messages coming from local senders
- * (default is false).
- * By local sender we mean a return-path with a local server part (server listed
- * in <servernames> in config.xml).
- * -->
- * <ignoreLocalSender>true</ignoreLocalSender>
- * <!--
- * Set this to the maximum message size (in bytes) that a message may have
- * to be considered spam (default is 100000).
- * -->
- * <maxSize>100000</maxSize>
- * <!--
- * Set this to false if you not want to tag the message if spam is detected (Default is true).
- * -->
- * <tagSubject>true</tagSubject>
- * </mailet>
- * </code>
- * </pre>
- *
- * <p>
- * The probability of being spam is pre-pended to the subject if it is > 0.1
- * (10%).
- * </p>
- *
- * <p>
- * The required tables are automatically created if not already there (see
- * sqlResources.xml). The token field in both the ham and spam tables is <b>case
- * sensitive</b>.
- * </p>
- *
- * @see BayesianAnalysisFeeder
- * @see org.apache.james.util.bayesian.BayesianAnalyzer
- * @see org.apache.james.util.bayesian.JDBCBayesianAnalyzer
- * @since 2.3.0
- */
-@Experimental
-public class BayesianAnalysis extends GenericMailet {
- private static final Logger LOGGER = LoggerFactory.getLogger(BayesianAnalysis.class);
-
- /**
- * The JDBCUtil helper class
- */
- private final JDBCUtil theJDBCUtil = new JDBCUtil();
-
- /**
- * The JDBCBayesianAnalyzer class that does all the work.
- */
- private final JDBCBayesianAnalyzer analyzer = new JDBCBayesianAnalyzer();
-
- private DataSource datasource;
-
- private static final AttributeName MAIL_ATTRIBUTE_NAME = AttributeName.of("org.apache.james.spam.probability");
- private static final String HEADER_NAME = "X-MessageIsSpamProbability";
- private static final long CORPUS_RELOAD_INTERVAL = 600000;
- private String headerName;
- private boolean ignoreLocalSender = false;
- private boolean tagSubject = true;
-
- /**
- * Return a string describing this mailet.
- *
- * @return a string describing this mailet
- */
- @Override
- public String getMailetInfo() {
- return "BayesianAnalysis Mailet";
- }
-
- /**
- * Holds value of property maxSize.
- */
- private int maxSize = 100000;
-
- /**
- * Holds value of property lastCorpusLoadTime.
- */
- private long lastCorpusLoadTime;
-
- private FileSystem fs;
-
- /**
- * Getter for property maxSize.
- *
- * @return Value of property maxSize.
- */
- public int getMaxSize() {
-
- return this.maxSize;
- }
-
- /**
- * Setter for property maxSize.
- *
- * @param maxSize
- * New value of property maxSize.
- */
- public void setMaxSize(int maxSize) {
-
- this.maxSize = maxSize;
- }
-
- /**
- * Getter for property lastCorpusLoadTime.
- *
- * @return Value of property lastCorpusLoadTime.
- */
- public long getLastCorpusLoadTime() {
-
- return this.lastCorpusLoadTime;
- }
-
- @Inject
- public void setDataSource(DataSource datasource) {
- this.datasource = datasource;
- }
-
- @Inject
- public void setFileSystem(FileSystem fs) {
- this.fs = fs;
- }
-
- /**
- * Sets lastCorpusLoadTime to System.currentTimeMillis().
- */
- private void touchLastCorpusLoadTime() {
-
- this.lastCorpusLoadTime = System.currentTimeMillis();
- }
-
- /**
- * Mailet initialization routine.
- *
- * @throws MessagingException
- * if a problem arises
- */
- @Override
- public void init() throws MessagingException {
- String repositoryPath = getInitParameter("repositoryPath");
-
- if (repositoryPath == null) {
- throw new MessagingException("repositoryPath is null");
- }
-
- headerName = getInitParameter("headerName", HEADER_NAME);
-
- ignoreLocalSender = Boolean.parseBoolean(getInitParameter("ignoreLocalSender"));
-
- if (ignoreLocalSender) {
- LOGGER.debug("Will ignore messages coming from local senders");
- } else {
- LOGGER.debug("Will analyze messages coming from local senders");
- }
-
- String maxSizeParam = getInitParameter("maxSize");
- if (maxSizeParam != null) {
- setMaxSize(Integer.parseInt(maxSizeParam));
- }
- LOGGER.debug("maxSize: {}", getMaxSize());
-
- String tag = getInitParameter("tagSubject");
- if (tag != null && tag.equals("false")) {
- tagSubject = false;
- }
-
- initDb();
-
- CorpusLoader corpusLoader = new CorpusLoader(this);
- corpusLoader.setDaemon(true);
- corpusLoader.start();
-
- }
-
- private void initDb() throws MessagingException {
-
- try {
- analyzer.initSqlQueries(datasource.getConnection(), fs.getFile("file://conf/sqlResources.xml"));
- } catch (Exception e) {
- throw new MessagingException("Exception initializing queries", e);
- }
-
- try {
- loadData(datasource.getConnection());
- } catch (java.sql.SQLException se) {
- throw new MessagingException("SQLException loading data", se);
- }
- }
-
- /**
- * Scans the mail and determines the spam probability.
- *
- * @param mail
- * The Mail message to be scanned.
- * @throws MessagingException
- * if a problem arises
- */
- @Override
- public void service(Mail mail) throws MessagingException {
-
- try {
- MimeMessage message = mail.getMessage();
-
- if (ignoreLocalSender) {
- // ignore the message if the sender is local
- if (isSenderLocal(mail)) {
- return;
- }
- }
-
- String[] headerArray = message.getHeader(headerName);
- // ignore the message if already analyzed
- if (headerArray != null && headerArray.length > 0) {
- return;
- }
-
- ByteArrayOutputStream baos = new ByteArrayOutputStream();
-
- double probability;
-
- if (message.getSize() < getMaxSize()) {
- message.writeTo(baos);
- probability = analyzer.computeSpamProbability(new BufferedReader(new StringReader(baos.toString())));
- } else {
- probability = 0.0;
- }
-
- mail.setAttribute(new Attribute(MAIL_ATTRIBUTE_NAME, AttributeValue.of(probability)));
- message.setHeader(headerName, Double.toString(probability));
-
- DecimalFormat probabilityForm = (DecimalFormat) DecimalFormat.getInstance();
- probabilityForm.applyPattern("##0.##%");
- String probabilityString = probabilityForm.format(probability);
-
- String senderString = mail.getMaybeSender().asString("null");
-
- if (probability > 0.1) {
- if (LOGGER.isDebugEnabled()) {
- LOGGER.debug(headerName + ": " + probabilityString + "; From: " + senderString + "; Recipient(s): " + getAddressesString(mail.getRecipients()));
- }
-
- // Check if we should tag the subject
- if (tagSubject) {
- appendToSubject(message, " [" + probabilityString + (probability > 0.9 ? " SPAM" : " spam") + "]");
- }
- }
-
- saveChanges(message);
-
- } catch (Exception e) {
- LOGGER.error("Exception: {}", e.getMessage(), e);
- throw new MessagingException("Exception thrown", e);
- }
- }
-
- private boolean isSenderLocal(Mail mail) {
- return mail.getMaybeSender().asOptional()
- .map(MailAddress::getDomain)
- .map(domain -> getMailetContext().isLocalServer(domain))
- .orElse(false);
- }
-
- private void loadData(Connection conn) throws java.sql.SQLException {
-
- try {
- // this is synchronized to avoid concurrent update of the corpus
- synchronized (JDBCBayesianAnalyzer.DATABASE_LOCK) {
- analyzer.tokenCountsClear();
- analyzer.loadHamNSpam(conn);
- analyzer.buildCorpus();
- analyzer.tokenCountsClear();
- }
-
- LOGGER.error("BayesianAnalysis Corpus loaded");
-
- touchLastCorpusLoadTime();
-
- } finally {
- if (conn != null) {
- theJDBCUtil.closeJDBCConnection(conn);
- }
- }
-
- }
-
- private String getAddressesString(Collection<MailAddress> addresses) {
- if (addresses == null) {
- return "null";
- }
-
- StringBuilder sb = new StringBuilder();
- sb.append('[');
- Joiner.on(", ").appendTo(sb, addresses);
- sb.append(']');
- return sb.toString();
- }
-
- private void appendToSubject(MimeMessage message, String toAppend) {
- try {
- String subject = message.getSubject();
-
- if (subject == null) {
- message.setSubject(toAppend, "iso-8859-1");
- } else {
- message.setSubject(toAppend + " " + subject, "iso-8859-1");
- }
- } catch (MessagingException ex) {
- LOGGER.error("Ignored error while modifying subject", ex);
- }
- }
-
- /**
- * Saves changes resetting the original message id.
- */
- private void saveChanges(MimeMessage message) throws MessagingException {
- String messageId = message.getMessageID();
- message.saveChanges();
- if (messageId != null) {
- message.setHeader(RFC2822Headers.MESSAGE_ID, messageId);
- }
- }
-
- private static class CorpusLoader extends Thread {
-
- private final BayesianAnalysis analysis;
-
- private CorpusLoader(BayesianAnalysis analysis) {
- super("BayesianAnalysis Corpus Loader");
- this.analysis = analysis;
- }
-
- /**
- * Thread entry point.
- */
- @Override
- public void run() {
- LOGGER.info("CorpusLoader thread started: will wake up every " + CORPUS_RELOAD_INTERVAL + " ms");
-
- try {
- Thread.sleep(CORPUS_RELOAD_INTERVAL);
-
- while (true) {
- if (analysis.getLastCorpusLoadTime() < JDBCBayesianAnalyzer.getLastDatabaseUpdateTime()) {
- LOGGER.info("Reloading Corpus ...");
- try {
- analysis.loadData(analysis.datasource.getConnection());
- LOGGER.info("Corpus reloaded");
- } catch (java.sql.SQLException se) {
- LOGGER.error("SQLException: ", se);
- }
-
- }
-
- if (Thread.interrupted()) {
- break;
- }
- Thread.sleep(CORPUS_RELOAD_INTERVAL);
- }
- } catch (InterruptedException ex) {
- interrupt();
- }
- }
-
- }
-
-}
diff --git a/server/mailet/mailets/src/main/java/org/apache/james/transport/mailets/BayesianAnalysisFeeder.java b/server/mailet/mailets/src/main/java/org/apache/james/transport/mailets/BayesianAnalysisFeeder.java
deleted file mode 100644
index 38d8717..0000000
--- a/server/mailet/mailets/src/main/java/org/apache/james/transport/mailets/BayesianAnalysisFeeder.java
+++ /dev/null
@@ -1,336 +0,0 @@
-/****************************************************************
- * 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.transport.mailets;
-
-import java.io.BufferedReader;
-import java.io.ByteArrayOutputStream;
-import java.io.StringReader;
-import java.sql.Connection;
-import java.util.Enumeration;
-
-import javax.inject.Inject;
-import javax.mail.Header;
-import javax.mail.MessagingException;
-import javax.mail.internet.MimeMessage;
-import javax.sql.DataSource;
-
-import org.apache.james.filesystem.api.FileSystem;
-import org.apache.james.util.bayesian.JDBCBayesianAnalyzer;
-import org.apache.james.util.sql.JDBCUtil;
-import org.apache.mailet.Experimental;
-import org.apache.mailet.Mail;
-import org.apache.mailet.base.GenericMailet;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * <p>
- * Feeds ham OR spam messages to train the {@link BayesianAnalysis} mailet.
- * </p>
- *
- * <p>
- * The new token frequencies will be stored in a JDBC database.
- * </p>
- *
- * <p>
- * Sample configuration:
- * </p>
- *
- * <pre>
- * <code>
- * <processor name="root">
- *
- * <mailet match="RecipientIs=not.spam@thisdomain.com" class="BayesianAnalysisFeeder">
- * <repositoryPath> db://maildb </repositoryPath>
- * <feedType>ham</feedType>
- * <!--
- * Set this to the maximum message size (in bytes) that a message may have
- * to be analyzed (default is 100000).
- * -->
- * <maxSize>100000</maxSize>
- * </mailet>
- *
- * <mailet match="RecipientIs=spam@thisdomain.com" class="BayesianAnalysisFeeder">
- * <repositoryPath> db://maildb </repositoryPath>
- * <feedType>spam</feedType>
- * <!--
- * Set this to the maximum message size (in bytes) that a message may have
- * to be analyzed (default is 100000).
- * -->
- * <maxSize>100000</maxSize>
- * </mailet>
- *
- * <processor>
- * </code>
- * </pre>
- *
- * <p>
- * The previous example will allow the user to send messages to the server and
- * use the recipient email address as the indicator for whether the message is
- * ham or spam.
- * </p>
- *
- * <p>
- * Using the example above, send good messages (ham not spam) to the email
- * address "not.spam@thisdomain.com" to pump good messages into the feeder, and
- * send spam messages (spam not ham) to the email address "spam@thisdomain.com"
- * to pump spam messages into the feeder.
- * </p>
- *
- * <p>
- * The bayesian database tables will be updated during the training reflecting
- * the new data
- * </p>
- *
- * <p>
- * At the end the mail will be destroyed (ghosted).
- * </p>
- *
- * <p>
- * <b>The correct approach is to send the original ham/spam message as an
- * attachment to another message sent to the feeder; all the headers of the
- * enveloping message will be removed and only the original message's tokens
- * will be analyzed.</b>
- * </p>
- *
- * <p>
- * After a training session, the frequency <i>Corpus</i> used by
- * <code>BayesianAnalysis</code> must be rebuilt from the database, in order to
- * take advantage of the new token frequencies. Every 10 minutes a special
- * thread in the <code>BayesianAnalysis</code> mailet will check if any change
- * was made to the database, and rebuild the corpus if necessary.
- * </p>
- *
- * <p>
- * Only one message at a time is scanned (the database update activity is
- * <i>synchronized</i>) in order to avoid too much database locking, as
- * thousands of rows may be updated just for one message fed.
- * </p>
- *
- * @see BayesianAnalysis
- * @see org.apache.james.util.bayesian.BayesianAnalyzer
- * @see org.apache.james.util.bayesian.JDBCBayesianAnalyzer
- * @since 2.3.0
- */
-@Experimental
-public class BayesianAnalysisFeeder extends GenericMailet {
- private static final Logger LOGGER = LoggerFactory.getLogger(BayesianAnalysisFeeder.class);
-
- /**
- * The JDBCUtil helper class
- */
- private final JDBCUtil theJDBCUtil = new JDBCUtil();
-
- /**
- * The JDBCBayesianAnalyzer class that does all the work.
- */
- private final JDBCBayesianAnalyzer analyzer = new JDBCBayesianAnalyzer();
-
- private DataSource datasource;
-
- private String feedType;
-
- @Override
- public String getMailetInfo() {
- return "BayesianAnalysisFeeder Mailet";
- }
-
- /**
- * Holds value of property maxSize.
- */
- private int maxSize = 100000;
-
- private FileSystem fs;
-
- /**
- * Getter for property maxSize.
- *
- * @return Value of property maxSize.
- */
- public int getMaxSize() {
-
- return this.maxSize;
- }
-
- @Inject
- public void setDataSource(DataSource datasource) {
- this.datasource = datasource;
- }
-
- /**
- * Setter for property maxSize.
- *
- * @param maxSize
- * New value of property maxSize.
- */
- public void setMaxSize(int maxSize) {
-
- this.maxSize = maxSize;
- }
-
- @Inject
- public void setFileSystem(FileSystem fs) {
- this.fs = fs;
- }
-
- @Override
- public void init() throws MessagingException {
- String repositoryPath = getInitParameter("repositoryPath");
-
- if (repositoryPath == null) {
- throw new MessagingException("repositoryPath is null");
- }
-
- feedType = getInitParameter("feedType");
- if (feedType == null) {
- throw new MessagingException("feedType is null");
- }
-
- String maxSizeParam = getInitParameter("maxSize");
- if (maxSizeParam != null) {
- setMaxSize(Integer.parseInt(maxSizeParam));
- }
- LOGGER.debug("maxSize: {}", getMaxSize());
-
- initDb();
-
- }
-
- private void initDb() throws MessagingException {
-
- try {
- analyzer.initSqlQueries(datasource.getConnection(), fs.getFile("file://conf/sqlResources.xml"));
- } catch (Exception e) {
- throw new MessagingException("Exception initializing queries", e);
- }
-
- }
-
- /**
- * Scans the mail and updates the token frequencies in the database.
- *
- * The method is synchronized in order to avoid too much database locking,
- * as thousands of rows may be updated just for one message fed.
- *
- * @param mail
- * The Mail message to be scanned.
- */
- @Override
- public void service(Mail mail) {
- boolean dbUpdated = false;
-
- mail.setState(Mail.GHOST);
-
- ByteArrayOutputStream baos = new ByteArrayOutputStream();
-
- Connection conn = null;
-
- try {
-
- MimeMessage message = mail.getMessage();
-
- String messageId = message.getMessageID();
-
- if (message.getSize() > getMaxSize()) {
- LOGGER.debug("{} Feeding HAM/SPAM ignored because message size > {}: {}", messageId, getMaxSize(), message.getSize());
- return;
- }
-
- clearAllHeaders(message);
-
- message.writeTo(baos);
-
- BufferedReader br = new BufferedReader(new StringReader(baos.toString()));
-
- // this is synchronized to avoid concurrent update of the corpus
- synchronized (JDBCBayesianAnalyzer.DATABASE_LOCK) {
-
- conn = datasource.getConnection();
-
- if (conn.getAutoCommit()) {
- conn.setAutoCommit(false);
- }
-
- dbUpdated = true;
-
- // Clear out any existing word/counts etc..
- analyzer.clear();
-
- if ("ham".equalsIgnoreCase(feedType)) {
- LOGGER.debug("{} Feeding HAM", messageId);
- // Process the stream as ham (not spam).
- analyzer.addHam(br);
-
- // Update storage statistics.
- analyzer.updateHamTokens(conn);
- } else {
- LOGGER.debug("{} Feeding SPAM", messageId);
- // Process the stream as spam.
- analyzer.addSpam(br);
-
- // Update storage statistics.
- analyzer.updateSpamTokens(conn);
- }
-
- // Commit our changes if necessary.
- if (conn != null && dbUpdated && !conn.getAutoCommit()) {
- conn.commit();
- dbUpdated = false;
- LOGGER.debug("{} Training ended successfully", messageId);
- JDBCBayesianAnalyzer.touchLastDatabaseUpdateTime();
- }
-
- }
-
- } catch (java.sql.SQLException se) {
- LOGGER.error("SQLException: ", se);
- } catch (java.io.IOException ioe) {
- LOGGER.error("IOException: ", ioe);
- } catch (javax.mail.MessagingException me) {
- LOGGER.error("MessagingException: ", me);
- } finally {
- // Rollback our changes if necessary.
- try {
- if (conn != null && dbUpdated && !conn.getAutoCommit()) {
- conn.rollback();
- dbUpdated = false;
- }
- } catch (Exception e) {
- LOGGER.error("Ignored error while rolling back", e);
- }
- theJDBCUtil.closeJDBCConnection(conn);
- }
- }
-
- private void clearAllHeaders(MimeMessage message) throws javax.mail.MessagingException {
- Enumeration<Header> headers = message.getAllHeaders();
-
- while (headers.hasMoreElements()) {
- Header header = headers.nextElement();
- try {
- message.removeHeader(header.getName());
- } catch (MessagingException me) {
- LOGGER.error("Ignored error while removing header", me);
- }
- }
- message.saveChanges();
- }
-
-}
diff --git a/server/mailet/mailets/src/main/java/org/apache/james/transport/mailets/ToRecipientFolder.java b/server/mailet/mailets/src/main/java/org/apache/james/transport/mailets/ToRecipientFolder.java
deleted file mode 100644
index 1ab3a2f..0000000
--- a/server/mailet/mailets/src/main/java/org/apache/james/transport/mailets/ToRecipientFolder.java
+++ /dev/null
@@ -1,99 +0,0 @@
-/****************************************************************
- * 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.transport.mailets;
-
-import static org.apache.james.transport.mailets.LocalDelivery.LOCAL_DELIVERED_MAILS_METRIC_NAME;
-
-import javax.inject.Inject;
-import javax.inject.Named;
-import javax.mail.MessagingException;
-
-import org.apache.james.mailbox.MailboxManager;
-import org.apache.james.mailbox.model.MailboxConstants;
-import org.apache.james.metrics.api.MetricFactory;
-import org.apache.james.transport.mailets.delivery.MailDispatcher;
-import org.apache.james.transport.mailets.delivery.MailboxAppender;
-import org.apache.james.transport.mailets.delivery.SimpleMailStore;
-import org.apache.james.user.api.UsersRepository;
-import org.apache.mailet.Mail;
-import org.apache.mailet.base.GenericMailet;
-
-/**
- * Receives a Mail from the Queue and takes care to deliver the message
- * to a defined folder of the recipient(s).
- *
- * You have to define the folder name of the recipient(s).
- * The flag 'consume' will tell is the mail will be further
- * processed by the upcoming processor mailets, or not.
- *
- * <pre>
- * <mailet match="RecipientIsLocal" class="ToRecipientFolder">
- * <folder> <i>Junk</i> </folder>
- * <consume> <i>false</i> </consume>
- * </mailet>
- * </pre>
- *
- * @Deprecated This mailet duplicates the behaviour of LocalDelivery. Prefer using WithStorageDirective
- * in order to tell LocalDelivery in which folder the mail should be appended.
- *
- */
-@Deprecated
-public class ToRecipientFolder extends GenericMailet {
-
- public static final String FOLDER_PARAMETER = "folder";
- public static final String CONSUME_PARAMETER = "consume";
-
- private final MailboxManager mailboxManager;
- private final UsersRepository usersRepository;
- private final MetricFactory metricFactory;
- private MailDispatcher mailDispatcher;
-
- @Inject
- public ToRecipientFolder(@Named("mailboxmanager")MailboxManager mailboxManager, UsersRepository usersRepository,
- MetricFactory metricFactory) {
- this.metricFactory = metricFactory;
- this.mailboxManager = mailboxManager;
- this.usersRepository = usersRepository;
- }
-
- @Override
- public void service(Mail mail) throws MessagingException {
- mailDispatcher.dispatch(mail);
- }
-
- @Override
- public void init() throws MessagingException {
- mailDispatcher = MailDispatcher.builder()
- .mailStore(SimpleMailStore.builder()
- .mailboxAppender(new MailboxAppender(mailboxManager))
- .usersRepository(usersRepository)
- .folder(getInitParameter(FOLDER_PARAMETER, MailboxConstants.INBOX))
- .metric(metricFactory.generate(LOCAL_DELIVERED_MAILS_METRIC_NAME))
- .build())
- .consume(getInitParameter(CONSUME_PARAMETER, false))
- .mailetContext(getMailetContext())
- .build();
- }
-
- @Override
- public String getMailetInfo() {
- return ToRecipientFolder.class.getName() + " Mailet";
- }
-
-}
diff --git a/server/mailet/mailets/src/test/java/org/apache/james/transport/mailets/delivery/ToRecipientFolderTest.java b/server/mailet/mailets/src/test/java/org/apache/james/transport/mailets/delivery/ToRecipientFolderTest.java
deleted file mode 100644
index 7108eb6..0000000
--- a/server/mailet/mailets/src/test/java/org/apache/james/transport/mailets/delivery/ToRecipientFolderTest.java
+++ /dev/null
@@ -1,198 +0,0 @@
-/****************************************************************
- * 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.transport.mailets.delivery;
-
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.eq;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-
-import java.io.IOException;
-
-import javax.mail.MessagingException;
-import javax.mail.internet.MimeBodyPart;
-
-import org.apache.james.core.MailAddress;
-import org.apache.james.core.Username;
-import org.apache.james.core.builder.MimeMessageBuilder;
-import org.apache.james.mailbox.MailboxManager;
-import org.apache.james.mailbox.MailboxSession;
-import org.apache.james.mailbox.MessageManager;
-import org.apache.james.mailbox.model.MailboxPath;
-import org.apache.james.metrics.api.MetricFactory;
-import org.apache.james.metrics.tests.RecordingMetricFactory;
-import org.apache.james.transport.mailets.ToRecipientFolder;
-import org.apache.james.user.api.UsersRepository;
-import org.apache.mailet.Mail;
-import org.apache.mailet.base.test.FakeMail;
-import org.apache.mailet.base.test.FakeMailContext;
-import org.apache.mailet.base.test.FakeMailetConfig;
-import org.junit.Before;
-import org.junit.Test;
-import org.slf4j.Logger;
-
-import reactor.core.publisher.Mono;
-
-@Deprecated
-public class ToRecipientFolderTest {
-
- public static final String USER_LOCAL_PART = "receiver";
- public static final String USER = USER_LOCAL_PART + "@domain.com";
- public static final Username USERNAME = Username.of(USER);
- public static final Username USERNAME_LOCAL_PART = Username.of(USER_LOCAL_PART);
- public static final MailboxPath INBOX = MailboxPath.inbox(USERNAME);
- public static final MailboxPath JUNK = MailboxPath.forUser(USERNAME_LOCAL_PART, "Junk");
- public static final MailboxPath JUNK_VIRTUAL_HOSTING = MailboxPath.forUser(USERNAME, "Junk");
- public static final String MAILET_NAME = "RecipientFolderTest";
-
- private MessageManager messageManager;
- private UsersRepository usersRepository;
- private MailboxManager mailboxManager;
- private ToRecipientFolder testee;
- private FakeMailContext mailetContext;
- private MailboxSession session;
-
- @Before
- public void setUp() throws Exception {
- mailetContext = FakeMailContext.builder().logger(mock(Logger.class)).build();
- messageManager = mock(MessageManager.class);
- usersRepository = mock(UsersRepository.class);
- mailboxManager = mock(MailboxManager.class);
-
-
- MetricFactory metricFactory = new RecordingMetricFactory();
- testee = new ToRecipientFolder(mailboxManager, usersRepository, metricFactory);
-
- session = mock(MailboxSession.class);
- when(session.getPathDelimiter()).thenReturn('.');
- when(mailboxManager.createSystemSession(any(Username.class))).thenReturn(session);
- when(mailboxManager.mailboxExists(any(), any())).thenReturn(Mono.just(true));
- when(session.getUser()).thenReturn(Username.of(USER));
- }
-
- @Test
- public void initParameterTesting() throws Exception {
- testee.init(FakeMailetConfig.builder()
- .mailetName(MAILET_NAME)
- .mailetContext(mailetContext)
- .setProperty(ToRecipientFolder.FOLDER_PARAMETER, "Junk")
- .build());
-
- assertThat(testee.getInitParameter(ToRecipientFolder.FOLDER_PARAMETER)).isEqualTo("Junk");
- }
-
- @Test
- public void consumeOptionShouldGhostTheMail() throws Exception {
- testee.init(FakeMailetConfig.builder()
- .mailetName(MAILET_NAME)
- .mailetContext(mailetContext)
- .setProperty(ToRecipientFolder.CONSUME_PARAMETER, "true")
- .build());
-
- Mail mail = createMail();
- testee.service(mail);
-
- assertThat(mail.getState()).isEqualTo(Mail.GHOST);
- }
-
- @Test
- public void consumeOptionShouldNotGhostTheMailByDefault() throws Exception {
- testee.init(FakeMailetConfig.builder()
- .mailetName(MAILET_NAME)
- .mailetContext(mailetContext)
- .build());
-
- Mail mail = createMail();
- testee.service(mail);
-
- assertThat(mail.getState()).isEqualTo(Mail.DEFAULT);
- }
-
- @Test
- public void folderParameterShouldIndicateDestinationFolder() throws Exception {
- when(usersRepository.supportVirtualHosting()).thenReturn(true);
- when(usersRepository.getUsername(new MailAddress(USER))).thenReturn(USERNAME);
- when(mailboxManager.getMailbox(eq(JUNK_VIRTUAL_HOSTING), any(MailboxSession.class))).thenReturn(messageManager);
-
- testee.init(FakeMailetConfig.builder()
- .mailetName(MAILET_NAME)
- .mailetContext(mailetContext)
- .setProperty(ToRecipientFolder.FOLDER_PARAMETER, "Junk")
- .build());
- testee.service(createMail());
-
- verify(messageManager).appendMessage(any(MessageManager.AppendCommand.class), any(MailboxSession.class));
- }
-
- @Test
- public void folderParameterShouldBeInboxByDefault() throws Exception {
- when(usersRepository.supportVirtualHosting()).thenReturn(true);
- when(usersRepository.getUsername(new MailAddress(USER))).thenReturn(USERNAME);
- when(mailboxManager.getMailbox(eq(INBOX), any(MailboxSession.class))).thenReturn(messageManager);
-
- testee.init(FakeMailetConfig.builder()
- .mailetName(MAILET_NAME)
- .mailetContext(mailetContext)
- .build());
- testee.service(createMail());
-
- verify(messageManager).appendMessage(any(MessageManager.AppendCommand.class), any(MailboxSession.class));
- }
-
- @Test
- public void folderParameterShouldWorkWhenVirtualHostingIsTurnedOff() throws Exception {
- when(usersRepository.supportVirtualHosting()).thenReturn(false);
- when(usersRepository.getUsername(new MailAddress(USER_LOCAL_PART + "@localhost"))).thenReturn(USERNAME_LOCAL_PART);
- when(usersRepository.getUsername(new MailAddress(USER))).thenReturn(USERNAME_LOCAL_PART);
- when(mailboxManager.getMailbox(eq(JUNK), any(MailboxSession.class))).thenReturn(messageManager);
- when(session.getUser()).thenReturn(Username.of(USER_LOCAL_PART));
-
- testee.init(FakeMailetConfig.builder()
- .mailetName(MAILET_NAME)
- .mailetContext(mailetContext)
- .setProperty(ToRecipientFolder.FOLDER_PARAMETER, "Junk")
- .setProperty(ToRecipientFolder.CONSUME_PARAMETER, "true")
- .build());
- testee.service(createMail());
-
- verify(messageManager).appendMessage(any(MessageManager.AppendCommand.class), any(MailboxSession.class));
- }
-
- private Mail createMail() throws MessagingException, IOException {
- return FakeMail.builder()
- .name("name")
- .mimeMessage(MimeMessageBuilder.mimeMessageBuilder()
- .setSender("sender@any.com")
- .setSubject("Subject")
- .addToRecipient(USER)
- .setMultipartWithBodyParts(
- MimeMessageBuilder.bodyPartBuilder()
- .data("toto")
- .disposition(MimeBodyPart.ATTACHMENT)
- .filename("file.txt")
- .addHeader("Content-Type", "application/sieve; charset=UTF-8")))
- .state(Mail.DEFAULT)
- .recipient("receiver@domain.com")
- .build();
- }
-
-}
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org