You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@james.apache.org by bt...@apache.org on 2022/09/22 02:50:57 UTC
[james-project] branch 3.7.x updated (e8da16cd5e -> ed7660c8b1)
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a change to branch 3.7.x
in repository https://gitbox.apache.org/repos/asf/james-project.git
from e8da16cd5e [Upgrade] Maven slf4j 1.7.32 -> 2.0.1
new 2ea97b0bc0 JAMES-3723 Allow to not consume emails upon reprocessing
new 61a1b756ac JAMES-3723 Fix a typo Reprocessing/Reindexing
new 112bf71c95 JAMES-3723 Document reprocessing with consume = false
new 474222f41b JAMES-3784 HealthCheck /var/mail/error repository size
new d12208333f JAMES-3784 WebAdmin: Provide RunningOptions (rateLimit) for Redeliver event task, Reprocessing mail task
new cd418ddfd4 JAMES-3784 WebAdmin: Document - Provide RunningOptions (limit) for Redeliver event task, Reprocessing mail task
new ed7660c8b1 task/task-distributed - fixing NullPointerException when executeTask
The 7 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:
.../tools/indexer/ErrorRecoveryIndexationTask.java | 14 +-
.../mailbox/tools/indexer/FullReindexingTask.java | 14 +-
.../mailbox/tools/indexer/ReIndexerPerformer.java | 38 +++---
...ocessingContext.java => ReIndexingContext.java} | 4 +-
...tion.java => ReIndexingContextInformation.java} | 6 +-
...O.java => ReIndexingContextInformationDTO.java} | 58 ++++----
.../tools/indexer/SingleMailboxReindexingTask.java | 14 +-
...lboxReindexingTaskAdditionalInformationDTO.java | 28 ++--
.../tools/indexer/SingleMessageReindexingTask.java | 2 +-
.../mailbox/tools/indexer/UserReindexingTask.java | 14 +-
...UserReindexingTaskAdditionalInformationDTO.java | 24 ++--
.../tools/indexer/CassandraReIndexerImplTest.java | 12 +-
...rorRecoveryIndexationTaskSerializationTest.java | 14 +-
.../FullReindexingTaskSerializationTest.java | 10 +-
.../docs/modules/ROOT/pages/operate/guide.adoc | 7 +
.../docs/modules/ROOT/pages/operate/webadmin.adoc | 48 ++++++-
.../apache/james/modules/CommonServicesModule.java | 1 +
...ErrorMailRepositoryEmptyHealthCheckModule.java} | 25 ++--
.../mailbox/ReIndexingTaskSerializationModule.java | 6 +-
.../WebAdminReIndexingTaskSerializationModule.java | 6 +-
.../api/EmptyErrorMailRepositoryHealthCheck.java | 37 +++--
...mptyErrorMailRepositoryHealthCheckContract.java | 94 +++++++++++++
...ryEmptyErrorMailRepositoryHealthCheckTest.java} | 51 +++----
...dminServerTaskSerializationIntegrationTest.java | 1 +
.../james/webadmin/utils/ParametersExtractor.java | 9 ++
...> WebAdminIndexationContextInformationDTO.java} | 22 +--
...lboxReindexingTaskAdditionalInformationDTO.java | 4 +-
...UserReindexingTaskAdditionalInformationDTO.java | 4 +-
.../webadmin/routes/EventDeadLettersRoutes.java | 15 +-
.../service/EventDeadLettersRedeliverAllTask.java | 15 +-
.../EventDeadLettersRedeliverAllTaskDTO.java | 17 ++-
.../EventDeadLettersRedeliverGroupTask.java | 15 +-
.../EventDeadLettersRedeliverGroupTaskDTO.java | 18 ++-
.../service/EventDeadLettersRedeliverOneTask.java | 2 +-
.../service/EventDeadLettersRedeliverService.java | 33 ++++-
...tersRedeliveryTaskAdditionalInformationDTO.java | 71 +++++++---
.../webadmin/service/EventDeadLettersService.java | 8 +-
...bAdminIndexationContextInformationDTOTest.java} | 20 +--
...a => WebAdminSingleMailboxIndexingDTOTest.java} | 2 +-
...est.java => WebAdminUserReIndexingDTOTest.java} | 2 +-
.../routes/EventDeadLettersRoutesTest.java | 91 ++++++++++++
.../james/webadmin/routes/MailboxesRoutesTest.java | 88 ++++++------
.../james/webadmin/routes/MessageRoutesTest.java | 2 +-
.../webadmin/routes/UserMailboxesRoutesTest.java | 3 +-
.../service/EventDeadLettersRedeliverTaskTest.java | 152 ++++++++++++++++++---
.../webadmin/routes/MailRepositoriesRoutes.java | 25 +++-
.../webadmin/service/ReprocessingAllMailsTask.java | 47 ++-----
...essingAllMailsTaskAdditionalInformationDTO.java | 32 ++++-
.../service/ReprocessingAllMailsTaskDTO.java | 33 ++++-
.../webadmin/service/ReprocessingOneMailTask.java | 38 ++----
...cessingOneMailTaskAdditionalInformationDTO.java | 20 ++-
.../service/ReprocessingOneMailTaskDTO.java | 26 ++--
.../webadmin/service/ReprocessingService.java | 85 +++++++++---
.../routes/MailRepositoriesRoutesTest.java | 147 ++++++++++++++++++++
.../service/ReprocessingAllMailsTaskTest.java | 69 ++++++++--
.../service/ReprocessingOneMailTaskTest.java | 52 ++++++-
.../webadmin/service/ReprocessingServiceTest.java | 13 +-
.../distributed/RabbitMQWorkQueue.java | 7 +-
.../distributed/DistributedTaskManagerTest.java | 8 +-
.../server/manage-guice-distributed-james.md | 7 +
src/site/markdown/server/manage-webadmin.md | 6 +
61 files changed, 1279 insertions(+), 457 deletions(-)
rename mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/{ReprocessingContext.java => ReIndexingContext.java} (98%)
rename mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/{ReprocessingContextInformation.java => ReIndexingContextInformation.java} (87%)
rename mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/{ReprocessingContextInformationDTO.java => ReIndexingContextInformationDTO.java} (80%)
copy server/container/guice/common/src/main/java/org/apache/james/modules/{IsStartedProbeModule.java => ErrorMailRepositoryEmptyHealthCheckModule.java} (66%)
copy event-bus/api/src/main/java/org/apache/james/events/EventDeadLettersHealthCheck.java => server/mailrepository/mailrepository-api/src/main/java/org/apache/james/mailrepository/api/EmptyErrorMailRepositoryHealthCheck.java (56%)
create mode 100644 server/mailrepository/mailrepository-api/src/test/java/org/apache/james/mailrepository/api/EmptyErrorMailRepositoryHealthCheckContract.java
copy server/{container/guice/configuration/src/test/java/org/apache/james/utils/PropertiesProviderFromEnvVariablesTest.java => mailrepository/mailrepository-memory/src/test/java/org/apache/james/mailrepository/memory/MemoryEmptyErrorMailRepositoryHealthCheckTest.java} (52%)
rename server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/dto/{WebAdminReprocessingContextInformationDTO.java => WebAdminIndexationContextInformationDTO.java} (82%)
rename server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/dto/{WebAdminReprocessingContextInformationDTOTest.java => WebAdminIndexationContextInformationDTOTest.java} (85%)
rename server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/dto/{WebAdminSingleMailboxReprocessingDTOTest.java => WebAdminSingleMailboxIndexingDTOTest.java} (98%)
rename server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/dto/{WebAdminUserReprocessingDTOTest.java => WebAdminUserReIndexingDTOTest.java} (99%)
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 04/07: JAMES-3784 HealthCheck /var/mail/error repository size
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch 3.7.x
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 474222f41b20ade6e48e4690d0459a9a484d53a5
Author: Tung Van TRAN <vt...@linagora.com>
AuthorDate: Thu Jun 23 17:51:35 2022 +0700
JAMES-3784 HealthCheck /var/mail/error repository size
---
.../ErrorMailRepositoryEmptyHealthCheckModule.java | 45 +++++++++++
.../api/EmptyErrorMailRepositoryHealthCheck.java | 56 +++++++++++++
...mptyErrorMailRepositoryHealthCheckContract.java | 94 ++++++++++++++++++++++
...oryEmptyErrorMailRepositoryHealthCheckTest.java | 65 +++++++++++++++
4 files changed, 260 insertions(+)
diff --git a/server/container/guice/common/src/main/java/org/apache/james/modules/ErrorMailRepositoryEmptyHealthCheckModule.java b/server/container/guice/common/src/main/java/org/apache/james/modules/ErrorMailRepositoryEmptyHealthCheckModule.java
new file mode 100644
index 0000000000..753d98989c
--- /dev/null
+++ b/server/container/guice/common/src/main/java/org/apache/james/modules/ErrorMailRepositoryEmptyHealthCheckModule.java
@@ -0,0 +1,45 @@
+/****************************************************************
+ * 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.modules;
+
+import static org.apache.james.mailrepository.api.MailRepositoryEmptyHealthCheck.ErrorMailRepositoryEmptyHealthCheck;
+
+import org.apache.james.core.healthcheck.HealthCheck;
+import org.apache.james.mailrepository.api.MailRepositoryStore;
+
+import com.google.inject.AbstractModule;
+import com.google.inject.Provides;
+import com.google.inject.Singleton;
+import com.google.inject.multibindings.Multibinder;
+
+public class ErrorMailRepositoryEmptyHealthCheckModule extends AbstractModule {
+
+ @Override
+ protected void configure() {
+ Multibinder.newSetBinder(binder(), HealthCheck.class).addBinding().to(ErrorMailRepositoryEmptyHealthCheck.class);
+ }
+
+ @Singleton
+ @Provides
+ ErrorMailRepositoryEmptyHealthCheck provideErrorMailRepositoryEmptyHealthCheck(MailRepositoryStore mailRepositoryStore) {
+ return new ErrorMailRepositoryEmptyHealthCheck(mailRepositoryStore);
+ }
+
+}
diff --git a/server/mailrepository/mailrepository-api/src/main/java/org/apache/james/mailrepository/api/EmptyErrorMailRepositoryHealthCheck.java b/server/mailrepository/mailrepository-api/src/main/java/org/apache/james/mailrepository/api/EmptyErrorMailRepositoryHealthCheck.java
new file mode 100644
index 0000000000..dd8752b0ca
--- /dev/null
+++ b/server/mailrepository/mailrepository-api/src/main/java/org/apache/james/mailrepository/api/EmptyErrorMailRepositoryHealthCheck.java
@@ -0,0 +1,56 @@
+/****************************************************************
+ * 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.mailrepository.api;
+
+import org.apache.james.core.healthcheck.ComponentName;
+import org.apache.james.core.healthcheck.HealthCheck;
+import org.apache.james.core.healthcheck.Result;
+import org.apache.james.util.FunctionalUtils;
+
+import com.github.fge.lambdas.Throwing;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+public class EmptyErrorMailRepositoryHealthCheck implements HealthCheck {
+ public static final ComponentName COMPONENT_NAME = new ComponentName("EmptyErrorMailRepository");
+ private final MailRepositoryStore repositoryStore;
+ private final MailRepositoryPath errorRepositoryPath;
+
+ public EmptyErrorMailRepositoryHealthCheck(MailRepositoryPath errorRepositoryPath, MailRepositoryStore repositoryStore) {
+ this.repositoryStore = repositoryStore;
+ this.errorRepositoryPath = errorRepositoryPath;
+ }
+
+ @Override
+ public ComponentName componentName() {
+ return COMPONENT_NAME;
+ }
+
+ @Override
+ public Mono<Result> check() {
+ // TODO: use .sizeReactive of https://github.com/apache/james-project/pull/1049/
+ return Flux.fromStream(Throwing.supplier(() -> repositoryStore.getByPath(errorRepositoryPath)))
+ .any(Throwing.predicate(repository -> repository.size() > 0))
+ .filter(FunctionalUtils.identityPredicate())
+ .map(hasSize -> Result.degraded(COMPONENT_NAME, "MailRepository is not empty"))
+ .switchIfEmpty(Mono.just(Result.healthy(COMPONENT_NAME)));
+ }
+}
diff --git a/server/mailrepository/mailrepository-api/src/test/java/org/apache/james/mailrepository/api/EmptyErrorMailRepositoryHealthCheckContract.java b/server/mailrepository/mailrepository-api/src/test/java/org/apache/james/mailrepository/api/EmptyErrorMailRepositoryHealthCheckContract.java
new file mode 100644
index 0000000000..ae9975de61
--- /dev/null
+++ b/server/mailrepository/mailrepository-api/src/test/java/org/apache/james/mailrepository/api/EmptyErrorMailRepositoryHealthCheckContract.java
@@ -0,0 +1,94 @@
+/****************************************************************
+ * 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.mailrepository.api;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.apache.james.core.builder.MimeMessageBuilder;
+import org.apache.james.core.healthcheck.ComponentName;
+import org.apache.james.core.healthcheck.Result;
+import org.apache.mailet.base.test.FakeMail;
+import org.junit.jupiter.api.Test;
+
+public interface EmptyErrorMailRepositoryHealthCheckContract {
+ MailRepositoryPath ERROR_REPOSITORY_PATH = MailRepositoryPath.from("var/mail/error");
+
+ MailRepositoryStore repositoryStore();
+
+ void createRepository();
+
+ default EmptyErrorMailRepositoryHealthCheck testee() {
+ createRepository();
+ return new EmptyErrorMailRepositoryHealthCheck(ERROR_REPOSITORY_PATH, repositoryStore());
+ }
+
+ @Test
+ default void componentNameShouldReturnTheRightValue() {
+ assertThat(testee().componentName().getName())
+ .isEqualTo("EmptyErrorMailRepository");
+ }
+
+ @Test
+ default void checkShouldReturnHealthyWhenRepositorySizeIsEmpty() {
+ EmptyErrorMailRepositoryHealthCheck testee = testee();
+ assertThat(testee.check().block())
+ .isEqualTo(Result.healthy(new ComponentName("EmptyErrorMailRepository")));
+ }
+
+ @Test
+ default void checkShouldReturnHealthyWhenRepositoryIsNotCreated() {
+ EmptyErrorMailRepositoryHealthCheck testee = new EmptyErrorMailRepositoryHealthCheck(ERROR_REPOSITORY_PATH, repositoryStore());
+ assertThat(testee.check().block())
+ .isEqualTo(Result.healthy(new ComponentName("EmptyErrorMailRepository")));
+ }
+
+ @Test
+ default void checkShouldReturnDegradedWhenRepositorySizeIsNotEmpty() throws Exception {
+ EmptyErrorMailRepositoryHealthCheck testee = testee();
+ repositoryStore().getByPath(ERROR_REPOSITORY_PATH)
+ .findFirst().orElseThrow()
+ .store(FakeMail.builder()
+ .name("name1")
+ .mimeMessage(MimeMessageBuilder.mimeMessageBuilder()
+ .setText("Any body"))
+ .build());
+
+ assertThat(testee.check().block().isDegraded())
+ .isTrue();
+ }
+
+ @Test
+ default void checkShouldReturnHealthyWhenRepositorySizeReturnEmptyAgain() throws Exception {
+ EmptyErrorMailRepositoryHealthCheck testee = testee();
+ MailRepository mailRepository = repositoryStore().getByPath(ERROR_REPOSITORY_PATH).findFirst().orElseThrow();
+ mailRepository.store(FakeMail.builder()
+ .name("name1")
+ .mimeMessage(MimeMessageBuilder.mimeMessageBuilder()
+ .setText("Any body"))
+ .build());
+
+ assertThat(testee.check().block().isDegraded())
+ .isTrue();
+
+ mailRepository.removeAll();
+ assertThat(testee.check().block().isHealthy())
+ .isTrue();
+ }
+}
diff --git a/server/mailrepository/mailrepository-memory/src/test/java/org/apache/james/mailrepository/memory/MemoryEmptyErrorMailRepositoryHealthCheckTest.java b/server/mailrepository/mailrepository-memory/src/test/java/org/apache/james/mailrepository/memory/MemoryEmptyErrorMailRepositoryHealthCheckTest.java
new file mode 100644
index 0000000000..651bb84916
--- /dev/null
+++ b/server/mailrepository/mailrepository-memory/src/test/java/org/apache/james/mailrepository/memory/MemoryEmptyErrorMailRepositoryHealthCheckTest.java
@@ -0,0 +1,65 @@
+/****************************************************************
+ * 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.mailrepository.memory;
+
+import org.apache.james.mailrepository.api.EmptyErrorMailRepositoryHealthCheckContract;
+import org.apache.james.mailrepository.api.MailRepositoryStore;
+import org.apache.james.mailrepository.api.MailRepositoryUrl;
+import org.apache.james.server.core.configuration.Configuration;
+import org.apache.james.server.core.configuration.FileConfigurationProvider;
+import org.apache.james.server.core.filesystem.FileSystemImpl;
+import org.junit.jupiter.api.BeforeEach;
+
+public class MemoryEmptyErrorMailRepositoryHealthCheckTest implements EmptyErrorMailRepositoryHealthCheckContract {
+
+ private MemoryMailRepositoryStore repositoryStore;
+
+ @BeforeEach
+ void setup() throws Exception {
+ Configuration.Basic configuration = Configuration.builder()
+ .workingDirectory("../")
+ .configurationFromClasspath()
+ .build();
+
+ FileSystemImpl fileSystem = new FileSystemImpl(configuration.directories());
+
+ MailRepositoryStoreConfiguration storeConfiguration = MailRepositoryStoreConfiguration.parse(
+ new FileConfigurationProvider(fileSystem, configuration).getConfiguration("mailrepositorystore"));
+
+ repositoryStore = new MemoryMailRepositoryStore(new MemoryMailRepositoryUrlStore(),
+ new SimpleMailRepositoryLoader(),
+ storeConfiguration);
+ repositoryStore.init();
+ }
+
+ @Override
+ public MailRepositoryStore repositoryStore() {
+ return repositoryStore;
+ }
+
+ @Override
+ public void createRepository() {
+ try {
+ repositoryStore.create(MailRepositoryUrl.fromPathAndProtocol(ERROR_REPOSITORY_PATH, "memory1"));
+ } catch (MailRepositoryStore.MailRepositoryStoreException e) {
+ throw new RuntimeException(e);
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 03/07: JAMES-3723 Document reprocessing with consume = false
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch 3.7.x
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 112bf71c95b7ab733d0888f5ecf46d918c2562ea
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Thu Mar 10 16:50:09 2022 +0700
JAMES-3723 Document reprocessing with consume = false
---
.../docs/modules/ROOT/pages/operate/webadmin.adoc | 20 ++++++++++++++++----
src/site/markdown/server/manage-webadmin.md | 6 ++++++
2 files changed, 22 insertions(+), 4 deletions(-)
diff --git a/server/apps/distributed-app/docs/modules/ROOT/pages/operate/webadmin.adoc b/server/apps/distributed-app/docs/modules/ROOT/pages/operate/webadmin.adoc
index 4ea5dd82d5..47ec81b266 100644
--- a/server/apps/distributed-app/docs/modules/ROOT/pages/operate/webadmin.adoc
+++ b/server/apps/distributed-app/docs/modules/ROOT/pages/operate/webadmin.adoc
@@ -3299,11 +3299,17 @@ For instance:
curl -XPATCH http://ip:port/mailRepositories/var%2Fmail%2Ferror%2F/mails?action=reprocess
....
-Additional query parameters are supported: - `queue` allows you to
+Additional query parameters are supported:
+
+- `queue` allows you to
target the mail queue you want to enqueue the mails in. Defaults to
-`spool`. - `processor` allows you to overwrite the state of the
+`spool`.
+- `processor` allows you to overwrite the state of the
reprocessing mails, and thus select the processors they will start their
processing in. Defaults to the `state` field of each processed email.
+- `consume` (boolean defaulting to `true`) whether the reprocessing should consume the mail in its originating mail repository. Passing
+this value to `false` allows non destructive reprocessing as you keep a copy of the email in the mail repository and can be valuable
+when debugging.
For instance:
@@ -3353,11 +3359,17 @@ For instance:
curl -XPATCH http://ip:port/mailRepositories/var%2Fmail%2Ferror%2F/mails/name1?action=reprocess
....
-Additional query parameters are supported: - `queue` allows you to
+Additional query parameters are supported:
+
+- `queue` allows you to
target the mail queue you want to enqueue the mails in. Defaults to
-`spool`. - `processor` allows you to overwrite the state of the
+`spool`.
+- `processor` allows you to overwrite the state of the
reprocessing mails, and thus select the processors they will start their
processing in. Defaults to the `state` field of each processed email.
+- `consume` (boolean defaulting to `true`) whether the reprocessing should consume the mail in its originating mail repository. Passing
+this value to `false` allows non destructive reprocessing as you keep a copy of the email in the mail repository and can be valuable
+when debugging.
While `processor` being an optional parameter, not specifying it will
result reprocessing the mails in their current state
diff --git a/src/site/markdown/server/manage-webadmin.md b/src/site/markdown/server/manage-webadmin.md
index 2f0f6f09cf..c7b2496d09 100644
--- a/src/site/markdown/server/manage-webadmin.md
+++ b/src/site/markdown/server/manage-webadmin.md
@@ -2985,6 +2985,9 @@ Additional query parameters are supported:
- `queue` allows you to target the mail queue you want to enqueue the mails in. Defaults to `spool`.
- `processor` allows you to overwrite the state of the reprocessing mails, and thus select the processors they will start their processing in.
Defaults to the `state` field of each processed email.
+ - `consume` (boolean defaulting to `true`) whether the reprocessing should consume the mail in its originating mail repository. Passing
+ this value to `false` allows non destructive reprocessing as you keep a copy of the email in the mail repository and can be valuable
+ when debugging.
For instance:
@@ -3034,6 +3037,9 @@ Additional query parameters are supported:
- `queue` allows you to target the mail queue you want to enqueue the mails in. Defaults to `spool`.
- `processor` allows you to overwrite the state of the reprocessing mails, and thus select the processors they will start their processing in.
Defaults to the `state` field of each processed email.
+ - `consume` (boolean defaulting to `true`) whether the reprocessing should consume the mail in its originating mail repository. Passing
+ this value to `false` allows non destructive reprocessing as you keep a copy of the email in the mail repository and can be valuable
+ when debugging.
While `processor` being an optional parameter, not specifying it will result reprocessing the mails in their current state ([see documentation about processors and state](https://james.apache.org/server/feature-mailetcontainer.html#Processors)).
Consequently, only few cases will give a different result, definitively storing them out of the mail repository.
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 07/07: task/task-distributed - fixing NullPointerException when executeTask
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch 3.7.x
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit ed7660c8b12cc644933cce21bef55d60489a4748
Author: Tung Van TRAN <vt...@linagora.com>
AuthorDate: Thu Jul 28 00:44:55 2022 +0700
task/task-distributed - fixing NullPointerException when executeTask
---
.../james/task/eventsourcing/distributed/RabbitMQWorkQueue.java | 7 +++++--
.../eventsourcing/distributed/DistributedTaskManagerTest.java | 8 +++++++-
2 files changed, 12 insertions(+), 3 deletions(-)
diff --git a/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQWorkQueue.java b/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQWorkQueue.java
index 60c9a3b5ae..1b3efd113f 100644
--- a/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQWorkQueue.java
+++ b/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQWorkQueue.java
@@ -151,8 +151,11 @@ public class RabbitMQWorkQueue implements WorkQueue {
}
private Mono<Task.Result> executeTask(AcknowledgableDelivery delivery) {
- return Mono.fromCallable(() -> TaskId.fromString(delivery.getProperties().getHeaders().get(TASK_ID).toString()))
- .flatMap(taskId -> deserialize(new String(delivery.getBody(), StandardCharsets.UTF_8), taskId)
+ return Mono.fromCallable(() -> delivery.getProperties().getHeaders())
+ .map(headers -> headers.get(TASK_ID))
+ .map(taskIdValue -> TaskId.fromString(taskIdValue.toString()))
+ .flatMap(taskId -> Mono.fromCallable(() -> new String(delivery.getBody(), StandardCharsets.UTF_8))
+ .flatMap(bodyValue -> deserialize(bodyValue, taskId))
.doOnNext(task -> delivery.ack())
.flatMap(task -> executeOnWorker(taskId, task)))
.onErrorResume(error -> {
diff --git a/server/task/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/DistributedTaskManagerTest.java b/server/task/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/DistributedTaskManagerTest.java
index de2a3caf4d..d220a5f451 100644
--- a/server/task/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/DistributedTaskManagerTest.java
+++ b/server/task/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/DistributedTaskManagerTest.java
@@ -46,6 +46,7 @@ import org.apache.james.backends.cassandra.components.CassandraModule;
import org.apache.james.backends.cassandra.init.CassandraZonedDateTimeModule;
import org.apache.james.backends.cassandra.versions.CassandraSchemaVersionModule;
import org.apache.james.backends.rabbitmq.RabbitMQExtension;
+import org.apache.james.backends.rabbitmq.RabbitMQManagementAPI;
import org.apache.james.backends.rabbitmq.ReceiverProvider;
import org.apache.james.eventsourcing.Event;
import org.apache.james.eventsourcing.EventSourcingSystem;
@@ -189,6 +190,8 @@ class DistributedTaskManagerTest implements TaskManagerContract {
@BeforeEach
void setUp(EventStore eventStore) {
+ memoryReferenceTaskStore = new MemoryReferenceTaskStore();
+ memoryReferenceWithCounterTaskStore = new MemoryReferenceWithCounterTaskStore();
CassandraCluster cassandra = CASSANDRA_CLUSTER.getCassandraCluster();
CassandraTaskExecutionDetailsProjectionDAO projectionDAO = new CassandraTaskExecutionDetailsProjectionDAO(cassandra.getConf(), cassandra.getTypesProvider(), JSON_TASK_ADDITIONAL_INFORMATION_SERIALIZER);
this.executionDetailsProjection = new CassandraTaskExecutionDetailsProjection(projectionDAO);
@@ -203,9 +206,12 @@ class DistributedTaskManagerTest implements TaskManagerContract {
}
@AfterEach
- void tearDown() {
+ void tearDown() throws Exception {
terminationSubscribers.forEach(RabbitMQTerminationSubscriber::close);
workQueueSupplier.stopWorkQueues();
+ RabbitMQManagementAPI managementAPI = rabbitMQExtension.managementAPI();
+ managementAPI.listQueues()
+ .forEach(queue -> managementAPI.deleteQueue("/", queue.getName()));
}
public EventSourcingTaskManager taskManager() {
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 02/07: JAMES-3723 Fix a typo Reprocessing/Reindexing
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch 3.7.x
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 61a1b756ac304ee04b452526bef0a200b8da097f
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Sat Mar 5 22:08:14 2022 +0700
JAMES-3723 Fix a typo Reprocessing/Reindexing
---
.../tools/indexer/ErrorRecoveryIndexationTask.java | 14 ++--
.../mailbox/tools/indexer/FullReindexingTask.java | 14 ++--
.../mailbox/tools/indexer/ReIndexerPerformer.java | 38 +++++-----
...ocessingContext.java => ReIndexingContext.java} | 4 +-
...tion.java => ReIndexingContextInformation.java} | 6 +-
...O.java => ReIndexingContextInformationDTO.java} | 58 +++++++-------
.../tools/indexer/SingleMailboxReindexingTask.java | 14 ++--
...lboxReindexingTaskAdditionalInformationDTO.java | 28 +++----
.../tools/indexer/SingleMessageReindexingTask.java | 2 +-
.../mailbox/tools/indexer/UserReindexingTask.java | 14 ++--
...UserReindexingTaskAdditionalInformationDTO.java | 24 +++---
.../tools/indexer/CassandraReIndexerImplTest.java | 12 +--
...rorRecoveryIndexationTaskSerializationTest.java | 14 ++--
.../FullReindexingTaskSerializationTest.java | 10 +--
.../mailbox/ReIndexingTaskSerializationModule.java | 6 +-
.../WebAdminReIndexingTaskSerializationModule.java | 6 +-
...> WebAdminIndexationContextInformationDTO.java} | 22 +++---
...lboxReindexingTaskAdditionalInformationDTO.java | 4 +-
...UserReindexingTaskAdditionalInformationDTO.java | 4 +-
...bAdminIndexationContextInformationDTOTest.java} | 20 ++---
...a => WebAdminSingleMailboxIndexingDTOTest.java} | 2 +-
...est.java => WebAdminUserReIndexingDTOTest.java} | 2 +-
.../james/webadmin/routes/MailboxesRoutesTest.java | 88 +++++++++++-----------
.../james/webadmin/routes/MessageRoutesTest.java | 2 +-
.../webadmin/routes/UserMailboxesRoutesTest.java | 3 +-
25 files changed, 205 insertions(+), 206 deletions(-)
diff --git a/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/ErrorRecoveryIndexationTask.java b/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/ErrorRecoveryIndexationTask.java
index 5f5c1e41ed..28c13a20c3 100644
--- a/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/ErrorRecoveryIndexationTask.java
+++ b/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/ErrorRecoveryIndexationTask.java
@@ -77,20 +77,20 @@ public class ErrorRecoveryIndexationTask implements Task {
}
private final ReIndexerPerformer reIndexerPerformer;
- private final ReprocessingContext reprocessingContext;
+ private final ReIndexingContext reIndexingContext;
private final ReIndexingExecutionFailures previousFailures;
private final RunningOptions runningOptions;
public ErrorRecoveryIndexationTask(ReIndexerPerformer reIndexerPerformer, ReIndexingExecutionFailures previousFailures, RunningOptions runningOptions) {
this.reIndexerPerformer = reIndexerPerformer;
this.previousFailures = previousFailures;
- this.reprocessingContext = new ReprocessingContext();
+ this.reIndexingContext = new ReIndexingContext();
this.runningOptions = runningOptions;
}
@Override
public Result run() {
- return reIndexerPerformer.reIndexErrors(reprocessingContext, previousFailures, runningOptions).block();
+ return reIndexerPerformer.reIndexErrors(reIndexingContext, previousFailures, runningOptions).block();
}
@Override
@@ -108,10 +108,10 @@ public class ErrorRecoveryIndexationTask implements Task {
@Override
public Optional<TaskExecutionDetails.AdditionalInformation> details() {
- return Optional.of(new ReprocessingContextInformationDTO.ReprocessingContextInformationForErrorRecoveryIndexationTask(
- reprocessingContext.successfullyReprocessedMailCount(),
- reprocessingContext.failedReprocessingMailCount(),
- reprocessingContext.failures(),
+ return Optional.of(new ReIndexingContextInformationDTO.ReIndexingContextInformationForErrorRecoveryIndexationTask(
+ reIndexingContext.successfullyReprocessedMailCount(),
+ reIndexingContext.failedReprocessingMailCount(),
+ reIndexingContext.failures(),
Clock.systemUTC().instant(),
runningOptions));
}
diff --git a/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/FullReindexingTask.java b/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/FullReindexingTask.java
index b99eb5facb..f974cb35b2 100644
--- a/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/FullReindexingTask.java
+++ b/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/FullReindexingTask.java
@@ -36,19 +36,19 @@ public class FullReindexingTask implements Task {
public static final TaskType FULL_RE_INDEXING = TaskType.of("full-reindexing");
private final ReIndexerPerformer reIndexerPerformer;
- private final ReprocessingContext reprocessingContext;
+ private final ReIndexingContext reIndexingContext;
private final RunningOptions runningOptions;
@Inject
public FullReindexingTask(ReIndexerPerformer reIndexerPerformer, RunningOptions runningOptions) {
this.reIndexerPerformer = reIndexerPerformer;
- this.reprocessingContext = new ReprocessingContext();
+ this.reIndexingContext = new ReIndexingContext();
this.runningOptions = runningOptions;
}
@Override
public Result run() {
- return reIndexerPerformer.reIndexAllMessages(reprocessingContext, runningOptions)
+ return reIndexerPerformer.reIndexAllMessages(reIndexingContext, runningOptions)
.onErrorResume(e -> Mono.just(Result.PARTIAL))
.block();
}
@@ -64,10 +64,10 @@ public class FullReindexingTask implements Task {
@Override
public Optional<TaskExecutionDetails.AdditionalInformation> details() {
- return Optional.of(new ReprocessingContextInformationDTO.ReprocessingContextInformationForFullReindexingTask(
- reprocessingContext.successfullyReprocessedMailCount(),
- reprocessingContext.failedReprocessingMailCount(),
- reprocessingContext.failures(),
+ return Optional.of(new ReIndexingContextInformationDTO.ReIndexingContextInformationForFullReindexingTask(
+ reIndexingContext.successfullyReprocessedMailCount(),
+ reIndexingContext.failedReprocessingMailCount(),
+ reIndexingContext.failures(),
Clock.systemUTC().instant(),
runningOptions));
}
diff --git a/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/ReIndexerPerformer.java b/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/ReIndexerPerformer.java
index 5f32f55d7b..9844ab7fb4 100644
--- a/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/ReIndexerPerformer.java
+++ b/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/ReIndexerPerformer.java
@@ -84,7 +84,7 @@ public class ReIndexerPerformer {
}
private interface Failure {
- void recordFailure(ReprocessingContext context);
+ void recordFailure(ReIndexingContext context);
}
private static class MailboxFailure implements Failure {
@@ -99,7 +99,7 @@ public class ReIndexerPerformer {
}
@Override
- public void recordFailure(ReprocessingContext context) {
+ public void recordFailure(ReIndexingContext context) {
context.recordMailboxFailure(mailboxId);
}
}
@@ -122,7 +122,7 @@ public class ReIndexerPerformer {
}
@Override
- public void recordFailure(ReprocessingContext context) {
+ public void recordFailure(ReIndexingContext context) {
context.recordFailureDetailsForMessage(mailboxId, uid);
}
}
@@ -146,28 +146,28 @@ public class ReIndexerPerformer {
this.mailboxSessionMapperFactory = mailboxSessionMapperFactory;
}
- Mono<Result> reIndexAllMessages(ReprocessingContext reprocessingContext, RunningOptions runningOptions) {
+ Mono<Result> reIndexAllMessages(ReIndexingContext reIndexingContext, RunningOptions runningOptions) {
MailboxSession mailboxSession = mailboxManager.createSystemSession(RE_INDEXER_PERFORMER_USER);
LOGGER.info("Starting a full reindex");
Flux<Either<Failure, ReIndexingEntry>> entriesToIndex = mailboxSessionMapperFactory.getMailboxMapper(mailboxSession).list()
.flatMap(mailbox -> reIndexingEntriesForMailbox(mailbox, mailboxSession, runningOptions), MAILBOX_CONCURRENCY);
- return reIndexMessages(entriesToIndex, runningOptions, reprocessingContext)
+ return reIndexMessages(entriesToIndex, runningOptions, reIndexingContext)
.doFinally(any -> LOGGER.info("Full reindex finished"));
}
- Mono<Result> reIndexSingleMailbox(MailboxId mailboxId, ReprocessingContext reprocessingContext, RunningOptions runningOptions) {
+ Mono<Result> reIndexSingleMailbox(MailboxId mailboxId, ReIndexingContext reIndexingContext, RunningOptions runningOptions) {
MailboxSession mailboxSession = mailboxManager.createSystemSession(RE_INDEXER_PERFORMER_USER);
Flux<Either<Failure, ReIndexingEntry>> entriesToIndex = mailboxSessionMapperFactory.getMailboxMapper(mailboxSession)
.findMailboxById(mailboxId)
.flatMapMany(mailbox -> reIndexingEntriesForMailbox(mailbox, mailboxSession, runningOptions));
- return reIndexMessages(entriesToIndex, runningOptions, reprocessingContext);
+ return reIndexMessages(entriesToIndex, runningOptions, reIndexingContext);
}
- Mono<Result> reIndexUserMailboxes(Username username, ReprocessingContext reprocessingContext, RunningOptions runningOptions) {
+ Mono<Result> reIndexUserMailboxes(Username username, ReIndexingContext reIndexingContext, RunningOptions runningOptions) {
MailboxSession mailboxSession = mailboxManager.createSystemSession(username);
MailboxMapper mailboxMapper = mailboxSessionMapperFactory.getMailboxMapper(mailboxSession);
LOGGER.info("Starting a reindex for user {}", username.asString());
@@ -178,7 +178,7 @@ public class ReIndexerPerformer {
Flux<Either<Failure, ReIndexingEntry>> entriesToIndex = mailboxMapper.findMailboxWithPathLike(mailboxQuery.asUserBound())
.flatMap(mailbox -> reIndexingEntriesForMailbox(mailbox, mailboxSession, runningOptions), MAILBOX_CONCURRENCY);
- return reIndexMessages(entriesToIndex, runningOptions, reprocessingContext)
+ return reIndexMessages(entriesToIndex, runningOptions, reIndexingContext)
.doFinally(any -> LOGGER.info("User {} reindex finished", username.asString()));
} catch (Exception e) {
LOGGER.error("Error fetching mailboxes for user: {}", username.asString());
@@ -186,7 +186,7 @@ public class ReIndexerPerformer {
}
}
- Mono<Result> reIndexSingleMessage(MailboxId mailboxId, MessageUid uid, ReprocessingContext reprocessingContext) {
+ Mono<Result> reIndexSingleMessage(MailboxId mailboxId, MessageUid uid, ReIndexingContext reIndexingContext) {
MailboxSession mailboxSession = mailboxManager.createSystemSession(RE_INDEXER_PERFORMER_USER);
return mailboxSessionMapperFactory.getMailboxMapper(mailboxSession)
@@ -211,7 +211,7 @@ public class ReIndexerPerformer {
});
}
- Mono<Result> reIndexErrors(ReprocessingContext reprocessingContext, ReIndexingExecutionFailures previousReIndexingFailures, RunningOptions runningOptions) {
+ Mono<Result> reIndexErrors(ReIndexingContext reIndexingContext, ReIndexingExecutionFailures previousReIndexingFailures, RunningOptions runningOptions) {
MailboxSession mailboxSession = mailboxManager.createSystemSession(RE_INDEXER_PERFORMER_USER);
MailboxMapper mapper = mailboxSessionMapperFactory.getMailboxMapper(mailboxSession);
@@ -226,7 +226,7 @@ public class ReIndexerPerformer {
return Mono.just(Either.left(new MailboxFailure(mailboxId)));
}), MAILBOX_CONCURRENCY));
- return reIndexMessages(entriesToIndex, runningOptions, reprocessingContext);
+ return reIndexMessages(entriesToIndex, runningOptions, reIndexingContext);
}
private Mono<Result> reIndex(MailboxMessage mailboxMessage, MailboxSession session) {
@@ -281,29 +281,29 @@ public class ReIndexerPerformer {
}
}
- private Mono<Task.Result> reIndexMessages(Flux<Either<Failure, ReIndexingEntry>> entriesToIndex, RunningOptions runningOptions, ReprocessingContext reprocessingContext) {
+ private Mono<Task.Result> reIndexMessages(Flux<Either<Failure, ReIndexingEntry>> entriesToIndex, RunningOptions runningOptions, ReIndexingContext reIndexingContext) {
return entriesToIndex.transform(
ReactorUtils.<Either<Failure, ReIndexingEntry>, Task.Result>throttle()
.elements(runningOptions.getMessagesPerSecond())
.per(Duration.ofSeconds(1))
- .forOperation(entry -> reIndex(entry, reprocessingContext, runningOptions)))
+ .forOperation(entry -> reIndex(entry, reIndexingContext, runningOptions)))
.reduce(Task::combine)
.switchIfEmpty(Mono.just(Result.COMPLETED));
}
- private Mono<Task.Result> reIndex(Either<Failure, ReIndexingEntry> failureOrEntry, ReprocessingContext reprocessingContext, RunningOptions runningOptions) {
+ private Mono<Task.Result> reIndex(Either<Failure, ReIndexingEntry> failureOrEntry, ReIndexingContext reIndexingContext, RunningOptions runningOptions) {
return toMono(failureOrEntry.map(entry -> reIndex(entry, runningOptions)))
.map(this::flatten)
- .map(failureOrTaskResult -> recordIndexingResult(failureOrTaskResult, reprocessingContext));
+ .map(failureOrTaskResult -> recordIndexingResult(failureOrTaskResult, reIndexingContext));
}
- private Result recordIndexingResult(Either<Failure, Result> failureOrTaskResult, ReprocessingContext reprocessingContext) {
+ private Result recordIndexingResult(Either<Failure, Result> failureOrTaskResult, ReIndexingContext reIndexingContext) {
return failureOrTaskResult.fold(
failure -> {
- failure.recordFailure(reprocessingContext);
+ failure.recordFailure(reIndexingContext);
return Result.PARTIAL;
},
- result -> result.onComplete(reprocessingContext::recordSuccess));
+ result -> result.onComplete(reIndexingContext::recordSuccess));
}
private Mono<Either<Failure, Result>> reIndex(ReIndexingEntry entry, RunningOptions runningOptions) {
diff --git a/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/ReprocessingContext.java b/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/ReIndexingContext.java
similarity index 98%
rename from mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/ReprocessingContext.java
rename to mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/ReIndexingContext.java
index bc4facb1eb..5a8b0a7576 100644
--- a/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/ReprocessingContext.java
+++ b/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/ReIndexingContext.java
@@ -28,13 +28,13 @@ import org.apache.james.mailbox.model.MailboxId;
import com.google.common.collect.ImmutableList;
-class ReprocessingContext {
+class ReIndexingContext {
private final AtomicInteger successfullyReprocessedMails;
private final AtomicInteger failedReprocessingMails;
private final ConcurrentLinkedDeque<ReIndexingExecutionFailures.ReIndexingFailure> failures;
private final ConcurrentLinkedDeque<MailboxId> mailboxFailures;
- ReprocessingContext() {
+ ReIndexingContext() {
failedReprocessingMails = new AtomicInteger(0);
successfullyReprocessedMails = new AtomicInteger(0);
failures = new ConcurrentLinkedDeque<>();
diff --git a/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/ReprocessingContextInformation.java b/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/ReIndexingContextInformation.java
similarity index 87%
rename from mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/ReprocessingContextInformation.java
rename to mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/ReIndexingContextInformation.java
index 0dc02b0ea9..41aace0305 100644
--- a/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/ReprocessingContextInformation.java
+++ b/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/ReIndexingContextInformation.java
@@ -26,7 +26,7 @@ import org.apache.james.mailbox.indexer.ReIndexer.RunningOptions;
import org.apache.james.mailbox.indexer.ReIndexingExecutionFailures;
import org.apache.james.task.TaskExecutionDetails;
-public class ReprocessingContextInformation implements TaskExecutionDetails.AdditionalInformation, IndexingDetailInformation {
+public class ReIndexingContextInformation implements TaskExecutionDetails.AdditionalInformation, IndexingDetailInformation {
private final int successfullyReprocessedMailCount;
private final int failedReprocessedMailCount;
@@ -34,8 +34,8 @@ public class ReprocessingContextInformation implements TaskExecutionDetails.Addi
private final Instant timestamp;
private final RunningOptions runningOptions;
- ReprocessingContextInformation(int successfullyReprocessedMailCount, int failedReprocessedMailCount,
- ReIndexingExecutionFailures failures, Instant timestamp, RunningOptions runningOptions) {
+ ReIndexingContextInformation(int successfullyReprocessedMailCount, int failedReprocessedMailCount,
+ ReIndexingExecutionFailures failures, Instant timestamp, RunningOptions runningOptions) {
this.successfullyReprocessedMailCount = successfullyReprocessedMailCount;
this.failedReprocessedMailCount = failedReprocessedMailCount;
this.failures = failures;
diff --git a/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/ReprocessingContextInformationDTO.java b/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/ReIndexingContextInformationDTO.java
similarity index 80%
rename from mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/ReprocessingContextInformationDTO.java
rename to mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/ReIndexingContextInformationDTO.java
index d22ea8396d..c4586301c7 100644
--- a/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/ReprocessingContextInformationDTO.java
+++ b/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/ReIndexingContextInformationDTO.java
@@ -40,7 +40,7 @@ import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableListMultimap;
-public class ReprocessingContextInformationDTO implements AdditionalInformationDTO {
+public class ReIndexingContextInformationDTO implements AdditionalInformationDTO {
public static class ReindexingFailureDTO {
@@ -61,9 +61,9 @@ public class ReprocessingContextInformationDTO implements AdditionalInformationD
}
}
- public static class ReprocessingContextInformationForErrorRecoveryIndexationTask extends ReprocessingContextInformation {
+ public static class ReIndexingContextInformationForErrorRecoveryIndexationTask extends ReIndexingContextInformation {
- public static class DTO extends ReprocessingContextInformationDTO {
+ public static class DTO extends ReIndexingContextInformationDTO {
DTO(@JsonProperty("type") String type,
@JsonProperty("successfullyReprocessedMailCount") int successfullyReprocessedMailCount,
@@ -77,10 +77,10 @@ public class ReprocessingContextInformationDTO implements AdditionalInformationD
}
}
- public static AdditionalInformationDTOModule<ReprocessingContextInformationForErrorRecoveryIndexationTask, DTO> module(MailboxId.Factory mailboxIdFactory) {
- return DTOModule.forDomainObject(ReprocessingContextInformationForErrorRecoveryIndexationTask.class)
+ public static AdditionalInformationDTOModule<ReIndexingContextInformationForErrorRecoveryIndexationTask, DTO> module(MailboxId.Factory mailboxIdFactory) {
+ return DTOModule.forDomainObject(ReIndexingContextInformationForErrorRecoveryIndexationTask.class)
.convertToDTO(DTO.class)
- .toDomainObjectConverter(dto -> new ReprocessingContextInformationForErrorRecoveryIndexationTask(
+ .toDomainObjectConverter(dto -> new ReIndexingContextInformationForErrorRecoveryIndexationTask(
dto.successfullyReprocessedMailCount,
dto.failedReprocessedMailCount,
deserializeFailures(mailboxIdFactory, dto.messageFailures, dto.mailboxFailures.orElse(ImmutableList.of())),
@@ -103,18 +103,18 @@ public class ReprocessingContextInformationDTO implements AdditionalInformationD
}
@VisibleForTesting
- public ReprocessingContextInformationForErrorRecoveryIndexationTask(int successfullyReprocessedMailCount,
- int failedReprocessedMailCount,
- ReIndexingExecutionFailures failures,
- Instant timestamp,
- RunningOptions runningOptions) {
+ public ReIndexingContextInformationForErrorRecoveryIndexationTask(int successfullyReprocessedMailCount,
+ int failedReprocessedMailCount,
+ ReIndexingExecutionFailures failures,
+ Instant timestamp,
+ RunningOptions runningOptions) {
super(successfullyReprocessedMailCount, failedReprocessedMailCount, failures, timestamp, runningOptions);
}
}
- public static class ReprocessingContextInformationForFullReindexingTask extends ReprocessingContextInformation {
+ public static class ReIndexingContextInformationForFullReindexingTask extends ReIndexingContextInformation {
- public static class DTO extends ReprocessingContextInformationDTO {
+ public static class DTO extends ReIndexingContextInformationDTO {
DTO(@JsonProperty("type") String type,
@JsonProperty("successfullyReprocessedMailCount") int successfullyReprocessedMailCount,
@@ -128,10 +128,10 @@ public class ReprocessingContextInformationDTO implements AdditionalInformationD
}
}
- public static AdditionalInformationDTOModule<ReprocessingContextInformationForFullReindexingTask, DTO> module(MailboxId.Factory mailboxIdFactory) {
- return DTOModule.forDomainObject(ReprocessingContextInformationForFullReindexingTask.class)
+ public static AdditionalInformationDTOModule<ReIndexingContextInformationForFullReindexingTask, DTO> module(MailboxId.Factory mailboxIdFactory) {
+ return DTOModule.forDomainObject(ReIndexingContextInformationForFullReindexingTask.class)
.convertToDTO(DTO.class)
- .toDomainObjectConverter(dto -> new ReprocessingContextInformationForFullReindexingTask(
+ .toDomainObjectConverter(dto -> new ReIndexingContextInformationForFullReindexingTask(
dto.successfullyReprocessedMailCount,
dto.failedReprocessedMailCount,
deserializeFailures(mailboxIdFactory, dto.messageFailures, dto.mailboxFailures.orElse(ImmutableList.of())),
@@ -153,11 +153,11 @@ public class ReprocessingContextInformationDTO implements AdditionalInformationD
}
@VisibleForTesting
- public ReprocessingContextInformationForFullReindexingTask(int successfullyReprocessedMailCount,
- int failedReprocessedMailCount,
- ReIndexingExecutionFailures failures,
- Instant timestamp,
- RunningOptions runningOptions) {
+ public ReIndexingContextInformationForFullReindexingTask(int successfullyReprocessedMailCount,
+ int failedReprocessedMailCount,
+ ReIndexingExecutionFailures failures,
+ Instant timestamp,
+ RunningOptions runningOptions) {
super(successfullyReprocessedMailCount, failedReprocessedMailCount, failures, timestamp, runningOptions);
}
}
@@ -223,14 +223,14 @@ public class ReprocessingContextInformationDTO implements AdditionalInformationD
protected final Instant timestamp;
protected final Optional<RunningOptionsDTO> runningOptions;
- ReprocessingContextInformationDTO(@JsonProperty("type") String type,
- @JsonProperty("successfullyReprocessedMailCount") int successfullyReprocessedMailCount,
- @JsonProperty("failedReprocessedMailCount") int failedReprocessedMailCount,
- @JsonProperty("failures") Optional<List<ReindexingFailureDTO>> failures,
- @JsonProperty("messageFailures") Optional<List<ReindexingFailureDTO>> messageFailures,
- @JsonProperty("mailboxFailures") Optional<List<String>> mailboxFailures,
- @JsonProperty("timestamp") Instant timestamp,
- @JsonProperty("runningOptions") Optional<RunningOptionsDTO> runningOptions) {
+ ReIndexingContextInformationDTO(@JsonProperty("type") String type,
+ @JsonProperty("successfullyReprocessedMailCount") int successfullyReprocessedMailCount,
+ @JsonProperty("failedReprocessedMailCount") int failedReprocessedMailCount,
+ @JsonProperty("failures") Optional<List<ReindexingFailureDTO>> failures,
+ @JsonProperty("messageFailures") Optional<List<ReindexingFailureDTO>> messageFailures,
+ @JsonProperty("mailboxFailures") Optional<List<String>> mailboxFailures,
+ @JsonProperty("timestamp") Instant timestamp,
+ @JsonProperty("runningOptions") Optional<RunningOptionsDTO> runningOptions) {
this.type = type;
this.successfullyReprocessedMailCount = successfullyReprocessedMailCount;
this.failedReprocessedMailCount = failedReprocessedMailCount;
diff --git a/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/SingleMailboxReindexingTask.java b/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/SingleMailboxReindexingTask.java
index 668b9341bc..3285d5af09 100644
--- a/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/SingleMailboxReindexingTask.java
+++ b/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/SingleMailboxReindexingTask.java
@@ -38,7 +38,7 @@ public class SingleMailboxReindexingTask implements Task {
public static final TaskType TYPE = TaskType.of("mailbox-reindexing");
- public static class AdditionalInformation extends ReprocessingContextInformation {
+ public static class AdditionalInformation extends ReIndexingContextInformation {
private final MailboxId mailboxId;
@VisibleForTesting
@@ -76,21 +76,21 @@ public class SingleMailboxReindexingTask implements Task {
private final ReIndexerPerformer reIndexerPerformer;
private final MailboxId mailboxId;
- private final ReprocessingContext reprocessingContext;
+ private final ReIndexingContext reIndexingContext;
private final RunningOptions runningOptions;
@Inject
public SingleMailboxReindexingTask(ReIndexerPerformer reIndexerPerformer, MailboxId mailboxId, RunningOptions runningOptions) {
this.reIndexerPerformer = reIndexerPerformer;
this.mailboxId = mailboxId;
- this.reprocessingContext = new ReprocessingContext();
+ this.reIndexingContext = new ReIndexingContext();
this.runningOptions = runningOptions;
}
@Override
public Result run() {
try {
- return reIndexerPerformer.reIndexSingleMailbox(mailboxId, reprocessingContext, runningOptions)
+ return reIndexerPerformer.reIndexSingleMailbox(mailboxId, reIndexingContext, runningOptions)
.block();
} catch (Exception e) {
return Result.PARTIAL;
@@ -115,9 +115,9 @@ public class SingleMailboxReindexingTask implements Task {
return Optional.of(
new SingleMailboxReindexingTask.AdditionalInformation(
mailboxId,
- reprocessingContext.successfullyReprocessedMailCount(),
- reprocessingContext.failedReprocessingMailCount(),
- reprocessingContext.failures(),
+ reIndexingContext.successfullyReprocessedMailCount(),
+ reIndexingContext.failedReprocessingMailCount(),
+ reIndexingContext.failures(),
Clock.systemUTC().instant(),
runningOptions)
);
diff --git a/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/SingleMailboxReindexingTaskAdditionalInformationDTO.java b/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/SingleMailboxReindexingTaskAdditionalInformationDTO.java
index faec1f48a4..d4ef440741 100644
--- a/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/SingleMailboxReindexingTaskAdditionalInformationDTO.java
+++ b/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/SingleMailboxReindexingTaskAdditionalInformationDTO.java
@@ -40,7 +40,7 @@ public class SingleMailboxReindexingTaskAdditionalInformationDTO implements Addi
.toDomainObjectConverter(dto -> new SingleMailboxReindexingTask.AdditionalInformation(factory.fromString(dto.getMailboxId()),
dto.getSuccessfullyReprocessedMailCount(),
dto.getFailedReprocessedMailCount(),
- ReprocessingContextInformationDTO.deserializeFailures(factory, dto.getMessageFailures(), dto.getMailboxFailures().orElse(ImmutableList.of())),
+ ReIndexingContextInformationDTO.deserializeFailures(factory, dto.getMessageFailures(), dto.getMailboxFailures().orElse(ImmutableList.of())),
dto.getTimestamp(),
dto.getRunningOptions()
.map(RunningOptionsDTO::toDomainObject)
@@ -52,7 +52,7 @@ public class SingleMailboxReindexingTaskAdditionalInformationDTO implements Addi
details.getSuccessfullyReprocessedMailCount(),
details.getFailedReprocessedMailCount(),
Optional.empty(),
- Optional.of(ReprocessingContextInformationDTO.serializeFailures(details.failures())),
+ Optional.of(ReIndexingContextInformationDTO.serializeFailures(details.failures())),
Optional.of(details.failures().mailboxFailures().stream().map(MailboxId::serialize).collect(ImmutableList.toImmutableList())),
details.timestamp(),
Optional.of(RunningOptionsDTO.toDTO(details.getRunningOptions()))
@@ -61,7 +61,7 @@ public class SingleMailboxReindexingTaskAdditionalInformationDTO implements Addi
.withFactory(AdditionalInformationDTOModule::new);
}
- private final ReprocessingContextInformationDTO reprocessingContextInformationDTO;
+ private final ReIndexingContextInformationDTO reIndexingContextInformationDTO;
private final String mailboxId;
@JsonCreator
@@ -69,13 +69,13 @@ public class SingleMailboxReindexingTaskAdditionalInformationDTO implements Addi
@JsonProperty("mailboxId") String mailboxId,
@JsonProperty("successfullyReprocessedMailCount") int successfullyReprocessedMailCount,
@JsonProperty("failedReprocessedMailCount") int failedReprocessedMailCount,
- @JsonProperty("failures") Optional<List<ReprocessingContextInformationDTO.ReindexingFailureDTO>> failures,
- @JsonProperty("messageFailures") Optional<List<ReprocessingContextInformationDTO.ReindexingFailureDTO>> messageFailures,
+ @JsonProperty("failures") Optional<List<ReIndexingContextInformationDTO.ReindexingFailureDTO>> failures,
+ @JsonProperty("messageFailures") Optional<List<ReIndexingContextInformationDTO.ReindexingFailureDTO>> messageFailures,
@JsonProperty("mailboxFailures") Optional<List<String>> mailboxFailures,
@JsonProperty("timestamp") Instant timestamp,
@JsonProperty("runningOptions") Optional<RunningOptionsDTO> runningOptions) {
this.mailboxId = mailboxId;
- this.reprocessingContextInformationDTO = new ReprocessingContextInformationDTO(
+ this.reIndexingContextInformationDTO = new ReIndexingContextInformationDTO(
type,
successfullyReprocessedMailCount,
failedReprocessedMailCount,
@@ -88,11 +88,11 @@ public class SingleMailboxReindexingTaskAdditionalInformationDTO implements Addi
@Override
public String getType() {
- return reprocessingContextInformationDTO.getType();
+ return reIndexingContextInformationDTO.getType();
}
public Instant getTimestamp() {
- return reprocessingContextInformationDTO.getTimestamp();
+ return reIndexingContextInformationDTO.getTimestamp();
}
public String getMailboxId() {
@@ -100,22 +100,22 @@ public class SingleMailboxReindexingTaskAdditionalInformationDTO implements Addi
}
public int getSuccessfullyReprocessedMailCount() {
- return reprocessingContextInformationDTO.getSuccessfullyReprocessedMailCount();
+ return reIndexingContextInformationDTO.getSuccessfullyReprocessedMailCount();
}
public int getFailedReprocessedMailCount() {
- return reprocessingContextInformationDTO.getFailedReprocessedMailCount();
+ return reIndexingContextInformationDTO.getFailedReprocessedMailCount();
}
- public List<ReprocessingContextInformationDTO.ReindexingFailureDTO> getMessageFailures() {
- return reprocessingContextInformationDTO.getMessageFailures();
+ public List<ReIndexingContextInformationDTO.ReindexingFailureDTO> getMessageFailures() {
+ return reIndexingContextInformationDTO.getMessageFailures();
}
public Optional<List<String>> getMailboxFailures() {
- return reprocessingContextInformationDTO.getMailboxFailures();
+ return reIndexingContextInformationDTO.getMailboxFailures();
}
public Optional<RunningOptionsDTO> getRunningOptions() {
- return reprocessingContextInformationDTO.getRunningOptions();
+ return reIndexingContextInformationDTO.getRunningOptions();
}
}
diff --git a/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/SingleMessageReindexingTask.java b/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/SingleMessageReindexingTask.java
index a5a787dfd7..5264100942 100644
--- a/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/SingleMessageReindexingTask.java
+++ b/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/SingleMessageReindexingTask.java
@@ -96,7 +96,7 @@ public class SingleMessageReindexingTask implements Task {
@Override
public Result run() {
- return reIndexerPerformer.reIndexSingleMessage(mailboxId, uid, new ReprocessingContext())
+ return reIndexerPerformer.reIndexSingleMessage(mailboxId, uid, new ReIndexingContext())
.onErrorResume(e -> {
LOGGER.warn("Error encountered while reindexing {} : {}", mailboxId, uid, e);
return Mono.just(Result.PARTIAL);
diff --git a/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/UserReindexingTask.java b/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/UserReindexingTask.java
index 5fe450bbf9..f7b82c9454 100644
--- a/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/UserReindexingTask.java
+++ b/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/UserReindexingTask.java
@@ -40,7 +40,7 @@ public class UserReindexingTask implements Task {
public static final TaskType USER_RE_INDEXING = TaskType.of("user-reindexing");
- public static class AdditionalInformation extends ReprocessingContextInformation {
+ public static class AdditionalInformation extends ReIndexingContextInformation {
private final Username username;
@VisibleForTesting
@@ -59,14 +59,14 @@ public class UserReindexingTask implements Task {
private final ReIndexerPerformer reIndexerPerformer;
private final Username username;
- private final ReprocessingContext reprocessingContext;
+ private final ReIndexingContext reIndexingContext;
private final RunningOptions runningOptions;
@Inject
public UserReindexingTask(ReIndexerPerformer reIndexerPerformer, Username username, RunningOptions runningOptions) {
this.reIndexerPerformer = reIndexerPerformer;
this.username = username;
- this.reprocessingContext = new ReprocessingContext();
+ this.reIndexingContext = new ReIndexingContext();
this.runningOptions = runningOptions;
}
@@ -90,7 +90,7 @@ public class UserReindexingTask implements Task {
@Override
public Result run() {
- return reIndexerPerformer.reIndexUserMailboxes(username, reprocessingContext, runningOptions)
+ return reIndexerPerformer.reIndexUserMailboxes(username, reIndexingContext, runningOptions)
.onErrorResume(e -> Mono.just(Result.PARTIAL))
.block();
}
@@ -111,9 +111,9 @@ public class UserReindexingTask implements Task {
@Override
public Optional<TaskExecutionDetails.AdditionalInformation> details() {
return Optional.of(new UserReindexingTask.AdditionalInformation(username,
- reprocessingContext.successfullyReprocessedMailCount(),
- reprocessingContext.failedReprocessingMailCount(),
- reprocessingContext.failures(),
+ reIndexingContext.successfullyReprocessedMailCount(),
+ reIndexingContext.failedReprocessingMailCount(),
+ reIndexingContext.failures(),
Clock.systemUTC().instant(),
runningOptions)
);
diff --git a/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/UserReindexingTaskAdditionalInformationDTO.java b/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/UserReindexingTaskAdditionalInformationDTO.java
index ce952b6a1e..7817bef369 100644
--- a/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/UserReindexingTaskAdditionalInformationDTO.java
+++ b/mailbox/tools/indexer/src/main/java/org/apache/mailbox/tools/indexer/UserReindexingTaskAdditionalInformationDTO.java
@@ -28,7 +28,7 @@ import org.apache.james.mailbox.indexer.ReIndexer;
import org.apache.james.mailbox.model.MailboxId;
import org.apache.james.server.task.json.dto.AdditionalInformationDTO;
import org.apache.james.server.task.json.dto.AdditionalInformationDTOModule;
-import org.apache.mailbox.tools.indexer.ReprocessingContextInformationDTO.ReindexingFailureDTO;
+import org.apache.mailbox.tools.indexer.ReIndexingContextInformationDTO.ReindexingFailureDTO;
import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
@@ -42,7 +42,7 @@ public class UserReindexingTaskAdditionalInformationDTO implements AdditionalInf
.toDomainObjectConverter(dto -> new UserReindexingTask.AdditionalInformation(Username.of(dto.getUser()),
dto.getSuccessfullyReprocessedMailCount(),
dto.getFailedReprocessedMailCount(),
- ReprocessingContextInformationDTO.deserializeFailures(factory, dto.getMessageFailures(), dto.getMailboxFailures().orElse(ImmutableList.of())),
+ ReIndexingContextInformationDTO.deserializeFailures(factory, dto.getMessageFailures(), dto.getMailboxFailures().orElse(ImmutableList.of())),
dto.getTimestamp(),
dto.getRunningOptions()
.map(RunningOptionsDTO::toDomainObject)
@@ -54,7 +54,7 @@ public class UserReindexingTaskAdditionalInformationDTO implements AdditionalInf
details.getSuccessfullyReprocessedMailCount(),
details.getFailedReprocessedMailCount(),
Optional.empty(),
- Optional.of(ReprocessingContextInformationDTO.serializeFailures(details.failures())),
+ Optional.of(ReIndexingContextInformationDTO.serializeFailures(details.failures())),
Optional.of(details.failures().mailboxFailures().stream().map(MailboxId::serialize).collect(ImmutableList.toImmutableList())),
details.timestamp(),
Optional.of(RunningOptionsDTO.toDTO(details.getRunningOptions()))))
@@ -62,7 +62,7 @@ public class UserReindexingTaskAdditionalInformationDTO implements AdditionalInf
.withFactory(AdditionalInformationDTOModule::new);
}
- private final ReprocessingContextInformationDTO reprocessingContextInformationDTO;
+ private final ReIndexingContextInformationDTO reIndexingContextInformationDTO;
private final String user;
@JsonCreator
@@ -77,7 +77,7 @@ public class UserReindexingTaskAdditionalInformationDTO implements AdditionalInf
@JsonProperty("runningOptions") Optional<RunningOptionsDTO> runningOptions
) {
this.user = user;
- this.reprocessingContextInformationDTO = new ReprocessingContextInformationDTO(type,
+ this.reIndexingContextInformationDTO = new ReIndexingContextInformationDTO(type,
successfullyReprocessedMailCount,
failedReprocessedMailCount,
failures,
@@ -89,11 +89,11 @@ public class UserReindexingTaskAdditionalInformationDTO implements AdditionalInf
@Override
public String getType() {
- return reprocessingContextInformationDTO.getType();
+ return reIndexingContextInformationDTO.getType();
}
public Instant getTimestamp() {
- return reprocessingContextInformationDTO.getTimestamp();
+ return reIndexingContextInformationDTO.getTimestamp();
}
public String getUser() {
@@ -101,22 +101,22 @@ public class UserReindexingTaskAdditionalInformationDTO implements AdditionalInf
}
public int getSuccessfullyReprocessedMailCount() {
- return reprocessingContextInformationDTO.getSuccessfullyReprocessedMailCount();
+ return reIndexingContextInformationDTO.getSuccessfullyReprocessedMailCount();
}
public int getFailedReprocessedMailCount() {
- return reprocessingContextInformationDTO.getFailedReprocessedMailCount();
+ return reIndexingContextInformationDTO.getFailedReprocessedMailCount();
}
public List<ReindexingFailureDTO> getMessageFailures() {
- return reprocessingContextInformationDTO.getMessageFailures();
+ return reIndexingContextInformationDTO.getMessageFailures();
}
public Optional<List<String>> getMailboxFailures() {
- return reprocessingContextInformationDTO.getMailboxFailures();
+ return reIndexingContextInformationDTO.getMailboxFailures();
}
public Optional<RunningOptionsDTO> getRunningOptions() {
- return reprocessingContextInformationDTO.getRunningOptions();
+ return reIndexingContextInformationDTO.getRunningOptions();
}
}
diff --git a/mailbox/tools/indexer/src/test/java/org/apache/mailbox/tools/indexer/CassandraReIndexerImplTest.java b/mailbox/tools/indexer/src/test/java/org/apache/mailbox/tools/indexer/CassandraReIndexerImplTest.java
index 9ccdd1f29e..0ddf6df064 100644
--- a/mailbox/tools/indexer/src/test/java/org/apache/mailbox/tools/indexer/CassandraReIndexerImplTest.java
+++ b/mailbox/tools/indexer/src/test/java/org/apache/mailbox/tools/indexer/CassandraReIndexerImplTest.java
@@ -53,8 +53,8 @@ import org.apache.james.mailbox.store.mail.model.MailboxMessage;
import org.apache.james.mailbox.store.search.ListeningMessageSearchIndex;
import org.apache.james.task.Task;
import org.apache.james.util.concurrency.ConcurrentTestRunner;
-import org.apache.mailbox.tools.indexer.ReprocessingContextInformationDTO.ReprocessingContextInformationForErrorRecoveryIndexationTask;
-import org.apache.mailbox.tools.indexer.ReprocessingContextInformationDTO.ReprocessingContextInformationForFullReindexingTask;
+import org.apache.mailbox.tools.indexer.ReIndexingContextInformationDTO.ReIndexingContextInformationForErrorRecoveryIndexationTask;
+import org.apache.mailbox.tools.indexer.ReIndexingContextInformationDTO.ReIndexingContextInformationForFullReindexingTask;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Nested;
import org.junit.jupiter.api.Test;
@@ -154,7 +154,7 @@ public class CassandraReIndexerImplTest {
Task task = reIndexer.reIndex(ReIndexer.RunningOptions.DEFAULT);
task.run();
- ReprocessingContextInformationForFullReindexingTask information = (ReprocessingContextInformationForFullReindexingTask) task.details().get();
+ ReIndexingContextInformationForFullReindexingTask information = (ReIndexingContextInformationForFullReindexingTask) task.details().get();
assertThat(information.failures().mailboxFailures()).containsExactly(mailbox.getId());
}
@@ -279,7 +279,7 @@ public class CassandraReIndexerImplTest {
ReIndexer.RunningOptions.DEFAULT);
task.run();
- ReprocessingContextInformationForErrorRecoveryIndexationTask information = (ReprocessingContextInformationForErrorRecoveryIndexationTask) task.details().get();
+ ReIndexingContextInformationForErrorRecoveryIndexationTask information = (ReIndexingContextInformationForErrorRecoveryIndexationTask) task.details().get();
assertThat(information.failures().mailboxFailures()).containsExactly(mailbox.getId());
}
@@ -303,7 +303,7 @@ public class CassandraReIndexerImplTest {
ReIndexer.RunningOptions.DEFAULT);
task.run();
- ReprocessingContextInformationForErrorRecoveryIndexationTask information = (ReprocessingContextInformationForErrorRecoveryIndexationTask) task.details().get();
+ ReIndexingContextInformationForErrorRecoveryIndexationTask information = (ReIndexingContextInformationForErrorRecoveryIndexationTask) task.details().get();
assertThat(information.failures().messageFailures()).containsExactly(new ReIndexingFailure(mailbox.getId(), appendResult.getId().getUid()));
}
@@ -323,7 +323,7 @@ public class CassandraReIndexerImplTest {
Task task = reIndexer.reIndex(ReIndexer.RunningOptions.DEFAULT);
task.run();
- ReprocessingContextInformationForFullReindexingTask information = (ReprocessingContextInformationForFullReindexingTask) task.details().get();
+ ReIndexingContextInformationForFullReindexingTask information = (ReIndexingContextInformationForFullReindexingTask) task.details().get();
assertThat(information.failures().messageFailures()).containsExactly(new ReIndexingFailure(mailbox.getId(), appendResult.getId().getUid()));
}
}
diff --git a/mailbox/tools/indexer/src/test/java/org/apache/mailbox/tools/indexer/ErrorRecoveryIndexationTaskSerializationTest.java b/mailbox/tools/indexer/src/test/java/org/apache/mailbox/tools/indexer/ErrorRecoveryIndexationTaskSerializationTest.java
index 7ef471c4cd..a964c03bee 100644
--- a/mailbox/tools/indexer/src/test/java/org/apache/mailbox/tools/indexer/ErrorRecoveryIndexationTaskSerializationTest.java
+++ b/mailbox/tools/indexer/src/test/java/org/apache/mailbox/tools/indexer/ErrorRecoveryIndexationTaskSerializationTest.java
@@ -30,7 +30,7 @@ import org.apache.james.mailbox.MessageUid;
import org.apache.james.mailbox.indexer.ReIndexer.RunningOptions;
import org.apache.james.mailbox.indexer.ReIndexingExecutionFailures;
import org.apache.james.mailbox.model.TestId;
-import org.apache.mailbox.tools.indexer.ReprocessingContextInformationDTO.ReprocessingContextInformationForErrorRecoveryIndexationTask;
+import org.apache.mailbox.tools.indexer.ReIndexingContextInformationDTO.ReIndexingContextInformationForErrorRecoveryIndexationTask;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -90,8 +90,8 @@ class ErrorRecoveryIndexationTaskSerializationTest {
@Test
void additionalInformationShouldBeSerializable() throws Exception {
- ReprocessingContextInformationForErrorRecoveryIndexationTask details = new ReprocessingContextInformationForErrorRecoveryIndexationTask(successfullyReprocessedMailCount, failedReprocessedMailCount, executionFailures, TIMESTAMP, RunningOptions.DEFAULT);
- JsonSerializationVerifier.dtoModule(ReprocessingContextInformationForErrorRecoveryIndexationTask.module(mailboxIdFactory))
+ ReIndexingContextInformationForErrorRecoveryIndexationTask details = new ReIndexingContextInformationForErrorRecoveryIndexationTask(successfullyReprocessedMailCount, failedReprocessedMailCount, executionFailures, TIMESTAMP, RunningOptions.DEFAULT);
+ JsonSerializationVerifier.dtoModule(ReIndexingContextInformationForErrorRecoveryIndexationTask.module(mailboxIdFactory))
.bean(details)
.json(serializedAdditionalInformation)
.verify();
@@ -102,8 +102,8 @@ class ErrorRecoveryIndexationTaskSerializationTest {
RunningOptions runningOptions = RunningOptions.builder()
.mode(RunningOptions.Mode.FIX_OUTDATED)
.build();
- ReprocessingContextInformationForErrorRecoveryIndexationTask details = new ReprocessingContextInformationForErrorRecoveryIndexationTask(successfullyReprocessedMailCount, failedReprocessedMailCount, executionFailures, TIMESTAMP, runningOptions);
- JsonSerializationVerifier.dtoModule(ReprocessingContextInformationForErrorRecoveryIndexationTask.module(mailboxIdFactory))
+ ReIndexingContextInformationForErrorRecoveryIndexationTask details = new ReIndexingContextInformationForErrorRecoveryIndexationTask(successfullyReprocessedMailCount, failedReprocessedMailCount, executionFailures, TIMESTAMP, runningOptions);
+ JsonSerializationVerifier.dtoModule(ReIndexingContextInformationForErrorRecoveryIndexationTask.module(mailboxIdFactory))
.bean(details)
.json(serializedAdditionalInformationWithCorrectMode)
.verify();
@@ -111,11 +111,11 @@ class ErrorRecoveryIndexationTaskSerializationTest {
@Test
void legacyAdditionalInformationShouldBeDeserializable() throws Exception {
- ReprocessingContextInformationForErrorRecoveryIndexationTask legacyAdditionalInformation = JsonGenericSerializer.forModules(ReprocessingContextInformationDTO.ReprocessingContextInformationForErrorRecoveryIndexationTask.module(new TestId.Factory()))
+ ReIndexingContextInformationForErrorRecoveryIndexationTask legacyAdditionalInformation = JsonGenericSerializer.forModules(ReIndexingContextInformationForErrorRecoveryIndexationTask.module(new TestId.Factory()))
.withoutNestedType()
.deserialize(legacySerializedAdditionalInformation);
- ReprocessingContextInformationDTO.ReprocessingContextInformationForFullReindexingTask expected = new ReprocessingContextInformationDTO.ReprocessingContextInformationForFullReindexingTask(
+ ReIndexingContextInformationDTO.ReIndexingContextInformationForFullReindexingTask expected = new ReIndexingContextInformationDTO.ReIndexingContextInformationForFullReindexingTask(
42,
2,
new ReIndexingExecutionFailures(failures, ImmutableList.of()),
diff --git a/mailbox/tools/indexer/src/test/java/org/apache/mailbox/tools/indexer/FullReindexingTaskSerializationTest.java b/mailbox/tools/indexer/src/test/java/org/apache/mailbox/tools/indexer/FullReindexingTaskSerializationTest.java
index 81ae7826af..a540f48bf2 100644
--- a/mailbox/tools/indexer/src/test/java/org/apache/mailbox/tools/indexer/FullReindexingTaskSerializationTest.java
+++ b/mailbox/tools/indexer/src/test/java/org/apache/mailbox/tools/indexer/FullReindexingTaskSerializationTest.java
@@ -29,7 +29,7 @@ import org.apache.james.mailbox.MessageUid;
import org.apache.james.mailbox.indexer.ReIndexer.RunningOptions;
import org.apache.james.mailbox.indexer.ReIndexingExecutionFailures;
import org.apache.james.mailbox.model.TestId;
-import org.apache.mailbox.tools.indexer.ReprocessingContextInformationDTO.ReprocessingContextInformationForFullReindexingTask;
+import org.apache.mailbox.tools.indexer.ReIndexingContextInformationDTO.ReIndexingContextInformationForFullReindexingTask;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -91,9 +91,9 @@ class FullReindexingTaskSerializationTest {
RunningOptions runningOptions = RunningOptions.builder()
.mode(RunningOptions.Mode.FIX_OUTDATED)
.build();
- ReprocessingContextInformationForFullReindexingTask details = new ReprocessingContextInformationForFullReindexingTask(successfullyReprocessedMailCount, failedReprocessedMailCount, reIndexingExecutionFailures, TIMESTAMP, runningOptions);
+ ReIndexingContextInformationForFullReindexingTask details = new ReIndexingContextInformationForFullReindexingTask(successfullyReprocessedMailCount, failedReprocessedMailCount, reIndexingExecutionFailures, TIMESTAMP, runningOptions);
- JsonSerializationVerifier.dtoModule(ReprocessingContextInformationForFullReindexingTask.module(new TestId.Factory()))
+ JsonSerializationVerifier.dtoModule(ReIndexingContextInformationForFullReindexingTask.module(new TestId.Factory()))
.bean(details)
.json(serializedAdditionalInformation)
.verify();
@@ -101,11 +101,11 @@ class FullReindexingTaskSerializationTest {
@Test
void legacyAdditionalInformationShouldBeDeserializable() throws Exception {
- ReprocessingContextInformationForFullReindexingTask legacyAdditionalInformation = JsonGenericSerializer.forModules(ReprocessingContextInformationForFullReindexingTask.module(new TestId.Factory()))
+ ReIndexingContextInformationForFullReindexingTask legacyAdditionalInformation = JsonGenericSerializer.forModules(ReIndexingContextInformationForFullReindexingTask.module(new TestId.Factory()))
.withoutNestedType()
.deserialize(legacySerializedAdditionalInformation);
- ReprocessingContextInformationForFullReindexingTask expected = new ReprocessingContextInformationForFullReindexingTask(
+ ReIndexingContextInformationForFullReindexingTask expected = new ReIndexingContextInformationForFullReindexingTask(
42,
2,
new ReIndexingExecutionFailures(ImmutableList.of(
diff --git a/server/container/guice/mailbox/src/main/java/org/apache/james/modules/mailbox/ReIndexingTaskSerializationModule.java b/server/container/guice/mailbox/src/main/java/org/apache/james/modules/mailbox/ReIndexingTaskSerializationModule.java
index cc90eb6cdb..0ec7c4ef95 100644
--- a/server/container/guice/mailbox/src/main/java/org/apache/james/modules/mailbox/ReIndexingTaskSerializationModule.java
+++ b/server/container/guice/mailbox/src/main/java/org/apache/james/modules/mailbox/ReIndexingTaskSerializationModule.java
@@ -33,7 +33,7 @@ import org.apache.mailbox.tools.indexer.MessageIdReIndexingTask;
import org.apache.mailbox.tools.indexer.MessageIdReindexingTaskAdditionalInformationDTO;
import org.apache.mailbox.tools.indexer.MessageIdReindexingTaskDTO;
import org.apache.mailbox.tools.indexer.ReIndexerPerformer;
-import org.apache.mailbox.tools.indexer.ReprocessingContextInformationDTO;
+import org.apache.mailbox.tools.indexer.ReIndexingContextInformationDTO;
import org.apache.mailbox.tools.indexer.SingleMailboxReindexingTask;
import org.apache.mailbox.tools.indexer.SingleMailboxReindexingTaskAdditionalInformationDTO;
import org.apache.mailbox.tools.indexer.SingleMailboxReindexingTaskDTO;
@@ -80,12 +80,12 @@ public class ReIndexingTaskSerializationModule extends AbstractModule {
@ProvidesIntoSet
public AdditionalInformationDTOModule<? extends TaskExecutionDetails.AdditionalInformation, ? extends AdditionalInformationDTO> errorRecoveryAdditionalInformation(MailboxId.Factory mailboxIdFactory) {
- return ReprocessingContextInformationDTO.ReprocessingContextInformationForErrorRecoveryIndexationTask.module(mailboxIdFactory);
+ return ReIndexingContextInformationDTO.ReIndexingContextInformationForErrorRecoveryIndexationTask.module(mailboxIdFactory);
}
@ProvidesIntoSet
public AdditionalInformationDTOModule<? extends TaskExecutionDetails.AdditionalInformation, ? extends AdditionalInformationDTO> fullReindexAdditionalInformation(MailboxId.Factory mailboxIdFactory) {
- return ReprocessingContextInformationDTO.ReprocessingContextInformationForFullReindexingTask.module(mailboxIdFactory);
+ return ReIndexingContextInformationDTO.ReIndexingContextInformationForFullReindexingTask.module(mailboxIdFactory);
}
@ProvidesIntoSet
diff --git a/server/container/guice/protocols/webadmin-mailbox/src/main/java/org/apache/james/modules/server/WebAdminReIndexingTaskSerializationModule.java b/server/container/guice/protocols/webadmin-mailbox/src/main/java/org/apache/james/modules/server/WebAdminReIndexingTaskSerializationModule.java
index 537d758a21..960878dcb4 100644
--- a/server/container/guice/protocols/webadmin-mailbox/src/main/java/org/apache/james/modules/server/WebAdminReIndexingTaskSerializationModule.java
+++ b/server/container/guice/protocols/webadmin-mailbox/src/main/java/org/apache/james/modules/server/WebAdminReIndexingTaskSerializationModule.java
@@ -24,7 +24,7 @@ import org.apache.james.server.task.json.dto.AdditionalInformationDTO;
import org.apache.james.server.task.json.dto.AdditionalInformationDTOModule;
import org.apache.james.task.TaskExecutionDetails;
import org.apache.james.webadmin.dto.DTOModuleInjections;
-import org.apache.james.webadmin.dto.WebAdminReprocessingContextInformationDTO;
+import org.apache.james.webadmin.dto.WebAdminIndexationContextInformationDTO;
import org.apache.james.webadmin.dto.WebAdminSingleMailboxReindexingTaskAdditionalInformationDTO;
import org.apache.james.webadmin.dto.WebAdminUserReindexingTaskAdditionalInformationDTO;
import org.apache.mailbox.tools.indexer.MessageIdReindexingTaskAdditionalInformationDTO;
@@ -38,13 +38,13 @@ public class WebAdminReIndexingTaskSerializationModule extends AbstractModule {
@Named(DTOModuleInjections.WEBADMIN_DTO)
@ProvidesIntoSet
public AdditionalInformationDTOModule<? extends TaskExecutionDetails.AdditionalInformation, ? extends AdditionalInformationDTO> errorRecoveryAdditionalInformation(MailboxId.Factory mailboxIdFactory) {
- return WebAdminReprocessingContextInformationDTO.WebAdminErrorRecoveryIndexationDTO.serializationModule();
+ return WebAdminIndexationContextInformationDTO.WebAdminErrorRecoveryIndexationDTO.serializationModule();
}
@Named(DTOModuleInjections.WEBADMIN_DTO)
@ProvidesIntoSet
public AdditionalInformationDTOModule<? extends TaskExecutionDetails.AdditionalInformation, ? extends AdditionalInformationDTO> fullReindexAdditionalInformation() {
- return WebAdminReprocessingContextInformationDTO.WebAdminFullIndexationDTO.serializationModule();
+ return WebAdminIndexationContextInformationDTO.WebAdminFullIndexationDTO.serializationModule();
}
@Named(DTOModuleInjections.WEBADMIN_DTO)
diff --git a/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/dto/WebAdminReprocessingContextInformationDTO.java b/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/dto/WebAdminIndexationContextInformationDTO.java
similarity index 82%
rename from server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/dto/WebAdminReprocessingContextInformationDTO.java
rename to server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/dto/WebAdminIndexationContextInformationDTO.java
index 35aa0b2596..de6e1e1f48 100644
--- a/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/dto/WebAdminReprocessingContextInformationDTO.java
+++ b/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/dto/WebAdminIndexationContextInformationDTO.java
@@ -29,15 +29,15 @@ import org.apache.james.server.task.json.dto.AdditionalInformationDTO;
import org.apache.james.server.task.json.dto.AdditionalInformationDTOModule;
import org.apache.mailbox.tools.indexer.ErrorRecoveryIndexationTask;
import org.apache.mailbox.tools.indexer.FullReindexingTask;
-import org.apache.mailbox.tools.indexer.ReprocessingContextInformationDTO;
+import org.apache.mailbox.tools.indexer.ReIndexingContextInformationDTO;
import org.apache.mailbox.tools.indexer.RunningOptionsDTO;
import com.google.common.collect.ImmutableList;
-public class WebAdminReprocessingContextInformationDTO implements AdditionalInformationDTO {
- public static class WebAdminErrorRecoveryIndexationDTO extends WebAdminReprocessingContextInformationDTO {
- public static AdditionalInformationDTOModule<ReprocessingContextInformationDTO.ReprocessingContextInformationForErrorRecoveryIndexationTask, WebAdminErrorRecoveryIndexationDTO> serializationModule() {
- return DTOModule.forDomainObject(ReprocessingContextInformationDTO.ReprocessingContextInformationForErrorRecoveryIndexationTask.class)
+public class WebAdminIndexationContextInformationDTO implements AdditionalInformationDTO {
+ public static class WebAdminErrorRecoveryIndexationDTO extends WebAdminIndexationContextInformationDTO {
+ public static AdditionalInformationDTOModule<ReIndexingContextInformationDTO.ReIndexingContextInformationForErrorRecoveryIndexationTask, WebAdminErrorRecoveryIndexationDTO> serializationModule() {
+ return DTOModule.forDomainObject(ReIndexingContextInformationDTO.ReIndexingContextInformationForErrorRecoveryIndexationTask.class)
.convertToDTO(WebAdminErrorRecoveryIndexationDTO.class)
.toDomainObjectConverter(dto -> {
throw new NotImplementedException("Deserialization not implemented for this DTO");
@@ -59,9 +59,9 @@ public class WebAdminReprocessingContextInformationDTO implements AdditionalInfo
}
}
- public static class WebAdminFullIndexationDTO extends WebAdminReprocessingContextInformationDTO {
- public static AdditionalInformationDTOModule<ReprocessingContextInformationDTO.ReprocessingContextInformationForFullReindexingTask, WebAdminFullIndexationDTO> serializationModule() {
- return DTOModule.forDomainObject(ReprocessingContextInformationDTO.ReprocessingContextInformationForFullReindexingTask.class)
+ public static class WebAdminFullIndexationDTO extends WebAdminIndexationContextInformationDTO {
+ public static AdditionalInformationDTOModule<ReIndexingContextInformationDTO.ReIndexingContextInformationForFullReindexingTask, WebAdminFullIndexationDTO> serializationModule() {
+ return DTOModule.forDomainObject(ReIndexingContextInformationDTO.ReIndexingContextInformationForFullReindexingTask.class)
.convertToDTO(WebAdminFullIndexationDTO.class)
.toDomainObjectConverter(dto -> {
throw new NotImplementedException("Deserialization not implemented for this DTO");
@@ -91,9 +91,9 @@ public class WebAdminReprocessingContextInformationDTO implements AdditionalInfo
private final List<String> mailboxFailures;
protected final Instant timestamp;
- WebAdminReprocessingContextInformationDTO(String type, RunningOptionsDTO runningOptions, int successfullyReprocessedMailCount, int failedReprocessedMailCount,
- ReIndexingExecutionFailures failures,
- Instant timestamp) {
+ WebAdminIndexationContextInformationDTO(String type, RunningOptionsDTO runningOptions, int successfullyReprocessedMailCount, int failedReprocessedMailCount,
+ ReIndexingExecutionFailures failures,
+ Instant timestamp) {
this.type = type;
this.runningOptions = runningOptions;
this.successfullyReprocessedMailCount = successfullyReprocessedMailCount;
diff --git a/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/dto/WebAdminSingleMailboxReindexingTaskAdditionalInformationDTO.java b/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/dto/WebAdminSingleMailboxReindexingTaskAdditionalInformationDTO.java
index 49c2692945..d4ddc124d1 100644
--- a/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/dto/WebAdminSingleMailboxReindexingTaskAdditionalInformationDTO.java
+++ b/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/dto/WebAdminSingleMailboxReindexingTaskAdditionalInformationDTO.java
@@ -51,7 +51,7 @@ public class WebAdminSingleMailboxReindexingTaskAdditionalInformationDTO impleme
.withFactory(AdditionalInformationDTOModule::new);
}
- private final WebAdminReprocessingContextInformationDTO reprocessingContextInformationDTO;
+ private final WebAdminIndexationContextInformationDTO reprocessingContextInformationDTO;
private final String mailboxId;
@JsonCreator
@@ -63,7 +63,7 @@ public class WebAdminSingleMailboxReindexingTaskAdditionalInformationDTO impleme
ReIndexingExecutionFailures failures,
Instant timestamp) {
this.mailboxId = mailboxId;
- this.reprocessingContextInformationDTO = new WebAdminReprocessingContextInformationDTO(
+ this.reprocessingContextInformationDTO = new WebAdminIndexationContextInformationDTO(
type, runningOptions, successfullyReprocessedMailCount, failedReprocessedMailCount, failures, timestamp);
}
diff --git a/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/dto/WebAdminUserReindexingTaskAdditionalInformationDTO.java b/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/dto/WebAdminUserReindexingTaskAdditionalInformationDTO.java
index e1c6f62f94..fe0485bc7c 100644
--- a/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/dto/WebAdminUserReindexingTaskAdditionalInformationDTO.java
+++ b/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/dto/WebAdminUserReindexingTaskAdditionalInformationDTO.java
@@ -51,7 +51,7 @@ public class WebAdminUserReindexingTaskAdditionalInformationDTO implements Addit
.withFactory(AdditionalInformationDTOModule::new);
}
- private final WebAdminReprocessingContextInformationDTO reprocessingContextInformationDTO;
+ private final WebAdminIndexationContextInformationDTO reprocessingContextInformationDTO;
private final String username;
@JsonCreator
@@ -63,7 +63,7 @@ public class WebAdminUserReindexingTaskAdditionalInformationDTO implements Addit
ReIndexingExecutionFailures failures,
Instant timestamp) {
this.username = username;
- this.reprocessingContextInformationDTO = new WebAdminReprocessingContextInformationDTO(
+ this.reprocessingContextInformationDTO = new WebAdminIndexationContextInformationDTO(
type,
runningOptions,
successfullyReprocessedMailCount,
diff --git a/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/dto/WebAdminReprocessingContextInformationDTOTest.java b/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/dto/WebAdminIndexationContextInformationDTOTest.java
similarity index 85%
rename from server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/dto/WebAdminReprocessingContextInformationDTOTest.java
rename to server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/dto/WebAdminIndexationContextInformationDTOTest.java
index 3f45a2dda4..0c2188af9d 100644
--- a/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/dto/WebAdminReprocessingContextInformationDTOTest.java
+++ b/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/dto/WebAdminIndexationContextInformationDTOTest.java
@@ -30,13 +30,13 @@ import org.apache.james.mailbox.indexer.ReIndexer;
import org.apache.james.mailbox.indexer.ReIndexingExecutionFailures;
import org.apache.james.mailbox.model.MailboxId;
import org.apache.james.mailbox.model.TestId;
-import org.apache.mailbox.tools.indexer.ReprocessingContextInformationDTO;
+import org.apache.mailbox.tools.indexer.ReIndexingContextInformationDTO;
import org.junit.jupiter.api.Test;
import com.fasterxml.jackson.databind.exc.InvalidDefinitionException;
import com.google.common.collect.ImmutableList;
-class WebAdminReprocessingContextInformationDTOTest {
+class WebAdminIndexationContextInformationDTOTest {
private static final Instant TIMESTAMP = Instant.parse("2018-11-13T12:00:55Z");
private final String serializedErrorRecoveryAdditionalInformation = "{" +
@@ -68,8 +68,8 @@ class WebAdminReprocessingContextInformationDTOTest {
@Test
void shouldSerializeErrorRecoveryAdditionalInformation() throws Exception {
- ReprocessingContextInformationDTO.ReprocessingContextInformationForErrorRecoveryIndexationTask domainObject =
- new ReprocessingContextInformationDTO.ReprocessingContextInformationForErrorRecoveryIndexationTask(
+ ReIndexingContextInformationDTO.ReIndexingContextInformationForErrorRecoveryIndexationTask domainObject =
+ new ReIndexingContextInformationDTO.ReIndexingContextInformationForErrorRecoveryIndexationTask(
42,
2,
executionFailures,
@@ -77,7 +77,7 @@ class WebAdminReprocessingContextInformationDTOTest {
ReIndexer.RunningOptions.DEFAULT);
String json =
- JsonGenericSerializer.forModules(WebAdminReprocessingContextInformationDTO.WebAdminErrorRecoveryIndexationDTO
+ JsonGenericSerializer.forModules(WebAdminIndexationContextInformationDTO.WebAdminErrorRecoveryIndexationDTO
.serializationModule())
.withoutNestedType()
.serialize(domainObject);
@@ -88,7 +88,7 @@ class WebAdminReprocessingContextInformationDTOTest {
@Test
void deserializeErrorRecoveryShouldNotBeSupported() {
- assertThatThrownBy(() -> JsonGenericSerializer.forModules(WebAdminReprocessingContextInformationDTO.WebAdminErrorRecoveryIndexationDTO
+ assertThatThrownBy(() -> JsonGenericSerializer.forModules(WebAdminIndexationContextInformationDTO.WebAdminErrorRecoveryIndexationDTO
.serializationModule())
.withoutNestedType()
.deserialize(serializedErrorRecoveryAdditionalInformation))
@@ -97,8 +97,8 @@ class WebAdminReprocessingContextInformationDTOTest {
@Test
void shouldSerializeFullAdditionalInformation() throws Exception {
- ReprocessingContextInformationDTO.ReprocessingContextInformationForFullReindexingTask domainObject =
- new ReprocessingContextInformationDTO.ReprocessingContextInformationForFullReindexingTask(
+ ReIndexingContextInformationDTO.ReIndexingContextInformationForFullReindexingTask domainObject =
+ new ReIndexingContextInformationDTO.ReIndexingContextInformationForFullReindexingTask(
42,
2,
executionFailures,
@@ -106,7 +106,7 @@ class WebAdminReprocessingContextInformationDTOTest {
ReIndexer.RunningOptions.DEFAULT);
String json =
- JsonGenericSerializer.forModules(WebAdminReprocessingContextInformationDTO.WebAdminFullIndexationDTO
+ JsonGenericSerializer.forModules(WebAdminIndexationContextInformationDTO.WebAdminFullIndexationDTO
.serializationModule())
.withoutNestedType()
.serialize(domainObject);
@@ -117,7 +117,7 @@ class WebAdminReprocessingContextInformationDTOTest {
@Test
void deserializeFullShouldNotBeSupported() {
- assertThatThrownBy(() -> JsonGenericSerializer.forModules(WebAdminReprocessingContextInformationDTO.WebAdminFullIndexationDTO
+ assertThatThrownBy(() -> JsonGenericSerializer.forModules(WebAdminIndexationContextInformationDTO.WebAdminFullIndexationDTO
.serializationModule())
.withoutNestedType()
.deserialize(serializedFullAdditionalInformation))
diff --git a/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/dto/WebAdminSingleMailboxReprocessingDTOTest.java b/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/dto/WebAdminSingleMailboxIndexingDTOTest.java
similarity index 98%
rename from server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/dto/WebAdminSingleMailboxReprocessingDTOTest.java
rename to server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/dto/WebAdminSingleMailboxIndexingDTOTest.java
index 03459e5315..51361fb77b 100644
--- a/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/dto/WebAdminSingleMailboxReprocessingDTOTest.java
+++ b/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/dto/WebAdminSingleMailboxIndexingDTOTest.java
@@ -36,7 +36,7 @@ import org.junit.jupiter.api.Test;
import com.fasterxml.jackson.databind.exc.InvalidDefinitionException;
import com.google.common.collect.ImmutableList;
-class WebAdminSingleMailboxReprocessingDTOTest {
+class WebAdminSingleMailboxIndexingDTOTest {
private static final Instant TIMESTAMP = Instant.parse("2018-11-13T12:00:55Z");
private final String serializedAdditionalInformation = "{" +
diff --git a/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/dto/WebAdminUserReprocessingDTOTest.java b/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/dto/WebAdminUserReIndexingDTOTest.java
similarity index 99%
rename from server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/dto/WebAdminUserReprocessingDTOTest.java
rename to server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/dto/WebAdminUserReIndexingDTOTest.java
index bce12a955f..6a1fa55024 100644
--- a/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/dto/WebAdminUserReprocessingDTOTest.java
+++ b/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/dto/WebAdminUserReIndexingDTOTest.java
@@ -37,7 +37,7 @@ import org.junit.jupiter.api.Test;
import com.fasterxml.jackson.databind.exc.InvalidDefinitionException;
import com.google.common.collect.ImmutableList;
-class WebAdminUserReprocessingDTOTest {
+class WebAdminUserReIndexingDTOTest {
private static final Instant TIMESTAMP = Instant.parse("2018-11-13T12:00:55Z");
private final String serializedAdditionalInformation = "{" +
diff --git a/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/MailboxesRoutesTest.java b/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/MailboxesRoutesTest.java
index 935d0754a1..344bd1239f 100644
--- a/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/MailboxesRoutesTest.java
+++ b/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/MailboxesRoutesTest.java
@@ -82,8 +82,8 @@ import org.apache.james.task.Hostname;
import org.apache.james.task.MemoryTaskManager;
import org.apache.james.webadmin.WebAdminServer;
import org.apache.james.webadmin.WebAdminUtils;
-import org.apache.james.webadmin.dto.WebAdminReprocessingContextInformationDTO.WebAdminErrorRecoveryIndexationDTO;
-import org.apache.james.webadmin.dto.WebAdminReprocessingContextInformationDTO.WebAdminFullIndexationDTO;
+import org.apache.james.webadmin.dto.WebAdminIndexationContextInformationDTO.WebAdminErrorRecoveryIndexationDTO;
+import org.apache.james.webadmin.dto.WebAdminIndexationContextInformationDTO.WebAdminFullIndexationDTO;
import org.apache.james.webadmin.dto.WebAdminSingleMailboxReindexingTaskAdditionalInformationDTO;
import org.apache.james.webadmin.service.PreviousReIndexingService;
import org.apache.james.webadmin.utils.ErrorResponder;
@@ -202,11 +202,11 @@ class MailboxesRoutesTest {
}
@Nested
- class FullReprocessing {
+ class FullReIndexing {
@Nested
class Validation {
@Test
- void fullReprocessingShouldFailWithNoTask() {
+ void fullReIndexingShouldFailWithNoTask() {
when()
.post("/mailboxes")
.then()
@@ -218,7 +218,7 @@ class MailboxesRoutesTest {
}
@Test
- void fullReprocessingShouldFailWithBadTask() {
+ void fullReIndexingShouldFailWithBadTask() {
when()
.post("/mailboxes?task=bad")
.then()
@@ -233,7 +233,7 @@ class MailboxesRoutesTest {
@Nested
class TaskDetails {
@Test
- void fullReprocessingShouldNotFailWhenNoMail() {
+ void fullReIndexingShouldNotFailWhenNoMail() {
String taskId = with()
.post("/mailboxes?task=reIndex")
.jsonPath()
@@ -257,7 +257,7 @@ class MailboxesRoutesTest {
}
@Test
- void fullReprocessingShouldReturnTaskDetailsWhenMail() throws Exception {
+ void fullReIndexingShouldReturnTaskDetailsWhenMail() throws Exception {
MailboxSession systemSession = mailboxManager.createSystemSession(USERNAME);
mailboxManager.createMailbox(INBOX, systemSession).get();
mailboxManager.getMailbox(INBOX, systemSession)
@@ -288,7 +288,7 @@ class MailboxesRoutesTest {
}
@Test
- void fullReprocessingWithMessagesPerSecondShouldReturnTaskDetailsWhenMail() throws Exception {
+ void fullReIndexingWithMessagesPerSecondShouldReturnTaskDetailsWhenMail() throws Exception {
MailboxSession systemSession = mailboxManager.createSystemSession(USERNAME);
mailboxManager.createMailbox(INBOX, systemSession).get();
mailboxManager.getMailbox(INBOX, systemSession)
@@ -320,7 +320,7 @@ class MailboxesRoutesTest {
}
@Test
- void fullReprocessingShouldReturnTaskDetailsWhenFailing() throws Exception {
+ void fullReIndexingShouldReturnTaskDetailsWhenFailing() throws Exception {
MailboxSession systemSession = mailboxManager.createSystemSession(USERNAME);
MailboxId mailboxId = mailboxManager.createMailbox(INBOX, systemSession).get();
ComposedMessageId composedMessageId = mailboxManager.getMailbox(INBOX, systemSession)
@@ -356,7 +356,7 @@ class MailboxesRoutesTest {
}
@Test
- void userReprocessingShouldReturnTaskDetailsWhenFailingAtTheMailboxLevel() throws Exception {
+ void userReIndexingShouldReturnTaskDetailsWhenFailingAtTheMailboxLevel() throws Exception {
MailboxSession systemSession = mailboxManager.createSystemSession(USERNAME);
MailboxId mailboxId = mailboxManager.createMailbox(INBOX, systemSession).get();
@@ -380,7 +380,7 @@ class MailboxesRoutesTest {
}
@Test
- void fullReprocessingWithCorrectModeShouldReturnTaskDetailsWhenMails() throws Exception {
+ void fullReIndexingWithCorrectModeShouldReturnTaskDetailsWhenMails() throws Exception {
MailboxSession systemSession = mailboxManager.createSystemSession(USERNAME);
MailboxId mailboxId = mailboxManager.createMailbox(INBOX, systemSession).get();
Mailbox mailbox = mailboxManager.getMailbox(mailboxId, systemSession).getMailboxEntity();
@@ -431,7 +431,7 @@ class MailboxesRoutesTest {
}
@Test
- void fullReprocessingShouldAcceptRebuildAllNoCleanupMode() throws Exception {
+ void fullReIndexingShouldAcceptRebuildAllNoCleanupMode() throws Exception {
MailboxSession systemSession = mailboxManager.createSystemSession(USERNAME);
MailboxId mailboxId = mailboxManager.createMailbox(INBOX, systemSession).get();
Mailbox mailbox = mailboxManager.getMailbox(mailboxId, systemSession).getMailboxEntity();
@@ -485,7 +485,7 @@ class MailboxesRoutesTest {
}
@Test
- void fullReprocessingWithCorrectModeShouldFixInconsistenciesInES() throws Exception {
+ void fullReIndexingWithCorrectModeShouldFixInconsistenciesInES() throws Exception {
MailboxSession systemSession = mailboxManager.createSystemSession(USERNAME);
MailboxId mailboxId = mailboxManager.createMailbox(INBOX, systemSession).get();
Mailbox mailbox = mailboxManager.getMailbox(mailboxId, systemSession).getMailboxEntity();
@@ -526,7 +526,7 @@ class MailboxesRoutesTest {
}
@Test
- void fullReprocessingNoCleanupShouldNoopWhenNoInconsistencies() throws Exception {
+ void fullReIndexingNoCleanupShouldNoopWhenNoInconsistencies() throws Exception {
MailboxSession systemSession = mailboxManager.createSystemSession(USERNAME);
MailboxId mailboxId = mailboxManager.createMailbox(INBOX, systemSession).get();
Mailbox mailbox = mailboxManager.getMailbox(mailboxId, systemSession).getMailboxEntity();
@@ -554,7 +554,7 @@ class MailboxesRoutesTest {
}
@Test
- void fullReprocessingNoCleanupShouldSolveInconsistencies() throws Exception {
+ void fullReIndexingNoCleanupShouldSolveInconsistencies() throws Exception {
MailboxSession systemSession = mailboxManager.createSystemSession(USERNAME);
MailboxId mailboxId = mailboxManager.createMailbox(INBOX, systemSession).get();
Mailbox mailbox = mailboxManager.getMailbox(mailboxId, systemSession).getMailboxEntity();
@@ -591,7 +591,7 @@ class MailboxesRoutesTest {
}
@Test
- void fullReprocessingWithCorrectModeShouldNotChangeDocumentsInESWhenNoInconsistencies() throws Exception {
+ void fullReIndexingWithCorrectModeShouldNotChangeDocumentsInESWhenNoInconsistencies() throws Exception {
MailboxSession systemSession = mailboxManager.createSystemSession(USERNAME);
MailboxId mailboxId = mailboxManager.createMailbox(INBOX, systemSession).get();
Mailbox mailbox = mailboxManager.getMailbox(mailboxId, systemSession).getMailboxEntity();
@@ -621,7 +621,7 @@ class MailboxesRoutesTest {
@Disabled("JAMES-3202 Limitation of the current correct mode reindexation. We only check metadata and fix "
+ "inconsistencies with ES, but we don't check for inconsistencies from ES to metadata")
@Test
- void fullReprocessingWithCorrectModeShouldRemoveOrphanMessagesInES() throws Exception {
+ void fullReIndexingWithCorrectModeShouldRemoveOrphanMessagesInES() throws Exception {
MailboxSession systemSession = mailboxManager.createSystemSession(USERNAME);
MailboxId mailboxId = mailboxManager.createMailbox(INBOX, systemSession).get();
Mailbox mailbox = mailboxManager.getMailbox(mailboxId, systemSession).getMailboxEntity();
@@ -662,7 +662,7 @@ class MailboxesRoutesTest {
@Nested
class SideEffects {
@Test
- void fullReprocessingShouldPerformReprocessingWhenMail() throws Exception {
+ void fullReIndexingShouldPerformReIndexingWhenMail() throws Exception {
MailboxSession systemSession = mailboxManager.createSystemSession(USERNAME);
MailboxId mailboxId = mailboxManager.createMailbox(INBOX, systemSession).get();
ComposedMessageId createdMessage = mailboxManager.getMailbox(INBOX, systemSession)
@@ -699,11 +699,11 @@ class MailboxesRoutesTest {
}
@Nested
- class MailboxReprocessing {
+ class MailboxReIndexing {
@Nested
class Validation {
@Test
- void mailboxReprocessingShouldFailWithNoTask() throws Exception {
+ void mailboxReIndexingShouldFailWithNoTask() throws Exception {
MailboxSession systemSession = mailboxManager.createSystemSession(USERNAME);
MailboxId mailboxId = mailboxManager.createMailbox(INBOX, systemSession).get();
@@ -718,7 +718,7 @@ class MailboxesRoutesTest {
}
@Test
- void mailboxReprocessingShouldFailWithBadTask() throws Exception {
+ void mailboxReIndexingShouldFailWithBadTask() throws Exception {
MailboxSession systemSession = mailboxManager.createSystemSession(USERNAME);
MailboxId mailboxId = mailboxManager.createMailbox(INBOX, systemSession).get();
@@ -733,7 +733,7 @@ class MailboxesRoutesTest {
}
@Test
- void mailboxReprocessingShouldFailWithBadMailboxId() {
+ void mailboxReIndexingShouldFailWithBadMailboxId() {
when()
.post("/mailboxes/bad?task=reIndex")
.then()
@@ -744,7 +744,7 @@ class MailboxesRoutesTest {
}
@Test
- void mailboxReprocessingShouldFailWithNonExistentMailboxId() {
+ void mailboxReIndexingShouldFailWithNonExistentMailboxId() {
when()
.post("/mailboxes/36?task=reIndex")
.then()
@@ -758,7 +758,7 @@ class MailboxesRoutesTest {
@Nested
class TaskDetails {
@Test
- void mailboxReprocessingShouldNotFailWhenNoMail() throws Exception {
+ void mailboxReIndexingShouldNotFailWhenNoMail() throws Exception {
MailboxSession systemSession = mailboxManager.createSystemSession(USERNAME);
MailboxId mailboxId = mailboxManager.createMailbox(INBOX, systemSession).get();
@@ -786,7 +786,7 @@ class MailboxesRoutesTest {
}
@Test
- void mailboxReprocessingShouldReturnTaskDetailsWhenMail() throws Exception {
+ void mailboxReIndexingShouldReturnTaskDetailsWhenMail() throws Exception {
MailboxSession systemSession = mailboxManager.createSystemSession(USERNAME);
MailboxId mailboxId = mailboxManager.createMailbox(INBOX, systemSession).get();
mailboxManager.getMailbox(INBOX, systemSession)
@@ -817,7 +817,7 @@ class MailboxesRoutesTest {
}
@Test
- void mailboxReprocessingWithMessagesPerSecondShouldReturnTaskDetailsWhenMail() throws Exception {
+ void mailboxReIndexingWithMessagesPerSecondShouldReturnTaskDetailsWhenMail() throws Exception {
MailboxSession systemSession = mailboxManager.createSystemSession(USERNAME);
MailboxId mailboxId = mailboxManager.createMailbox(INBOX, systemSession).get();
mailboxManager.getMailbox(INBOX, systemSession)
@@ -850,7 +850,7 @@ class MailboxesRoutesTest {
}
@Test
- void mailboxReprocessingShouldReturnTaskDetailsWhenFailing() throws Exception {
+ void mailboxReIndexingShouldReturnTaskDetailsWhenFailing() throws Exception {
MailboxSession systemSession = mailboxManager.createSystemSession(USERNAME);
MailboxId mailboxId = mailboxManager.createMailbox(INBOX, systemSession).get();
ComposedMessageId composedMessageId = mailboxManager.getMailbox(INBOX, systemSession)
@@ -887,7 +887,7 @@ class MailboxesRoutesTest {
}
@Test
- void userReprocessingShouldReturnTaskDetailsWhenFailingAtTheMailboxLevel() throws Exception {
+ void userReIndexingShouldReturnTaskDetailsWhenFailingAtTheMailboxLevel() throws Exception {
MailboxSession systemSession = mailboxManager.createSystemSession(USERNAME);
MailboxId mailboxId = mailboxManager.createMailbox(INBOX, systemSession).get();
@@ -913,7 +913,7 @@ class MailboxesRoutesTest {
@Test
- void mailboxReprocessingWithCorrectModeShouldReturnTaskDetailsWhenMails() throws Exception {
+ void mailboxReIndexingWithCorrectModeShouldReturnTaskDetailsWhenMails() throws Exception {
MailboxSession systemSession = mailboxManager.createSystemSession(USERNAME);
MailboxId mailboxId = mailboxManager.createMailbox(INBOX, systemSession).get();
Mailbox mailbox = mailboxManager.getMailbox(mailboxId, systemSession).getMailboxEntity();
@@ -966,7 +966,7 @@ class MailboxesRoutesTest {
}
@Test
- void mailboxReprocessingWithCorrectModeShouldFixInconsistenciesInES() throws Exception {
+ void mailboxReIndexingWithCorrectModeShouldFixInconsistenciesInES() throws Exception {
MailboxSession systemSession = mailboxManager.createSystemSession(USERNAME);
MailboxId mailboxId = mailboxManager.createMailbox(INBOX, systemSession).get();
Mailbox mailbox = mailboxManager.getMailbox(mailboxId, systemSession).getMailboxEntity();
@@ -1009,7 +1009,7 @@ class MailboxesRoutesTest {
}
@Test
- void mailboxReprocessingWithCorrectModeShouldNotChangeDocumentsInESWhenNoInconsistencies() throws Exception {
+ void mailboxReIndexingWithCorrectModeShouldNotChangeDocumentsInESWhenNoInconsistencies() throws Exception {
MailboxSession systemSession = mailboxManager.createSystemSession(USERNAME);
MailboxId mailboxId = mailboxManager.createMailbox(INBOX, systemSession).get();
Mailbox mailbox = mailboxManager.getMailbox(mailboxId, systemSession).getMailboxEntity();
@@ -1041,7 +1041,7 @@ class MailboxesRoutesTest {
@Disabled("JAMES-3202 Limitation of the current correct mode reindexation. We only check metadata and fix "
+ "inconsistencies with ES, but we don't check for inconsistencies from ES to metadata")
@Test
- void mailboxReprocessingWithCorrectModeShouldRemoveOrphanMessagesInES() throws Exception {
+ void mailboxReIndexingWithCorrectModeShouldRemoveOrphanMessagesInES() throws Exception {
MailboxSession systemSession = mailboxManager.createSystemSession(USERNAME);
MailboxId mailboxId = mailboxManager.createMailbox(INBOX, systemSession).get();
Mailbox mailbox = mailboxManager.getMailbox(mailboxId, systemSession).getMailboxEntity();
@@ -1084,7 +1084,7 @@ class MailboxesRoutesTest {
@Nested
class SideEffects {
@Test
- void mailboxReprocessingShouldPerformReprocessingWhenMail() throws Exception {
+ void mailboxReIndexingShouldPerformReIndexingWhenMail() throws Exception {
MailboxSession systemSession = mailboxManager.createSystemSession(USERNAME);
MailboxId mailboxId = mailboxManager.createMailbox(INBOX, systemSession).get();
ComposedMessageId createdMessage = mailboxManager.getMailbox(INBOX, systemSession)
@@ -1122,11 +1122,11 @@ class MailboxesRoutesTest {
}
@Nested
- class MessageReprocessing {
+ class MessageReIndexing {
@Nested
class Validation {
@Test
- void messageReprocessingShouldFailWithNoTask() throws Exception {
+ void messageReIndexingShouldFailWithNoTask() throws Exception {
MailboxSession systemSession = mailboxManager.createSystemSession(USERNAME);
MailboxId mailboxId = mailboxManager.createMailbox(INBOX, systemSession).get();
@@ -1141,7 +1141,7 @@ class MailboxesRoutesTest {
}
@Test
- void messageReprocessingShouldFailWithBadTask() throws Exception {
+ void messageReIndexingShouldFailWithBadTask() throws Exception {
MailboxSession systemSession = mailboxManager.createSystemSession(USERNAME);
MailboxId mailboxId = mailboxManager.createMailbox(INBOX, systemSession).get();
@@ -1156,7 +1156,7 @@ class MailboxesRoutesTest {
}
@Test
- void messageReprocessingShouldFailWithBadMailboxId() {
+ void messageReIndexingShouldFailWithBadMailboxId() {
when()
.post("/mailboxes/bad/mails/7?task=reIndex")
.then()
@@ -1167,7 +1167,7 @@ class MailboxesRoutesTest {
}
@Test
- void messageReprocessingShouldFailWithNonExistentMailboxId() {
+ void messageReIndexingShouldFailWithNonExistentMailboxId() {
when()
.post("/mailboxes/36/mails/7?task=reIndex")
.then()
@@ -1178,7 +1178,7 @@ class MailboxesRoutesTest {
}
@Test
- void messageReprocessingShouldFailWithBadUid() {
+ void messageReIndexingShouldFailWithBadUid() {
when()
.post("/mailboxes/36/mails/bad?task=reIndex")
.then()
@@ -1192,7 +1192,7 @@ class MailboxesRoutesTest {
@Nested
class TaskDetails {
@Test
- void messageReprocessingShouldNotFailWhenUidNotFound() throws Exception {
+ void messageReIndexingShouldNotFailWhenUidNotFound() throws Exception {
MailboxSession systemSession = mailboxManager.createSystemSession(USERNAME);
MailboxId mailboxId = mailboxManager.createMailbox(INBOX, systemSession).get();
@@ -1217,7 +1217,7 @@ class MailboxesRoutesTest {
}
@Test
- void messageReprocessingShouldReturnTaskDetailsWhenMail() throws Exception {
+ void messageReIndexingShouldReturnTaskDetailsWhenMail() throws Exception {
MailboxSession systemSession = mailboxManager.createSystemSession(USERNAME);
MailboxId mailboxId = mailboxManager.createMailbox(INBOX, systemSession).get();
ComposedMessageId composedMessageId = mailboxManager.getMailbox(INBOX, systemSession)
@@ -1250,7 +1250,7 @@ class MailboxesRoutesTest {
@Nested
class SideEffects {
@Test
- void mailboxReprocessingShouldPerformReprocessingWhenMail() throws Exception {
+ void mailboxReIndexingShouldPerformReIndexingWhenMail() throws Exception {
MailboxSession systemSession = mailboxManager.createSystemSession(USERNAME);
MailboxId mailboxId = mailboxManager.createMailbox(INBOX, systemSession).get();
ComposedMessageId createdMessage = mailboxManager.getMailbox(INBOX, systemSession)
@@ -1530,7 +1530,7 @@ class MailboxesRoutesTest {
}
@Test
- void userReprocessingShouldReturnTaskDetailsWhenFailingAtTheMailboxLevel() throws Exception {
+ void userReIndexingShouldReturnTaskDetailsWhenFailingAtTheMailboxLevel() throws Exception {
MailboxSession systemSession = mailboxManager.createSystemSession(USERNAME);
MailboxId mailboxId = mailboxManager.createMailbox(INBOX, systemSession).get();
@@ -1570,7 +1570,7 @@ class MailboxesRoutesTest {
@Nested
class SideEffects {
@Test
- void fixingReprocessingShouldPerformReprocessingWhenMail() throws Exception {
+ void fixingReIndexingShouldPerformReIndexingWhenMail() throws Exception {
MailboxSession systemSession = mailboxManager.createSystemSession(USERNAME);
MailboxId mailboxId = mailboxManager.createMailbox(INBOX, systemSession).get();
ComposedMessageId createdMessage = mailboxManager.getMailbox(INBOX, systemSession)
diff --git a/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/MessageRoutesTest.java b/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/MessageRoutesTest.java
index 27058fade3..f86221fab1 100644
--- a/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/MessageRoutesTest.java
+++ b/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/MessageRoutesTest.java
@@ -108,7 +108,7 @@ class MessageRoutesTest {
}
@Nested
- class MessageReprocessing {
+ class MessageReIndexing {
@Nested
class Validation {
@Test
diff --git a/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/UserMailboxesRoutesTest.java b/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/UserMailboxesRoutesTest.java
index fe79047209..861cb7bd54 100644
--- a/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/UserMailboxesRoutesTest.java
+++ b/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/UserMailboxesRoutesTest.java
@@ -1444,8 +1444,7 @@ class UserMailboxesRoutesTest {
}
@Nested
- class UserReprocessing {
- static final int BATCH_SIZE = 1;
+ class UserReIndexing {
static final int SEARCH_SIZE = 1;
@RegisterExtension
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 05/07: JAMES-3784 WebAdmin: Provide RunningOptions (rateLimit) for Redeliver event task, Reprocessing mail task
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch 3.7.x
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit d12208333fd9bc9fb813974898bf1509d1453282
Author: Tung Van TRAN <vt...@linagora.com>
AuthorDate: Thu Jun 23 17:59:33 2022 +0700
JAMES-3784 WebAdmin: Provide RunningOptions (rateLimit) for Redeliver event task, Reprocessing mail task
---
.../apache/james/modules/CommonServicesModule.java | 1 +
.../ErrorMailRepositoryEmptyHealthCheckModule.java | 10 +-
.../james/webadmin/utils/ParametersExtractor.java | 9 ++
.../webadmin/routes/EventDeadLettersRoutes.java | 15 +-
.../service/EventDeadLettersRedeliverAllTask.java | 15 +-
.../EventDeadLettersRedeliverAllTaskDTO.java | 17 ++-
.../EventDeadLettersRedeliverGroupTask.java | 15 +-
.../EventDeadLettersRedeliverGroupTaskDTO.java | 18 ++-
.../service/EventDeadLettersRedeliverOneTask.java | 2 +-
.../service/EventDeadLettersRedeliverService.java | 33 ++++-
...tersRedeliveryTaskAdditionalInformationDTO.java | 71 +++++++---
.../webadmin/service/EventDeadLettersService.java | 8 +-
.../routes/EventDeadLettersRoutesTest.java | 91 ++++++++++++
.../service/EventDeadLettersRedeliverTaskTest.java | 152 ++++++++++++++++++---
.../webadmin/routes/MailRepositoriesRoutes.java | 7 +-
.../webadmin/service/ReprocessingAllMailsTask.java | 12 +-
...essingAllMailsTaskAdditionalInformationDTO.java | 16 ++-
.../service/ReprocessingAllMailsTaskDTO.java | 17 ++-
...cessingOneMailTaskAdditionalInformationDTO.java | 4 +-
.../service/ReprocessingOneMailTaskDTO.java | 4 +-
.../webadmin/service/ReprocessingService.java | 47 +++++--
.../routes/MailRepositoriesRoutesTest.java | 56 ++++++++
.../service/ReprocessingAllMailsTaskTest.java | 32 ++++-
.../service/ReprocessingOneMailTaskTest.java | 20 ++-
.../webadmin/service/ReprocessingServiceTest.java | 11 +-
25 files changed, 574 insertions(+), 109 deletions(-)
diff --git a/server/container/guice/common/src/main/java/org/apache/james/modules/CommonServicesModule.java b/server/container/guice/common/src/main/java/org/apache/james/modules/CommonServicesModule.java
index f9cc271cd0..7aede854d3 100644
--- a/server/container/guice/common/src/main/java/org/apache/james/modules/CommonServicesModule.java
+++ b/server/container/guice/common/src/main/java/org/apache/james/modules/CommonServicesModule.java
@@ -61,6 +61,7 @@ public class CommonServicesModule extends AbstractModule {
install(new MimeMessageModule());
install(new ClockModule());
install(new PeriodicalHealthChecksModule());
+ install(new ErrorMailRepositoryEmptyHealthCheckModule());
bind(FileSystem.class).toInstance(fileSystem);
bind(Configuration.class).toInstance(configuration);
diff --git a/server/container/guice/common/src/main/java/org/apache/james/modules/ErrorMailRepositoryEmptyHealthCheckModule.java b/server/container/guice/common/src/main/java/org/apache/james/modules/ErrorMailRepositoryEmptyHealthCheckModule.java
index 753d98989c..c85f9803d4 100644
--- a/server/container/guice/common/src/main/java/org/apache/james/modules/ErrorMailRepositoryEmptyHealthCheckModule.java
+++ b/server/container/guice/common/src/main/java/org/apache/james/modules/ErrorMailRepositoryEmptyHealthCheckModule.java
@@ -19,9 +19,9 @@
package org.apache.james.modules;
-import static org.apache.james.mailrepository.api.MailRepositoryEmptyHealthCheck.ErrorMailRepositoryEmptyHealthCheck;
-
import org.apache.james.core.healthcheck.HealthCheck;
+import org.apache.james.mailetcontainer.impl.JamesMailSpooler;
+import org.apache.james.mailrepository.api.EmptyErrorMailRepositoryHealthCheck;
import org.apache.james.mailrepository.api.MailRepositoryStore;
import com.google.inject.AbstractModule;
@@ -33,13 +33,13 @@ public class ErrorMailRepositoryEmptyHealthCheckModule extends AbstractModule {
@Override
protected void configure() {
- Multibinder.newSetBinder(binder(), HealthCheck.class).addBinding().to(ErrorMailRepositoryEmptyHealthCheck.class);
+ Multibinder.newSetBinder(binder(), HealthCheck.class).addBinding().to(EmptyErrorMailRepositoryHealthCheck.class);
}
@Singleton
@Provides
- ErrorMailRepositoryEmptyHealthCheck provideErrorMailRepositoryEmptyHealthCheck(MailRepositoryStore mailRepositoryStore) {
- return new ErrorMailRepositoryEmptyHealthCheck(mailRepositoryStore);
+ EmptyErrorMailRepositoryHealthCheck provideErrorMailRepositoryEmptyHealthCheck(MailRepositoryStore mailRepositoryStore, JamesMailSpooler.Configuration mailSpoolerConfiguration) {
+ return new EmptyErrorMailRepositoryHealthCheck(mailSpoolerConfiguration.getErrorRepositoryURL().getPath(), mailRepositoryStore);
}
}
diff --git a/server/protocols/webadmin/webadmin-core/src/main/java/org/apache/james/webadmin/utils/ParametersExtractor.java b/server/protocols/webadmin/webadmin-core/src/main/java/org/apache/james/webadmin/utils/ParametersExtractor.java
index 0cc0fc14de..7d16a1b691 100644
--- a/server/protocols/webadmin/webadmin-core/src/main/java/org/apache/james/webadmin/utils/ParametersExtractor.java
+++ b/server/protocols/webadmin/webadmin-core/src/main/java/org/apache/james/webadmin/utils/ParametersExtractor.java
@@ -18,9 +18,12 @@
****************************************************************/
package org.apache.james.webadmin.utils;
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
import java.util.Optional;
import java.util.function.Function;
+import org.apache.james.util.DurationParser;
import org.apache.james.util.streams.Limit;
import org.apache.james.util.streams.Offset;
import org.eclipse.jetty.http.HttpStatus;
@@ -51,6 +54,12 @@ public class ParametersExtractor {
return extractPositiveNumber(request, parameterName, Integer::valueOf);
}
+ public static Optional<Duration> extractDuration(Request request, String parameterName) {
+ return Optional.ofNullable(request.queryParams(parameterName))
+ .filter(s -> !s.isEmpty())
+ .map(raw -> DurationParser.parse(raw, ChronoUnit.SECONDS));
+ }
+
private static <T extends Number> Optional<T> extractPositiveNumber(Request request, String parameterName, Function<String, T> toNumber) {
try {
return Optional.ofNullable(request.queryParams(parameterName))
diff --git a/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/routes/EventDeadLettersRoutes.java b/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/routes/EventDeadLettersRoutes.java
index c5d88e229b..03ee708553 100644
--- a/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/routes/EventDeadLettersRoutes.java
+++ b/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/routes/EventDeadLettersRoutes.java
@@ -19,18 +19,22 @@
package org.apache.james.webadmin.routes;
+import static org.apache.james.webadmin.service.EventDeadLettersRedeliverService.RunningOptions;
+
import javax.inject.Inject;
import org.apache.james.events.EventDeadLetters;
import org.apache.james.events.EventSerializer;
import org.apache.james.events.Group;
import org.apache.james.task.TaskManager;
+import org.apache.james.util.streams.Limit;
import org.apache.james.webadmin.Routes;
import org.apache.james.webadmin.service.EventDeadLettersService;
import org.apache.james.webadmin.tasks.TaskFromRequestRegistry;
import org.apache.james.webadmin.tasks.TaskRegistrationKey;
import org.apache.james.webadmin.utils.ErrorResponder;
import org.apache.james.webadmin.utils.JsonTransformer;
+import org.apache.james.webadmin.utils.ParametersExtractor;
import org.apache.james.webadmin.utils.Responses;
import org.eclipse.jetty.http.HttpStatus;
@@ -40,6 +44,7 @@ import spark.Route;
import spark.Service;
public class EventDeadLettersRoutes implements Routes {
+
public static final String BASE_PATH = "/events/deadLetter";
private static final String GROUP_PARAM = ":group";
private static final String INSERTION_ID_PARAMETER = ":insertionId";
@@ -76,7 +81,7 @@ public class EventDeadLettersRoutes implements Routes {
}
public Route performActionOnAllEvents() {
- return TaskFromRequestRegistry.of(RE_DELIVER, request -> eventDeadLettersService.redeliverAllEvents())
+ return TaskFromRequestRegistry.of(RE_DELIVER, request -> eventDeadLettersService.redeliverAllEvents(parseRunningOptions(request)))
.asRoute(taskManager);
}
@@ -90,7 +95,7 @@ public class EventDeadLettersRoutes implements Routes {
}
public Route performActionOnGroupEvents() {
- return TaskFromRequestRegistry.of(RE_DELIVER, request -> eventDeadLettersService.redeliverGroupEvents(parseGroup(request)))
+ return TaskFromRequestRegistry.of(RE_DELIVER, request -> eventDeadLettersService.redeliverGroupEvents(parseGroup(request), parseRunningOptions(request)))
.asRoute(taskManager);
}
@@ -144,4 +149,10 @@ public class EventDeadLettersRoutes implements Routes {
.haltError();
}
}
+
+ private RunningOptions parseRunningOptions(Request request) {
+ return ParametersExtractor.extractPositiveInteger(request, "limit")
+ .map(limit -> new RunningOptions(Limit.from(limit)))
+ .orElse(RunningOptions.DEFAULT);
+ }
}
diff --git a/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/EventDeadLettersRedeliverAllTask.java b/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/EventDeadLettersRedeliverAllTask.java
index 3b89e38899..4c9e201554 100644
--- a/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/EventDeadLettersRedeliverAllTask.java
+++ b/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/EventDeadLettersRedeliverAllTask.java
@@ -19,6 +19,8 @@
package org.apache.james.webadmin.service;
+import static org.apache.james.webadmin.service.EventDeadLettersRedeliverService.RunningOptions;
+
import java.time.Clock;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicLong;
@@ -35,17 +37,19 @@ public class EventDeadLettersRedeliverAllTask implements Task {
private final EventRetriever eventRetriever;
private final AtomicLong successfulRedeliveriesCount;
private final AtomicLong failedRedeliveriesCount;
+ private final RunningOptions runningOptions;
- EventDeadLettersRedeliverAllTask(EventDeadLettersRedeliverService service) {
+ EventDeadLettersRedeliverAllTask(EventDeadLettersRedeliverService service, RunningOptions runningOptions) {
this.service = service;
this.eventRetriever = EventRetriever.allEvents();
this.successfulRedeliveriesCount = new AtomicLong(0L);
this.failedRedeliveriesCount = new AtomicLong(0L);
+ this.runningOptions = runningOptions;
}
@Override
public Result run() {
- return service.redeliverEvents(eventRetriever)
+ return service.redeliverEvents(eventRetriever, runningOptions)
.map(this::updateCounters)
.reduce(Result.COMPLETED, Task::combine)
.block();
@@ -70,6 +74,10 @@ public class EventDeadLettersRedeliverAllTask implements Task {
return TYPE;
}
+ public RunningOptions getRunningOptions() {
+ return runningOptions;
+ }
+
@Override
public Optional<TaskExecutionDetails.AdditionalInformation> details() {
return Optional.of(createAdditionalInformation());
@@ -79,6 +87,7 @@ public class EventDeadLettersRedeliverAllTask implements Task {
return new EventDeadLettersRedeliveryTaskAdditionalInformationForAll(
successfulRedeliveriesCount.get(),
failedRedeliveriesCount.get(),
- Clock.systemUTC().instant());
+ Clock.systemUTC().instant(),
+ getRunningOptions());
}
}
diff --git a/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/EventDeadLettersRedeliverAllTaskDTO.java b/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/EventDeadLettersRedeliverAllTaskDTO.java
index 020bbf57ad..ab6968251d 100644
--- a/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/EventDeadLettersRedeliverAllTaskDTO.java
+++ b/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/EventDeadLettersRedeliverAllTaskDTO.java
@@ -18,6 +18,10 @@
****************************************************************/
package org.apache.james.webadmin.service;
+import static org.apache.james.webadmin.service.EventDeadLettersRedeliverService.RunningOptions;
+
+import java.util.Optional;
+
import org.apache.james.json.DTOModule;
import org.apache.james.server.task.json.dto.TaskDTO;
import org.apache.james.server.task.json.dto.TaskDTOModule;
@@ -30,20 +34,27 @@ public class EventDeadLettersRedeliverAllTaskDTO implements TaskDTO {
return DTOModule
.forDomainObject(EventDeadLettersRedeliverAllTask.class)
.convertToDTO(EventDeadLettersRedeliverAllTaskDTO.class)
- .toDomainObjectConverter(dto -> new EventDeadLettersRedeliverAllTask(service))
- .toDTOConverter((domainObject, typeName) -> new EventDeadLettersRedeliverAllTaskDTO(typeName))
+ .toDomainObjectConverter(dto -> new EventDeadLettersRedeliverAllTask(service, dto.getRunningOptions()))
+ .toDTOConverter((domainObject, typeName) -> new EventDeadLettersRedeliverAllTaskDTO(typeName, domainObject.getRunningOptions()))
.typeName(EventDeadLettersRedeliverAllTask.TYPE.asString())
.withFactory(TaskDTOModule::new);
}
private final String type;
+ private final RunningOptions runningOptions;
- public EventDeadLettersRedeliverAllTaskDTO(@JsonProperty("type") String type) {
+ public EventDeadLettersRedeliverAllTaskDTO(@JsonProperty("type") String type,
+ @JsonProperty("runningOptions") RunningOptions runningOptions) {
this.type = type;
+ this.runningOptions = Optional.ofNullable(runningOptions).orElse(RunningOptions.DEFAULT);
}
@Override
public String getType() {
return type;
}
+
+ public RunningOptions getRunningOptions() {
+ return runningOptions;
+ }
}
diff --git a/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/EventDeadLettersRedeliverGroupTask.java b/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/EventDeadLettersRedeliverGroupTask.java
index 935ecc736f..27d0666771 100644
--- a/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/EventDeadLettersRedeliverGroupTask.java
+++ b/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/EventDeadLettersRedeliverGroupTask.java
@@ -19,6 +19,8 @@
package org.apache.james.webadmin.service;
+import static org.apache.james.webadmin.service.EventDeadLettersRedeliverService.RunningOptions;
+
import java.time.Clock;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicLong;
@@ -37,18 +39,20 @@ public class EventDeadLettersRedeliverGroupTask implements Task {
private final AtomicLong successfulRedeliveriesCount;
private final AtomicLong failedRedeliveriesCount;
private final Group group;
+ private final RunningOptions runningOptions;
- EventDeadLettersRedeliverGroupTask(EventDeadLettersRedeliverService service, Group group) {
+ EventDeadLettersRedeliverGroupTask(EventDeadLettersRedeliverService service, Group group, RunningOptions runningOptions) {
this.service = service;
this.group = group;
this.eventRetriever = EventRetriever.groupEvents(group);
+ this.runningOptions = runningOptions;
this.successfulRedeliveriesCount = new AtomicLong(0L);
this.failedRedeliveriesCount = new AtomicLong(0L);
}
@Override
public Result run() {
- return service.redeliverEvents(eventRetriever)
+ return service.redeliverEvents(eventRetriever, runningOptions)
.map(this::updateCounters)
.reduce(Result.COMPLETED, Task::combine)
.block();
@@ -82,11 +86,16 @@ public class EventDeadLettersRedeliverGroupTask implements Task {
return group;
}
+ public RunningOptions getRunningOptions() {
+ return runningOptions;
+ }
+
private EventDeadLettersRedeliveryTaskAdditionalInformation createAdditionalInformation() {
return new EventDeadLettersRedeliveryTaskAdditionalInformationForGroup(
successfulRedeliveriesCount.get(),
failedRedeliveriesCount.get(),
eventRetriever.forGroup(),
- Clock.systemUTC().instant());
+ Clock.systemUTC().instant(),
+ getRunningOptions());
}
}
diff --git a/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/EventDeadLettersRedeliverGroupTaskDTO.java b/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/EventDeadLettersRedeliverGroupTaskDTO.java
index 62ff4e3f2d..aa3912279e 100644
--- a/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/EventDeadLettersRedeliverGroupTaskDTO.java
+++ b/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/EventDeadLettersRedeliverGroupTaskDTO.java
@@ -18,6 +18,10 @@
****************************************************************/
package org.apache.james.webadmin.service;
+import static org.apache.james.webadmin.service.EventDeadLettersRedeliverService.RunningOptions;
+
+import java.util.Optional;
+
import org.apache.james.events.Group;
import org.apache.james.json.DTOModule;
import org.apache.james.server.task.json.dto.TaskDTO;
@@ -32,22 +36,26 @@ public class EventDeadLettersRedeliverGroupTaskDTO implements TaskDTO {
.forDomainObject(EventDeadLettersRedeliverGroupTask.class)
.convertToDTO(EventDeadLettersRedeliverGroupTaskDTO.class)
.toDomainObjectConverter(dto -> dto.toDomainObject(service))
- .toDTOConverter((domainObject, typeName) -> new EventDeadLettersRedeliverGroupTaskDTO(typeName, domainObject.getGroup().asString()))
+ .toDTOConverter((domainObject, typeName) -> new EventDeadLettersRedeliverGroupTaskDTO(typeName, domainObject.getGroup().asString(), domainObject.getRunningOptions()))
.typeName(EventDeadLettersRedeliverGroupTask.TYPE.asString())
.withFactory(TaskDTOModule::new);
}
private final String type;
private final String group;
+ private final RunningOptions runningOptions;
- public EventDeadLettersRedeliverGroupTaskDTO(@JsonProperty("type") String type, @JsonProperty("group") String group) {
+ public EventDeadLettersRedeliverGroupTaskDTO(@JsonProperty("type") String type,
+ @JsonProperty("group") String group,
+ @JsonProperty("runningOptions") RunningOptions runningOptions) {
this.type = type;
this.group = group;
+ this.runningOptions = Optional.ofNullable(runningOptions).orElse(RunningOptions.DEFAULT);
}
EventDeadLettersRedeliverGroupTask toDomainObject(EventDeadLettersRedeliverService service) {
try {
- return new EventDeadLettersRedeliverGroupTask(service, Group.deserialize(getGroup()));
+ return new EventDeadLettersRedeliverGroupTask(service, Group.deserialize(getGroup()), runningOptions);
} catch (Group.GroupDeserializationException e) {
throw new RuntimeException(e);
}
@@ -61,4 +69,8 @@ public class EventDeadLettersRedeliverGroupTaskDTO implements TaskDTO {
public String getGroup() {
return group;
}
+
+ public RunningOptions getRunningOptions() {
+ return runningOptions;
+ }
}
diff --git a/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/EventDeadLettersRedeliverOneTask.java b/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/EventDeadLettersRedeliverOneTask.java
index b644055a80..2ce833cc09 100644
--- a/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/EventDeadLettersRedeliverOneTask.java
+++ b/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/EventDeadLettersRedeliverOneTask.java
@@ -50,7 +50,7 @@ public class EventDeadLettersRedeliverOneTask implements Task {
@Override
public Result run() {
- return service.redeliverEvents(eventRetriever)
+ return service.redeliverEvents(eventRetriever, EventDeadLettersRedeliverService.RunningOptions.DEFAULT)
.map(this::updateCounters)
.reduce(Result.COMPLETED, Task::combine)
.block();
diff --git a/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/EventDeadLettersRedeliverService.java b/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/EventDeadLettersRedeliverService.java
index c6f239f390..47c7f53ecf 100644
--- a/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/EventDeadLettersRedeliverService.java
+++ b/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/EventDeadLettersRedeliverService.java
@@ -19,7 +19,7 @@
package org.apache.james.webadmin.service;
-import static org.apache.james.util.ReactorUtils.DEFAULT_CONCURRENCY;
+import java.util.Optional;
import javax.inject.Inject;
@@ -28,15 +28,40 @@ import org.apache.james.events.EventBus;
import org.apache.james.events.EventDeadLetters;
import org.apache.james.events.Group;
import org.apache.james.task.Task;
+import org.apache.james.util.streams.Limit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.annotations.VisibleForTesting;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
public class EventDeadLettersRedeliverService {
+ public static class RunningOptions {
+ public static RunningOptions DEFAULT = new RunningOptions(Limit.unlimited());
+ private final Limit limit;
+
+ public RunningOptions(Limit limit) {
+ this.limit = limit;
+ }
+
+ @JsonCreator
+ public RunningOptions(@JsonProperty("limit") Integer limit) {
+ this.limit = Limit.from(Optional.ofNullable(limit));
+ }
+
+ @JsonProperty("limit")
+ public Optional<Integer> limitValue() {
+ return limit.getLimit();
+ }
+
+ public Limit limit() {
+ return limit;
+ }
+ }
private static final Logger LOGGER = LoggerFactory.getLogger(EventDeadLettersRedeliverService.class);
@@ -50,9 +75,9 @@ public class EventDeadLettersRedeliverService {
this.deadLetters = deadLetters;
}
- Flux<Task.Result> redeliverEvents(EventRetriever eventRetriever) {
- return eventRetriever.retrieveEvents(deadLetters)
- .flatMap(entry -> redeliverGroupEvents(entry.getT1(), entry.getT2(), entry.getT3()), DEFAULT_CONCURRENCY);
+ Flux<Task.Result> redeliverEvents(EventRetriever eventRetriever, RunningOptions runningOptions) {
+ return runningOptions.limit().applyOnFlux(eventRetriever.retrieveEvents(deadLetters))
+ .flatMap(tuple3 -> redeliverGroupEvents(tuple3.getT1(), tuple3.getT2(), tuple3.getT3()));
}
private Mono<Task.Result> redeliverGroupEvents(Group group, Event event, EventDeadLetters.InsertionId insertionId) {
diff --git a/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/EventDeadLettersRedeliveryTaskAdditionalInformationDTO.java b/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/EventDeadLettersRedeliveryTaskAdditionalInformationDTO.java
index 40035d395f..5f2c622fc0 100644
--- a/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/EventDeadLettersRedeliveryTaskAdditionalInformationDTO.java
+++ b/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/EventDeadLettersRedeliveryTaskAdditionalInformationDTO.java
@@ -1,5 +1,7 @@
package org.apache.james.webadmin.service;
+import static org.apache.james.webadmin.service.EventDeadLettersRedeliverService.RunningOptions;
+
import java.time.Instant;
import java.util.Optional;
@@ -16,13 +18,22 @@ public class EventDeadLettersRedeliveryTaskAdditionalInformationDTO implements A
public static class EventDeadLettersRedeliveryTaskAdditionalInformationForAll extends EventDeadLettersRedeliveryTaskAdditionalInformation {
public static class DTO extends EventDeadLettersRedeliveryTaskAdditionalInformationDTO {
+
+ private final RunningOptions runningOptions;
+
public DTO(@JsonProperty("type") String type,
@JsonProperty("successfulRedeliveriesCount") long successfulRedeliveriesCount,
@JsonProperty("failedRedeliveriesCount") long failedRedeliveriesCount,
@JsonProperty("group") Optional<String> group,
@JsonProperty("insertionId") Optional<String> insertionId,
- @JsonProperty("timestamp") Instant timestamp) {
- super(type, successfulRedeliveriesCount, failedRedeliveriesCount, group,insertionId, timestamp);
+ @JsonProperty("timestamp") Instant timestamp,
+ @JsonProperty("runningOptions") RunningOptions runningOptions) {
+ super(type, successfulRedeliveriesCount, failedRedeliveriesCount, group, insertionId, timestamp);
+ this.runningOptions = runningOptions;
+ }
+
+ public RunningOptions getRunningOptions() {
+ return runningOptions;
}
}
@@ -36,27 +47,42 @@ public class EventDeadLettersRedeliveryTaskAdditionalInformationDTO implements A
domainObject.getFailedRedeliveriesCount(),
domainObject.getGroup(),
domainObject.getInsertionId(),
- domainObject.timestamp()))
+ domainObject.timestamp(),
+ domainObject.getRunningOptions()))
.typeName(EventDeadLettersRedeliverAllTask.TYPE.asString())
.withFactory(AdditionalInformationDTOModule::new);
}
+ private final RunningOptions runningOptions;
- EventDeadLettersRedeliveryTaskAdditionalInformationForAll(long successfulRedeliveriesCount, long failedRedeliveriesCount, Instant timestamp) {
+ EventDeadLettersRedeliveryTaskAdditionalInformationForAll(long successfulRedeliveriesCount, long failedRedeliveriesCount, Instant timestamp, RunningOptions runningOptions) {
super(successfulRedeliveriesCount, failedRedeliveriesCount, Optional.empty(), Optional.empty(), timestamp);
+ this.runningOptions = runningOptions;
+ }
+
+ public RunningOptions getRunningOptions() {
+ return runningOptions;
}
}
public static class EventDeadLettersRedeliveryTaskAdditionalInformationForGroup extends EventDeadLettersRedeliveryTaskAdditionalInformation {
public static class DTO extends EventDeadLettersRedeliveryTaskAdditionalInformationDTO {
+ private final RunningOptions runningOptions;
+
public DTO(@JsonProperty("type") String type,
@JsonProperty("successfulRedeliveriesCount") long successfulRedeliveriesCount,
@JsonProperty("failedRedeliveriesCount") long failedRedeliveriesCount,
@JsonProperty("group") Optional<String> group,
@JsonProperty("insertionId") Optional<String> insertionId,
- @JsonProperty("timestamp") Instant timestamp) {
- super(type, successfulRedeliveriesCount, failedRedeliveriesCount, group,insertionId, timestamp);
+ @JsonProperty("timestamp") Instant timestamp,
+ @JsonProperty("runningOptions") RunningOptions runningOptions) {
+ super(type, successfulRedeliveriesCount, failedRedeliveriesCount, group, insertionId, timestamp);
+ this.runningOptions = runningOptions;
+ }
+
+ public RunningOptions getRunningOptions() {
+ return runningOptions;
}
}
@@ -69,14 +95,21 @@ public class EventDeadLettersRedeliveryTaskAdditionalInformationDTO implements A
domainObject.getFailedRedeliveriesCount(),
domainObject.getGroup(),
domainObject.getInsertionId(),
- domainObject.timestamp()))
+ domainObject.timestamp(),
+ domainObject.getRunningOptions()))
.typeName(EventDeadLettersRedeliverGroupTask.TYPE.asString())
.withFactory(AdditionalInformationDTOModule::new);
}
+ private final RunningOptions runningOptions;
- EventDeadLettersRedeliveryTaskAdditionalInformationForGroup(long successfulRedeliveriesCount, long failedRedeliveriesCount, Optional<Group> group, Instant timestamp) {
+ EventDeadLettersRedeliveryTaskAdditionalInformationForGroup(long successfulRedeliveriesCount, long failedRedeliveriesCount, Optional<Group> group, Instant timestamp, RunningOptions runningOptions) {
super(successfulRedeliveriesCount, failedRedeliveriesCount, group, Optional.empty(), timestamp);
+ this.runningOptions = runningOptions;
+ }
+
+ public RunningOptions getRunningOptions() {
+ return runningOptions;
}
}
@@ -88,7 +121,7 @@ public class EventDeadLettersRedeliveryTaskAdditionalInformationDTO implements A
@JsonProperty("group") Optional<String> group,
@JsonProperty("insertionId") Optional<String> insertionId,
@JsonProperty("timestamp") Instant timestamp) {
- super(type, successfulRedeliveriesCount, failedRedeliveriesCount, group,insertionId, timestamp);
+ super(type, successfulRedeliveriesCount, failedRedeliveriesCount, group, insertionId, timestamp);
}
}
@@ -117,19 +150,21 @@ public class EventDeadLettersRedeliveryTaskAdditionalInformationDTO implements A
}
}
- private static EventDeadLettersRedeliveryTaskAdditionalInformationForAll fromAll(EventDeadLettersRedeliveryTaskAdditionalInformationDTO dto) {
+ private static EventDeadLettersRedeliveryTaskAdditionalInformationForAll fromAll(EventDeadLettersRedeliveryTaskAdditionalInformationForAll.DTO dto) {
return new EventDeadLettersRedeliveryTaskAdditionalInformationForAll(
- dto.successfulRedeliveriesCount,
- dto.failedRedeliveriesCount,
- dto.timestamp);
+ dto.getSuccessfulRedeliveriesCount(),
+ dto.getFailedRedeliveriesCount(),
+ dto.getTimestamp(),
+ dto.runningOptions);
}
- private static EventDeadLettersRedeliveryTaskAdditionalInformationForGroup fromGroup(EventDeadLettersRedeliveryTaskAdditionalInformationDTO dto) {
+ private static EventDeadLettersRedeliveryTaskAdditionalInformationForGroup fromGroup(EventDeadLettersRedeliveryTaskAdditionalInformationForGroup.DTO dto) {
return new EventDeadLettersRedeliveryTaskAdditionalInformationForGroup(
- dto.successfulRedeliveriesCount,
- dto.failedRedeliveriesCount,
- dto.group.map(Throwing.function(Group::deserialize).sneakyThrow()),
- dto.timestamp);
+ dto.getSuccessfulRedeliveriesCount(),
+ dto.getFailedRedeliveriesCount(),
+ dto.getGroup().map(Throwing.function(Group::deserialize).sneakyThrow()),
+ dto.getTimestamp(),
+ dto.runningOptions);
}
private static EventDeadLettersRedeliveryTaskAdditionalInformationForOne fromOne(EventDeadLettersRedeliveryTaskAdditionalInformationDTO dto) {
diff --git a/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/EventDeadLettersService.java b/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/EventDeadLettersService.java
index a48dbffdde..57ba1f2737 100644
--- a/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/EventDeadLettersService.java
+++ b/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/EventDeadLettersService.java
@@ -68,12 +68,12 @@ public class EventDeadLettersService {
deadLetters.remove(group, insertionId).block();
}
- public Task redeliverAllEvents() {
- return new EventDeadLettersRedeliverAllTask(redeliverService);
+ public Task redeliverAllEvents(EventDeadLettersRedeliverService.RunningOptions runningOptions) {
+ return new EventDeadLettersRedeliverAllTask(redeliverService, runningOptions);
}
- public Task redeliverGroupEvents(Group group) {
- return new EventDeadLettersRedeliverGroupTask(redeliverService, group);
+ public Task redeliverGroupEvents(Group group, EventDeadLettersRedeliverService.RunningOptions runningOptions) {
+ return new EventDeadLettersRedeliverGroupTask(redeliverService, group, runningOptions);
}
public Task redeliverSingleEvent(Group group, EventDeadLetters.InsertionId insertionId) {
diff --git a/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/EventDeadLettersRoutesTest.java b/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/EventDeadLettersRoutesTest.java
index e0eb519dd2..40f798b31e 100644
--- a/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/EventDeadLettersRoutesTest.java
+++ b/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/EventDeadLettersRoutesTest.java
@@ -568,6 +568,52 @@ class EventDeadLettersRoutesTest {
.body("message", is("Invalid arguments supplied in the user request"))
.body("details", is("'action' query parameter is compulsory. Supported values are [reDeliver]"));
}
+
+ @Test
+ void postRedeliverAllEventsShouldSuccessWhenProvideLimitParameter() {
+ deadLetters.store(groupA, EVENT_2).block();
+ deadLetters.store(groupA, EVENT_2).block();
+
+ String taskId = with()
+ .queryParam("action", EVENTS_ACTION)
+ .queryParam("limit", 1)
+ .post("/events/deadLetter")
+ .jsonPath()
+ .get("taskId");
+
+ given()
+ .basePath(TasksRoutes.BASE)
+ .when()
+ .get(taskId + "/await")
+ .then()
+ .body("status", is("completed"))
+ .body("additionalInformation.successfulRedeliveriesCount", is(1))
+ .body("additionalInformation.failedRedeliveriesCount", is(0))
+ .body("additionalInformation.runningOptions.limit", is(1));
+
+ when()
+ .get("/events/deadLetter/groups")
+ .then()
+ .statusCode(HttpStatus.OK_200)
+ .contentType(ContentType.JSON)
+ .body(".", hasSize(1));
+
+ assertThat(eventCollectorA.getEvents()).hasSize(1);
+ }
+
+ @Test
+ void postRedeliverAllEventsShouldSuccessWhenInvalidLimitParameter() {
+ with()
+ .queryParam("action", EVENTS_ACTION)
+ .queryParam("limit", "invalid")
+ .post("/events/deadLetter")
+ .then()
+ .statusCode(HttpStatus.BAD_REQUEST_400)
+ .contentType(ContentType.JSON)
+ .body("statusCode", is(400))
+ .body("type", is(ErrorResponder.ErrorType.INVALID_ARGUMENT.getType()))
+ .body("message", is("Can not parse limit"));
+ }
}
@Nested
@@ -788,6 +834,51 @@ class EventDeadLettersRoutesTest {
.contentType(ContentType.JSON)
.body(".", hasSize(1));
}
+
+ @Test
+ void postRedeliverGroupEventsShouldSuccessWhenProvideLimitParameter() {
+ deadLetters.store(groupA, EVENT_1).block();
+ deadLetters.store(groupA, EVENT_2).block();
+
+ String taskId = with()
+ .queryParam("action", EVENTS_ACTION)
+ .queryParam("limit", 1)
+ .post("/events/deadLetter/groups/" + SERIALIZED_GROUP_A)
+ .jsonPath()
+ .get("taskId");
+
+ given()
+ .basePath(TasksRoutes.BASE)
+ .when()
+ .get(taskId + "/await")
+ .then()
+ .body("status", is("completed"))
+ .body("additionalInformation.successfulRedeliveriesCount", is(1))
+ .body("additionalInformation.failedRedeliveriesCount", is(0))
+ .body("additionalInformation.runningOptions.limit", is(1))
+ .body("additionalInformation.group", is(SERIALIZED_GROUP_A));
+
+ when()
+ .get("/events/deadLetter/groups/" + SERIALIZED_GROUP_A + "/" + INSERTION_UUID_1)
+ .then()
+ .statusCode(HttpStatus.NOT_FOUND_404);
+
+ assertThat(eventCollector.getEvents()).hasSize(1);
+ }
+
+ @Test
+ void postRedeliverGroupEventsShouldFailWhenInvalidLimitParameter() {
+ with()
+ .queryParam("action", EVENTS_ACTION)
+ .queryParam("limit", "invalid")
+ .post("/events/deadLetter/groups/" + SERIALIZED_GROUP_A)
+ .then()
+ .statusCode(HttpStatus.BAD_REQUEST_400)
+ .contentType(ContentType.JSON)
+ .body("statusCode", is(400))
+ .body("type", is(ErrorResponder.ErrorType.INVALID_ARGUMENT.getType()))
+ .body("message", is("Can not parse limit"));
+ }
}
@Nested
diff --git a/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/service/EventDeadLettersRedeliverTaskTest.java b/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/service/EventDeadLettersRedeliverTaskTest.java
index bd70718a53..48f0d2cc99 100644
--- a/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/service/EventDeadLettersRedeliverTaskTest.java
+++ b/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/service/EventDeadLettersRedeliverTaskTest.java
@@ -19,6 +19,7 @@
package org.apache.james.webadmin.service;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
import static org.mockito.Mockito.mock;
import java.time.Instant;
@@ -28,6 +29,9 @@ import org.apache.james.JsonSerializationVerifier;
import org.apache.james.events.EventDeadLetters;
import org.apache.james.events.Group;
import org.apache.james.mailbox.events.GenericGroup;
+import org.apache.james.server.task.json.JsonTaskSerializer;
+import org.apache.james.util.streams.Limit;
+import org.apache.james.webadmin.service.EventDeadLettersRedeliverService.RunningOptions;
import org.apache.james.webadmin.service.EventDeadLettersRedeliveryTaskAdditionalInformationDTO.EventDeadLettersRedeliveryTaskAdditionalInformationForAll;
import org.apache.james.webadmin.service.EventDeadLettersRedeliveryTaskAdditionalInformationDTO.EventDeadLettersRedeliveryTaskAdditionalInformationForGroup;
import org.apache.james.webadmin.service.EventDeadLettersRedeliveryTaskAdditionalInformationDTO.EventDeadLettersRedeliveryTaskAdditionalInformationForOne;
@@ -36,15 +40,9 @@ import org.junit.jupiter.api.Test;
class EventDeadLettersRedeliverTaskTest {
private static final Instant TIMESTAMP = Instant.parse("2018-11-13T12:00:55Z");
- private static final String SERIALIZED_ALL = "{\"type\":\"event-dead-letters-redeliver-all\"}";
- private static final String SERIALIZED_GROUP = "{\"type\":\"event-dead-letters-redeliver-group\",\"group\":\"org.apache.james.mailbox.events.GenericGroup-abc\"}";
- private static final String SERIALIZED_ONE = "{\"type\":\"event-dead-letters-redeliver-one\",\"group\":\"org.apache.james.mailbox.events.GenericGroup-abc\",\"insertionId\":\"fcbc3c92-e9a0-4ece-94ed-6e6b45045258\"}";
- private static final String SERIALIZED_TASK_ADDITIONAL_INFORMATION_ALL = "{\"type\":\"event-dead-letters-redeliver-all\",\"successfulRedeliveriesCount\":10,\"failedRedeliveriesCount\":4, \"timestamp\":\"2018-11-13T12:00:55Z\"}";
- private static final String SERIALIZED_TASK_ADDITIONAL_INFORMATION_GROUP = "{\"type\":\"event-dead-letters-redeliver-group\",\"successfulRedeliveriesCount\":10,\"failedRedeliveriesCount\":4,\"group\":\"org.apache.james.mailbox.events.GenericGroup-foo\", \"timestamp\":\"2018-11-13T12:00:55Z\"}";
- private static final String SERIALIZED_TASK_ADDITIONAL_INFORMATION_ONE = "{\"type\":\"event-dead-letters-redeliver-one\",\"successfulRedeliveriesCount\":10,\"failedRedeliveriesCount\":4,\"group\":\"org.apache.james.mailbox.events.GenericGroup-foo\",\"insertionId\":\"53db3dd9-80eb-476f-b25a-722ad364905a\", \"timestamp\":\"2018-11-13T12:00:55Z\"}";
private static final EventDeadLettersRedeliverService SERVICE = mock(EventDeadLettersRedeliverService.class);
- private static final EventDeadLettersRedeliverAllTask TASK_ALL = new EventDeadLettersRedeliverAllTask(SERVICE);
- private static final EventDeadLettersRedeliverGroupTask TASK_GROUP = new EventDeadLettersRedeliverGroupTask(SERVICE, new GenericGroup("abc"));
+ private static final EventDeadLettersRedeliverAllTask TASK_ALL = new EventDeadLettersRedeliverAllTask(SERVICE, EventDeadLettersRedeliverService.RunningOptions.DEFAULT);
+ private static final EventDeadLettersRedeliverGroupTask TASK_GROUP = new EventDeadLettersRedeliverGroupTask(SERVICE, new GenericGroup("abc"), EventDeadLettersRedeliverService.RunningOptions.DEFAULT);
private static final EventDeadLettersRedeliverOneTask TASK_ONE = new EventDeadLettersRedeliverOneTask(SERVICE, new GenericGroup("abc"), EventDeadLetters.InsertionId.of("fcbc3c92-e9a0-4ece-94ed-6e6b45045258"));
private static final long SUCCESSFUL_REDELIVERY_COUNT = 10L;
@@ -62,39 +60,154 @@ class EventDeadLettersRedeliverTaskTest {
void redeliverAllTaskShouldMatchJsonSerializationContract() throws Exception {
JsonSerializationVerifier.dtoModule(EventDeadLettersRedeliverAllTaskDTO.module(SERVICE))
.bean(TASK_ALL)
- .json(SERIALIZED_ALL)
+ .json("{" +
+ " \"type\": \"event-dead-letters-redeliver-all\"," +
+ " \"runningOptions\": {}" +
+ "}")
.verify();
+
+ EventDeadLettersRedeliverAllTask taskAllWithLimit = new EventDeadLettersRedeliverAllTask(SERVICE, new RunningOptions(Limit.limit(10)));
+
+ JsonSerializationVerifier.dtoModule(EventDeadLettersRedeliverAllTaskDTO.module(SERVICE))
+ .bean(taskAllWithLimit)
+ .json("{\"type\":\"event-dead-letters-redeliver-all\", \"runningOptions\":{\"limit\": 10}}")
+ .verify();
+ }
+
+ @Test
+ void redeliverAllTaskShouldDeserializationSuccess() throws Exception {
+ JsonTaskSerializer serializer = JsonTaskSerializer.of(EventDeadLettersRedeliverAllTaskDTO.module(SERVICE));
+
+ assertThat(serializer.deserialize("{\"type\":\"event-dead-letters-redeliver-all\", \"runningOptions\":{\"limit\": 10}}"))
+ .usingRecursiveComparison()
+ .isEqualTo(new EventDeadLettersRedeliverAllTask(SERVICE, new RunningOptions(Limit.limit(10))));
+
+ assertThat(serializer.deserialize("{\"type\":\"event-dead-letters-redeliver-all\", \"runningOptions\":{}}"))
+ .usingRecursiveComparison()
+ .isEqualTo(new EventDeadLettersRedeliverAllTask(SERVICE, new RunningOptions(Limit.unlimited())));
+
+ assertThat(serializer.deserialize("{\"type\":\"event-dead-letters-redeliver-all\"}"))
+ .usingRecursiveComparison()
+ .isEqualTo(new EventDeadLettersRedeliverAllTask(SERVICE, new RunningOptions(Limit.unlimited())));
}
@Test
void redeliverGroupTaskShouldMatchJsonSerializationContract() throws Exception {
JsonSerializationVerifier.dtoModule(EventDeadLettersRedeliverGroupTaskDTO.module(SERVICE))
.bean(TASK_GROUP)
- .json(SERIALIZED_GROUP)
+ .json("{" +
+ " \"type\": \"event-dead-letters-redeliver-group\"," +
+ " \"group\": \"org.apache.james.mailbox.events.GenericGroup-abc\"," +
+ " \"runningOptions\": {}" +
+ "}")
+ .verify();
+
+ EventDeadLettersRedeliverGroupTask taskGroupWithLimit = new EventDeadLettersRedeliverGroupTask(SERVICE, new GenericGroup("abc"), new RunningOptions(Limit.limit(10)));
+
+ JsonSerializationVerifier.dtoModule(EventDeadLettersRedeliverGroupTaskDTO.module(SERVICE))
+ .bean(taskGroupWithLimit)
+ .json("{" +
+ " \"type\": \"event-dead-letters-redeliver-group\"," +
+ " \"group\": \"org.apache.james.mailbox.events.GenericGroup-abc\"," +
+ " \"runningOptions\": {" +
+ " \"limit\": 10" +
+ " }" +
+ "}")
.verify();
}
+ @Test
+ void redeliverGroupTaskShouldDeserializationSuccess() throws Exception {
+ JsonTaskSerializer serializer = JsonTaskSerializer.of(EventDeadLettersRedeliverGroupTaskDTO.module(SERVICE));
+
+ assertThat(serializer.deserialize("{" +
+ " \"type\": \"event-dead-letters-redeliver-group\"," +
+ " \"group\": \"org.apache.james.mailbox.events.GenericGroup-abc\"," +
+ " \"runningOptions\": {" +
+ " \"limit\": 10" +
+ " }" +
+ "}"))
+ .usingRecursiveComparison()
+ .isEqualTo(new EventDeadLettersRedeliverGroupTask(SERVICE, new GenericGroup("abc"), new RunningOptions(Limit.limit(10))));
+
+ assertThat(serializer.deserialize("{" +
+ " \"type\": \"event-dead-letters-redeliver-group\"," +
+ " \"group\": \"org.apache.james.mailbox.events.GenericGroup-abc\"," +
+ " \"runningOptions\": {}" +
+ "}"))
+ .usingRecursiveComparison()
+ .isEqualTo(new EventDeadLettersRedeliverGroupTask(SERVICE, new GenericGroup("abc"), new RunningOptions(Limit.unlimited())));
+
+
+ assertThat(serializer.deserialize("{" +
+ " \"type\": \"event-dead-letters-redeliver-group\"," +
+ " \"group\": \"org.apache.james.mailbox.events.GenericGroup-abc\"" +
+ "}"))
+ .usingRecursiveComparison()
+ .isEqualTo(new EventDeadLettersRedeliverGroupTask(SERVICE, new GenericGroup("abc"), new RunningOptions(Limit.unlimited())));
+ }
+
@Test
void redeliverOneTaskShouldMatchJsonSerializationContract() throws Exception {
JsonSerializationVerifier.dtoModule(EventDeadLettersRedeliverOneTaskDTO.module(SERVICE))
.bean(TASK_ONE)
- .json(SERIALIZED_ONE)
+ .json("{" +
+ " \"type\": \"event-dead-letters-redeliver-one\"," +
+ " \"group\": \"org.apache.james.mailbox.events.GenericGroup-abc\"," +
+ " \"insertionId\": \"fcbc3c92-e9a0-4ece-94ed-6e6b45045258\"" +
+ "}")
.verify();
}
@Test
void redeliverAllAdditionalInformationShouldMatchJsonSerializationContract() throws Exception {
JsonSerializationVerifier.dtoModule(EventDeadLettersRedeliveryTaskAdditionalInformationForAll.module())
- .bean(new EventDeadLettersRedeliveryTaskAdditionalInformationForAll(SUCCESSFUL_REDELIVERY_COUNT, FAILED_REDELIVERY_COUNT, TIMESTAMP))
- .json(SERIALIZED_TASK_ADDITIONAL_INFORMATION_ALL)
+ .bean(new EventDeadLettersRedeliveryTaskAdditionalInformationForAll(SUCCESSFUL_REDELIVERY_COUNT, FAILED_REDELIVERY_COUNT, TIMESTAMP, RunningOptions.DEFAULT))
+ .json("{" +
+ " \"type\": \"event-dead-letters-redeliver-all\"," +
+ " \"successfulRedeliveriesCount\": 10," +
+ " \"failedRedeliveriesCount\": 4," +
+ " \"timestamp\": \"2018-11-13T12:00:55Z\"," +
+ " \"runningOptions\":{}" +
+ "}")
+ .verify();
+
+ JsonSerializationVerifier.dtoModule(EventDeadLettersRedeliveryTaskAdditionalInformationForAll.module())
+ .bean(new EventDeadLettersRedeliveryTaskAdditionalInformationForAll(SUCCESSFUL_REDELIVERY_COUNT, FAILED_REDELIVERY_COUNT, TIMESTAMP, new RunningOptions(Limit.limit(10))))
+ .json("{" +
+ " \"type\": \"event-dead-letters-redeliver-all\"," +
+ " \"successfulRedeliveriesCount\": 10," +
+ " \"failedRedeliveriesCount\": 4," +
+ " \"timestamp\": \"2018-11-13T12:00:55Z\"," +
+ " \"runningOptions\":{ \"limit\": 10}" +
+ "}")
.verify();
}
@Test
void redeliverGroupAdditionalInformationShouldMatchJsonSerializationContract() throws Exception {
JsonSerializationVerifier.dtoModule(EventDeadLettersRedeliveryTaskAdditionalInformationForGroup.module())
- .bean(new EventDeadLettersRedeliveryTaskAdditionalInformationForGroup(SUCCESSFUL_REDELIVERY_COUNT, FAILED_REDELIVERY_COUNT, SOME_GROUP, TIMESTAMP))
- .json(SERIALIZED_TASK_ADDITIONAL_INFORMATION_GROUP)
+ .bean(new EventDeadLettersRedeliveryTaskAdditionalInformationForGroup(SUCCESSFUL_REDELIVERY_COUNT, FAILED_REDELIVERY_COUNT, SOME_GROUP, TIMESTAMP, RunningOptions.DEFAULT))
+ .json("{" +
+ " \"type\": \"event-dead-letters-redeliver-group\"," +
+ " \"successfulRedeliveriesCount\": 10," +
+ " \"failedRedeliveriesCount\": 4," +
+ " \"group\": \"org.apache.james.mailbox.events.GenericGroup-foo\"," +
+ " \"timestamp\": \"2018-11-13T12:00:55Z\"," +
+ " \"runningOptions\": {}" +
+ "}")
+ .verify();
+
+ JsonSerializationVerifier.dtoModule(EventDeadLettersRedeliveryTaskAdditionalInformationForGroup.module())
+ .bean(new EventDeadLettersRedeliveryTaskAdditionalInformationForGroup(SUCCESSFUL_REDELIVERY_COUNT, FAILED_REDELIVERY_COUNT, SOME_GROUP, TIMESTAMP, new RunningOptions(Limit.limit(10))))
+ .json("{" +
+ " \"type\": \"event-dead-letters-redeliver-group\"," +
+ " \"successfulRedeliveriesCount\": 10," +
+ " \"failedRedeliveriesCount\": 4," +
+ " \"group\": \"org.apache.james.mailbox.events.GenericGroup-foo\"," +
+ " \"timestamp\": \"2018-11-13T12:00:55Z\"," +
+ " \"runningOptions\": {\"limit\": 10}" +
+ "}")
.verify();
}
@@ -102,7 +215,14 @@ class EventDeadLettersRedeliverTaskTest {
void redeliverOneAdditionalInformationShouldMatchJsonSerializationContract() throws Exception {
JsonSerializationVerifier.dtoModule(EventDeadLettersRedeliveryTaskAdditionalInformationForOne.module())
.bean(new EventDeadLettersRedeliveryTaskAdditionalInformationForOne(SUCCESSFUL_REDELIVERY_COUNT, FAILED_REDELIVERY_COUNT, SOME_GROUP, SOME_INSERTION_ID, TIMESTAMP))
- .json(SERIALIZED_TASK_ADDITIONAL_INFORMATION_ONE)
+ .json("{" +
+ " \"type\": \"event-dead-letters-redeliver-one\"," +
+ " \"successfulRedeliveriesCount\": 10," +
+ " \"failedRedeliveriesCount\": 4," +
+ " \"group\": \"org.apache.james.mailbox.events.GenericGroup-foo\"," +
+ " \"insertionId\": \"53db3dd9-80eb-476f-b25a-722ad364905a\"," +
+ " \"timestamp\": \"2018-11-13T12:00:55Z\"" +
+ "}")
.verify();
}
}
\ No newline at end of file
diff --git a/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/routes/MailRepositoriesRoutes.java b/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/routes/MailRepositoriesRoutes.java
index 97c563425d..1e416f990f 100644
--- a/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/routes/MailRepositoriesRoutes.java
+++ b/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/routes/MailRepositoriesRoutes.java
@@ -330,7 +330,8 @@ public class MailRepositoriesRoutes implements Routes {
private ReprocessingService.Configuration extractConfiguration(Request request) {
return new ReprocessingService.Configuration(parseTargetQueue(request),
parseTargetProcessor(request),
- parseConsume(request).orElse(true));
+ parseConsume(request).orElse(true),
+ parseLimit(request));
}
public void defineReprocessOne() {
@@ -376,4 +377,8 @@ public class MailRepositoriesRoutes implements Routes {
private MailRepositoryPath decodedRepositoryPath(Request request) throws UnsupportedEncodingException {
return MailRepositoryPath.fromEncoded(request.params("encodedPath"));
}
+
+ private Limit parseLimit(Request request) {
+ return Limit.from(ParametersExtractor.extractPositiveInteger(request, "limit"));
+ }
}
diff --git a/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingAllMailsTask.java b/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingAllMailsTask.java
index ef930f94fd..4b7ab090a4 100644
--- a/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingAllMailsTask.java
+++ b/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingAllMailsTask.java
@@ -24,11 +24,8 @@ import java.time.Instant;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicLong;
-import javax.mail.MessagingException;
-
import org.apache.james.mailrepository.api.MailKey;
import org.apache.james.mailrepository.api.MailRepositoryPath;
-import org.apache.james.mailrepository.api.MailRepositoryStore;
import org.apache.james.task.Task;
import org.apache.james.task.TaskExecutionDetails;
import org.apache.james.task.TaskType;
@@ -109,13 +106,8 @@ public class ReprocessingAllMailsTask implements Task {
@Override
public Result run() {
- try {
- reprocessingService.reprocessAll(repositoryPath, configuration, this::notifyProgress);
- return Result.COMPLETED;
- } catch (MessagingException | MailRepositoryStore.MailRepositoryStoreException e) {
- LOGGER.error("Encountered error while reprocessing repository", e);
- return Result.PARTIAL;
- }
+ return reprocessingService.reprocessAll(repositoryPath, configuration, this::notifyProgress)
+ .block();
}
MailRepositoryPath getRepositoryPath() {
diff --git a/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingAllMailsTaskAdditionalInformationDTO.java b/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingAllMailsTaskAdditionalInformationDTO.java
index 55dcdb76fb..a20a852c51 100644
--- a/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingAllMailsTaskAdditionalInformationDTO.java
+++ b/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingAllMailsTaskAdditionalInformationDTO.java
@@ -26,6 +26,7 @@ import org.apache.james.mailrepository.api.MailRepositoryPath;
import org.apache.james.queue.api.MailQueueName;
import org.apache.james.server.task.json.dto.AdditionalInformationDTO;
import org.apache.james.server.task.json.dto.AdditionalInformationDTOModule;
+import org.apache.james.util.streams.Limit;
import com.fasterxml.jackson.annotation.JsonProperty;
@@ -38,7 +39,8 @@ public class ReprocessingAllMailsTaskAdditionalInformationDTO implements Additio
new ReprocessingService.Configuration(
MailQueueName.of(dto.getTargetQueue()),
dto.getTargetProcessor(),
- dto.isConsume()),
+ dto.isConsume(),
+ Limit.from(dto.getLimit())),
dto.initialCount,
dto.remainingCount,
dto.timestamp))
@@ -50,7 +52,8 @@ public class ReprocessingAllMailsTaskAdditionalInformationDTO implements Additio
Optional.of(details.getConfiguration().isConsume()),
details.getInitialCount(),
details.getRemainingCount(),
- details.timestamp()))
+ details.timestamp(),
+ details.getConfiguration().getLimit().getLimit()))
.typeName(ReprocessingAllMailsTask.TYPE.asString())
.withFactory(AdditionalInformationDTOModule::new);
}
@@ -63,6 +66,7 @@ public class ReprocessingAllMailsTaskAdditionalInformationDTO implements Additio
private final long initialCount;
private final long remainingCount;
private final Instant timestamp;
+ private final Optional<Integer> limit;
public ReprocessingAllMailsTaskAdditionalInformationDTO(
@JsonProperty("type") String type,
@@ -72,7 +76,8 @@ public class ReprocessingAllMailsTaskAdditionalInformationDTO implements Additio
@JsonProperty("consume") Optional<Boolean> consume,
@JsonProperty("initialCount") long initialCount,
@JsonProperty("remainingCount") long remainingCount,
- @JsonProperty("timestamp") Instant timestamp) {
+ @JsonProperty("timestamp") Instant timestamp,
+ @JsonProperty("limit") Optional<Integer> limit) {
this.type = type;
this.repositoryPath = repositoryPath;
this.targetQueue = targetQueue;
@@ -81,6 +86,7 @@ public class ReprocessingAllMailsTaskAdditionalInformationDTO implements Additio
this.remainingCount = remainingCount;
this.timestamp = timestamp;
this.consume = consume.orElse(true);
+ this.limit = limit;
}
public boolean isConsume() {
@@ -115,4 +121,8 @@ public class ReprocessingAllMailsTaskAdditionalInformationDTO implements Additio
public Optional<String> getTargetProcessor() {
return targetProcessor;
}
+
+ public Optional<Integer> getLimit() {
+ return limit;
+ }
}
diff --git a/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingAllMailsTaskDTO.java b/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingAllMailsTaskDTO.java
index 7a7b722aee..f081567bb3 100644
--- a/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingAllMailsTaskDTO.java
+++ b/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingAllMailsTaskDTO.java
@@ -25,6 +25,7 @@ import org.apache.james.mailrepository.api.MailRepositoryPath;
import org.apache.james.queue.api.MailQueueName;
import org.apache.james.server.task.json.dto.TaskDTO;
import org.apache.james.server.task.json.dto.TaskDTOModule;
+import org.apache.james.util.streams.Limit;
import com.fasterxml.jackson.annotation.JsonProperty;
@@ -48,7 +49,8 @@ public class ReprocessingAllMailsTaskDTO implements TaskDTO {
domainObject.getRepositoryPath().urlEncoded(),
domainObject.getConfiguration().getMailQueueName().asString(),
Optional.of(domainObject.getConfiguration().isConsume()),
- domainObject.getConfiguration().getTargetProcessor());
+ domainObject.getConfiguration().getTargetProcessor(),
+ domainObject.getConfiguration().getLimit().getLimit());
} catch (Exception e) {
throw new ReprocessingAllMailsTask.UrlEncodingFailureSerializationException(domainObject.getRepositoryPath());
}
@@ -61,18 +63,22 @@ public class ReprocessingAllMailsTaskDTO implements TaskDTO {
private final boolean consume;
private final Optional<String> targetProcessor;
+ private final Optional<Integer> limit;
+
public ReprocessingAllMailsTaskDTO(@JsonProperty("type") String type,
@JsonProperty("repositorySize") long repositorySize,
@JsonProperty("repositoryPath") String repositoryPath,
@JsonProperty("targetQueue") String targetQueue,
@JsonProperty("consume") Optional<Boolean> consume,
- @JsonProperty("targetProcessor") Optional<String> targetProcessor) {
+ @JsonProperty("targetProcessor") Optional<String> targetProcessor,
+ @JsonProperty("limit") Optional<Integer> limit) {
this.type = type;
this.repositorySize = repositorySize;
this.repositoryPath = repositoryPath;
this.targetQueue = targetQueue;
this.consume = consume.orElse(true);
this.targetProcessor = targetProcessor;
+ this.limit = limit;
}
private ReprocessingAllMailsTask fromDTO(ReprocessingService reprocessingService) {
@@ -84,7 +90,8 @@ public class ReprocessingAllMailsTaskDTO implements TaskDTO {
new ReprocessingService.Configuration(
MailQueueName.of(targetQueue),
targetProcessor,
- consume));
+ consume,
+ Limit.from(limit)));
} catch (Exception e) {
throw new ReprocessingAllMailsTask.InvalidMailRepositoryPathDeserializationException(repositoryPath);
}
@@ -114,4 +121,8 @@ public class ReprocessingAllMailsTaskDTO implements TaskDTO {
public Optional<String> getTargetProcessor() {
return targetProcessor;
}
+
+ public Optional<Integer> getLimit() {
+ return limit;
+ }
}
diff --git a/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingOneMailTaskAdditionalInformationDTO.java b/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingOneMailTaskAdditionalInformationDTO.java
index 40802d8bf3..a61e150d68 100644
--- a/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingOneMailTaskAdditionalInformationDTO.java
+++ b/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingOneMailTaskAdditionalInformationDTO.java
@@ -27,6 +27,7 @@ import org.apache.james.mailrepository.api.MailRepositoryPath;
import org.apache.james.queue.api.MailQueueName;
import org.apache.james.server.task.json.dto.AdditionalInformationDTO;
import org.apache.james.server.task.json.dto.AdditionalInformationDTOModule;
+import org.apache.james.util.streams.Limit;
import com.fasterxml.jackson.annotation.JsonProperty;
@@ -39,7 +40,8 @@ public class ReprocessingOneMailTaskAdditionalInformationDTO implements Addition
new ReprocessingService.Configuration(
MailQueueName.of(dto.targetQueue),
dto.targetProcessor,
- dto.isConsume()),
+ dto.isConsume(),
+ Limit.unlimited()),
new MailKey(dto.mailKey),
dto.timestamp))
.toDTOConverter((details, type) -> new ReprocessingOneMailTaskAdditionalInformationDTO(
diff --git a/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingOneMailTaskDTO.java b/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingOneMailTaskDTO.java
index 9ced35507e..bd938687c8 100644
--- a/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingOneMailTaskDTO.java
+++ b/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingOneMailTaskDTO.java
@@ -27,6 +27,7 @@ import org.apache.james.mailrepository.api.MailRepositoryPath;
import org.apache.james.queue.api.MailQueueName;
import org.apache.james.server.task.json.dto.TaskDTO;
import org.apache.james.server.task.json.dto.TaskDTOModule;
+import org.apache.james.util.streams.Limit;
import com.fasterxml.jackson.annotation.JsonProperty;
@@ -84,7 +85,8 @@ public class ReprocessingOneMailTaskDTO implements TaskDTO {
new ReprocessingService.Configuration(
MailQueueName.of(targetQueue),
targetProcessor,
- consume),
+ consume,
+ Limit.unlimited()),
new MailKey(mailKey),
clock);
}
diff --git a/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingService.java b/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingService.java
index 9c443d2365..9d7de3347c 100644
--- a/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingService.java
+++ b/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingService.java
@@ -36,13 +36,18 @@ import org.apache.james.mailrepository.api.MailRepositoryStore;
import org.apache.james.queue.api.MailQueue;
import org.apache.james.queue.api.MailQueueFactory;
import org.apache.james.queue.api.MailQueueName;
+import org.apache.james.task.Task;
import org.apache.james.util.streams.Iterators;
+import org.apache.james.util.streams.Limit;
import org.apache.mailet.Mail;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.github.fge.lambdas.Throwing;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
public class ReprocessingService {
private static final Logger LOGGER = LoggerFactory.getLogger(ReprocessingService.class);
@@ -57,11 +62,13 @@ public class ReprocessingService {
private final MailQueueName mailQueueName;
private final Optional<String> targetProcessor;
private final boolean consume;
+ private final Limit limit;
- public Configuration(MailQueueName mailQueueName, Optional<String> targetProcessor, boolean consume) {
+ public Configuration(MailQueueName mailQueueName, Optional<String> targetProcessor, boolean consume, Limit limit) {
this.mailQueueName = mailQueueName;
this.targetProcessor = targetProcessor;
this.consume = consume;
+ this.limit = limit;
}
public MailQueueName getMailQueueName() {
@@ -75,6 +82,10 @@ public class ReprocessingService {
public boolean isConsume() {
return consume;
}
+
+ public Limit getLimit() {
+ return limit;
+ }
}
static class Reprocessor implements Closeable {
@@ -120,17 +131,29 @@ public class ReprocessingService {
this.mailRepositoryStoreService = mailRepositoryStoreService;
}
- public void reprocessAll(MailRepositoryPath path, Configuration configuration, Consumer<MailKey> keyListener) throws MailRepositoryStore.MailRepositoryStoreException, MessagingException {
- try (Reprocessor reprocessor = new Reprocessor(getMailQueue(configuration.getMailQueueName()), configuration)) {
- mailRepositoryStoreService
- .getRepositories(path)
- .forEach(Throwing.consumer((MailRepository repository) ->
- Iterators.toStream(repository.list())
- .peek(keyListener)
- .map(Throwing.function(key -> Optional.ofNullable(repository.retrieve(key))))
- .flatMap(Optional::stream)
- .forEach(mail -> reprocessor.reprocess(repository, mail))));
- }
+ public Mono<Task.Result> reprocessAll(MailRepositoryPath path, Configuration configuration, Consumer<MailKey> keyListener) {
+ return Mono.using(() -> new Reprocessor(getMailQueue(configuration.getMailQueueName()), configuration),
+ reprocessor -> reprocessAll(reprocessor, path, configuration, keyListener),
+ Reprocessor::close);
+ }
+
+ private Mono<Task.Result> reprocessAll(Reprocessor reprocessor, MailRepositoryPath path, Configuration configuration, Consumer<MailKey> keyListener) {
+ return configuration.limit.applyOnFlux(Flux.fromStream(Throwing.supplier(() -> mailRepositoryStoreService.getRepositories(path)))
+ .flatMap(Throwing.function((MailRepository repository) -> Iterators.toFlux(repository.list())
+ .doOnNext(keyListener)
+ .map(mailKey -> Pair.of(repository, mailKey)))))
+ .flatMap(pair -> reprocess(pair.getRight(), pair.getLeft(), reprocessor))
+ .reduce(Task.Result.COMPLETED, Task::combine);
+ }
+
+ private Mono<Task.Result> reprocess(MailKey key, MailRepository repository, Reprocessor reprocessor) {
+ return Mono.fromCallable(() -> repository.retrieve(key))
+ .doOnNext(mail -> reprocessor.reprocess(repository, mail))
+ .thenReturn(Task.Result.COMPLETED)
+ .onErrorResume(error -> {
+ LOGGER.warn("Failed when reprocess mail {}", key.asString(), error);
+ return Mono.just(Task.Result.PARTIAL);
+ });
}
public void reprocess(MailRepositoryPath path, MailKey key, Configuration configuration) throws MailRepositoryStore.MailRepositoryStoreException, MessagingException {
diff --git a/server/protocols/webadmin/webadmin-mailrepository/src/test/java/org/apache/james/webadmin/routes/MailRepositoriesRoutesTest.java b/server/protocols/webadmin/webadmin-mailrepository/src/test/java/org/apache/james/webadmin/routes/MailRepositoriesRoutesTest.java
index 16bd869e8d..499afc7ba1 100644
--- a/server/protocols/webadmin/webadmin-mailrepository/src/test/java/org/apache/james/webadmin/routes/MailRepositoriesRoutesTest.java
+++ b/server/protocols/webadmin/webadmin-mailrepository/src/test/java/org/apache/james/webadmin/routes/MailRepositoriesRoutesTest.java
@@ -54,6 +54,7 @@ import org.apache.james.json.DTOConverter;
import org.apache.james.mailrepository.api.MailKey;
import org.apache.james.mailrepository.api.MailRepository;
import org.apache.james.mailrepository.api.MailRepositoryPath;
+import org.apache.james.mailrepository.api.MailRepositoryStore;
import org.apache.james.mailrepository.api.MailRepositoryUrl;
import org.apache.james.mailrepository.api.Protocol;
import org.apache.james.mailrepository.memory.MailRepositoryStoreConfiguration;
@@ -1159,6 +1160,7 @@ class MailRepositoriesRoutesTest {
.param("queue", CUSTOM_QUEUE.asString())
.param("processor", transport)
.param("consume", false)
+ .param("limit", 100)
.patch(PATH_ESCAPED_MY_REPO + "/mails")
.jsonPath()
.get("taskId");
@@ -1177,6 +1179,7 @@ class MailRepositoriesRoutesTest {
.body("additionalInformation.targetProcessor", is(transport))
.body("additionalInformation.targetQueue", is(CUSTOM_QUEUE.asString()))
.body("additionalInformation.consume", is(false))
+ .body("additionalInformation.limit", is(100))
.body("startedDate", is(notNullValue()))
.body("submitDate", is(notNullValue()))
.body("completedDate", is(notNullValue()));
@@ -1246,6 +1249,59 @@ class MailRepositoriesRoutesTest {
.isEmpty();
}
+ @Test
+ void reprocessingAllTaskShouldClearMailInLimitedWhenProvideLimitParameter() throws Exception {
+ MailRepository mailRepository = mailRepositoryStore.create(URL_MY_REPO);
+ String name1 = "name1";
+ String name2 = "name2";
+ mailRepository.store(FakeMail.builder()
+ .name(name1)
+ .mimeMessage(MimeMessageUtil.mimeMessageFromBytes(MESSAGE_BYTES))
+ .build());
+ mailRepository.store(FakeMail.builder()
+ .name(name2)
+ .mimeMessage(MimeMessageUtil.mimeMessageFromBytes(MESSAGE_BYTES))
+ .build());
+
+ String taskId = with()
+ .param("action", "reprocess")
+ .param("limit", 1)
+ .patch(PATH_ESCAPED_MY_REPO + "/mails")
+ .jsonPath()
+ .get("taskId");
+
+ given()
+ .basePath(TasksRoutes.BASE)
+ .when()
+ .get(taskId + "/await")
+ .then()
+ .body("status", is("completed"))
+ .body("taskId", is(notNullValue()))
+ .body("type", is(ReprocessingAllMailsTask.TYPE.asString()))
+ .body("additionalInformation.initialCount", is(2))
+ .body("additionalInformation.remainingCount", is(1))
+ .body("additionalInformation.limit", is(1));
+
+ assertThat(mailRepository.list())
+ .toIterable()
+ .hasSize(1);
+ }
+
+ @Test
+ void reprocessingAllTaskShouldFailWhenInvalidLimitParameter() throws Exception {
+ MailRepository mailRepository = mailRepositoryStore.create(URL_MY_REPO);
+ with()
+ .param("action", "reprocess")
+ .param("limit", "invalid")
+ .patch(PATH_ESCAPED_MY_REPO + "/mails")
+ .then()
+ .statusCode(HttpStatus.BAD_REQUEST_400)
+ .contentType(ContentType.JSON)
+ .body("statusCode", is(400))
+ .body("type", is(ErrorResponder.ErrorType.INVALID_ARGUMENT.getType()))
+ .body("message", is("Can not parse limit"));
+ }
+
@Test
void reprocessingAllTaskShouldNotClearMailRepositoryWhenNotConsume() throws Exception {
MailRepository mailRepository = mailRepositoryStore.create(URL_MY_REPO);
diff --git a/server/protocols/webadmin/webadmin-mailrepository/src/test/java/org/apache/james/webadmin/service/ReprocessingAllMailsTaskTest.java b/server/protocols/webadmin/webadmin-mailrepository/src/test/java/org/apache/james/webadmin/service/ReprocessingAllMailsTaskTest.java
index 64b5fe33e3..dc8a68c382 100644
--- a/server/protocols/webadmin/webadmin-mailrepository/src/test/java/org/apache/james/webadmin/service/ReprocessingAllMailsTaskTest.java
+++ b/server/protocols/webadmin/webadmin-mailrepository/src/test/java/org/apache/james/webadmin/service/ReprocessingAllMailsTaskTest.java
@@ -31,6 +31,7 @@ import org.apache.james.mailrepository.api.MailRepositoryPath;
import org.apache.james.queue.api.MailQueueName;
import org.apache.james.server.task.json.JsonTaskAdditionalInformationSerializer;
import org.apache.james.server.task.json.JsonTaskSerializer;
+import org.apache.james.util.streams.Limit;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
@@ -55,12 +56,15 @@ class ReprocessingAllMailsTaskTest {
@Test
void taskShouldBeSerializable() throws Exception {
- ReprocessingAllMailsTask taskWithTargetProcessor = new ReprocessingAllMailsTask(REPROCESSING_SERVICE, REPOSITORY_SIZE, REPOSITORY_PATH, new ReprocessingService.Configuration(TARGET_QUEUE, SOME_TARGET_PROCESSOR, CONSUME));
- ReprocessingAllMailsTask taskWithoutTargetProcessor = new ReprocessingAllMailsTask(REPROCESSING_SERVICE, REPOSITORY_SIZE, REPOSITORY_PATH, new ReprocessingService.Configuration(TARGET_QUEUE, Optional.empty(), !CONSUME));
+ ReprocessingAllMailsTask taskWithTargetProcessor = new ReprocessingAllMailsTask(REPROCESSING_SERVICE, REPOSITORY_SIZE, REPOSITORY_PATH, new ReprocessingService.Configuration(TARGET_QUEUE, SOME_TARGET_PROCESSOR, CONSUME, Limit.unlimited()));
+ ReprocessingAllMailsTask taskWithoutTargetProcessor = new ReprocessingAllMailsTask(REPROCESSING_SERVICE, REPOSITORY_SIZE, REPOSITORY_PATH, new ReprocessingService.Configuration(TARGET_QUEUE, Optional.empty(), !CONSUME, Limit.unlimited()));
+
+ ReprocessingAllMailsTask taskWithLimit = new ReprocessingAllMailsTask(REPROCESSING_SERVICE, REPOSITORY_SIZE, REPOSITORY_PATH, new ReprocessingService.Configuration(TARGET_QUEUE, SOME_TARGET_PROCESSOR, CONSUME, Limit.limit(10)));
JsonSerializationVerifier.dtoModule(ReprocessingAllMailsTaskDTO.module(REPROCESSING_SERVICE))
.testCase(taskWithTargetProcessor, SERIALIZED_TASK_WITH_TARGET_PROCESSOR)
.testCase(taskWithoutTargetProcessor, SERIALIZED_TASK_WITHOUT_TARGET_PROCESSOR)
+ .testCase(taskWithLimit,"{\"type\":\"reprocessing-all\",\"repositorySize\":5,\"repositoryPath\":\"a\",\"targetQueue\":\"queue\",\"targetProcessor\":\"targetProcessor\",\"consume\":true, \"limit\":10}" )
.verify();
}
@@ -78,35 +82,51 @@ class ReprocessingAllMailsTaskTest {
@Test
void additionalInformationShouldBeSerializable() throws Exception {
ReprocessingAllMailsTask.AdditionalInformation details = new ReprocessingAllMailsTask.AdditionalInformation(REPOSITORY_PATH,
- new ReprocessingService.Configuration(TARGET_QUEUE, SOME_TARGET_PROCESSOR, CONSUME),
+ new ReprocessingService.Configuration(TARGET_QUEUE, SOME_TARGET_PROCESSOR, CONSUME, Limit.unlimited()),
REPOSITORY_SIZE, REMAINING_COUNT, TIMESTAMP);
ReprocessingAllMailsTask.AdditionalInformation detailsWithoutProcessor = new ReprocessingAllMailsTask.AdditionalInformation(REPOSITORY_PATH,
- new ReprocessingService.Configuration(TARGET_QUEUE, Optional.empty(), !CONSUME),
+ new ReprocessingService.Configuration(TARGET_QUEUE, Optional.empty(), !CONSUME, Limit.unlimited()),
+ REPOSITORY_SIZE, REMAINING_COUNT, TIMESTAMP);
+
+ ReprocessingAllMailsTask.AdditionalInformation detailWithLimit = new ReprocessingAllMailsTask.AdditionalInformation(REPOSITORY_PATH,
+ new ReprocessingService.Configuration(TARGET_QUEUE, SOME_TARGET_PROCESSOR, CONSUME, Limit.limit(10)),
REPOSITORY_SIZE, REMAINING_COUNT, TIMESTAMP);
JsonSerializationVerifier.dtoModule(ReprocessingAllMailsTaskAdditionalInformationDTO.module())
.testCase(details, SERIALIZED_TASK_ADDITIONAL_INFORMATION_WITH_TARGET_PROCESSOR)
.testCase(detailsWithoutProcessor, SERIALIZED_TASK_ADDITIONAL_INFORMATION_WITHOUT_TARGET_PROCESSOR)
+ .testCase(detailWithLimit, "{\"type\":\"reprocessing-all\", \"repositoryPath\":\"a\",\"targetQueue\":\"queue\",\"targetProcessor\":\"targetProcessor\",\"initialCount\":5,\"remainingCount\":3, \"timestamp\":\"2018-11-13T12:00:55Z\",\"consume\":true, \"limit\": 10}")
.verify();
}
@Test
void shouldDeserializePreviousTaskFormat() throws Exception {
- ReprocessingAllMailsTask taskWithTargetProcessor = new ReprocessingAllMailsTask(REPROCESSING_SERVICE, REPOSITORY_SIZE, REPOSITORY_PATH, new ReprocessingService.Configuration(TARGET_QUEUE, SOME_TARGET_PROCESSOR, CONSUME));
+ ReprocessingAllMailsTask taskWithTargetProcessor = new ReprocessingAllMailsTask(REPROCESSING_SERVICE, REPOSITORY_SIZE, REPOSITORY_PATH, new ReprocessingService.Configuration(TARGET_QUEUE, SOME_TARGET_PROCESSOR, CONSUME, Limit.unlimited()));
JsonTaskSerializer testee = JsonTaskSerializer.of(ReprocessingAllMailsTaskDTO.module(REPROCESSING_SERVICE));
assertThat(testee.deserialize(OLD_SERIALIZED_TASK))
.isEqualToComparingFieldByFieldRecursively(taskWithTargetProcessor);
+
+ String serializedTaskWithLimit = "{\"type\":\"reprocessing-all\",\"repositorySize\":5,\"repositoryPath\":\"a\",\"targetQueue\":\"queue\",\"targetProcessor\":\"targetProcessor\", \"limit\":10}";
+ assertThat(testee.deserialize(serializedTaskWithLimit))
+ .isEqualToComparingFieldByFieldRecursively(new ReprocessingAllMailsTask(REPROCESSING_SERVICE, REPOSITORY_SIZE, REPOSITORY_PATH, new ReprocessingService.Configuration(TARGET_QUEUE, SOME_TARGET_PROCESSOR, CONSUME, Limit.limit(10))));
}
@Test
void shouldDeserializePreviousAdditionalInformationFormat() throws Exception {
ReprocessingAllMailsTask.AdditionalInformation details = new ReprocessingAllMailsTask.AdditionalInformation(REPOSITORY_PATH,
- new ReprocessingService.Configuration(TARGET_QUEUE, SOME_TARGET_PROCESSOR, CONSUME),
+ new ReprocessingService.Configuration(TARGET_QUEUE, SOME_TARGET_PROCESSOR, CONSUME, Limit.unlimited()),
REPOSITORY_SIZE, REMAINING_COUNT, TIMESTAMP);
JsonTaskAdditionalInformationSerializer testee = JsonTaskAdditionalInformationSerializer.of(ReprocessingAllMailsTaskAdditionalInformationDTO.module());
assertThat(testee.deserialize(OLD_SERIALIZED_TASK_ADDITIONAL_INFORMATION))
.isEqualToComparingFieldByFieldRecursively(details);
+
+ String serializedTaskAdditionalInformation = "{\"type\":\"reprocessing-all\", \"repositoryPath\":\"a\",\"targetQueue\":\"queue\",\"targetProcessor\":\"targetProcessor\",\"initialCount\":5,\"remainingCount\":3, \"timestamp\":\"2018-11-13T12:00:55Z\", \"limit\": 10}";
+
+ assertThat(testee.deserialize(serializedTaskAdditionalInformation))
+ .isEqualToComparingFieldByFieldRecursively(new ReprocessingAllMailsTask.AdditionalInformation(REPOSITORY_PATH,
+ new ReprocessingService.Configuration(TARGET_QUEUE, SOME_TARGET_PROCESSOR, CONSUME, Limit.limit(10)),
+ REPOSITORY_SIZE, REMAINING_COUNT, TIMESTAMP));
}
}
\ No newline at end of file
diff --git a/server/protocols/webadmin/webadmin-mailrepository/src/test/java/org/apache/james/webadmin/service/ReprocessingOneMailTaskTest.java b/server/protocols/webadmin/webadmin-mailrepository/src/test/java/org/apache/james/webadmin/service/ReprocessingOneMailTaskTest.java
index cf0746befb..bdeaaceec0 100644
--- a/server/protocols/webadmin/webadmin-mailrepository/src/test/java/org/apache/james/webadmin/service/ReprocessingOneMailTaskTest.java
+++ b/server/protocols/webadmin/webadmin-mailrepository/src/test/java/org/apache/james/webadmin/service/ReprocessingOneMailTaskTest.java
@@ -35,6 +35,7 @@ import org.apache.james.mailrepository.api.MailRepositoryPath;
import org.apache.james.queue.api.MailQueueName;
import org.apache.james.server.task.json.JsonTaskAdditionalInformationSerializer;
import org.apache.james.server.task.json.JsonTaskSerializer;
+import org.apache.james.util.streams.Limit;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
@@ -55,8 +56,8 @@ class ReprocessingOneMailTaskTest {
@Test
void taskShouldBeSerializable() throws Exception {
- ReprocessingOneMailTask taskWithTargetProcessor = new ReprocessingOneMailTask(REPROCESSING_SERVICE, REPOSITORY_PATH, new ReprocessingService.Configuration(TARGET_QUEUE, TARGET_PROCESSOR, CONSUME), MAIL_KEY, CLOCK);
- ReprocessingOneMailTask taskWithoutTargetProcessor = new ReprocessingOneMailTask(REPROCESSING_SERVICE, REPOSITORY_PATH, new ReprocessingService.Configuration(TARGET_QUEUE, Optional.empty(), !CONSUME), MAIL_KEY, CLOCK);
+ ReprocessingOneMailTask taskWithTargetProcessor = new ReprocessingOneMailTask(REPROCESSING_SERVICE, REPOSITORY_PATH, new ReprocessingService.Configuration(TARGET_QUEUE, TARGET_PROCESSOR, CONSUME, Limit.unlimited()), MAIL_KEY, CLOCK);
+ ReprocessingOneMailTask taskWithoutTargetProcessor = new ReprocessingOneMailTask(REPROCESSING_SERVICE, REPOSITORY_PATH, new ReprocessingService.Configuration(TARGET_QUEUE, Optional.empty(), !CONSUME, Limit.unlimited()), MAIL_KEY, CLOCK);
JsonSerializationVerifier.dtoModule(ReprocessingOneMailTaskDTO.module(CLOCK, REPROCESSING_SERVICE))
.testCase(taskWithTargetProcessor, SERIALIZED_TASK_1)
@@ -65,6 +66,15 @@ class ReprocessingOneMailTaskTest {
.verify();
}
+ @Test
+ void taskShouldBeDeserializable() throws Exception {
+ JsonTaskSerializer serializer = JsonTaskSerializer.of(ReprocessingOneMailTaskDTO.module(CLOCK, REPROCESSING_SERVICE));
+
+ assertThat(serializer.deserialize("{\"type\":\"reprocessing-one\",\"repositoryPath\":\"a\",\"targetQueue\":\"queue\",\"mailKey\": \"myMail\", \"consume\":false}"))
+ .usingRecursiveComparison()
+ .isEqualTo(new ReprocessingOneMailTask(REPROCESSING_SERVICE, REPOSITORY_PATH, new ReprocessingService.Configuration(TARGET_QUEUE, Optional.empty(), !CONSUME, Limit.unlimited()), MAIL_KEY, CLOCK));
+ }
+
@ParameterizedTest
@ValueSource(strings = {
"{\"type\":\"reprocessing-one\",\"repositoryPath\":\"%\",\"targetQueue\":\"queue\",\"mailKey\": \"myMail\",\"targetProcessor\":\"targetProcessor\"}",
@@ -79,7 +89,7 @@ class ReprocessingOneMailTaskTest {
@Test
void additionalInformationShouldBeSerializable() throws IOException {
ReprocessingOneMailTask.AdditionalInformation details = new ReprocessingOneMailTask.AdditionalInformation(REPOSITORY_PATH,
- new ReprocessingService.Configuration(TARGET_QUEUE, TARGET_PROCESSOR, true), MAIL_KEY, TIMESTAMP);
+ new ReprocessingService.Configuration(TARGET_QUEUE, TARGET_PROCESSOR, true, Limit.unlimited()), MAIL_KEY, TIMESTAMP);
JsonSerializationVerifier.dtoModule(ReprocessingOneMailTaskAdditionalInformationDTO.module())
.bean(details)
.json(SERIALIZED_TASK_1_ADDITIONAL_INFORMATION)
@@ -89,7 +99,7 @@ class ReprocessingOneMailTaskTest {
@Test
void shouldDeserializePreviousTaskFormat() throws Exception {
- ReprocessingOneMailTask taskWithTargetProcessor = new ReprocessingOneMailTask(REPROCESSING_SERVICE, REPOSITORY_PATH, new ReprocessingService.Configuration(TARGET_QUEUE, TARGET_PROCESSOR, CONSUME), MAIL_KEY, CLOCK);
+ ReprocessingOneMailTask taskWithTargetProcessor = new ReprocessingOneMailTask(REPROCESSING_SERVICE, REPOSITORY_PATH, new ReprocessingService.Configuration(TARGET_QUEUE, TARGET_PROCESSOR, CONSUME, Limit.unlimited()), MAIL_KEY, CLOCK);
JsonTaskSerializer testee = JsonTaskSerializer.of(ReprocessingOneMailTaskDTO.module(CLOCK, REPROCESSING_SERVICE));
assertThat(testee.deserialize(SERIALIZED_TASK_OLD))
@@ -99,7 +109,7 @@ class ReprocessingOneMailTaskTest {
@Test
void shouldDeserializePreviousAdditionalInformationFormat() throws Exception {
ReprocessingOneMailTask.AdditionalInformation details = new ReprocessingOneMailTask.AdditionalInformation(REPOSITORY_PATH,
- new ReprocessingService.Configuration(TARGET_QUEUE, TARGET_PROCESSOR, true), MAIL_KEY, TIMESTAMP);
+ new ReprocessingService.Configuration(TARGET_QUEUE, TARGET_PROCESSOR, true, Limit.unlimited()), MAIL_KEY, TIMESTAMP);
JsonTaskAdditionalInformationSerializer testee = JsonTaskAdditionalInformationSerializer.of(ReprocessingOneMailTaskAdditionalInformationDTO.module());
assertThat(testee.deserialize(SERIALIZED_TASK_OLD_ADDITIONAL_INFORMATION))
diff --git a/server/protocols/webadmin/webadmin-mailrepository/src/test/java/org/apache/james/webadmin/service/ReprocessingServiceTest.java b/server/protocols/webadmin/webadmin-mailrepository/src/test/java/org/apache/james/webadmin/service/ReprocessingServiceTest.java
index 1ad1e9fd82..3fdbf572b9 100644
--- a/server/protocols/webadmin/webadmin-mailrepository/src/test/java/org/apache/james/webadmin/service/ReprocessingServiceTest.java
+++ b/server/protocols/webadmin/webadmin-mailrepository/src/test/java/org/apache/james/webadmin/service/ReprocessingServiceTest.java
@@ -43,6 +43,7 @@ import org.apache.james.queue.api.ManageableMailQueue;
import org.apache.james.queue.api.RawMailQueueItemDecoratorFactory;
import org.apache.james.queue.memory.MemoryMailQueueFactory;
import org.apache.james.util.MimeMessageUtil;
+import org.apache.james.util.streams.Limit;
import org.apache.james.webadmin.service.ReprocessingService.Configuration;
import org.apache.mailet.base.test.FakeMail;
import org.junit.jupiter.api.BeforeEach;
@@ -108,7 +109,7 @@ class ReprocessingServiceTest {
repository.store(mail2);
repository.store(mail3);
- reprocessingService.reprocess(PATH, KEY_2, new ReprocessingService.Configuration(SPOOL, NO_TARGET_PROCESSOR, CONSUME));
+ reprocessingService.reprocess(PATH, KEY_2, new ReprocessingService.Configuration(SPOOL, NO_TARGET_PROCESSOR, CONSUME, Limit.unlimited()));
assertThat(queueFactory.getQueue(SPOOL).get().browse())
.toIterable()
@@ -123,7 +124,7 @@ class ReprocessingServiceTest {
repository.store(mail2);
repository.store(mail3);
- reprocessingService.reprocess(PATH, KEY_2, new ReprocessingService.Configuration(SPOOL, NO_TARGET_PROCESSOR, CONSUME));
+ reprocessingService.reprocess(PATH, KEY_2, new ReprocessingService.Configuration(SPOOL, NO_TARGET_PROCESSOR, CONSUME, Limit.unlimited()));
assertThat(repository.list()).toIterable()
.containsOnly(KEY_1, KEY_3);
@@ -136,7 +137,7 @@ class ReprocessingServiceTest {
repository.store(mail2);
repository.store(mail3);
- reprocessingService.reprocessAll(PATH, new Configuration(SPOOL, NO_TARGET_PROCESSOR, CONSUME), NOOP_CONSUMER);
+ reprocessingService.reprocessAll(PATH, new Configuration(SPOOL, NO_TARGET_PROCESSOR, CONSUME, Limit.unlimited()), NOOP_CONSUMER).block();
assertThat(repository.list()).toIterable()
.isEmpty();
@@ -149,7 +150,7 @@ class ReprocessingServiceTest {
repository.store(mail2);
repository.store(mail3);
- reprocessingService.reprocessAll(PATH, new Configuration(SPOOL, NO_TARGET_PROCESSOR, CONSUME), NOOP_CONSUMER);
+ reprocessingService.reprocessAll(PATH, new Configuration(SPOOL, NO_TARGET_PROCESSOR, CONSUME, Limit.unlimited()), NOOP_CONSUMER).block();
assertThat(queueFactory.getQueue(SPOOL).get().browse())
.toIterable()
@@ -180,7 +181,7 @@ class ReprocessingServiceTest {
}
});
- reprocessingService.reprocessAll(PATH, new ReprocessingService.Configuration(SPOOL, NO_TARGET_PROCESSOR, CONSUME), concurrentRemoveConsumer);
+ reprocessingService.reprocessAll(PATH, new ReprocessingService.Configuration(SPOOL, NO_TARGET_PROCESSOR, CONSUME, Limit.unlimited()), concurrentRemoveConsumer).block();
assertThat(queueFactory.getQueue(SPOOL).get().browse())
.toIterable()
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 06/07: JAMES-3784 WebAdmin: Document - Provide RunningOptions (limit) for Redeliver event task, Reprocessing mail task
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch 3.7.x
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit cd418ddfd448227d471a9eb126bacb142b8f20be
Author: Tung Van TRAN <vt...@linagora.com>
AuthorDate: Mon Jun 27 14:13:58 2022 +0700
JAMES-3784 WebAdmin: Document - Provide RunningOptions (limit) for Redeliver event task, Reprocessing mail task
---
.../docs/modules/ROOT/pages/operate/guide.adoc | 7 ++++++
.../docs/modules/ROOT/pages/operate/webadmin.adoc | 28 ++++++++++++++++++++--
.../server/manage-guice-distributed-james.md | 7 ++++++
3 files changed, 40 insertions(+), 2 deletions(-)
diff --git a/server/apps/distributed-app/docs/modules/ROOT/pages/operate/guide.adoc b/server/apps/distributed-app/docs/modules/ROOT/pages/operate/guide.adoc
index 50b5bbb0c7..1b74424961 100644
--- a/server/apps/distributed-app/docs/modules/ROOT/pages/operate/guide.adoc
+++ b/server/apps/distributed-app/docs/modules/ROOT/pages/operate/guide.adoc
@@ -40,6 +40,10 @@ all mails in a mail repository] or
xref:operate/webadmin.adoc#_reprocessing_a_specific_mail_from_a_mail_repository[reprocessing
a single mail in a mail repository].
+In order to prevent unbounded processing that could consume unbounded resources. We can provide a CRON with `limit` parameter.
+Ex: 10 reprocessed per minute
+Note that it only support the reprocessing all mails.
+
Also, one can decide to
xref:operate/webadmin.adoc#_removing_all_mails_from_a_mail_repository[delete
all the mails of a mail repository] or
@@ -114,6 +118,9 @@ xref:operate/webadmin.adoc#_redeliver_all_events[redeliver all events]
task. It will start reprocessing all the failed events registered in
event dead letters.
+In order to prevent unbounded processing that could consume unbounded resources. We can provide a CRON with `limit` parameter.
+Ex: 10 redelivery per minute
+
If for some other reason you don’t need to redeliver all events, you
have more fine-grained operations allowing you to
xref:operate/webadmin.adoc#_redeliver_group_events[redeliver group events]
diff --git a/server/apps/distributed-app/docs/modules/ROOT/pages/operate/webadmin.adoc b/server/apps/distributed-app/docs/modules/ROOT/pages/operate/webadmin.adoc
index 47ec81b266..eaebfa653d 100644
--- a/server/apps/distributed-app/docs/modules/ROOT/pages/operate/webadmin.adoc
+++ b/server/apps/distributed-app/docs/modules/ROOT/pages/operate/webadmin.adoc
@@ -3310,8 +3310,10 @@ processing in. Defaults to the `state` field of each processed email.
- `consume` (boolean defaulting to `true`) whether the reprocessing should consume the mail in its originating mail repository. Passing
this value to `false` allows non destructive reprocessing as you keep a copy of the email in the mail repository and can be valuable
when debugging.
+- `limit` (integer value. Optional, default is empty). It enables to limit the count of elements reprocessed.
+If unspecified the count of the processed elements is unbounded
-For instance:
+redeliver_group_events
....
curl -XPATCH 'http://ip:port/mailRepositories/var%2Fmail%2Ferror%2F/mails?action=reprocess&processor=transport&queue=spool'
@@ -3697,6 +3699,17 @@ Response codes:
curl -XPOST http://ip:port/events/deadLetter?action=redeliver
....
+Additional query parameters are supported:
+
+- `limit` (integer value. Optional, default is empty). It enables to limit the count of elements redelivered.
+If unspecified the count of the processed elements is unbounded
+
+For instance:
+
+....
+curl -XPOST http://ip:port/events/deadLetter?action=redeliver&limit=10
+....
+
Will create a task that will attempt to redeliver all events stored in
``Event Dead Letter''. If successful, redelivered events will then be
removed from ``Dead Letter''.
@@ -3712,13 +3725,24 @@ Response codes:
=== Redeliver group events
....
-curl -XPOST http://ip:port/events/deadLetter/groups/org.apache.james.mailbox.events.EventBusTestFixture$GroupA
+curl -XPOST http://ip:port/events/deadLetter/groups/org.apache.james.mailbox.events.EventBusTestFixture$GroupA?action=redeliver
....
Will create a task that will attempt to redeliver all events of a
particular group stored in ``Event Dead Letter''. If successful,
redelivered events will then be removed from ``Dead Letter''.
+Additional query parameters are supported:
+
+- `limit` (integer value. Optional, default is empty). It enables to limit the count of elements redelivered.
+If unspecified the count of the processed elements is unbounded
+
+For instance:
+
+....
+curl -XPOST http://ip:port/events/deadLetter/groups/org.apache.james.mailbox.events.EventBusTestFixture$GroupA?action=redeliver&limit=10
+....
+
link:#_endpoints_returning_a_task[More details about endpoints returning
a task].
diff --git a/src/site/markdown/server/manage-guice-distributed-james.md b/src/site/markdown/server/manage-guice-distributed-james.md
index d5d21cec6f..d7a889dfcb 100644
--- a/src/site/markdown/server/manage-guice-distributed-james.md
+++ b/src/site/markdown/server/manage-guice-distributed-james.md
@@ -148,6 +148,10 @@ WebAdmin exposes all utilities for
or
[reprocessing a single mail in a mail repository](manage-webadmin.html#Reprocessing_a_specific_mail_from_a_mail_repository).
+In order to prevent unbounded processing that could consume unbounded resources. We can provide a CRON with `limit` parameter.
+Ex: 10 reprocessed per minute
+Note that it only support the reprocessing all mails.
+
Also, one can decide to
[delete all the mails of a mail repository](manage-webadmin.html#Removing_all_mails_from_a_mail_repository)
or [delete a single mail of a mail repository](manage-webadmin.html#Removing_a_mail_from_a_mail_repository).
@@ -209,6 +213,9 @@ An easy way to solve this is just to trigger then the
[redeliver all events](manage-webadmin.html#Redeliver_all_events) task. It will start
reprocessing all the failed events registered in event dead letters.
+In order to prevent unbounded processing that could consume unbounded resources. We can provide a CRON with `limit` parameter.
+Ex: 10 redelivery per minute
+
If for some other reason you don't need to redeliver all events, you have more fine-grained operations allowing you to
[redeliver group events](manage-webadmin.html#Redeliver_group_events) or even just
[redeliver a single event](manage-webadmin.html#Redeliver_a_single_event).
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 01/07: JAMES-3723 Allow to not consume emails upon reprocessing
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch 3.7.x
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 2ea97b0bc0c99d5ca2612b50576dcd17c3ed670b
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Sat Mar 5 22:06:45 2022 +0700
JAMES-3723 Allow to not consume emails upon reprocessing
---
...dminServerTaskSerializationIntegrationTest.java | 1 +
.../webadmin/routes/MailRepositoriesRoutes.java | 20 +++--
.../webadmin/service/ReprocessingAllMailsTask.java | 37 +++------
...essingAllMailsTaskAdditionalInformationDTO.java | 18 ++++-
.../service/ReprocessingAllMailsTaskDTO.java | 20 +++--
.../webadmin/service/ReprocessingOneMailTask.java | 38 +++------
...cessingOneMailTaskAdditionalInformationDTO.java | 18 ++++-
.../service/ReprocessingOneMailTaskDTO.java | 24 ++++--
.../webadmin/service/ReprocessingService.java | 44 ++++++++---
.../routes/MailRepositoriesRoutesTest.java | 91 ++++++++++++++++++++++
.../service/ReprocessingAllMailsTaskTest.java | 49 +++++++++---
.../service/ReprocessingOneMailTaskTest.java | 42 ++++++++--
.../webadmin/service/ReprocessingServiceTest.java | 12 +--
13 files changed, 304 insertions(+), 110 deletions(-)
diff --git a/server/protocols/webadmin-integration-test/distributed-webadmin-integration-test/src/test/java/org/apache/james/webadmin/integration/rabbitmq/RabbitMQWebAdminServerTaskSerializationIntegrationTest.java b/server/protocols/webadmin-integration-test/distributed-webadmin-integration-test/src/test/java/org/apache/james/webadmin/integration/rabbitmq/RabbitMQWebAdminServerTaskSerializationIntegrationTest.java
index 4e3ac973b3..4e5ba18230 100644
--- a/server/protocols/webadmin-integration-test/distributed-webadmin-integration-test/src/test/java/org/apache/james/webadmin/integration/rabbitmq/RabbitMQWebAdminServerTaskSerializationIntegrationTest.java
+++ b/server/protocols/webadmin-integration-test/distributed-webadmin-integration-test/src/test/java/org/apache/james/webadmin/integration/rabbitmq/RabbitMQWebAdminServerTaskSerializationIntegrationTest.java
@@ -344,6 +344,7 @@ class RabbitMQWebAdminServerTaskSerializationIntegrationTest {
.body("type", is("reprocessing-one"))
.body("additionalInformation.repositoryPath", is(mailRepositoryUrl.asString()))
.body("additionalInformation.targetQueue", is(notNullValue()))
+ .body("additionalInformation.consume", is(notNullValue()))
.body("additionalInformation.mailKey", is(mailKey))
.body("additionalInformation.targetProcessor", is(nullValue()));
}
diff --git a/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/routes/MailRepositoriesRoutes.java b/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/routes/MailRepositoriesRoutes.java
index aea7771580..97c563425d 100644
--- a/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/routes/MailRepositoriesRoutes.java
+++ b/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/routes/MailRepositoriesRoutes.java
@@ -322,11 +322,15 @@ public class MailRepositoriesRoutes implements Routes {
private Task reprocessAll(Request request) throws UnsupportedEncodingException, MailRepositoryStore.MailRepositoryStoreException {
MailRepositoryPath path = decodedRepositoryPath(request);
- Optional<String> targetProcessor = parseTargetProcessor(request);
- MailQueueName targetQueue = parseTargetQueue(request);
Long repositorySize = repositoryStoreService.size(path).orElse(0L);
- return new ReprocessingAllMailsTask(reprocessingService, repositorySize, path, targetQueue, targetProcessor);
+ return new ReprocessingAllMailsTask(reprocessingService, repositorySize, path, extractConfiguration(request));
+ }
+
+ private ReprocessingService.Configuration extractConfiguration(Request request) {
+ return new ReprocessingService.Configuration(parseTargetQueue(request),
+ parseTargetProcessor(request),
+ parseConsume(request).orElse(true));
}
public void defineReprocessOne() {
@@ -340,10 +344,7 @@ public class MailRepositoriesRoutes implements Routes {
MailRepositoryPath path = decodedRepositoryPath(request);
MailKey key = new MailKey(request.params("key"));
- Optional<String> targetProcessor = parseTargetProcessor(request);
- MailQueueName targetQueue = parseTargetQueue(request);
-
- return new ReprocessingOneMailTask(reprocessingService, path, targetQueue, key, targetProcessor, Clock.systemUTC());
+ return new ReprocessingOneMailTask(reprocessingService, path, extractConfiguration(request), key, Clock.systemUTC());
}
private Set<AdditionalField> extractAdditionalFields(String additionalFieldsParam) throws IllegalArgumentException {
@@ -361,6 +362,11 @@ public class MailRepositoriesRoutes implements Routes {
return Optional.ofNullable(request.queryParams("processor"));
}
+ private Optional<Boolean> parseConsume(Request request) {
+ return Optional.ofNullable(request.queryParams("consume"))
+ .map(Boolean::parseBoolean);
+ }
+
private MailQueueName parseTargetQueue(Request request) {
return Optional.ofNullable(request.queryParams("queue"))
.map(MailQueueName::of)
diff --git a/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingAllMailsTask.java b/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingAllMailsTask.java
index 5d8ea395e6..ef930f94fd 100644
--- a/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingAllMailsTask.java
+++ b/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingAllMailsTask.java
@@ -29,7 +29,6 @@ import javax.mail.MessagingException;
import org.apache.james.mailrepository.api.MailKey;
import org.apache.james.mailrepository.api.MailRepositoryPath;
import org.apache.james.mailrepository.api.MailRepositoryStore;
-import org.apache.james.queue.api.MailQueueName;
import org.apache.james.task.Task;
import org.apache.james.task.TaskExecutionDetails;
import org.apache.james.task.TaskType;
@@ -40,27 +39,21 @@ public class ReprocessingAllMailsTask implements Task {
public static class AdditionalInformation implements TaskExecutionDetails.AdditionalInformation {
private final MailRepositoryPath repositoryPath;
- private final String targetQueue;
- private final Optional<String> targetProcessor;
+ private final ReprocessingService.Configuration configuration;
private final long initialCount;
private final long remainingCount;
private final Instant timestamp;
- public AdditionalInformation(MailRepositoryPath repositoryPath, MailQueueName targetQueue, Optional<String> targetProcessor, long initialCount, long remainingCount, Instant timestamp) {
+ public AdditionalInformation(MailRepositoryPath repositoryPath, ReprocessingService.Configuration configuration, long initialCount, long remainingCount, Instant timestamp) {
this.repositoryPath = repositoryPath;
- this.targetQueue = targetQueue.asString();
- this.targetProcessor = targetProcessor;
+ this.configuration = configuration;
this.initialCount = initialCount;
this.remainingCount = remainingCount;
this.timestamp = timestamp;
}
- public String getTargetQueue() {
- return targetQueue;
- }
-
- public Optional<String> getTargetProcessor() {
- return targetProcessor;
+ public ReprocessingService.Configuration getConfiguration() {
+ return configuration;
}
public String getRepositoryPath() {
@@ -97,17 +90,15 @@ public class ReprocessingAllMailsTask implements Task {
private final ReprocessingService reprocessingService;
private final MailRepositoryPath repositoryPath;
- private final MailQueueName targetQueue;
- private final Optional<String> targetProcessor;
+ private final ReprocessingService.Configuration configuration;
private final long repositorySize;
private final AtomicLong processedCount;
public ReprocessingAllMailsTask(ReprocessingService reprocessingService, long repositorySize,
- MailRepositoryPath repositoryPath, MailQueueName targetQueue, Optional<String> targetProcessor) {
+ MailRepositoryPath repositoryPath, ReprocessingService.Configuration configuration) {
this.reprocessingService = reprocessingService;
this.repositoryPath = repositoryPath;
- this.targetQueue = targetQueue;
- this.targetProcessor = targetProcessor;
+ this.configuration = configuration;
this.repositorySize = repositorySize;
this.processedCount = new AtomicLong(0);
}
@@ -119,7 +110,7 @@ public class ReprocessingAllMailsTask implements Task {
@Override
public Result run() {
try {
- reprocessingService.reprocessAll(repositoryPath, targetProcessor, targetQueue, this::notifyProgress);
+ reprocessingService.reprocessAll(repositoryPath, configuration, this::notifyProgress);
return Result.COMPLETED;
} catch (MessagingException | MailRepositoryStore.MailRepositoryStoreException e) {
LOGGER.error("Encountered error while reprocessing repository", e);
@@ -135,12 +126,8 @@ public class ReprocessingAllMailsTask implements Task {
return repositorySize;
}
- Optional<String> getTargetProcessor() {
- return targetProcessor;
- }
-
- MailQueueName getTargetQueue() {
- return targetQueue;
+ ReprocessingService.Configuration getConfiguration() {
+ return configuration;
}
@Override
@@ -151,7 +138,7 @@ public class ReprocessingAllMailsTask implements Task {
@Override
public Optional<TaskExecutionDetails.AdditionalInformation> details() {
return Optional.of(new AdditionalInformation(
- repositoryPath, targetQueue, targetProcessor, repositorySize, repositorySize - processedCount.get(),
+ repositoryPath, configuration, repositorySize, repositorySize - processedCount.get(),
Clock.systemUTC().instant()));
}
diff --git a/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingAllMailsTaskAdditionalInformationDTO.java b/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingAllMailsTaskAdditionalInformationDTO.java
index adacac9f6a..55dcdb76fb 100644
--- a/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingAllMailsTaskAdditionalInformationDTO.java
+++ b/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingAllMailsTaskAdditionalInformationDTO.java
@@ -35,16 +35,19 @@ public class ReprocessingAllMailsTaskAdditionalInformationDTO implements Additio
.convertToDTO(ReprocessingAllMailsTaskAdditionalInformationDTO.class)
.toDomainObjectConverter(dto -> new ReprocessingAllMailsTask.AdditionalInformation(
MailRepositoryPath.from(dto.repositoryPath),
- MailQueueName.of(dto.targetQueue),
- dto.targetProcessor,
+ new ReprocessingService.Configuration(
+ MailQueueName.of(dto.getTargetQueue()),
+ dto.getTargetProcessor(),
+ dto.isConsume()),
dto.initialCount,
dto.remainingCount,
dto.timestamp))
.toDTOConverter((details, type) -> new ReprocessingAllMailsTaskAdditionalInformationDTO(
type,
details.getRepositoryPath(),
- details.getTargetQueue(),
- details.getTargetProcessor(),
+ details.getConfiguration().getMailQueueName().asString(),
+ details.getConfiguration().getTargetProcessor(),
+ Optional.of(details.getConfiguration().isConsume()),
details.getInitialCount(),
details.getRemainingCount(),
details.timestamp()))
@@ -56,6 +59,7 @@ public class ReprocessingAllMailsTaskAdditionalInformationDTO implements Additio
private final String repositoryPath;
private final String targetQueue;
private final Optional<String> targetProcessor;
+ private final boolean consume;
private final long initialCount;
private final long remainingCount;
private final Instant timestamp;
@@ -65,6 +69,7 @@ public class ReprocessingAllMailsTaskAdditionalInformationDTO implements Additio
@JsonProperty("repositoryPath") String repositoryPath,
@JsonProperty("targetQueue") String targetQueue,
@JsonProperty("targetProcessor") Optional<String> targetProcessor,
+ @JsonProperty("consume") Optional<Boolean> consume,
@JsonProperty("initialCount") long initialCount,
@JsonProperty("remainingCount") long remainingCount,
@JsonProperty("timestamp") Instant timestamp) {
@@ -75,6 +80,11 @@ public class ReprocessingAllMailsTaskAdditionalInformationDTO implements Additio
this.initialCount = initialCount;
this.remainingCount = remainingCount;
this.timestamp = timestamp;
+ this.consume = consume.orElse(true);
+ }
+
+ public boolean isConsume() {
+ return consume;
}
@Override
diff --git a/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingAllMailsTaskDTO.java b/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingAllMailsTaskDTO.java
index d81f2a8ee1..7a7b722aee 100644
--- a/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingAllMailsTaskDTO.java
+++ b/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingAllMailsTaskDTO.java
@@ -46,9 +46,9 @@ public class ReprocessingAllMailsTaskDTO implements TaskDTO {
typeName,
domainObject.getRepositorySize(),
domainObject.getRepositoryPath().urlEncoded(),
- domainObject.getTargetQueue().asString(),
- domainObject.getTargetProcessor()
- );
+ domainObject.getConfiguration().getMailQueueName().asString(),
+ Optional.of(domainObject.getConfiguration().isConsume()),
+ domainObject.getConfiguration().getTargetProcessor());
} catch (Exception e) {
throw new ReprocessingAllMailsTask.UrlEncodingFailureSerializationException(domainObject.getRepositoryPath());
}
@@ -58,17 +58,20 @@ public class ReprocessingAllMailsTaskDTO implements TaskDTO {
private final long repositorySize;
private final String repositoryPath;
private final String targetQueue;
+ private final boolean consume;
private final Optional<String> targetProcessor;
public ReprocessingAllMailsTaskDTO(@JsonProperty("type") String type,
@JsonProperty("repositorySize") long repositorySize,
@JsonProperty("repositoryPath") String repositoryPath,
@JsonProperty("targetQueue") String targetQueue,
+ @JsonProperty("consume") Optional<Boolean> consume,
@JsonProperty("targetProcessor") Optional<String> targetProcessor) {
this.type = type;
this.repositorySize = repositorySize;
this.repositoryPath = repositoryPath;
this.targetQueue = targetQueue;
+ this.consume = consume.orElse(true);
this.targetProcessor = targetProcessor;
}
@@ -78,9 +81,10 @@ public class ReprocessingAllMailsTaskDTO implements TaskDTO {
reprocessingService,
repositorySize,
MailRepositoryPath.fromEncoded(repositoryPath),
- MailQueueName.of(targetQueue),
- targetProcessor
- );
+ new ReprocessingService.Configuration(
+ MailQueueName.of(targetQueue),
+ targetProcessor,
+ consume));
} catch (Exception e) {
throw new ReprocessingAllMailsTask.InvalidMailRepositoryPathDeserializationException(repositoryPath);
}
@@ -103,6 +107,10 @@ public class ReprocessingAllMailsTaskDTO implements TaskDTO {
return targetQueue;
}
+ public boolean isConsume() {
+ return consume;
+ }
+
public Optional<String> getTargetProcessor() {
return targetProcessor;
}
diff --git a/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingOneMailTask.java b/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingOneMailTask.java
index 2777ce24ae..bb81055ca0 100644
--- a/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingOneMailTask.java
+++ b/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingOneMailTask.java
@@ -28,7 +28,6 @@ import javax.mail.MessagingException;
import org.apache.james.mailrepository.api.MailKey;
import org.apache.james.mailrepository.api.MailRepositoryPath;
import org.apache.james.mailrepository.api.MailRepositoryStore;
-import org.apache.james.queue.api.MailQueueName;
import org.apache.james.task.Task;
import org.apache.james.task.TaskExecutionDetails;
import org.apache.james.task.TaskType;
@@ -39,16 +38,14 @@ public class ReprocessingOneMailTask implements Task {
public static class AdditionalInformation implements TaskExecutionDetails.AdditionalInformation {
private final MailRepositoryPath repositoryPath;
- private final String targetQueue;
+ private final ReprocessingService.Configuration configuration;
private final MailKey mailKey;
- private final Optional<String> targetProcessor;
private final Instant timestamp;
- public AdditionalInformation(MailRepositoryPath repositoryPath, MailQueueName targetQueue, MailKey mailKey, Optional<String> targetProcessor, Instant timestamp) {
+ public AdditionalInformation(MailRepositoryPath repositoryPath, ReprocessingService.Configuration configuration, MailKey mailKey, Instant timestamp) {
this.repositoryPath = repositoryPath;
- this.targetQueue = targetQueue.asString();
+ this.configuration = configuration;
this.mailKey = mailKey;
- this.targetProcessor = targetProcessor;
this.timestamp = timestamp;
}
@@ -56,12 +53,8 @@ public class ReprocessingOneMailTask implements Task {
return mailKey.asString();
}
- public String getTargetQueue() {
- return targetQueue;
- }
-
- public Optional<String> getTargetProcessor() {
- return targetProcessor;
+ public ReprocessingService.Configuration getConfiguration() {
+ return configuration;
}
public String getRepositoryPath() {
@@ -90,29 +83,26 @@ public class ReprocessingOneMailTask implements Task {
private final ReprocessingService reprocessingService;
private final MailRepositoryPath repositoryPath;
- private final MailQueueName targetQueue;
+ private final ReprocessingService.Configuration configuration;
private final MailKey mailKey;
- private final Optional<String> targetProcessor;
private final AdditionalInformation additionalInformation;
public ReprocessingOneMailTask(ReprocessingService reprocessingService,
MailRepositoryPath repositoryPath,
- MailQueueName targetQueue,
+ ReprocessingService.Configuration configuration,
MailKey mailKey,
- Optional<String> targetProcessor,
Clock clock) {
this.reprocessingService = reprocessingService;
this.repositoryPath = repositoryPath;
- this.targetQueue = targetQueue;
+ this.configuration = configuration;
this.mailKey = mailKey;
- this.targetProcessor = targetProcessor;
- this.additionalInformation = new AdditionalInformation(repositoryPath, targetQueue, mailKey, targetProcessor, clock.instant());
+ this.additionalInformation = new AdditionalInformation(repositoryPath, configuration, mailKey, clock.instant());
}
@Override
public Result run() {
try {
- reprocessingService.reprocess(repositoryPath, mailKey, targetProcessor, targetQueue);
+ reprocessingService.reprocess(repositoryPath, mailKey, configuration);
return Result.COMPLETED;
} catch (MessagingException | MailRepositoryStore.MailRepositoryStoreException e) {
LOGGER.error("Encountered error while reprocessing repository", e);
@@ -129,16 +119,12 @@ public class ReprocessingOneMailTask implements Task {
return repositoryPath;
}
- MailQueueName getTargetQueue() {
- return targetQueue;
- }
-
MailKey getMailKey() {
return mailKey;
}
- Optional<String> getTargetProcessor() {
- return targetProcessor;
+ public ReprocessingService.Configuration getConfiguration() {
+ return configuration;
}
@Override
diff --git a/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingOneMailTaskAdditionalInformationDTO.java b/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingOneMailTaskAdditionalInformationDTO.java
index f84ae9d33a..40802d8bf3 100644
--- a/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingOneMailTaskAdditionalInformationDTO.java
+++ b/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingOneMailTaskAdditionalInformationDTO.java
@@ -36,16 +36,19 @@ public class ReprocessingOneMailTaskAdditionalInformationDTO implements Addition
.convertToDTO(ReprocessingOneMailTaskAdditionalInformationDTO.class)
.toDomainObjectConverter(dto -> new ReprocessingOneMailTask.AdditionalInformation(
MailRepositoryPath.from(dto.repositoryPath),
- MailQueueName.of(dto.targetQueue),
+ new ReprocessingService.Configuration(
+ MailQueueName.of(dto.targetQueue),
+ dto.targetProcessor,
+ dto.isConsume()),
new MailKey(dto.mailKey),
- dto.targetProcessor,
dto.timestamp))
.toDTOConverter((details, type) -> new ReprocessingOneMailTaskAdditionalInformationDTO(
type,
details.getRepositoryPath(),
- details.getTargetQueue(),
+ details.getConfiguration().getMailQueueName().asString(),
details.getMailKey(),
- details.getTargetProcessor(),
+ Optional.of(details.getConfiguration().isConsume()),
+ details.getConfiguration().getTargetProcessor(),
details.timestamp()))
.typeName(ReprocessingOneMailTask.TYPE.asString())
.withFactory(AdditionalInformationDTOModule::new);
@@ -56,15 +59,18 @@ public class ReprocessingOneMailTaskAdditionalInformationDTO implements Addition
private final String targetQueue;
private final String mailKey;
private final Optional<String> targetProcessor;
+ private final boolean consume;
private final Instant timestamp;
public ReprocessingOneMailTaskAdditionalInformationDTO(@JsonProperty("type") String type,
@JsonProperty("repositoryPath") String repositoryPath,
@JsonProperty("targetQueue") String targetQueue,
@JsonProperty("mailKey") String mailKey,
+ @JsonProperty("consume") Optional<Boolean> consume,
@JsonProperty("targetProcessor") Optional<String> targetProcessor,
@JsonProperty("timestamp") Instant timestamp) {
this.type = type;
+ this.consume = consume.orElse(true);
this.repositoryPath = repositoryPath;
this.targetQueue = targetQueue;
this.mailKey = mailKey;
@@ -72,6 +78,10 @@ public class ReprocessingOneMailTaskAdditionalInformationDTO implements Addition
this.timestamp = timestamp;
}
+ public boolean isConsume() {
+ return consume;
+ }
+
@Override
public String getType() {
return type;
diff --git a/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingOneMailTaskDTO.java b/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingOneMailTaskDTO.java
index ea09c18bee..9ced35507e 100644
--- a/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingOneMailTaskDTO.java
+++ b/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingOneMailTaskDTO.java
@@ -47,10 +47,10 @@ public class ReprocessingOneMailTaskDTO implements TaskDTO {
return new ReprocessingOneMailTaskDTO(
typeName,
domainObject.getRepositoryPath().urlEncoded(),
- domainObject.getTargetQueue().asString(),
+ domainObject.getConfiguration().getMailQueueName().asString(),
domainObject.getMailKey().asString(),
- domainObject.getTargetProcessor()
- );
+ domainObject.getConfiguration().getTargetProcessor(),
+ Optional.of(domainObject.getConfiguration().isConsume()));
} catch (Exception e) {
throw new ReprocessingOneMailTask.UrlEncodingFailureSerializationException(domainObject.getRepositoryPath());
}
@@ -61,28 +61,32 @@ public class ReprocessingOneMailTaskDTO implements TaskDTO {
private final String targetQueue;
private final String mailKey;
private final Optional<String> targetProcessor;
+ private final boolean consume;
public ReprocessingOneMailTaskDTO(@JsonProperty("type") String type,
@JsonProperty("repositoryPath") String repositoryPath,
@JsonProperty("targetQueue") String targetQueue,
@JsonProperty("mailKey") String mailKey,
- @JsonProperty("targetProcessor") Optional<String> targetProcessor) {
+ @JsonProperty("targetProcessor") Optional<String> targetProcessor,
+ @JsonProperty("boolean") Optional<Boolean> consume) {
this.type = type;
this.repositoryPath = repositoryPath;
this.mailKey = mailKey;
this.targetQueue = targetQueue;
this.targetProcessor = targetProcessor;
+ this.consume = consume.orElse(true);
}
public ReprocessingOneMailTask fromDTO(ReprocessingService reprocessingService, Clock clock) {
return new ReprocessingOneMailTask(
reprocessingService,
getMailRepositoryPath(),
- MailQueueName.of(targetQueue),
+ new ReprocessingService.Configuration(
+ MailQueueName.of(targetQueue),
+ targetProcessor,
+ consume),
new MailKey(mailKey),
- targetProcessor,
- clock
- );
+ clock);
}
private MailRepositoryPath getMailRepositoryPath() {
@@ -110,6 +114,10 @@ public class ReprocessingOneMailTaskDTO implements TaskDTO {
return targetQueue;
}
+ public boolean isConsume() {
+ return consume;
+ }
+
public Optional<String> getTargetProcessor() {
return targetProcessor;
}
diff --git a/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingService.java b/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingService.java
index 5b7ed6baa6..9c443d2365 100644
--- a/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingService.java
+++ b/server/protocols/webadmin/webadmin-mailrepository/src/main/java/org/apache/james/webadmin/service/ReprocessingService.java
@@ -53,20 +53,46 @@ public class ReprocessingService {
}
}
+ public static class Configuration {
+ private final MailQueueName mailQueueName;
+ private final Optional<String> targetProcessor;
+ private final boolean consume;
+
+ public Configuration(MailQueueName mailQueueName, Optional<String> targetProcessor, boolean consume) {
+ this.mailQueueName = mailQueueName;
+ this.targetProcessor = targetProcessor;
+ this.consume = consume;
+ }
+
+ public MailQueueName getMailQueueName() {
+ return mailQueueName;
+ }
+
+ public Optional<String> getTargetProcessor() {
+ return targetProcessor;
+ }
+
+ public boolean isConsume() {
+ return consume;
+ }
+ }
+
static class Reprocessor implements Closeable {
private final MailQueue mailQueue;
- private final Optional<String> targetProcessor;
+ private final Configuration configuration;
- Reprocessor(MailQueue mailQueue, Optional<String> targetProcessor) {
+ Reprocessor(MailQueue mailQueue, Configuration configuration) {
this.mailQueue = mailQueue;
- this.targetProcessor = targetProcessor;
+ this.configuration = configuration;
}
private void reprocess(MailRepository repository, Mail mail) {
try {
- targetProcessor.ifPresent(mail::setState);
+ configuration.getTargetProcessor().ifPresent(mail::setState);
mailQueue.enQueue(mail);
- repository.remove(mail);
+ if (configuration.isConsume()) {
+ repository.remove(mail);
+ }
} catch (Exception e) {
throw new RuntimeException("Error encountered while reprocessing mail " + mail.getName(), e);
} finally {
@@ -94,8 +120,8 @@ public class ReprocessingService {
this.mailRepositoryStoreService = mailRepositoryStoreService;
}
- public void reprocessAll(MailRepositoryPath path, Optional<String> targetProcessor, MailQueueName targetQueue, Consumer<MailKey> keyListener) throws MailRepositoryStore.MailRepositoryStoreException, MessagingException {
- try (Reprocessor reprocessor = new Reprocessor(getMailQueue(targetQueue), targetProcessor)) {
+ public void reprocessAll(MailRepositoryPath path, Configuration configuration, Consumer<MailKey> keyListener) throws MailRepositoryStore.MailRepositoryStoreException, MessagingException {
+ try (Reprocessor reprocessor = new Reprocessor(getMailQueue(configuration.getMailQueueName()), configuration)) {
mailRepositoryStoreService
.getRepositories(path)
.forEach(Throwing.consumer((MailRepository repository) ->
@@ -107,8 +133,8 @@ public class ReprocessingService {
}
}
- public void reprocess(MailRepositoryPath path, MailKey key, Optional<String> targetProcessor, MailQueueName targetQueue) throws MailRepositoryStore.MailRepositoryStoreException, MessagingException {
- try (Reprocessor reprocessor = new Reprocessor(getMailQueue(targetQueue), targetProcessor)) {
+ public void reprocess(MailRepositoryPath path, MailKey key, Configuration configuration) throws MailRepositoryStore.MailRepositoryStoreException, MessagingException {
+ try (Reprocessor reprocessor = new Reprocessor(getMailQueue(configuration.getMailQueueName()), configuration)) {
Pair<MailRepository, Mail> mailPair = mailRepositoryStoreService
.getRepositories(path)
.map(Throwing.function(repository -> Pair.of(repository, Optional.ofNullable(repository.retrieve(key)))))
diff --git a/server/protocols/webadmin/webadmin-mailrepository/src/test/java/org/apache/james/webadmin/routes/MailRepositoriesRoutesTest.java b/server/protocols/webadmin/webadmin-mailrepository/src/test/java/org/apache/james/webadmin/routes/MailRepositoriesRoutesTest.java
index 95e8860d96..16bd869e8d 100644
--- a/server/protocols/webadmin/webadmin-mailrepository/src/test/java/org/apache/james/webadmin/routes/MailRepositoriesRoutesTest.java
+++ b/server/protocols/webadmin/webadmin-mailrepository/src/test/java/org/apache/james/webadmin/routes/MailRepositoriesRoutesTest.java
@@ -1133,6 +1133,7 @@ class MailRepositoriesRoutesTest {
.body("additionalInformation.remainingCount", is(0))
.body("additionalInformation.targetProcessor", is(emptyOrNullString()))
.body("additionalInformation.targetQueue", is(MailQueueFactory.SPOOL.asString()))
+ .body("additionalInformation.consume", is(true))
.body("startedDate", is(notNullValue()))
.body("submitDate", is(notNullValue()))
.body("completedDate", is(notNullValue()));
@@ -1157,6 +1158,7 @@ class MailRepositoriesRoutesTest {
.param("action", "reprocess")
.param("queue", CUSTOM_QUEUE.asString())
.param("processor", transport)
+ .param("consume", false)
.patch(PATH_ESCAPED_MY_REPO + "/mails")
.jsonPath()
.get("taskId");
@@ -1174,6 +1176,7 @@ class MailRepositoriesRoutesTest {
.body("additionalInformation.remainingCount", is(0))
.body("additionalInformation.targetProcessor", is(transport))
.body("additionalInformation.targetQueue", is(CUSTOM_QUEUE.asString()))
+ .body("additionalInformation.consume", is(false))
.body("startedDate", is(notNullValue()))
.body("submitDate", is(notNullValue()))
.body("completedDate", is(notNullValue()));
@@ -1243,6 +1246,39 @@ class MailRepositoriesRoutesTest {
.isEmpty();
}
+ @Test
+ void reprocessingAllTaskShouldNotClearMailRepositoryWhenNotConsume() throws Exception {
+ MailRepository mailRepository = mailRepositoryStore.create(URL_MY_REPO);
+ String name1 = "name1";
+ String name2 = "name2";
+ mailRepository.store(FakeMail.builder()
+ .name(name1)
+ .mimeMessage(MimeMessageUtil.mimeMessageFromBytes(MESSAGE_BYTES))
+ .build());
+ mailRepository.store(FakeMail.builder()
+ .name(name2)
+ .mimeMessage(MimeMessageUtil.mimeMessageFromBytes(MESSAGE_BYTES))
+ .build());
+
+ String transport = "transport";
+ String taskId = with()
+ .param("action", "reprocess")
+ .param("queue", CUSTOM_QUEUE.asString())
+ .param("processor", transport)
+ .param("consume", false)
+ .patch(PATH_ESCAPED_MY_REPO + "/mails")
+ .jsonPath()
+ .get("taskId");
+
+ with()
+ .basePath(TasksRoutes.BASE)
+ .get(taskId + "/await");
+
+ assertThat(mailRepository.list())
+ .toIterable()
+ .hasSize(2);
+ }
+
@Test
void reprocessingAllTaskShouldClearBothMailRepositoriesWhenSamePath() throws Exception {
MailRepository mailRepository1 = mailRepositoryStore.create(URL_MY_REPO);
@@ -1519,6 +1555,7 @@ class MailRepositoriesRoutesTest {
.body("additionalInformation.mailKey", is(NAME_1))
.body("additionalInformation.targetProcessor", is(emptyOrNullString()))
.body("additionalInformation.targetQueue", is(MailQueueFactory.SPOOL.asString()))
+ .body("additionalInformation.consume", is(true))
.body("startedDate", is(notNullValue()))
.body("submitDate", is(notNullValue()))
.body("completedDate", is(notNullValue()));
@@ -1543,6 +1580,7 @@ class MailRepositoriesRoutesTest {
.param("action", "reprocess")
.param("queue", CUSTOM_QUEUE.asString())
.param("processor", transport)
+ .param("consume", false)
.patch(PATH_ESCAPED_MY_REPO + "/mails/" + NAME_1)
.jsonPath()
.get("taskId");
@@ -1559,6 +1597,7 @@ class MailRepositoriesRoutesTest {
.body("additionalInformation.mailKey", is(NAME_1))
.body("additionalInformation.targetProcessor", is(transport))
.body("additionalInformation.targetQueue", is(CUSTOM_QUEUE.asString()))
+ .body("additionalInformation.consume", is(false))
.body("startedDate", is(notNullValue()))
.body("submitDate", is(notNullValue()))
.body("completedDate", is(notNullValue()));
@@ -1781,6 +1820,58 @@ class MailRepositoriesRoutesTest {
.isEmpty();
}
+ @Test
+ void reprocessingOneTaskShouldNotRemoveEmailWhenNotConsume() throws Exception {
+ MailRepository mailRepository = mailRepositoryStore.create(URL_MY_REPO);
+ mailRepository.store(FakeMail.builder()
+ .name(NAME_1)
+ .build());
+ mailRepository.store(FakeMail.builder()
+ .name(NAME_2)
+ .build());
+
+ String taskId = with()
+ .param("action", "reprocess")
+ .param("queue", CUSTOM_QUEUE.asString())
+ .param("consume", false)
+ .patch(PATH_ESCAPED_MY_REPO + "/mails/" + NAME_1)
+ .jsonPath()
+ .get("taskId");
+
+ with()
+ .basePath(TasksRoutes.BASE)
+ .get(taskId + "/await");
+
+ assertThat(mailRepository.size())
+ .isEqualTo(2);
+ }
+
+ @Test
+ void reprocessingOneTaskShouldRemoveEmailWhenConsume() throws Exception {
+ MailRepository mailRepository = mailRepositoryStore.create(URL_MY_REPO);
+ mailRepository.store(FakeMail.builder()
+ .name(NAME_1)
+ .build());
+ mailRepository.store(FakeMail.builder()
+ .name(NAME_2)
+ .build());
+
+ String taskId = with()
+ .param("action", "reprocess")
+ .param("queue", CUSTOM_QUEUE.asString())
+ .param("consume", true)
+ .patch(PATH_ESCAPED_MY_REPO + "/mails/" + NAME_1)
+ .jsonPath()
+ .get("taskId");
+
+ with()
+ .basePath(TasksRoutes.BASE)
+ .get(taskId + "/await");
+
+ assertThat(mailRepository.size())
+ .isEqualTo(1);
+ }
+
@Test
void reprocessingOneTaskShouldNotRemoveMailFromRepositoryWhenUnknownMailKey() throws Exception {
MailRepository mailRepository = mailRepositoryStore.create(URL_MY_REPO);
diff --git a/server/protocols/webadmin/webadmin-mailrepository/src/test/java/org/apache/james/webadmin/service/ReprocessingAllMailsTaskTest.java b/server/protocols/webadmin/webadmin-mailrepository/src/test/java/org/apache/james/webadmin/service/ReprocessingAllMailsTaskTest.java
index d9eb9ca308..64b5fe33e3 100644
--- a/server/protocols/webadmin/webadmin-mailrepository/src/test/java/org/apache/james/webadmin/service/ReprocessingAllMailsTaskTest.java
+++ b/server/protocols/webadmin/webadmin-mailrepository/src/test/java/org/apache/james/webadmin/service/ReprocessingAllMailsTaskTest.java
@@ -19,6 +19,7 @@
package org.apache.james.webadmin.service;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.Mockito.mock;
@@ -28,6 +29,7 @@ import java.util.Optional;
import org.apache.james.JsonSerializationVerifier;
import org.apache.james.mailrepository.api.MailRepositoryPath;
import org.apache.james.queue.api.MailQueueName;
+import org.apache.james.server.task.json.JsonTaskAdditionalInformationSerializer;
import org.apache.james.server.task.json.JsonTaskSerializer;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
@@ -41,15 +43,20 @@ class ReprocessingAllMailsTaskTest {
private static final MailQueueName TARGET_QUEUE = MailQueueName.of("queue");
private static final Optional<String> SOME_TARGET_PROCESSOR = Optional.of("targetProcessor");
private static final long REMAINING_COUNT = 3L;
- private static final String SERIALIZED_TASK_WITH_TARGET_PROCESSOR = "{\"type\":\"reprocessing-all\",\"repositorySize\":5,\"repositoryPath\":\"a\",\"targetQueue\":\"queue\",\"targetProcessor\":\"targetProcessor\"}";
- private static final String SERIALIZED_TASK_WITHOUT_TARGET_PROCESSOR = "{\"type\":\"reprocessing-all\",\"repositorySize\":5,\"repositoryPath\":\"a\",\"targetQueue\":\"queue\"}";
- private static final String SERIALIZED_TASK_ADDITIONAL_INFORMATION_WITH_TARGET_PROCESSOR = "{\"type\":\"reprocessing-all\", \"repositoryPath\":\"a\",\"targetQueue\":\"queue\",\"targetProcessor\":\"targetProcessor\",\"initialCount\":5,\"remainingCount\":3, \"timestamp\":\"2018-11-13T12:00:55Z\"}";
- private static final String SERIALIZED_TASK_ADDITIONAL_INFORMATION_WITHOUT_TARGET_PROCESSOR = "{\"type\":\"reprocessing-all\", \"repositoryPath\":\"a\",\"targetQueue\":\"queue\", \"initialCount\":5,\"remainingCount\":3, \"timestamp\":\"2018-11-13T12:00:55Z\"}";
+ private static final boolean CONSUME = true;
+
+ private static final String SERIALIZED_TASK_WITH_TARGET_PROCESSOR = "{\"type\":\"reprocessing-all\",\"repositorySize\":5,\"repositoryPath\":\"a\",\"targetQueue\":\"queue\",\"targetProcessor\":\"targetProcessor\",\"consume\":true}";
+ private static final String SERIALIZED_TASK_WITHOUT_TARGET_PROCESSOR = "{\"type\":\"reprocessing-all\",\"repositorySize\":5,\"repositoryPath\":\"a\",\"targetQueue\":\"queue\",\"consume\":false}";
+ private static final String SERIALIZED_TASK_ADDITIONAL_INFORMATION_WITH_TARGET_PROCESSOR = "{\"type\":\"reprocessing-all\", \"repositoryPath\":\"a\",\"targetQueue\":\"queue\",\"targetProcessor\":\"targetProcessor\",\"initialCount\":5,\"remainingCount\":3, \"timestamp\":\"2018-11-13T12:00:55Z\",\"consume\":true}";
+ private static final String SERIALIZED_TASK_ADDITIONAL_INFORMATION_WITHOUT_TARGET_PROCESSOR = "{\"type\":\"reprocessing-all\", \"repositoryPath\":\"a\",\"targetQueue\":\"queue\", \"initialCount\":5,\"remainingCount\":3, \"timestamp\":\"2018-11-13T12:00:55Z\",\"consume\":false}";
+
+ private static final String OLD_SERIALIZED_TASK = "{\"type\":\"reprocessing-all\",\"repositorySize\":5,\"repositoryPath\":\"a\",\"targetQueue\":\"queue\",\"targetProcessor\":\"targetProcessor\"}";
+ private static final String OLD_SERIALIZED_TASK_ADDITIONAL_INFORMATION = "{\"type\":\"reprocessing-all\", \"repositoryPath\":\"a\",\"targetQueue\":\"queue\",\"targetProcessor\":\"targetProcessor\",\"initialCount\":5,\"remainingCount\":3, \"timestamp\":\"2018-11-13T12:00:55Z\"}";
@Test
void taskShouldBeSerializable() throws Exception {
- ReprocessingAllMailsTask taskWithTargetProcessor = new ReprocessingAllMailsTask(REPROCESSING_SERVICE, REPOSITORY_SIZE, REPOSITORY_PATH, TARGET_QUEUE, SOME_TARGET_PROCESSOR);
- ReprocessingAllMailsTask taskWithoutTargetProcessor = new ReprocessingAllMailsTask(REPROCESSING_SERVICE, REPOSITORY_SIZE, REPOSITORY_PATH, TARGET_QUEUE, Optional.empty());
+ ReprocessingAllMailsTask taskWithTargetProcessor = new ReprocessingAllMailsTask(REPROCESSING_SERVICE, REPOSITORY_SIZE, REPOSITORY_PATH, new ReprocessingService.Configuration(TARGET_QUEUE, SOME_TARGET_PROCESSOR, CONSUME));
+ ReprocessingAllMailsTask taskWithoutTargetProcessor = new ReprocessingAllMailsTask(REPROCESSING_SERVICE, REPOSITORY_SIZE, REPOSITORY_PATH, new ReprocessingService.Configuration(TARGET_QUEUE, Optional.empty(), !CONSUME));
JsonSerializationVerifier.dtoModule(ReprocessingAllMailsTaskDTO.module(REPROCESSING_SERVICE))
.testCase(taskWithTargetProcessor, SERIALIZED_TASK_WITH_TARGET_PROCESSOR)
@@ -58,7 +65,9 @@ class ReprocessingAllMailsTaskTest {
}
@ParameterizedTest
- @ValueSource(strings = {"{\"type\":\"reprocessing-all\",\"repositorySize\":5,\"repositoryPath\":\"%\",\"targetQueue\":\"queue\",\"targetProcessor\":\"targetProcessor\"}", "{\"type\":\"reprocessing-all\",\"repositorySize\":5,\"repositoryPath\":\"%\",\"targetQueue\":\"queue\"}"})
+ @ValueSource(strings = {
+ "{\"type\":\"reprocessing-all\",\"repositorySize\":5,\"repositoryPath\":\"%\",\"targetQueue\":\"queue\",\"targetProcessor\":\"targetProcessor\"}",
+ "{\"type\":\"reprocessing-all\",\"repositorySize\":5,\"repositoryPath\":\"%\",\"targetQueue\":\"queue\"}"})
void taskShouldThrowOnDeserializationUrlDecodingError(String serialized) {
JsonTaskSerializer testee = JsonTaskSerializer.of(ReprocessingAllMailsTaskDTO.module(REPROCESSING_SERVICE));
@@ -68,9 +77,11 @@ class ReprocessingAllMailsTaskTest {
@Test
void additionalInformationShouldBeSerializable() throws Exception {
- ReprocessingAllMailsTask.AdditionalInformation details = new ReprocessingAllMailsTask.AdditionalInformation(REPOSITORY_PATH, TARGET_QUEUE, SOME_TARGET_PROCESSOR,
+ ReprocessingAllMailsTask.AdditionalInformation details = new ReprocessingAllMailsTask.AdditionalInformation(REPOSITORY_PATH,
+ new ReprocessingService.Configuration(TARGET_QUEUE, SOME_TARGET_PROCESSOR, CONSUME),
REPOSITORY_SIZE, REMAINING_COUNT, TIMESTAMP);
- ReprocessingAllMailsTask.AdditionalInformation detailsWithoutProcessor = new ReprocessingAllMailsTask.AdditionalInformation(REPOSITORY_PATH, TARGET_QUEUE, Optional.empty(),
+ ReprocessingAllMailsTask.AdditionalInformation detailsWithoutProcessor = new ReprocessingAllMailsTask.AdditionalInformation(REPOSITORY_PATH,
+ new ReprocessingService.Configuration(TARGET_QUEUE, Optional.empty(), !CONSUME),
REPOSITORY_SIZE, REMAINING_COUNT, TIMESTAMP);
JsonSerializationVerifier.dtoModule(ReprocessingAllMailsTaskAdditionalInformationDTO.module())
@@ -78,4 +89,24 @@ class ReprocessingAllMailsTaskTest {
.testCase(detailsWithoutProcessor, SERIALIZED_TASK_ADDITIONAL_INFORMATION_WITHOUT_TARGET_PROCESSOR)
.verify();
}
+
+ @Test
+ void shouldDeserializePreviousTaskFormat() throws Exception {
+ ReprocessingAllMailsTask taskWithTargetProcessor = new ReprocessingAllMailsTask(REPROCESSING_SERVICE, REPOSITORY_SIZE, REPOSITORY_PATH, new ReprocessingService.Configuration(TARGET_QUEUE, SOME_TARGET_PROCESSOR, CONSUME));
+ JsonTaskSerializer testee = JsonTaskSerializer.of(ReprocessingAllMailsTaskDTO.module(REPROCESSING_SERVICE));
+
+ assertThat(testee.deserialize(OLD_SERIALIZED_TASK))
+ .isEqualToComparingFieldByFieldRecursively(taskWithTargetProcessor);
+ }
+
+ @Test
+ void shouldDeserializePreviousAdditionalInformationFormat() throws Exception {
+ ReprocessingAllMailsTask.AdditionalInformation details = new ReprocessingAllMailsTask.AdditionalInformation(REPOSITORY_PATH,
+ new ReprocessingService.Configuration(TARGET_QUEUE, SOME_TARGET_PROCESSOR, CONSUME),
+ REPOSITORY_SIZE, REMAINING_COUNT, TIMESTAMP);
+ JsonTaskAdditionalInformationSerializer testee = JsonTaskAdditionalInformationSerializer.of(ReprocessingAllMailsTaskAdditionalInformationDTO.module());
+
+ assertThat(testee.deserialize(OLD_SERIALIZED_TASK_ADDITIONAL_INFORMATION))
+ .isEqualToComparingFieldByFieldRecursively(details);
+ }
}
\ No newline at end of file
diff --git a/server/protocols/webadmin/webadmin-mailrepository/src/test/java/org/apache/james/webadmin/service/ReprocessingOneMailTaskTest.java b/server/protocols/webadmin/webadmin-mailrepository/src/test/java/org/apache/james/webadmin/service/ReprocessingOneMailTaskTest.java
index 1d89553bab..cf0746befb 100644
--- a/server/protocols/webadmin/webadmin-mailrepository/src/test/java/org/apache/james/webadmin/service/ReprocessingOneMailTaskTest.java
+++ b/server/protocols/webadmin/webadmin-mailrepository/src/test/java/org/apache/james/webadmin/service/ReprocessingOneMailTaskTest.java
@@ -19,6 +19,7 @@
package org.apache.james.webadmin.service;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.Mockito.mock;
@@ -32,6 +33,7 @@ import org.apache.james.JsonSerializationVerifier;
import org.apache.james.mailrepository.api.MailKey;
import org.apache.james.mailrepository.api.MailRepositoryPath;
import org.apache.james.queue.api.MailQueueName;
+import org.apache.james.server.task.json.JsonTaskAdditionalInformationSerializer;
import org.apache.james.server.task.json.JsonTaskSerializer;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
@@ -41,27 +43,32 @@ class ReprocessingOneMailTaskTest {
private static final Instant TIMESTAMP = Instant.parse("2018-11-13T12:00:55Z");
private static final Clock CLOCK = Clock.fixed(TIMESTAMP, ZoneId.of("UTC"));
private static final ReprocessingService REPROCESSING_SERVICE = mock(ReprocessingService.class);
- private static final String SERIALIZED_TASK_1 = "{\"type\":\"reprocessing-one\",\"repositoryPath\":\"a\",\"targetQueue\":\"queue\",\"mailKey\": \"myMail\",\"targetProcessor\":\"targetProcessor\"}";
- private static final String SERIALIZED_TASK_1_ADDITIONAL_INFORMATION = "{\"type\":\"reprocessing-one\", \"repositoryPath\":\"a\",\"targetQueue\":\"queue\",\"mailKey\": \"myMail\",\"targetProcessor\":\"targetProcessor\", \"timestamp\":\"2018-11-13T12:00:55Z\"}";
+ private static final String SERIALIZED_TASK_1 = "{\"type\":\"reprocessing-one\",\"repositoryPath\":\"a\",\"targetQueue\":\"queue\",\"mailKey\": \"myMail\",\"targetProcessor\":\"targetProcessor\", \"consume\":true}";
+ private static final String SERIALIZED_TASK_1_ADDITIONAL_INFORMATION = "{\"type\":\"reprocessing-one\", \"repositoryPath\":\"a\",\"targetQueue\":\"queue\",\"mailKey\": \"myMail\",\"targetProcessor\":\"targetProcessor\", \"timestamp\":\"2018-11-13T12:00:55Z\", \"consume\":true}";
+ private static final String SERIALIZED_TASK_OLD = "{\"type\":\"reprocessing-one\",\"repositoryPath\":\"a\",\"targetQueue\":\"queue\",\"mailKey\": \"myMail\",\"targetProcessor\":\"targetProcessor\"}";
+ private static final String SERIALIZED_TASK_OLD_ADDITIONAL_INFORMATION = "{\"type\":\"reprocessing-one\", \"repositoryPath\":\"a\",\"targetQueue\":\"queue\",\"mailKey\": \"myMail\",\"targetProcessor\":\"targetProcessor\", \"timestamp\":\"2018-11-13T12:00:55Z\"}";
private static final MailRepositoryPath REPOSITORY_PATH = MailRepositoryPath.from("a");
private static final MailQueueName TARGET_QUEUE = MailQueueName.of("queue");
private static final MailKey MAIL_KEY = new MailKey("myMail");
private static final Optional<String> TARGET_PROCESSOR = Optional.of("targetProcessor");
+ public static final boolean CONSUME = true;
@Test
void taskShouldBeSerializable() throws Exception {
- ReprocessingOneMailTask taskWithTargetProcessor = new ReprocessingOneMailTask(REPROCESSING_SERVICE, REPOSITORY_PATH, TARGET_QUEUE, MAIL_KEY, TARGET_PROCESSOR, CLOCK);
- ReprocessingOneMailTask taskWithoutTargetProcessor = new ReprocessingOneMailTask(REPROCESSING_SERVICE, REPOSITORY_PATH, TARGET_QUEUE, MAIL_KEY, Optional.empty(), CLOCK);
+ ReprocessingOneMailTask taskWithTargetProcessor = new ReprocessingOneMailTask(REPROCESSING_SERVICE, REPOSITORY_PATH, new ReprocessingService.Configuration(TARGET_QUEUE, TARGET_PROCESSOR, CONSUME), MAIL_KEY, CLOCK);
+ ReprocessingOneMailTask taskWithoutTargetProcessor = new ReprocessingOneMailTask(REPROCESSING_SERVICE, REPOSITORY_PATH, new ReprocessingService.Configuration(TARGET_QUEUE, Optional.empty(), !CONSUME), MAIL_KEY, CLOCK);
JsonSerializationVerifier.dtoModule(ReprocessingOneMailTaskDTO.module(CLOCK, REPROCESSING_SERVICE))
.testCase(taskWithTargetProcessor, SERIALIZED_TASK_1)
.testCase(taskWithoutTargetProcessor,
- "{\"type\":\"reprocessing-one\",\"repositoryPath\":\"a\",\"targetQueue\":\"queue\",\"mailKey\": \"myMail\"}")
+ "{\"type\":\"reprocessing-one\",\"repositoryPath\":\"a\",\"targetQueue\":\"queue\",\"mailKey\": \"myMail\", \"consume\":false}")
.verify();
}
@ParameterizedTest
- @ValueSource(strings = {"{\"type\":\"reprocessing-one\",\"repositoryPath\":\"%\",\"targetQueue\":\"queue\",\"mailKey\": \"myMail\",\"targetProcessor\":\"targetProcessor\"}", "{\"type\":\"reprocessing-one\",\"repositoryPath\":\"%\",\"targetQueue\":\"queue\",\"mailKey\": \"myMail\"}"})
+ @ValueSource(strings = {
+ "{\"type\":\"reprocessing-one\",\"repositoryPath\":\"%\",\"targetQueue\":\"queue\",\"mailKey\": \"myMail\",\"targetProcessor\":\"targetProcessor\"}",
+ "{\"type\":\"reprocessing-one\",\"repositoryPath\":\"%\",\"targetQueue\":\"queue\",\"mailKey\": \"myMail\"}"})
void taskShouldThrowOnDeserializationUrlDecodingError(String serialized) {
JsonTaskSerializer testee = JsonTaskSerializer.of(ReprocessingOneMailTaskDTO.module(CLOCK, REPROCESSING_SERVICE));
@@ -71,10 +78,31 @@ class ReprocessingOneMailTaskTest {
@Test
void additionalInformationShouldBeSerializable() throws IOException {
- ReprocessingOneMailTask.AdditionalInformation details = new ReprocessingOneMailTask.AdditionalInformation(REPOSITORY_PATH, TARGET_QUEUE, MAIL_KEY, TARGET_PROCESSOR, TIMESTAMP);
+ ReprocessingOneMailTask.AdditionalInformation details = new ReprocessingOneMailTask.AdditionalInformation(REPOSITORY_PATH,
+ new ReprocessingService.Configuration(TARGET_QUEUE, TARGET_PROCESSOR, true), MAIL_KEY, TIMESTAMP);
JsonSerializationVerifier.dtoModule(ReprocessingOneMailTaskAdditionalInformationDTO.module())
.bean(details)
.json(SERIALIZED_TASK_1_ADDITIONAL_INFORMATION)
.verify();
}
+
+
+ @Test
+ void shouldDeserializePreviousTaskFormat() throws Exception {
+ ReprocessingOneMailTask taskWithTargetProcessor = new ReprocessingOneMailTask(REPROCESSING_SERVICE, REPOSITORY_PATH, new ReprocessingService.Configuration(TARGET_QUEUE, TARGET_PROCESSOR, CONSUME), MAIL_KEY, CLOCK);
+ JsonTaskSerializer testee = JsonTaskSerializer.of(ReprocessingOneMailTaskDTO.module(CLOCK, REPROCESSING_SERVICE));
+
+ assertThat(testee.deserialize(SERIALIZED_TASK_OLD))
+ .isEqualToComparingFieldByFieldRecursively(taskWithTargetProcessor);
+ }
+
+ @Test
+ void shouldDeserializePreviousAdditionalInformationFormat() throws Exception {
+ ReprocessingOneMailTask.AdditionalInformation details = new ReprocessingOneMailTask.AdditionalInformation(REPOSITORY_PATH,
+ new ReprocessingService.Configuration(TARGET_QUEUE, TARGET_PROCESSOR, true), MAIL_KEY, TIMESTAMP);
+ JsonTaskAdditionalInformationSerializer testee = JsonTaskAdditionalInformationSerializer.of(ReprocessingOneMailTaskAdditionalInformationDTO.module());
+
+ assertThat(testee.deserialize(SERIALIZED_TASK_OLD_ADDITIONAL_INFORMATION))
+ .isEqualToComparingFieldByFieldRecursively(details);
+ }
}
diff --git a/server/protocols/webadmin/webadmin-mailrepository/src/test/java/org/apache/james/webadmin/service/ReprocessingServiceTest.java b/server/protocols/webadmin/webadmin-mailrepository/src/test/java/org/apache/james/webadmin/service/ReprocessingServiceTest.java
index d2ad1c58d9..1ad1e9fd82 100644
--- a/server/protocols/webadmin/webadmin-mailrepository/src/test/java/org/apache/james/webadmin/service/ReprocessingServiceTest.java
+++ b/server/protocols/webadmin/webadmin-mailrepository/src/test/java/org/apache/james/webadmin/service/ReprocessingServiceTest.java
@@ -43,6 +43,7 @@ import org.apache.james.queue.api.ManageableMailQueue;
import org.apache.james.queue.api.RawMailQueueItemDecoratorFactory;
import org.apache.james.queue.memory.MemoryMailQueueFactory;
import org.apache.james.util.MimeMessageUtil;
+import org.apache.james.webadmin.service.ReprocessingService.Configuration;
import org.apache.mailet.base.test.FakeMail;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -64,6 +65,7 @@ class ReprocessingServiceTest {
private static final Consumer<MailKey> NOOP_CONSUMER = key -> { };
private static final Optional<String> NO_TARGET_PROCESSOR = Optional.empty();
private static final byte[] MESSAGE_BYTES = "header: value \r\n".getBytes(UTF_8);
+ public static final boolean CONSUME = true;
private ReprocessingService reprocessingService;
private MemoryMailRepositoryStore mailRepositoryStore;
@@ -106,7 +108,7 @@ class ReprocessingServiceTest {
repository.store(mail2);
repository.store(mail3);
- reprocessingService.reprocess(PATH, KEY_2, NO_TARGET_PROCESSOR, SPOOL);
+ reprocessingService.reprocess(PATH, KEY_2, new ReprocessingService.Configuration(SPOOL, NO_TARGET_PROCESSOR, CONSUME));
assertThat(queueFactory.getQueue(SPOOL).get().browse())
.toIterable()
@@ -121,7 +123,7 @@ class ReprocessingServiceTest {
repository.store(mail2);
repository.store(mail3);
- reprocessingService.reprocess(PATH, KEY_2, NO_TARGET_PROCESSOR, SPOOL);
+ reprocessingService.reprocess(PATH, KEY_2, new ReprocessingService.Configuration(SPOOL, NO_TARGET_PROCESSOR, CONSUME));
assertThat(repository.list()).toIterable()
.containsOnly(KEY_1, KEY_3);
@@ -134,7 +136,7 @@ class ReprocessingServiceTest {
repository.store(mail2);
repository.store(mail3);
- reprocessingService.reprocessAll(PATH, NO_TARGET_PROCESSOR, SPOOL, NOOP_CONSUMER);
+ reprocessingService.reprocessAll(PATH, new Configuration(SPOOL, NO_TARGET_PROCESSOR, CONSUME), NOOP_CONSUMER);
assertThat(repository.list()).toIterable()
.isEmpty();
@@ -147,7 +149,7 @@ class ReprocessingServiceTest {
repository.store(mail2);
repository.store(mail3);
- reprocessingService.reprocessAll(PATH, NO_TARGET_PROCESSOR, SPOOL, NOOP_CONSUMER);
+ reprocessingService.reprocessAll(PATH, new Configuration(SPOOL, NO_TARGET_PROCESSOR, CONSUME), NOOP_CONSUMER);
assertThat(queueFactory.getQueue(SPOOL).get().browse())
.toIterable()
@@ -178,7 +180,7 @@ class ReprocessingServiceTest {
}
});
- reprocessingService.reprocessAll(PATH, NO_TARGET_PROCESSOR, SPOOL, concurrentRemoveConsumer);
+ reprocessingService.reprocessAll(PATH, new ReprocessingService.Configuration(SPOOL, NO_TARGET_PROCESSOR, CONSUME), concurrentRemoveConsumer);
assertThat(queueFactory.getQueue(SPOOL).get().browse())
.toIterable()
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org