You are viewing a plain text version of this content. The canonical link for it is here.
Posted to server-dev@james.apache.org by ma...@apache.org on 2020/03/17 09:35:33 UTC
[james-project] branch master updated (d3eb619 -> 69de2ec)
This is an automated email from the ASF dual-hosted git repository.
matthieu pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git.
from d3eb619 [Refactoring] replace old-school Guava FP with Java 8 Streams
new 84454de JAMES-3070 Don't use cache for RabbitMQ Mailqueue Consumer
new 30f8ecd JAMES-3070 listCreatedMailQueues should only return queue names
new 850da1e JAMES-3070 strong type MailQueueName
new 69de2ec [Refactoring] rename DeleteMailsFromMailQueueTask fields to make it clear arguments are optionals
The 4 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:
.../mailets/MailReprocessingIntegrationTest.java | 8 +-
.../mailetcontainer/impl/JamesMailSpooler.java | 9 +-
.../mailetcontainer/impl/JamesMailetContext.java | 15 ++-
.../james/transport/mailets/RemoteDelivery.java | 6 +
.../delivery/RemoteDeliveryConfiguration.java | 11 +-
.../delivery/RemoteDeliveryConfigurationTest.java | 2 +-
.../remote/delivery/RemoteDeliveryRunningTest.java | 5 +-
.../remote/delivery/RemoteDeliveryTest.java | 2 +-
.../org/apache/james/fetchmail/FetchScheduler.java | 7 +-
.../SetMessagesOutboxFlagUpdateTest.java | 13 +-
.../apache/james/jmap/draft/send/MailSpool.java | 19 ++-
.../apache/james/smtpserver/SendMailHandler.java | 11 ++
.../apache/james/smtpserver/SMTPServerTest.java | 2 +-
.../apache/james/webadmin/dto/MailQueueDTO.java | 2 +-
.../james/webadmin/routes/MailQueueRoutes.java | 75 +++++++----
.../james/webadmin/service/ClearMailQueueTask.java | 64 ++++++---
...ClearMailQueueTaskAdditionalInformationDTO.java | 3 +-
.../webadmin/service/ClearMailQueueTaskDTO.java | 16 ++-
.../service/DeleteMailsFromMailQueueTask.java | 117 ++++++++++------
...sFromMailQueueTaskAdditionalInformationDTO.java | 3 +-
.../service/DeleteMailsFromMailQueueTaskDTO.java | 13 +-
.../james/webadmin/routes/MailQueueRoutesTest.java | 147 +++++++++++----------
.../webadmin/service/ClearMailQueueTaskTest.java | 19 +--
.../service/DeleteMailsFromMailQueueTaskTest.java | 29 ++--
.../webadmin/routes/MailRepositoriesRoutes.java | 11 +-
.../webadmin/service/ReprocessingAllMailsTask.java | 11 +-
...essingAllMailsTaskAdditionalInformationDTO.java | 3 +-
.../service/ReprocessingAllMailsTaskDTO.java | 5 +-
.../webadmin/service/ReprocessingOneMailTask.java | 11 +-
...cessingOneMailTaskAdditionalInformationDTO.java | 3 +-
.../service/ReprocessingOneMailTaskDTO.java | 5 +-
.../webadmin/service/ReprocessingService.java | 69 ++++++----
.../routes/MailRepositoriesRoutesTest.java | 35 ++---
.../service/ReprocessingAllMailsTaskTest.java | 3 +-
.../service/ReprocessingOneMailTaskTest.java | 3 +-
.../webadmin/service/ReprocessingServiceTest.java | 3 +-
...lQueue.java => ActiveMQCacheableMailQueue.java} | 23 ++--
.../queue/activemq/ActiveMQMailQueueFactory.java | 7 +-
.../queue/activemq/ActiveMQMailQueueBlobTest.java | 7 +-
.../queue/activemq/ActiveMQMailQueueTest.java | 9 +-
.../java/org/apache/james/queue/api/MailQueue.java | 5 +-
.../apache/james/queue/api/MailQueueFactory.java | 8 +-
.../org/apache/james/queue/api/MailQueueName.java} | 27 ++--
.../james/queue/api/MailQueueFactoryContract.java | 9 +-
.../apache/james/queue/api/MailQueueNameTest.java} | 17 ++-
.../api/ManageableMailQueueFactoryContract.java | 2 +-
...eMailQueue.java => FileCacheableMailQueue.java} | 20 ++-
.../james/queue/file/FileMailQueueFactory.java | 37 +++---
...java => FileCacheableMailQueueFactoryTest.java} | 2 +-
...ueTest.java => FileCacheableMailQueueTest.java} | 7 +-
...MSMailQueue.java => JMSCacheableMailQueue.java} | 47 ++++---
.../james/queue/jms/JMSMailQueueFactory.java | 7 +-
.../apache/james/queue/jms/JMSMailQueueItem.java | 6 +-
.../queue/library/AbstractMailQueueFactory.java | 32 ++---
.../apache/james/queue/jms/BrokerExtension.java | 5 +-
....java => JMSCacheableMailQueueFactoryTest.java} | 2 +-
...eueTest.java => JMSCacheableMailQueueTest.java} | 11 +-
.../library/AbstractMailQueueFactoryTest.java | 29 ++--
.../james/queue/memory/MemoryMailQueueFactory.java | 38 +++---
...va => MemoryCacheableMailQueueFactoryTest.java} | 2 +-
...Test.java => MemoryCacheableMailQueueTest.java} | 7 +-
.../org/apache/james/queue/rabbitmq/Dequeuer.java | 17 ++-
.../james/queue/rabbitmq/RabbitMQMailQueue.java | 9 +-
.../queue/rabbitmq/RabbitMQMailQueueFactory.java | 39 ++----
.../RabbitMQMailQueueConfigurationChangeTest.java | 3 +-
.../queue/rabbitmq/RabbitMQMailQueueTest.java | 6 +-
.../rabbitmq/RabbitMqMailQueueFactoryTest.java | 17 ---
67 files changed, 712 insertions(+), 505 deletions(-)
rename server/queue/queue-activemq/src/main/java/org/apache/james/queue/activemq/{ActiveMQMailQueue.java => ActiveMQCacheableMailQueue.java} (89%)
copy server/{mailrepository/mailrepository-api/src/main/java/org/apache/james/mailrepository/api/MailKey.java => queue/queue-api/src/main/java/org/apache/james/queue/api/MailQueueName.java} (78%)
copy server/{task/task-api/src/test/java/org/apache/james/task/TaskTypeTest.java => queue/queue-api/src/test/java/org/apache/james/queue/api/MailQueueNameTest.java} (73%)
rename server/queue/queue-file/src/main/java/org/apache/james/queue/file/{FileMailQueue.java => FileCacheableMailQueue.java} (96%)
rename server/queue/queue-file/src/test/java/org/apache/james/queue/file/{FileMailQueueFactoryTest.java => FileCacheableMailQueueFactoryTest.java} (94%)
rename server/queue/queue-file/src/test/java/org/apache/james/queue/file/{FileMailQueueTest.java => FileCacheableMailQueueTest.java} (88%)
rename server/queue/queue-jms/src/main/java/org/apache/james/queue/jms/{JMSMailQueue.java => JMSCacheableMailQueue.java} (94%)
rename server/queue/queue-jms/src/test/java/org/apache/james/queue/jms/{JMSMailQueueFactoryTest.java => JMSCacheableMailQueueFactoryTest.java} (95%)
rename server/queue/queue-jms/src/test/java/org/apache/james/queue/jms/{JMSMailQueueTest.java => JMSCacheableMailQueueTest.java} (90%)
rename server/queue/queue-memory/src/test/java/org/apache/james/queue/memory/{MemoryMailQueueFactoryTest.java => MemoryCacheableMailQueueFactoryTest.java} (93%)
rename server/queue/queue-memory/src/test/java/org/apache/james/queue/memory/{MemoryMailQueueTest.java => MemoryCacheableMailQueueTest.java} (89%)
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 04/04: [Refactoring] rename
DeleteMailsFromMailQueueTask fields to make it clear arguments are
optionals
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
matthieu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 69de2ec081dec4511614e89a7a0887329be4377f
Author: Matthieu Baechler <ma...@apache.org>
AuthorDate: Mon Feb 24 12:00:37 2020 +0100
[Refactoring] rename DeleteMailsFromMailQueueTask fields to make it clear arguments are optionals
---
.../service/DeleteMailsFromMailQueueTask.java | 45 +++++++++++-----------
1 file changed, 23 insertions(+), 22 deletions(-)
diff --git a/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/DeleteMailsFromMailQueueTask.java b/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/DeleteMailsFromMailQueueTask.java
index a5abf75..36d3e55 100644
--- a/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/DeleteMailsFromMailQueueTask.java
+++ b/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/DeleteMailsFromMailQueueTask.java
@@ -50,15 +50,15 @@ public class DeleteMailsFromMailQueueTask implements Task {
private final Instant timestamp;
public AdditionalInformation(MailQueueName mailQueueName, long initialCount, long remainingCount,
- Optional<MailAddress> maybeSender, Optional<String> maybeName,
- Optional<MailAddress> maybeRecipient, Instant timestamp) {
+ Optional<MailAddress> optionalSender, Optional<String> optionalName,
+ Optional<MailAddress> optionalRecipient, Instant timestamp) {
this.mailQueueName = mailQueueName;
this.initialCount = initialCount;
this.remainingCount = remainingCount;
- sender = maybeSender.map(MailAddress::asString);
- name = maybeName;
- recipient = maybeRecipient.map(MailAddress::asString);
+ sender = optionalSender.map(MailAddress::asString);
+ name = optionalName;
+ recipient = optionalRecipient.map(MailAddress::asString);
this.timestamp = timestamp;
}
@@ -105,9 +105,9 @@ public class DeleteMailsFromMailQueueTask implements Task {
public static final TaskType TYPE = TaskType.of("delete-mails-from-mail-queue");
- private final Optional<MailAddress> maybeSender;
- private final Optional<String> maybeName;
- private final Optional<MailAddress> maybeRecipient;
+ private final Optional<MailAddress> optionalSender;
+ private final Optional<String> optionalName;
+ private final Optional<MailAddress> optionalRecipient;
private final MailQueueFactory factory;
private final MailQueueName queueName;
private Optional<Long> initialCount;
@@ -115,17 +115,18 @@ public class DeleteMailsFromMailQueueTask implements Task {
private Optional<TaskExecutionDetails.AdditionalInformation> lastAdditionalInformation;
public DeleteMailsFromMailQueueTask(MailQueueName queueName, MailQueueFactory factory,
- Optional<MailAddress> maybeSender,
- Optional<String> maybeName, Optional<MailAddress> maybeRecipient) {
+ Optional<MailAddress> optionalSender,
+ Optional<String> optionalName,
+ Optional<MailAddress> optionalRecipient) {
this.factory = factory;
this.queueName = queueName;
Preconditions.checkArgument(
- Booleans.countTrue(maybeSender.isPresent(), maybeName.isPresent(), maybeRecipient.isPresent()) == 1,
+ Booleans.countTrue(optionalSender.isPresent(), optionalName.isPresent(), optionalRecipient.isPresent()) == 1,
"You should provide one and only one of the query parameters 'sender', 'name' or 'recipient'.");
- this.maybeSender = maybeSender;
- this.maybeName = maybeName;
- this.maybeRecipient = maybeRecipient;
+ this.optionalSender = optionalSender;
+ this.optionalName = optionalName;
+ this.optionalRecipient = optionalRecipient;
this.initialCount = Optional.empty();
this.queue = Optional.empty();
this.lastAdditionalInformation = Optional.empty();
@@ -137,11 +138,11 @@ public class DeleteMailsFromMailQueueTask implements Task {
try (ManageableMailQueue queue = factory.create(queueName)) {
this.initialCount = Optional.of(getRemainingSize(queue));
this.queue = Optional.of(queue);
- maybeSender.ifPresent(Throwing.consumer(
+ optionalSender.ifPresent(Throwing.consumer(
(MailAddress sender) -> queue.remove(ManageableMailQueue.Type.Sender, sender.asString())));
- maybeName.ifPresent(Throwing.consumer(
+ optionalName.ifPresent(Throwing.consumer(
(String name) -> queue.remove(ManageableMailQueue.Type.Name, name)));
- maybeRecipient.ifPresent(Throwing.consumer(
+ optionalRecipient.ifPresent(Throwing.consumer(
(MailAddress recipient) -> queue.remove(ManageableMailQueue.Type.Recipient, recipient.asString())));
this.lastAdditionalInformation = details();
@@ -165,15 +166,15 @@ public class DeleteMailsFromMailQueueTask implements Task {
}
Optional<String> getMaybeName() {
- return maybeName;
+ return optionalName;
}
Optional<MailAddress> getMaybeRecipient() {
- return maybeRecipient;
+ return optionalRecipient;
}
Optional<MailAddress> getMaybeSender() {
- return maybeSender;
+ return optionalSender;
}
@Override
@@ -184,8 +185,8 @@ public class DeleteMailsFromMailQueueTask implements Task {
new AdditionalInformation(
queueName,
initialCount.get(),
- getRemainingSize(queue), maybeSender,
- maybeName, maybeRecipient, Clock.systemUTC().instant())));
+ getRemainingSize(queue), optionalSender,
+ optionalName, optionalRecipient, Clock.systemUTC().instant())));
}
public long getRemainingSize(ManageableMailQueue queue) {
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 02/04: JAMES-3070 listCreatedMailQueues should only
return queue names
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
matthieu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 30f8ecd4c257600a1bc4b41948e7a65d31d0cfa4
Author: Matthieu Baechler <ma...@apache.org>
AuthorDate: Wed Feb 19 16:45:22 2020 +0100
JAMES-3070 listCreatedMailQueues should only return queue names
---
.../methods/integration/SetMessagesOutboxFlagUpdateTest.java | 2 +-
.../org/apache/james/webadmin/routes/MailQueueRoutes.java | 7 +------
.../java/org/apache/james/queue/api/MailQueueFactory.java | 2 +-
.../org/apache/james/queue/api/MailQueueFactoryContract.java | 5 +----
.../org/apache/james/queue/file/FileMailQueueFactory.java | 11 +++++++----
.../apache/james/queue/library/AbstractMailQueueFactory.java | 9 ++++++---
.../org/apache/james/queue/memory/MemoryMailQueueFactory.java | 9 +++++----
.../apache/james/queue/rabbitmq/RabbitMQMailQueueFactory.java | 6 ++----
8 files changed, 24 insertions(+), 27 deletions(-)
diff --git a/server/protocols/jmap-draft-integration-testing/jmap-draft-integration-testing-common/src/test/java/org/apache/james/jmap/draft/methods/integration/SetMessagesOutboxFlagUpdateTest.java b/server/protocols/jmap-draft-integration-testing/jmap-draft-integration-testing-common/src/test/java/org/apache/james/jmap/draft/methods/integration/SetMessagesOutboxFlagUpdateTest.java
index db8ea1e..00dcf07 100644
--- a/server/protocols/jmap-draft-integration-testing/jmap-draft-integration-testing-common/src/test/java/org/apache/james/jmap/draft/methods/integration/SetMessagesOutboxFlagUpdateTest.java
+++ b/server/protocols/jmap-draft-integration-testing/jmap-draft-integration-testing-common/src/test/java/org/apache/james/jmap/draft/methods/integration/SetMessagesOutboxFlagUpdateTest.java
@@ -105,7 +105,7 @@ public abstract class SetMessagesOutboxFlagUpdateTest {
}
@Override
- public Set<MailQueue> listCreatedMailQueues() {
+ public Set<String> listCreatedMailQueues() {
throw new NotImplementedException("Minimalistic implementation. Please do not list queues");
}
};
diff --git a/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/routes/MailQueueRoutes.java b/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/routes/MailQueueRoutes.java
index 0338071..1764b6a 100644
--- a/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/routes/MailQueueRoutes.java
+++ b/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/routes/MailQueueRoutes.java
@@ -135,12 +135,7 @@ public class MailQueueRoutes implements Routes {
})
public void defineListQueues(Service service) {
service.get(BASE_URL,
- (request, response) ->
- mailQueueFactory
- .listCreatedMailQueues()
- .stream()
- .map(ManageableMailQueue::getName)
- .collect(Guavate.toImmutableList()),
+ (request, response) -> mailQueueFactory.listCreatedMailQueues(),
jsonTransformer);
}
diff --git a/server/queue/queue-api/src/main/java/org/apache/james/queue/api/MailQueueFactory.java b/server/queue/queue-api/src/main/java/org/apache/james/queue/api/MailQueueFactory.java
index 125dd5b..09850b3 100644
--- a/server/queue/queue-api/src/main/java/org/apache/james/queue/api/MailQueueFactory.java
+++ b/server/queue/queue-api/src/main/java/org/apache/james/queue/api/MailQueueFactory.java
@@ -42,5 +42,5 @@ public interface MailQueueFactory<T extends MailQueue> {
T createQueue(String name);
- Set<T> listCreatedMailQueues();
+ Set<String> listCreatedMailQueues();
}
diff --git a/server/queue/queue-api/src/test/java/org/apache/james/queue/api/MailQueueFactoryContract.java b/server/queue/queue-api/src/test/java/org/apache/james/queue/api/MailQueueFactoryContract.java
index dd3bd0d..20d99a6 100644
--- a/server/queue/queue-api/src/test/java/org/apache/james/queue/api/MailQueueFactoryContract.java
+++ b/server/queue/queue-api/src/test/java/org/apache/james/queue/api/MailQueueFactoryContract.java
@@ -43,9 +43,7 @@ public interface MailQueueFactoryContract<T extends MailQueue> {
mailQueueFactory.createQueue(NAME_1);
mailQueueFactory.createQueue(NAME_2);
- assertThat(mailQueueFactory.listCreatedMailQueues())
- .extracting(MailQueue::getName)
- .containsOnly(NAME_1, NAME_2);
+ assertThat(mailQueueFactory.listCreatedMailQueues()).containsOnly(NAME_1, NAME_2);
}
@Test
@@ -56,7 +54,6 @@ public interface MailQueueFactoryContract<T extends MailQueue> {
mailQueueFactory.createQueue(NAME_1);
assertThat(mailQueueFactory.listCreatedMailQueues())
- .extracting(MailQueue::getName)
.containsOnly(NAME_1);
}
diff --git a/server/queue/queue-file/src/main/java/org/apache/james/queue/file/FileMailQueueFactory.java b/server/queue/queue-file/src/main/java/org/apache/james/queue/file/FileMailQueueFactory.java
index 623e1e0..663272a 100644
--- a/server/queue/queue-file/src/main/java/org/apache/james/queue/file/FileMailQueueFactory.java
+++ b/server/queue/queue-file/src/main/java/org/apache/james/queue/file/FileMailQueueFactory.java
@@ -19,7 +19,6 @@
package org.apache.james.queue.file;
import java.io.IOException;
-import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
@@ -28,11 +27,12 @@ import java.util.concurrent.ConcurrentHashMap;
import javax.inject.Inject;
import org.apache.james.filesystem.api.FileSystem;
+import org.apache.james.queue.api.MailQueue;
import org.apache.james.queue.api.MailQueueFactory;
import org.apache.james.queue.api.MailQueueItemDecoratorFactory;
import org.apache.james.queue.api.ManageableMailQueue;
-import com.google.common.collect.ImmutableSet;
+import com.github.steveash.guavate.Guavate;
/**
* {@link MailQueueFactory} implementation which returns {@link FileCacheableMailQueue} instances
@@ -55,8 +55,11 @@ public class FileMailQueueFactory implements MailQueueFactory<ManageableMailQueu
}
@Override
- public Set<ManageableMailQueue> listCreatedMailQueues() {
- return ImmutableSet.copyOf(queues.values());
+ public Set<String> listCreatedMailQueues() {
+ return queues.values()
+ .stream()
+ .map(MailQueue::getName)
+ .collect(Guavate.toImmutableSet());
}
/**
diff --git a/server/queue/queue-jms/src/main/java/org/apache/james/queue/library/AbstractMailQueueFactory.java b/server/queue/queue-jms/src/main/java/org/apache/james/queue/library/AbstractMailQueueFactory.java
index fdd4b86..7d41b75 100644
--- a/server/queue/queue-jms/src/main/java/org/apache/james/queue/library/AbstractMailQueueFactory.java
+++ b/server/queue/queue-jms/src/main/java/org/apache/james/queue/library/AbstractMailQueueFactory.java
@@ -39,8 +39,8 @@ import org.apache.james.queue.api.ManageableMailQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.github.steveash.guavate.Guavate;
import com.google.common.annotations.VisibleForTesting;
-import com.google.common.collect.ImmutableSet;
/**
* {@link MailQueueFactory} abstract base class which take care of register the
@@ -71,8 +71,11 @@ public abstract class AbstractMailQueueFactory<T extends MailQueue> implements M
}
@Override
- public Set<T> listCreatedMailQueues() {
- return ImmutableSet.copyOf(queues.values());
+ public Set<String> listCreatedMailQueues() {
+ return queues.values()
+ .stream()
+ .map(MailQueue::getName)
+ .collect(Guavate.toImmutableSet());
}
@PreDestroy
diff --git a/server/queue/queue-memory/src/main/java/org/apache/james/queue/memory/MemoryMailQueueFactory.java b/server/queue/queue-memory/src/main/java/org/apache/james/queue/memory/MemoryMailQueueFactory.java
index 4aa75ec..6c3d8f3 100644
--- a/server/queue/queue-memory/src/main/java/org/apache/james/queue/memory/MemoryMailQueueFactory.java
+++ b/server/queue/queue-memory/src/main/java/org/apache/james/queue/memory/MemoryMailQueueFactory.java
@@ -19,7 +19,6 @@
package org.apache.james.queue.memory;
-import java.io.IOException;
import java.time.DateTimeException;
import java.time.Duration;
import java.time.Instant;
@@ -52,7 +51,6 @@ import com.github.fge.lambdas.Throwing;
import com.github.steveash.guavate.Guavate;
import com.google.common.base.Objects;
import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
import reactor.core.publisher.Flux;
@@ -71,8 +69,11 @@ public class MemoryMailQueueFactory implements MailQueueFactory<ManageableMailQu
}
@Override
- public Set<ManageableMailQueue> listCreatedMailQueues() {
- return ImmutableSet.copyOf(mailQueues.values());
+ public Set<String> listCreatedMailQueues() {
+ return mailQueues.values()
+ .stream()
+ .map(MemoryCacheableMailQueue::getName)
+ .collect(Guavate.toImmutableSet());
}
@Override
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueFactory.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueFactory.java
index 57bde2f..cc328c0 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueFactory.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueFactory.java
@@ -45,7 +45,6 @@ import org.apache.james.queue.api.MailQueueFactory;
import org.apache.james.queue.api.MailQueueItemDecoratorFactory;
import org.apache.james.queue.rabbitmq.view.RabbitMQMailQueueConfiguration;
import org.apache.james.queue.rabbitmq.view.api.MailQueueView;
-import org.apache.james.util.OptionalUtils;
import com.github.fge.lambdas.Throwing;
import com.google.common.annotations.VisibleForTesting;
@@ -146,10 +145,9 @@ public class RabbitMQMailQueueFactory implements MailQueueFactory<RabbitMQMailQu
}
@Override
- public Set<RabbitMQMailQueue> listCreatedMailQueues() {
- //TODO: it creates connections and leak them
+ public Set<String> listCreatedMailQueues() {
return mqManagementApi.listCreatedMailQueueNames()
- .flatMap(name -> OptionalUtils.toStream(getQueue(name.asString())))
+ .map(MailQueueName::asString)
.collect(ImmutableSet.toImmutableSet());
}
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 01/04: JAMES-3070 Don't use cache for RabbitMQ
Mailqueue Consumer
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
matthieu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 84454ded73591f899b7166cd036da6e306f260ef
Author: Matthieu Baechler <ma...@apache.org>
AuthorDate: Wed Feb 19 16:10:23 2020 +0100
JAMES-3070 Don't use cache for RabbitMQ Mailqueue Consumer
Make MailQueue Closeable to free resources when using non-cached
implementations.
For that we needed to take care about MailQueue lifecycle at every
usage place.
---
.../mailetcontainer/impl/JamesMailSpooler.java | 9 ++-
.../mailetcontainer/impl/JamesMailetContext.java | 15 ++++-
.../james/transport/mailets/RemoteDelivery.java | 6 ++
.../org/apache/james/fetchmail/FetchScheduler.java | 7 ++-
.../SetMessagesOutboxFlagUpdateTest.java | 4 ++
.../apache/james/jmap/draft/send/MailSpool.java | 19 ++++++-
.../apache/james/smtpserver/SendMailHandler.java | 11 ++++
.../apache/james/smtpserver/SMTPServerTest.java | 2 +-
.../james/webadmin/routes/MailQueueRoutes.java | 45 +++++++++------
.../james/webadmin/service/ClearMailQueueTask.java | 47 ++++++++++++----
.../webadmin/service/ClearMailQueueTaskDTO.java | 15 +++--
.../service/DeleteMailsFromMailQueueTask.java | 64 ++++++++++++++--------
.../service/DeleteMailsFromMailQueueTaskDTO.java | 12 ++--
.../james/webadmin/routes/MailQueueRoutesTest.java | 35 ++++++------
.../webadmin/service/ClearMailQueueTaskTest.java | 10 ++--
.../service/DeleteMailsFromMailQueueTaskTest.java | 21 ++++---
.../webadmin/service/ReprocessingService.java | 60 ++++++++++++--------
...lQueue.java => ActiveMQCacheableMailQueue.java} | 18 +++---
.../queue/activemq/ActiveMQMailQueueFactory.java | 6 +-
.../queue/activemq/ActiveMQMailQueueBlobTest.java | 4 +-
.../queue/activemq/ActiveMQMailQueueTest.java | 6 +-
.../java/org/apache/james/queue/api/MailQueue.java | 3 +-
...eMailQueue.java => FileCacheableMailQueue.java} | 13 +++--
.../james/queue/file/FileMailQueueFactory.java | 21 +++----
...java => FileCacheableMailQueueFactoryTest.java} | 2 +-
...ueTest.java => FileCacheableMailQueueTest.java} | 6 +-
...MSMailQueue.java => JMSCacheableMailQueue.java} | 18 ++++--
.../james/queue/jms/JMSMailQueueFactory.java | 6 +-
.../apache/james/queue/jms/JMSMailQueueItem.java | 6 +-
.../queue/library/AbstractMailQueueFactory.java | 10 ++--
....java => JMSCacheableMailQueueFactoryTest.java} | 2 +-
...eueTest.java => JMSCacheableMailQueueTest.java} | 8 +--
.../library/AbstractMailQueueFactoryTest.java | 2 +-
.../james/queue/memory/MemoryMailQueueFactory.java | 24 ++++----
...va => MemoryCacheableMailQueueFactoryTest.java} | 2 +-
...Test.java => MemoryCacheableMailQueueTest.java} | 6 +-
.../org/apache/james/queue/rabbitmq/Dequeuer.java | 17 +++++-
.../james/queue/rabbitmq/RabbitMQMailQueue.java | 5 ++
.../queue/rabbitmq/RabbitMQMailQueueFactory.java | 28 ++--------
.../rabbitmq/RabbitMqMailQueueFactoryTest.java | 17 ------
40 files changed, 370 insertions(+), 242 deletions(-)
diff --git a/server/mailet/mailetcontainer-camel/src/main/java/org/apache/james/mailetcontainer/impl/JamesMailSpooler.java b/server/mailet/mailetcontainer-camel/src/main/java/org/apache/james/mailetcontainer/impl/JamesMailSpooler.java
index 29d5395..28383cd 100644
--- a/server/mailet/mailetcontainer-camel/src/main/java/org/apache/james/mailetcontainer/impl/JamesMailSpooler.java
+++ b/server/mailet/mailetcontainer-camel/src/main/java/org/apache/james/mailetcontainer/impl/JamesMailSpooler.java
@@ -21,6 +21,7 @@ package org.apache.james.mailetcontainer.impl;
import static org.apache.james.metrics.api.TimeMetric.ExecutionResult.DEFAULT_100_MS_THRESHOLD;
+import java.io.IOException;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
@@ -59,7 +60,6 @@ public class JamesMailSpooler implements Disposable, Configurable, MailSpoolerMB
private static final Logger LOGGER = LoggerFactory.getLogger(JamesMailSpooler.class);
public static final String SPOOL_PROCESSING = "spoolProcessing";
- private MailQueue queue;
/**
* The number of threads used to move mail through the spool.
@@ -79,6 +79,8 @@ public class JamesMailSpooler implements Disposable, Configurable, MailSpoolerMB
private reactor.core.Disposable disposable;
private Scheduler spooler;
private int parallelismLevel;
+ private MailQueue queue;
+
@Inject
public JamesMailSpooler(MetricFactory metricFactory, MailProcessor mailProcessor, MailQueueFactory<?> queueFactory) {
@@ -173,6 +175,11 @@ public class JamesMailSpooler implements Disposable, Configurable, MailSpoolerMB
LOGGER.info("start dispose() ...");
disposable.dispose();
spooler.dispose();
+ try {
+ queue.close();
+ } catch (IOException e) {
+ LOGGER.debug("error closing queue", e);
+ }
LOGGER.info("thread shutdown completed.");
}
diff --git a/server/mailet/mailetcontainer-camel/src/main/java/org/apache/james/mailetcontainer/impl/JamesMailetContext.java b/server/mailet/mailetcontainer-camel/src/main/java/org/apache/james/mailetcontainer/impl/JamesMailetContext.java
index bceea55..e1c3483 100644
--- a/server/mailet/mailetcontainer-camel/src/main/java/org/apache/james/mailetcontainer/impl/JamesMailetContext.java
+++ b/server/mailet/mailetcontainer-camel/src/main/java/org/apache/james/mailetcontainer/impl/JamesMailetContext.java
@@ -19,6 +19,7 @@
package org.apache.james.mailetcontainer.impl;
+import java.io.IOException;
import java.util.Collection;
import java.util.Date;
import java.util.HashSet;
@@ -30,6 +31,7 @@ import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
+import javax.annotation.PreDestroy;
import javax.inject.Inject;
import javax.mail.Address;
import javax.mail.Message;
@@ -50,6 +52,7 @@ import org.apache.james.dnsservice.library.MXHostAddressIterator;
import org.apache.james.domainlist.api.DomainList;
import org.apache.james.domainlist.api.DomainListException;
import org.apache.james.lifecycle.api.Configurable;
+import org.apache.james.lifecycle.api.Disposable;
import org.apache.james.lifecycle.api.LifecycleUtil;
import org.apache.james.queue.api.MailQueue;
import org.apache.james.queue.api.MailQueueFactory;
@@ -66,7 +69,7 @@ import org.slf4j.LoggerFactory;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableSet;
-public class JamesMailetContext implements MailetContext, Configurable {
+public class JamesMailetContext implements MailetContext, Configurable, Disposable {
private static final Logger LOGGER = LoggerFactory.getLogger(JamesMailetContext.class);
/**
@@ -89,6 +92,16 @@ public class JamesMailetContext implements MailetContext, Configurable {
this.mailQueueFactory = mailQueueFactory;
}
+ @PreDestroy
+ @Override
+ public void dispose() {
+ try {
+ rootMailQueue.close();
+ } catch (IOException e) {
+ LOGGER.debug("error closing queue", e);
+ }
+ }
+
@Override
public Collection<String> getMailServers(Domain host) {
try {
diff --git a/server/mailet/mailets/src/main/java/org/apache/james/transport/mailets/RemoteDelivery.java b/server/mailet/mailets/src/main/java/org/apache/james/transport/mailets/RemoteDelivery.java
index ab597e9..5cb3889 100644
--- a/server/mailet/mailets/src/main/java/org/apache/james/transport/mailets/RemoteDelivery.java
+++ b/server/mailet/mailets/src/main/java/org/apache/james/transport/mailets/RemoteDelivery.java
@@ -19,6 +19,7 @@
package org.apache.james.transport.mailets;
+import java.io.IOException;
import java.net.UnknownHostException;
import java.util.Collection;
import java.util.Map;
@@ -258,6 +259,11 @@ public class RemoteDelivery extends GenericMailet {
if (startThreads == ThreadState.START_THREADS) {
deliveryRunnable.dispose();
}
+ try {
+ queue.close();
+ } catch (IOException e) {
+ LOGGER.debug("error closing queue", e);
+ }
}
}
diff --git a/server/protocols/fetchmail/src/main/java/org/apache/james/fetchmail/FetchScheduler.java b/server/protocols/fetchmail/src/main/java/org/apache/james/fetchmail/FetchScheduler.java
index a8175b6..869e5c1 100644
--- a/server/protocols/fetchmail/src/main/java/org/apache/james/fetchmail/FetchScheduler.java
+++ b/server/protocols/fetchmail/src/main/java/org/apache/james/fetchmail/FetchScheduler.java
@@ -19,6 +19,7 @@
package org.apache.james.fetchmail;
+import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ScheduledExecutorService;
@@ -67,6 +68,7 @@ public class FetchScheduler implements FetchSchedulerMBean, Configurable {
private MailQueueFactory<?> queueFactory;
private DomainList domainList;
+ private MailQueue queue;
@Inject
public void setMailQueueFactory(MailQueueFactory<?> queueFactory) {
@@ -105,7 +107,7 @@ public class FetchScheduler implements FetchSchedulerMBean, Configurable {
The scheduler service that is used to trigger fetch tasks.
*/
ScheduledExecutorService scheduler = new JMXEnabledScheduledThreadPoolExecutor(numThreads, jmxPath, "scheduler");
- MailQueue queue = queueFactory.createQueue(MailQueueFactory.SPOOL);
+ queue = queueFactory.createQueue(MailQueueFactory.SPOOL);
List<HierarchicalConfiguration<ImmutableNode>> fetchConfs = conf.configurationsAt("fetch");
for (HierarchicalConfiguration<ImmutableNode> fetchConf : fetchConfs) {
@@ -132,9 +134,10 @@ public class FetchScheduler implements FetchSchedulerMBean, Configurable {
}
@PreDestroy
- public void dispose() {
+ public void dispose() throws IOException {
if (enabled) {
LOGGER.info("FetchMail dispose...");
+ queue.close();
for (ScheduledFuture<?> scheduler1 : schedulers) {
scheduler1.cancel(false);
}
diff --git a/server/protocols/jmap-draft-integration-testing/jmap-draft-integration-testing-common/src/test/java/org/apache/james/jmap/draft/methods/integration/SetMessagesOutboxFlagUpdateTest.java b/server/protocols/jmap-draft-integration-testing/jmap-draft-integration-testing-common/src/test/java/org/apache/james/jmap/draft/methods/integration/SetMessagesOutboxFlagUpdateTest.java
index e9141ea..db8ea1e 100644
--- a/server/protocols/jmap-draft-integration-testing/jmap-draft-integration-testing-common/src/test/java/org/apache/james/jmap/draft/methods/integration/SetMessagesOutboxFlagUpdateTest.java
+++ b/server/protocols/jmap-draft-integration-testing/jmap-draft-integration-testing-common/src/test/java/org/apache/james/jmap/draft/methods/integration/SetMessagesOutboxFlagUpdateTest.java
@@ -80,6 +80,10 @@ public abstract class SetMessagesOutboxFlagUpdateTest {
public MailQueue createQueue(String name) {
return new MailQueue() {
@Override
+ public void close() throws IOException {
+ }
+
+ @Override
public String getName() {
return name;
}
diff --git a/server/protocols/jmap-draft/src/main/java/org/apache/james/jmap/draft/send/MailSpool.java b/server/protocols/jmap-draft/src/main/java/org/apache/james/jmap/draft/send/MailSpool.java
index c652bc9..afda197 100644
--- a/server/protocols/jmap-draft/src/main/java/org/apache/james/jmap/draft/send/MailSpool.java
+++ b/server/protocols/jmap-draft/src/main/java/org/apache/james/jmap/draft/send/MailSpool.java
@@ -19,8 +19,12 @@
package org.apache.james.jmap.draft.send;
+import java.io.IOException;
+
+import javax.annotation.PreDestroy;
import javax.inject.Inject;
+import org.apache.james.lifecycle.api.Disposable;
import org.apache.james.lifecycle.api.Startable;
import org.apache.james.queue.api.MailQueue;
import org.apache.james.queue.api.MailQueue.MailQueueException;
@@ -28,10 +32,14 @@ import org.apache.james.queue.api.MailQueueFactory;
import org.apache.mailet.Attribute;
import org.apache.mailet.AttributeValue;
import org.apache.mailet.Mail;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import com.google.common.annotations.VisibleForTesting;
-public class MailSpool implements Startable {
+public class MailSpool implements Startable, Disposable {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(MailSpool.class);
private final MailQueueFactory<?> queueFactory;
private MailQueue queue;
@@ -45,6 +53,15 @@ public class MailSpool implements Startable {
queue = queueFactory.createQueue(MailQueueFactory.SPOOL);
}
+ @PreDestroy
+ public void dispose() {
+ try {
+ queue.close();
+ } catch (IOException e) {
+ LOGGER.debug("error closing queue", e);
+ }
+ }
+
public void send(Mail mail, MailMetadata metadata) throws MailQueueException {
mail.setAttribute(new Attribute(MailMetadata.MAIL_METADATA_MESSAGE_ID_ATTRIBUTE, AttributeValue.of(metadata.getMessageId().serialize())));
mail.setAttribute(new Attribute(MailMetadata.MAIL_METADATA_USERNAME_ATTRIBUTE, AttributeValue.of(metadata.getUsername())));
diff --git a/server/protocols/protocols-smtp/src/main/java/org/apache/james/smtpserver/SendMailHandler.java b/server/protocols/protocols-smtp/src/main/java/org/apache/james/smtpserver/SendMailHandler.java
index 4ef8201..ea82d37 100644
--- a/server/protocols/protocols-smtp/src/main/java/org/apache/james/smtpserver/SendMailHandler.java
+++ b/server/protocols/protocols-smtp/src/main/java/org/apache/james/smtpserver/SendMailHandler.java
@@ -19,6 +19,8 @@
package org.apache.james.smtpserver;
+import java.io.IOException;
+
import javax.inject.Inject;
import javax.mail.MessagingException;
@@ -52,6 +54,15 @@ public class SendMailHandler implements JamesMessageHook {
queue = queueFactory.createQueue(MailQueueFactory.SPOOL);
}
+ @Override
+ public void destroy() {
+ try {
+ queue.close();
+ } catch (IOException e) {
+ LOGGER.debug("error close queue", e);
+ }
+ }
+
/**
* Adds header to the message
*/
diff --git a/server/protocols/protocols-smtp/src/test/java/org/apache/james/smtpserver/SMTPServerTest.java b/server/protocols/protocols-smtp/src/test/java/org/apache/james/smtpserver/SMTPServerTest.java
index 8b4b19d..e3ae55e 100644
--- a/server/protocols/protocols-smtp/src/test/java/org/apache/james/smtpserver/SMTPServerTest.java
+++ b/server/protocols/protocols-smtp/src/test/java/org/apache/james/smtpserver/SMTPServerTest.java
@@ -199,7 +199,7 @@ public class SMTPServerTest {
protected Configuration configuration;
protected MockProtocolHandlerLoader chain;
protected MemoryMailQueueFactory queueFactory;
- protected MemoryMailQueueFactory.MemoryMailQueue queue;
+ protected MemoryMailQueueFactory.MemoryCacheableMailQueue queue;
protected MemoryRecipientRewriteTable rewriteTable;
private AliasReverseResolver aliasReverseResolver;
protected CanSendFrom canSendFrom;
diff --git a/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/routes/MailQueueRoutes.java b/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/routes/MailQueueRoutes.java
index c2f6064..0338071 100644
--- a/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/routes/MailQueueRoutes.java
+++ b/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/routes/MailQueueRoutes.java
@@ -21,6 +21,8 @@ package org.apache.james.webadmin.routes;
import static org.apache.james.webadmin.Constants.SEPARATOR;
+import java.io.IOException;
+import java.io.UncheckedIOException;
import java.time.ZonedDateTime;
import java.util.List;
import java.util.Optional;
@@ -32,6 +34,7 @@ import javax.ws.rs.Path;
import javax.ws.rs.Produces;
import org.apache.james.core.MailAddress;
+import org.apache.james.queue.api.MailQueue;
import org.apache.james.queue.api.MailQueue.MailQueueException;
import org.apache.james.queue.api.MailQueueFactory;
import org.apache.james.queue.api.ManageableMailQueue;
@@ -223,7 +226,7 @@ public class MailQueueRoutes implements Routes {
private List<MailQueueItemDTO> listMails(Request request) {
String mailQueueName = request.params(MAIL_QUEUE_NAME);
return mailQueueFactory.getQueue(mailQueueName)
- .map(name -> listMails(name, isDelayed(request.queryParams(DELAYED_QUERY_PARAM)), ParametersExtractor.extractLimit(request)))
+ .map(queue -> listMails(queue, isDelayed(request.queryParams(DELAYED_QUERY_PARAM)), ParametersExtractor.extractLimit(request)))
.orElseThrow(
() -> ErrorResponder.builder()
.message("%s can not be found", mailQueueName)
@@ -238,7 +241,7 @@ public class MailQueueRoutes implements Routes {
}
private List<MailQueueItemDTO> listMails(ManageableMailQueue queue, Optional<Boolean> isDelayed, Limit limit) {
- try {
+ try (MailQueue closeable = queue) {
return limit.applyOnStream(Iterators.toStream(queue.browse()))
.map(Throwing.function(MailQueueItemDTO::from).sneakyThrow())
.filter(item -> filter(item, isDelayed))
@@ -250,6 +253,8 @@ public class MailQueueRoutes implements Routes {
.message("Invalid request for listing the mails from the mail queue %s", queue)
.cause(e)
.haltError();
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
}
}
@@ -304,17 +309,10 @@ public class MailQueueRoutes implements Routes {
private Task deleteMails(Request request) {
String mailQueueName = request.params(MAIL_QUEUE_NAME);
- return mailQueueFactory.getQueue(mailQueueName)
- .map(name -> deleteMailsTask(name,
+ return deleteMailsTask(mailQueueName,
sender(request.queryParams(SENDER_QUERY_PARAM)),
name(request.queryParams(NAME_QUERY_PARAM)),
- recipient(request.queryParams(RECIPIENT_QUERY_PARAM))))
- .orElseThrow(
- () -> ErrorResponder.builder()
- .message("%s can not be found", mailQueueName)
- .statusCode(HttpStatus.NOT_FOUND_404)
- .type(ErrorResponder.ErrorType.NOT_FOUND)
- .haltError());
+ recipient(request.queryParams(RECIPIENT_QUERY_PARAM)));
}
private Optional<MailAddress> sender(String senderAsString) throws HaltException {
@@ -383,10 +381,11 @@ public class MailQueueRoutes implements Routes {
private String forceDelayedMailsDelivery(Request request, Response response) throws JsonExtractException, MailQueueException {
assertDelayedParamIsTrue(request);
assertPayloadContainsDelayedEntry(request);
- ManageableMailQueue mailQueue = assertMailQueueExists(request);
-
- mailQueue.flush();
-
+ try (ManageableMailQueue mailQueue = assertMailQueueExists(request)) {
+ mailQueue.flush();
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
return Responses.returnNoContent(response);
}
@@ -421,13 +420,13 @@ public class MailQueueRoutes implements Routes {
}
}
- private Task deleteMailsTask(ManageableMailQueue queue, Optional<MailAddress> maybeSender, Optional<String> maybeName, Optional<MailAddress> maybeRecipient) {
+ private Task deleteMailsTask(String queueName, Optional<MailAddress> maybeSender, Optional<String> maybeName, Optional<MailAddress> maybeRecipient) {
int paramCount = Booleans.countTrue(maybeSender.isPresent(), maybeName.isPresent(), maybeRecipient.isPresent());
switch (paramCount) {
case 0:
- return new ClearMailQueueTask(queue);
+ return new ClearMailQueueTask(queueName, this::getQueueOrThrow);
case 1:
- return new DeleteMailsFromMailQueueTask(queue, maybeSender, maybeName, maybeRecipient);
+ return new DeleteMailsFromMailQueueTask(queueName, this::getQueueOrThrow, maybeSender, maybeName, maybeRecipient);
default:
throw ErrorResponder.builder()
.statusCode(HttpStatus.BAD_REQUEST_400)
@@ -438,6 +437,16 @@ public class MailQueueRoutes implements Routes {
}
}
+ private ManageableMailQueue getQueueOrThrow(String queueName) {
+ return mailQueueFactory.getQueue(queueName)
+ .orElseThrow(
+ () -> ErrorResponder.builder()
+ .message("%s can not be found", queueName)
+ .statusCode(HttpStatus.NOT_FOUND_404)
+ .type(ErrorResponder.ErrorType.NOT_FOUND)
+ .haltError());
+ }
+
private void assertDelayedParamIsTrue(Request request) {
if (!isDelayed(request.queryParams(DELAYED_QUERY_PARAM)).orElse(false)) {
throw ErrorResponder.builder()
diff --git a/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/ClearMailQueueTask.java b/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/ClearMailQueueTask.java
index ff4d777..ab9e25b 100644
--- a/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/ClearMailQueueTask.java
+++ b/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/ClearMailQueueTask.java
@@ -19,6 +19,7 @@
package org.apache.james.webadmin.service;
+import java.io.IOException;
import java.time.Clock;
import java.time.Instant;
import java.util.Optional;
@@ -31,9 +32,12 @@ import org.apache.james.task.TaskType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.google.common.annotations.VisibleForTesting;
+
public class ClearMailQueueTask implements Task {
public static class AdditionalInformation implements TaskExecutionDetails.AdditionalInformation {
+
private final String mailQueueName;
private final long initialCount;
private final long remainingCount;
@@ -62,30 +66,44 @@ public class ClearMailQueueTask implements Task {
public Instant timestamp() {
return timestamp;
}
+
}
public static class UnknownSerializedQueue extends RuntimeException {
+
public UnknownSerializedQueue(String queueName) {
super("Unable to retrieve '" + queueName + "' queue");
}
}
+ @FunctionalInterface
+ public interface MailQueueFactory {
+ ManageableMailQueue create(String mailQueueName);
+ }
+
private static final Logger LOGGER = LoggerFactory.getLogger(ClearMailQueueTask.class);
public static final TaskType TYPE = TaskType.of("clear-mail-queue");
- private final ManageableMailQueue queue;
- private final long initialCount;
+ private final String queueName;
+ private final ClearMailQueueTask.MailQueueFactory factory;
+ private Optional<Long> initialCount;
+ private Optional<ManageableMailQueue> queue;
- public ClearMailQueueTask(ManageableMailQueue queue) {
- this.queue = queue;
- initialCount = getRemainingSize();
+ public ClearMailQueueTask(String queueName, ClearMailQueueTask.MailQueueFactory factory) {
+ this.queueName = queueName;
+ this.factory = factory;
+ this.initialCount = Optional.empty();
+ this.queue = Optional.empty();
}
@Override
public Result run() {
- try {
+ try (ManageableMailQueue queue = factory.create(queueName)) {
+ this.initialCount = Optional.of(getRemainingSize(queue));
+ this.queue = Optional.of(queue);
queue.clear();
- } catch (MailQueue.MailQueueException e) {
+ this.queue = Optional.empty();
+ } catch (MailQueue.MailQueueException | IOException e) {
LOGGER.error("Clear MailQueue got an exception", e);
return Result.PARTIAL;
}
@@ -100,16 +118,21 @@ public class ClearMailQueueTask implements Task {
@Override
public Optional<TaskExecutionDetails.AdditionalInformation> details() {
- return Optional.of(new AdditionalInformation(queue.getName(), initialCount, getRemainingSize(), Clock.systemUTC().instant()));
+ return queue.map(q -> new AdditionalInformation(queueName, initialCount.get(), getRemainingSize(q), Clock.systemUTC().instant()));
+ }
+
+ String getQueueName() {
+ return queueName;
}
- ManageableMailQueue getQueue() {
- return queue;
+ @VisibleForTesting
+ Optional<Long> initialCount() {
+ return initialCount;
}
- private long getRemainingSize() {
+ private long getRemainingSize(ManageableMailQueue mailQueue) {
try {
- return queue.getSize();
+ return mailQueue.getSize();
} catch (MailQueue.MailQueueException e) {
throw new RuntimeException(e);
}
diff --git a/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/ClearMailQueueTaskDTO.java b/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/ClearMailQueueTaskDTO.java
index 8ba59f4..f4af298 100644
--- a/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/ClearMailQueueTaskDTO.java
+++ b/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/ClearMailQueueTaskDTO.java
@@ -39,19 +39,22 @@ public class ClearMailQueueTaskDTO implements TaskDTO {
}
public static ClearMailQueueTaskDTO toDTO(ClearMailQueueTask domainObject, String typeName) {
- return new ClearMailQueueTaskDTO(typeName, domainObject.getQueue().getName());
+ return new ClearMailQueueTaskDTO(typeName, domainObject.getQueueName());
}
private final String type;
- private final String queue;
+ private final String queueName;
- public ClearMailQueueTaskDTO(@JsonProperty("type") String type, @JsonProperty("queue") String queue) {
+ public ClearMailQueueTaskDTO(@JsonProperty("type") String type, @JsonProperty("queue") String queueName) {
this.type = type;
- this.queue = queue;
+ this.queueName = queueName;
}
public ClearMailQueueTask fromDTO(MailQueueFactory<? extends ManageableMailQueue> mailQueueFactory) {
- return new ClearMailQueueTask(mailQueueFactory.getQueue(queue).orElseThrow(() -> new ClearMailQueueTask.UnknownSerializedQueue(queue)));
+ return new ClearMailQueueTask(queueName,
+ name -> mailQueueFactory
+ .getQueue(name)
+ .orElseThrow(() -> new ClearMailQueueTask.UnknownSerializedQueue(queueName)));
}
@Override
@@ -60,6 +63,6 @@ public class ClearMailQueueTaskDTO implements TaskDTO {
}
public String getQueue() {
- return queue;
+ return queueName;
}
}
diff --git a/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/DeleteMailsFromMailQueueTask.java b/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/DeleteMailsFromMailQueueTask.java
index e55b7d6..0a1a184 100644
--- a/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/DeleteMailsFromMailQueueTask.java
+++ b/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/DeleteMailsFromMailQueueTask.java
@@ -19,6 +19,8 @@
package org.apache.james.webadmin.service;
+import java.io.IOException;
+import java.io.UncheckedIOException;
import java.time.Clock;
import java.time.Instant;
import java.util.Optional;
@@ -95,38 +97,60 @@ public class DeleteMailsFromMailQueueTask implements Task {
}
}
+ @FunctionalInterface
+ public interface MailQueueFactory {
+ ManageableMailQueue create(String mailQueueName);
+ }
+
public static final TaskType TYPE = TaskType.of("delete-mails-from-mail-queue");
- private final ManageableMailQueue queue;
private final Optional<MailAddress> maybeSender;
private final Optional<String> maybeName;
private final Optional<MailAddress> maybeRecipient;
+ private final MailQueueFactory factory;
+ private final String queueName;
+ private Optional<Long> initialCount;
+ private Optional<ManageableMailQueue> queue;
- private final long initialCount;
-
- public DeleteMailsFromMailQueueTask(ManageableMailQueue queue, Optional<MailAddress> maybeSender,
+ public DeleteMailsFromMailQueueTask(String queueName, MailQueueFactory factory,
+ Optional<MailAddress> maybeSender,
Optional<String> maybeName, Optional<MailAddress> maybeRecipient) {
+ this.factory = factory;
+ this.queueName = queueName;
Preconditions.checkArgument(
Booleans.countTrue(maybeSender.isPresent(), maybeName.isPresent(), maybeRecipient.isPresent()) == 1,
"You should provide one and only one of the query parameters 'sender', 'name' or 'recipient'.");
- this.queue = queue;
this.maybeSender = maybeSender;
this.maybeName = maybeName;
this.maybeRecipient = maybeRecipient;
- this.initialCount = getRemainingSize();
+ this.initialCount = Optional.empty();
+ this.queue = Optional.empty();
+
}
@Override
public Result run() {
- maybeSender.ifPresent(Throwing.consumer(
- (MailAddress sender) -> queue.remove(ManageableMailQueue.Type.Sender, sender.asString())));
- maybeName.ifPresent(Throwing.consumer(
- (String name) -> queue.remove(ManageableMailQueue.Type.Name, name)));
- maybeRecipient.ifPresent(Throwing.consumer(
- (MailAddress recipient) -> queue.remove(ManageableMailQueue.Type.Recipient, recipient.asString())));
-
- return Result.COMPLETED;
+ try (ManageableMailQueue queue = factory.create(queueName)) {
+ this.initialCount = Optional.of(getRemainingSize(queue));
+ this.queue = Optional.of(queue);
+ maybeSender.ifPresent(Throwing.consumer(
+ (MailAddress sender) -> queue.remove(ManageableMailQueue.Type.Sender, sender.asString())));
+ maybeName.ifPresent(Throwing.consumer(
+ (String name) -> queue.remove(ManageableMailQueue.Type.Name, name)));
+ maybeRecipient.ifPresent(Throwing.consumer(
+ (MailAddress recipient) -> queue.remove(ManageableMailQueue.Type.Recipient, recipient.asString())));
+
+ this.queue = Optional.empty();
+
+ return Result.COMPLETED;
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ public String getQueueName() {
+ return queueName;
}
@Override
@@ -134,10 +158,6 @@ public class DeleteMailsFromMailQueueTask implements Task {
return TYPE;
}
- ManageableMailQueue getQueue() {
- return queue;
- }
-
Optional<String> getMaybeName() {
return maybeName;
}
@@ -152,13 +172,13 @@ public class DeleteMailsFromMailQueueTask implements Task {
@Override
public Optional<TaskExecutionDetails.AdditionalInformation> details() {
- return Optional.of(new AdditionalInformation(queue.getName(),
- initialCount,
- getRemainingSize(), maybeSender,
+ return this.queue.map(queue -> new AdditionalInformation(queueName,
+ initialCount.get(),
+ getRemainingSize(queue), maybeSender,
maybeName, maybeRecipient, Clock.systemUTC().instant()));
}
- public long getRemainingSize() {
+ public long getRemainingSize(ManageableMailQueue queue) {
try {
return queue.getSize();
} catch (MailQueue.MailQueueException e) {
diff --git a/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/DeleteMailsFromMailQueueTaskDTO.java b/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/DeleteMailsFromMailQueueTaskDTO.java
index e5dfa9e..12f3485 100644
--- a/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/DeleteMailsFromMailQueueTaskDTO.java
+++ b/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/DeleteMailsFromMailQueueTaskDTO.java
@@ -27,7 +27,7 @@ public class DeleteMailsFromMailQueueTaskDTO implements TaskDTO {
public static DeleteMailsFromMailQueueTaskDTO toDTO(DeleteMailsFromMailQueueTask domainObject, String typeName) {
return new DeleteMailsFromMailQueueTaskDTO(
typeName,
- domainObject.getQueue().getName(),
+ domainObject.getQueueName(),
domainObject.getMaybeSender().map(MailAddress::asString),
domainObject.getMaybeName(),
domainObject.getMaybeRecipient().map(MailAddress::asString)
@@ -35,18 +35,18 @@ public class DeleteMailsFromMailQueueTaskDTO implements TaskDTO {
}
private final String type;
- private final String queue;
+ private final String queueName;
private final Optional<String> sender;
private final Optional<String> name;
private final Optional<String> recipient;
public DeleteMailsFromMailQueueTaskDTO(@JsonProperty("type") String type,
- @JsonProperty("queue") String queue,
+ @JsonProperty("queue") String queueName,
@JsonProperty("sender") Optional<String> sender,
@JsonProperty("name") Optional<String> name,
@JsonProperty("recipient") Optional<String> recipient) {
this.type = type;
- this.queue = queue;
+ this.queueName = queueName;
this.sender = sender;
this.name = name;
this.recipient = recipient;
@@ -54,7 +54,7 @@ public class DeleteMailsFromMailQueueTaskDTO implements TaskDTO {
public DeleteMailsFromMailQueueTask fromDTO(MailQueueFactory<? extends ManageableMailQueue> mailQueueFactory) {
return new DeleteMailsFromMailQueueTask(
- mailQueueFactory.getQueue(queue).orElseThrow(() -> new DeleteMailsFromMailQueueTask.UnknownSerializedQueue(queue)),
+ queueName, name -> mailQueueFactory.getQueue(name).orElseThrow(() -> new DeleteMailsFromMailQueueTask.UnknownSerializedQueue(queueName)),
sender.map(Throwing.<String, MailAddress>function(MailAddress::new).sneakyThrow()),
name,
recipient.map(Throwing.<String, MailAddress>function(MailAddress::new).sneakyThrow())
@@ -67,7 +67,7 @@ public class DeleteMailsFromMailQueueTaskDTO implements TaskDTO {
}
public String getQueue() {
- return queue;
+ return queueName;
}
public Optional<String> getSender() {
diff --git a/server/protocols/webadmin/webadmin-mailqueue/src/test/java/org/apache/james/webadmin/routes/MailQueueRoutesTest.java b/server/protocols/webadmin/webadmin-mailqueue/src/test/java/org/apache/james/webadmin/routes/MailQueueRoutesTest.java
index 2cd4f49..f40f1ac 100644
--- a/server/protocols/webadmin/webadmin-mailqueue/src/test/java/org/apache/james/webadmin/routes/MailQueueRoutesTest.java
+++ b/server/protocols/webadmin/webadmin-mailqueue/src/test/java/org/apache/james/webadmin/routes/MailQueueRoutesTest.java
@@ -42,7 +42,6 @@ import org.apache.james.queue.api.Mails;
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.queue.memory.MemoryMailQueueFactory.MemoryMailQueue;
import org.apache.james.task.Hostname;
import org.apache.james.task.MemoryTaskManager;
import org.apache.james.task.TaskManager;
@@ -229,7 +228,7 @@ class MailQueueRoutesTest {
@Test
public void listMailsShouldReturnMailsWhenSome() throws Exception {
- MemoryMailQueue queue = mailQueueFactory.createQueue(FIRST_QUEUE);
+ MemoryMailQueueFactory.MemoryCacheableMailQueue queue = mailQueueFactory.createQueue(FIRST_QUEUE);
queue.enQueue(Mails.defaultMail().name("name").build());
queue.enQueue(Mails.defaultMail().name("name").build());
@@ -243,7 +242,7 @@ class MailQueueRoutesTest {
@Test
public void listMailsShouldReturnMailDetailsWhenSome() throws Exception {
- MemoryMailQueue queue = mailQueueFactory.createQueue(FIRST_QUEUE);
+ MemoryMailQueueFactory.MemoryCacheableMailQueue queue = mailQueueFactory.createQueue(FIRST_QUEUE);
FakeMail mail = Mails.defaultMail().name("name").build();
queue.enQueue(mail);
@@ -265,7 +264,7 @@ class MailQueueRoutesTest {
@Test
public void listMailsShouldReturnEmptyWhenNoDelayedMailsAndAskFor() throws Exception {
- MemoryMailQueue queue = mailQueueFactory.createQueue(FIRST_QUEUE);
+ MemoryMailQueueFactory.MemoryCacheableMailQueue queue = mailQueueFactory.createQueue(FIRST_QUEUE);
FakeMail mail = Mails.defaultMail().name("name").build();
queue.enQueue(mail);
@@ -281,7 +280,7 @@ class MailQueueRoutesTest {
@Test
public void listMailsShouldReturnCurrentMailsWhenMailsAndAskForNotDelayed() throws Exception {
- MemoryMailQueue queue = mailQueueFactory.createQueue(FIRST_QUEUE);
+ MemoryMailQueueFactory.MemoryCacheableMailQueue queue = mailQueueFactory.createQueue(FIRST_QUEUE);
FakeMail mail = Mails.defaultMail().name("name").build();
queue.enQueue(mail);
@@ -297,7 +296,7 @@ class MailQueueRoutesTest {
@Test
public void listMailsShouldReturnDelayedMailsWhenAskFor() throws Exception {
- MemoryMailQueue queue = mailQueueFactory.createQueue(FIRST_QUEUE);
+ MemoryMailQueueFactory.MemoryCacheableMailQueue queue = mailQueueFactory.createQueue(FIRST_QUEUE);
FakeMail mail = Mails.defaultMail().name("name").build();
queue.enQueue(mail, 10, TimeUnit.MINUTES);
@@ -313,7 +312,7 @@ class MailQueueRoutesTest {
@Test
public void listMailsShouldReturnOneMailWhenMailsAndAskForALimitOfOne() throws Exception {
- MemoryMailQueue queue = mailQueueFactory.createQueue(FIRST_QUEUE);
+ MemoryMailQueueFactory.MemoryCacheableMailQueue queue = mailQueueFactory.createQueue(FIRST_QUEUE);
FakeMail mail = Mails.defaultMail().name("name").build();
queue.enQueue(mail);
queue.enQueue(mail);
@@ -336,7 +335,7 @@ class MailQueueRoutesTest {
@Test
public void getMailQueueShouldReturnTheMailQueueDataWhenMailQueueExists() throws Exception {
- MemoryMailQueue queue = mailQueueFactory.createQueue(FIRST_QUEUE);
+ MemoryMailQueueFactory.MemoryCacheableMailQueue queue = mailQueueFactory.createQueue(FIRST_QUEUE);
queue.enQueue(Mails.defaultMail().name("name").build());
when()
@@ -509,7 +508,7 @@ class MailQueueRoutesTest {
class SideEffects {
@Test
public void forcingDelayedMailsDeliveryShouldActuallyChangePropertyOnMails() throws Exception {
- MemoryMailQueue queue = mailQueueFactory.createQueue(FIRST_QUEUE);
+ MemoryMailQueueFactory.MemoryCacheableMailQueue queue = mailQueueFactory.createQueue(FIRST_QUEUE);
FakeMail mail = Mails.defaultMail().name("name").build();
queue.enQueue(mail, 10L, TimeUnit.MINUTES);
queue.enQueue(mail, 10L, TimeUnit.MINUTES);
@@ -656,7 +655,7 @@ class MailQueueRoutesTest {
@Test
public void deleteMailsTasksShouldHaveDetailsWhenSenderIsGiven() throws Exception {
- MemoryMailQueue queue = mailQueueFactory.createQueue(FIRST_QUEUE);
+ MemoryMailQueueFactory.MemoryCacheableMailQueue queue = mailQueueFactory.createQueue(FIRST_QUEUE);
queue.enQueue(FakeMail.builder()
.name(FAKE_MAIL_NAME_1)
@@ -693,7 +692,7 @@ class MailQueueRoutesTest {
@Test
public void deleteMailsTasksShouldHaveDetailsWhenNameIsGiven() throws Exception {
- MemoryMailQueue queue = mailQueueFactory.createQueue(FIRST_QUEUE);
+ MemoryMailQueueFactory.MemoryCacheableMailQueue queue = mailQueueFactory.createQueue(FIRST_QUEUE);
queue.enQueue(FakeMail.builder()
.name(FAKE_MAIL_NAME_1)
@@ -728,7 +727,7 @@ class MailQueueRoutesTest {
@Test
public void deleteMailsTasksShouldHaveDetailsWhenRecipientIsGiven() throws Exception {
- MemoryMailQueue queue = mailQueueFactory.createQueue(FIRST_QUEUE);
+ MemoryMailQueueFactory.MemoryCacheableMailQueue queue = mailQueueFactory.createQueue(FIRST_QUEUE);
queue.enQueue(FakeMail.builder()
.name(FAKE_MAIL_NAME_1)
@@ -774,7 +773,7 @@ class MailQueueRoutesTest {
@Test
public void deleteMailsShouldDeleteMailsWhenSenderIsGiven() throws Exception {
- MemoryMailQueue queue = mailQueueFactory.createQueue(FIRST_QUEUE);
+ MemoryMailQueueFactory.MemoryCacheableMailQueue queue = mailQueueFactory.createQueue(FIRST_QUEUE);
queue.enQueue(FakeMail.builder()
.name(FAKE_MAIL_NAME_1)
@@ -808,7 +807,7 @@ class MailQueueRoutesTest {
@Test
public void deleteMailsShouldDeleteMailsWhenNameIsGiven() throws Exception {
- MemoryMailQueue queue = mailQueueFactory.createQueue(FIRST_QUEUE);
+ MemoryMailQueueFactory.MemoryCacheableMailQueue queue = mailQueueFactory.createQueue(FIRST_QUEUE);
queue.enQueue(FakeMail.builder()
.name(FAKE_MAIL_NAME_1)
@@ -840,7 +839,7 @@ class MailQueueRoutesTest {
@Test
public void deleteMailsShouldDeleteMailsWhenRecipientIsGiven() throws Exception {
- MemoryMailQueue queue = mailQueueFactory.createQueue(FIRST_QUEUE);
+ MemoryMailQueueFactory.MemoryCacheableMailQueue queue = mailQueueFactory.createQueue(FIRST_QUEUE);
queue.enQueue(FakeMail.builder()
.name(FAKE_MAIL_NAME_1)
@@ -880,7 +879,7 @@ class MailQueueRoutesTest {
@Test
public void deleteMailsShouldDeleteMailsWhenTheyAreMatching() throws Exception {
- MemoryMailQueue queue = mailQueueFactory.createQueue(FIRST_QUEUE);
+ MemoryMailQueueFactory.MemoryCacheableMailQueue queue = mailQueueFactory.createQueue(FIRST_QUEUE);
String recipient = "recipient@james.org";
queue.enQueue(Mails.defaultMail()
.name("name")
@@ -945,7 +944,7 @@ class MailQueueRoutesTest {
@Test
public void clearMailQueueShouldHaveDetailsWhenNoQueryParameters() throws Exception {
- MemoryMailQueue queue = mailQueueFactory.createQueue(FIRST_QUEUE);
+ MemoryMailQueueFactory.MemoryCacheableMailQueue queue = mailQueueFactory.createQueue(FIRST_QUEUE);
queue.enQueue(FakeMail.builder()
.name(FAKE_MAIL_NAME_1)
@@ -982,7 +981,7 @@ class MailQueueRoutesTest {
@Test
void clearMailQueueShouldDeleteAllMailsInQueueWhenNoQueryParameters() throws Exception {
- MemoryMailQueue queue = mailQueueFactory.createQueue(FIRST_QUEUE);
+ MemoryMailQueueFactory.MemoryCacheableMailQueue queue = mailQueueFactory.createQueue(FIRST_QUEUE);
queue.enQueue(FakeMail.builder()
.name(FAKE_MAIL_NAME_1)
diff --git a/server/protocols/webadmin/webadmin-mailqueue/src/test/java/org/apache/james/webadmin/service/ClearMailQueueTaskTest.java b/server/protocols/webadmin/webadmin-mailqueue/src/test/java/org/apache/james/webadmin/service/ClearMailQueueTaskTest.java
index 6d04e01..94b4432 100644
--- a/server/protocols/webadmin/webadmin-mailqueue/src/test/java/org/apache/james/webadmin/service/ClearMailQueueTaskTest.java
+++ b/server/protocols/webadmin/webadmin-mailqueue/src/test/java/org/apache/james/webadmin/service/ClearMailQueueTaskTest.java
@@ -17,6 +17,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.ArgumentMatchers.anyString;
import static org.mockito.Mockito.mock;
@@ -46,23 +47,24 @@ class ClearMailQueueTaskTest {
when(mockedQueue.getName()).thenReturn(QUEUE_NAME);
when(mailQueueFactory.getQueue(anyString())).thenAnswer(arg -> Optional.of(mockedQueue));
- ManageableMailQueue queue = mailQueueFactory.getQueue(QUEUE_NAME).get();
- ClearMailQueueTask task = new ClearMailQueueTask(queue);
+ ClearMailQueueTask.MailQueueFactory factory = queueName -> mailQueueFactory.getQueue(queueName).orElseThrow(RuntimeException::new);
+ ClearMailQueueTask task = new ClearMailQueueTask(QUEUE_NAME, factory);
JsonSerializationVerifier.dtoModule(ClearMailQueueTaskDTO.module(mailQueueFactory))
.bean(task)
.json(SERIALIZED)
+ .equalityTester((a, b) -> assertThat(a.getQueueName()).isEqualTo(b.getQueueName()))
.verify();
}
@Test
- void taskShouldThrowWhenDeserializeAnUnknownQueue() {
+ void taskShouldThrowWhenRunOnAnUnknownQueue() {
MailQueueFactory<ManageableMailQueue> mailQueueFactory = mock(MailQueueFactory.class);
when(mailQueueFactory.getQueue(anyString())).thenReturn(Optional.empty());
JsonTaskSerializer testee = JsonTaskSerializer.of(ClearMailQueueTaskDTO.module(mailQueueFactory));
String serializedJson = "{\"type\": \"clear-mail-queue\", \"queue\": \"anyQueue\"}";
- assertThatThrownBy(() -> testee.deserialize(serializedJson))
+ assertThatThrownBy(() -> testee.deserialize(serializedJson).run())
.isInstanceOf(ClearMailQueueTask.UnknownSerializedQueue.class);
}
diff --git a/server/protocols/webadmin/webadmin-mailqueue/src/test/java/org/apache/james/webadmin/service/DeleteMailsFromMailQueueTaskTest.java b/server/protocols/webadmin/webadmin-mailqueue/src/test/java/org/apache/james/webadmin/service/DeleteMailsFromMailQueueTaskTest.java
index c6797c2..6e44829 100644
--- a/server/protocols/webadmin/webadmin-mailqueue/src/test/java/org/apache/james/webadmin/service/DeleteMailsFromMailQueueTaskTest.java
+++ b/server/protocols/webadmin/webadmin-mailqueue/src/test/java/org/apache/james/webadmin/service/DeleteMailsFromMailQueueTaskTest.java
@@ -17,6 +17,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.ArgumentMatchers.anyString;
import static org.mockito.Mockito.mock;
@@ -53,26 +54,32 @@ class DeleteMailsFromMailQueueTaskTest {
@Test
void taskShouldBeSerializable() throws Exception {
- ManageableMailQueue queue = mailQueueFactory.getQueue(queueName).get();
- DeleteMailsFromMailQueueTask taskSender = new DeleteMailsFromMailQueueTask(queue, Optional.of(new MailAddress("a@b.c")), Optional.empty(), Optional.empty());
- DeleteMailsFromMailQueueTask taskName = new DeleteMailsFromMailQueueTask(queue, Optional.empty(), Optional.of("name"), Optional.empty());
- DeleteMailsFromMailQueueTask taskRecipient = new DeleteMailsFromMailQueueTask(queue, Optional.empty(), Optional.empty(), Optional.of(new MailAddress("d@e.f")));
+ DeleteMailsFromMailQueueTask.MailQueueFactory factory = name -> this.mailQueueFactory.getQueue(name).orElseThrow(RuntimeException::new);
+ DeleteMailsFromMailQueueTask taskSender = new DeleteMailsFromMailQueueTask(queueName, factory, Optional.of(new MailAddress("a@b.c")), Optional.empty(), Optional.empty());
+ DeleteMailsFromMailQueueTask taskName = new DeleteMailsFromMailQueueTask(queueName, factory, Optional.empty(), Optional.of("name"), Optional.empty());
+ DeleteMailsFromMailQueueTask taskRecipient = new DeleteMailsFromMailQueueTask(queueName, factory, Optional.empty(), Optional.empty(), Optional.of(new MailAddress("d@e.f")));
- JsonSerializationVerifier.dtoModule(DeleteMailsFromMailQueueTaskDTO.module(mailQueueFactory))
+ JsonSerializationVerifier.dtoModule(DeleteMailsFromMailQueueTaskDTO.module(this.mailQueueFactory))
.testCase(taskSender, "{\"type\": \"delete-mails-from-mail-queue\", \"queue\": \"anyQueue\", \"sender\": \"a@b.c\"}")
.testCase(taskName, "{\"type\": \"delete-mails-from-mail-queue\", \"queue\": \"anyQueue\", \"name\": \"name\"}")
.testCase(taskRecipient, "{\"type\": \"delete-mails-from-mail-queue\", \"queue\": \"anyQueue\", \"recipient\": \"d@e.f\"}")
+ .equalityTester((a, b) -> {
+ assertThat(a.getQueueName()).isEqualTo(b.getQueueName());
+ assertThat(a.getMaybeName()).isEqualTo(b.getMaybeName());
+ assertThat(a.getMaybeSender()).isEqualTo(b.getMaybeSender());
+ assertThat(a.getMaybeRecipient()).isEqualTo(b.getMaybeRecipient());
+ })
.verify();
}
@Test
- void taskShouldThrowWhenDeserializeAnUnknownQueue() {
+ void taskShouldThrowWhenRunOnAnUnknownQueue() {
MailQueueFactory<ManageableMailQueue> mailQueueFactory = mock(MailQueueFactory.class);
when(mailQueueFactory.getQueue(anyString())).thenReturn(Optional.empty());
JsonTaskSerializer testee = JsonTaskSerializer.of(DeleteMailsFromMailQueueTaskDTO.module(mailQueueFactory));
String serializedJson = "{\"type\": \"delete-mails-from-mail-queue\", \"queue\": \"anyQueue\", \"sender\": \"a@b.c\"}";
- assertThatThrownBy(() -> testee.deserialize(serializedJson))
+ assertThatThrownBy(() -> testee.deserialize(serializedJson).run())
.isInstanceOf(DeleteMailsFromMailQueueTask.UnknownSerializedQueue.class);
}
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 6ccb045..6171b00 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
@@ -19,6 +19,8 @@
package org.apache.james.webadmin.service;
+import java.io.Closeable;
+import java.io.IOException;
import java.util.Optional;
import java.util.function.Consumer;
@@ -35,17 +37,22 @@ import org.apache.james.queue.api.MailQueueFactory;
import org.apache.james.util.OptionalUtils;
import org.apache.james.util.streams.Iterators;
import org.apache.mailet.Mail;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import com.github.fge.lambdas.Throwing;
public class ReprocessingService {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ReprocessingService.class);
+
public static class MissingKeyException extends RuntimeException {
MissingKeyException(MailKey key) {
super(key.asString() + " can not be found");
}
}
- static class Reprocessor {
+ static class Reprocessor implements Closeable {
private final MailQueue mailQueue;
private final Optional<String> targetProcessor;
@@ -63,6 +70,15 @@ public class ReprocessingService {
throw new RuntimeException("Error encountered while reprocessing mail " + mail.getName(), e);
}
}
+
+ @Override
+ public void close() {
+ try {
+ mailQueue.close();
+ } catch (IOException e) {
+ LOGGER.debug("error closing queue", e);
+ }
+ }
}
private final MailQueueFactory<?> mailQueueFactory;
@@ -76,30 +92,30 @@ public class ReprocessingService {
}
public void reprocessAll(MailRepositoryPath path, Optional<String> targetProcessor, String targetQueue, Consumer<MailKey> keyListener) throws MailRepositoryStore.MailRepositoryStoreException, MessagingException {
- Reprocessor reprocessor = new Reprocessor(getMailQueue(targetQueue), targetProcessor);
-
- mailRepositoryStoreService
- .getRepositories(path)
- .forEach(Throwing.consumer((MailRepository repository) ->
- Iterators.toStream(repository.list())
- .peek(keyListener)
- .map(Throwing.function(key -> Optional.ofNullable(repository.retrieve(key))))
- .flatMap(OptionalUtils::toStream)
- .forEach(mail -> reprocessor.reprocess(repository, mail))));
+ try (Reprocessor reprocessor = new Reprocessor(getMailQueue(targetQueue), targetProcessor)) {
+ mailRepositoryStoreService
+ .getRepositories(path)
+ .forEach(Throwing.consumer((MailRepository repository) ->
+ Iterators.toStream(repository.list())
+ .peek(keyListener)
+ .map(Throwing.function(key -> Optional.ofNullable(repository.retrieve(key))))
+ .flatMap(OptionalUtils::toStream)
+ .forEach(mail -> reprocessor.reprocess(repository, mail))));
+ }
}
public void reprocess(MailRepositoryPath path, MailKey key, Optional<String> targetProcessor, String targetQueue) throws MailRepositoryStore.MailRepositoryStoreException, MessagingException {
- Reprocessor reprocessor = new Reprocessor(getMailQueue(targetQueue), targetProcessor);
-
- Pair<MailRepository, Mail> mailPair = mailRepositoryStoreService
- .getRepositories(path)
- .map(Throwing.function(repository -> Pair.of(repository, Optional.ofNullable(repository.retrieve(key)))))
- .filter(pair -> pair.getRight().isPresent())
- .map(pair -> Pair.of(pair.getLeft(), pair.getRight().get()))
- .findFirst()
- .orElseThrow(() -> new MissingKeyException(key));
-
- reprocessor.reprocess(mailPair.getKey(), mailPair.getValue());
+ try (Reprocessor reprocessor = new Reprocessor(getMailQueue(targetQueue), targetProcessor)) {
+ Pair<MailRepository, Mail> mailPair = mailRepositoryStoreService
+ .getRepositories(path)
+ .map(Throwing.function(repository -> Pair.of(repository, Optional.ofNullable(repository.retrieve(key)))))
+ .filter(pair -> pair.getRight().isPresent())
+ .map(pair -> Pair.of(pair.getLeft(), pair.getRight().get()))
+ .findFirst()
+ .orElseThrow(() -> new MissingKeyException(key));
+
+ reprocessor.reprocess(mailPair.getKey(), mailPair.getValue());
+ }
}
private MailQueue getMailQueue(String targetQueue) {
diff --git a/server/queue/queue-activemq/src/main/java/org/apache/james/queue/activemq/ActiveMQMailQueue.java b/server/queue/queue-activemq/src/main/java/org/apache/james/queue/activemq/ActiveMQCacheableMailQueue.java
similarity index 91%
rename from server/queue/queue-activemq/src/main/java/org/apache/james/queue/activemq/ActiveMQMailQueue.java
rename to server/queue/queue-activemq/src/main/java/org/apache/james/queue/activemq/ActiveMQCacheableMailQueue.java
index c6e094b..145acd5 100644
--- a/server/queue/queue-activemq/src/main/java/org/apache/james/queue/activemq/ActiveMQMailQueue.java
+++ b/server/queue/queue-activemq/src/main/java/org/apache/james/queue/activemq/ActiveMQCacheableMailQueue.java
@@ -44,7 +44,7 @@ import org.apache.james.metrics.api.GaugeRegistry;
import org.apache.james.metrics.api.MetricFactory;
import org.apache.james.queue.api.MailQueue;
import org.apache.james.queue.api.MailQueueItemDecoratorFactory;
-import org.apache.james.queue.jms.JMSMailQueue;
+import org.apache.james.queue.jms.JMSCacheableMailQueue;
import org.apache.james.server.core.MailImpl;
import org.apache.james.server.core.MimeMessageCopyOnWriteProxy;
import org.apache.james.server.core.MimeMessageInputStream;
@@ -89,17 +89,17 @@ import org.slf4j.LoggerFactory;
* </p>
* To have a good throughput you should use a caching connection factory. </p>
*/
-public class ActiveMQMailQueue extends JMSMailQueue implements ActiveMQSupport {
- private static final Logger LOGGER = LoggerFactory.getLogger(ActiveMQMailQueue.class);
+public class ActiveMQCacheableMailQueue extends JMSCacheableMailQueue implements ActiveMQSupport {
+ private static final Logger LOGGER = LoggerFactory.getLogger(ActiveMQCacheableMailQueue.class);
private final boolean useBlob;
/**
- * Construct a {@link ActiveMQMailQueue} which only use {@link BlobMessage}
+ * Construct a {@link ActiveMQCacheableMailQueue} which only use {@link BlobMessage}
*
*/
- public ActiveMQMailQueue(ConnectionFactory connectionFactory, MailQueueItemDecoratorFactory mailQueueItemDecoratorFactory, String queuename, MetricFactory metricFactory,
- GaugeRegistry gaugeRegistry) {
+ public ActiveMQCacheableMailQueue(ConnectionFactory connectionFactory, MailQueueItemDecoratorFactory mailQueueItemDecoratorFactory, String queuename, MetricFactory metricFactory,
+ GaugeRegistry gaugeRegistry) {
this(connectionFactory, mailQueueItemDecoratorFactory, queuename, true, metricFactory, gaugeRegistry);
}
@@ -110,8 +110,8 @@ public class ActiveMQMailQueue extends JMSMailQueue implements ActiveMQSupport {
* @param queuename
* @param useBlob
*/
- public ActiveMQMailQueue(ConnectionFactory connectionFactory, MailQueueItemDecoratorFactory mailQueueItemDecoratorFactory, String queuename, boolean useBlob, MetricFactory metricFactory,
- GaugeRegistry gaugeRegistry) {
+ public ActiveMQCacheableMailQueue(ConnectionFactory connectionFactory, MailQueueItemDecoratorFactory mailQueueItemDecoratorFactory, String queuename, boolean useBlob, MetricFactory metricFactory,
+ GaugeRegistry gaugeRegistry) {
super(connectionFactory, mailQueueItemDecoratorFactory, queuename, metricFactory, gaugeRegistry);
this.useBlob = useBlob;
}
@@ -268,7 +268,7 @@ public class ActiveMQMailQueue extends JMSMailQueue implements ActiveMQSupport {
/**
* Try to use ActiveMQ StatisticsPlugin to get size and if that fails
- * fallback to {@link JMSMailQueue#getSize()}
+ * fallback to {@link JMSCacheableMailQueue#getSize()}
*/
@Override
public long getSize() throws MailQueueException {
diff --git a/server/queue/queue-activemq/src/main/java/org/apache/james/queue/activemq/ActiveMQMailQueueFactory.java b/server/queue/queue-activemq/src/main/java/org/apache/james/queue/activemq/ActiveMQMailQueueFactory.java
index 5008508..5b61ca5 100644
--- a/server/queue/queue-activemq/src/main/java/org/apache/james/queue/activemq/ActiveMQMailQueueFactory.java
+++ b/server/queue/queue-activemq/src/main/java/org/apache/james/queue/activemq/ActiveMQMailQueueFactory.java
@@ -30,7 +30,7 @@ import org.apache.james.queue.jms.JMSMailQueueFactory;
/**
* {@link MailQueueFactory} implementations which return
- * {@link ActiveMQMailQueue} instances
+ * {@link ActiveMQCacheableMailQueue} instances
*/
public class ActiveMQMailQueueFactory extends JMSMailQueueFactory {
@@ -51,7 +51,7 @@ public class ActiveMQMailQueueFactory extends JMSMailQueueFactory {
}
@Override
- protected ManageableMailQueue createMailQueue(String name) {
- return new ActiveMQMailQueue(connectionFactory, mailQueueItemDecoratorFactory, name, useBlob, metricFactory, gaugeRegistry);
+ protected ManageableMailQueue createCacheableMailQueue(String name) {
+ return new ActiveMQCacheableMailQueue(connectionFactory, mailQueueItemDecoratorFactory, name, useBlob, metricFactory, gaugeRegistry);
}
}
diff --git a/server/queue/queue-activemq/src/test/java/org/apache/james/queue/activemq/ActiveMQMailQueueBlobTest.java b/server/queue/queue-activemq/src/test/java/org/apache/james/queue/activemq/ActiveMQMailQueueBlobTest.java
index b2c3cea..da22eb7 100644
--- a/server/queue/queue-activemq/src/test/java/org/apache/james/queue/activemq/ActiveMQMailQueueBlobTest.java
+++ b/server/queue/queue-activemq/src/test/java/org/apache/james/queue/activemq/ActiveMQMailQueueBlobTest.java
@@ -59,7 +59,7 @@ public class ActiveMQMailQueueBlobTest implements DelayedManageableMailQueueCont
static final String BASE_DIR = "file://target/james-test";
static final boolean USE_BLOB = true;
- ActiveMQMailQueue mailQueue;
+ ActiveMQCacheableMailQueue mailQueue;
MyFileSystem fileSystem;
@BeforeEach
@@ -78,7 +78,7 @@ public class ActiveMQMailQueueBlobTest implements DelayedManageableMailQueueCont
MetricFactory metricFactory = metricTestSystem.getMetricFactory();
GaugeRegistry gaugeRegistry = metricTestSystem.getSpyGaugeRegistry();
String queueName = BrokerExtension.generateRandomQueueName(broker);
- mailQueue = new ActiveMQMailQueue(connectionFactory, mailQueueItemDecoratorFactory, queueName, USE_BLOB, metricFactory, gaugeRegistry);
+ mailQueue = new ActiveMQCacheableMailQueue(connectionFactory, mailQueueItemDecoratorFactory, queueName, USE_BLOB, metricFactory, gaugeRegistry);
}
@AfterEach
diff --git a/server/queue/queue-activemq/src/test/java/org/apache/james/queue/activemq/ActiveMQMailQueueTest.java b/server/queue/queue-activemq/src/test/java/org/apache/james/queue/activemq/ActiveMQMailQueueTest.java
index 609cef1..e7a08dd 100644
--- a/server/queue/queue-activemq/src/test/java/org/apache/james/queue/activemq/ActiveMQMailQueueTest.java
+++ b/server/queue/queue-activemq/src/test/java/org/apache/james/queue/activemq/ActiveMQMailQueueTest.java
@@ -18,8 +18,6 @@
****************************************************************/
package org.apache.james.queue.activemq;
-import javax.jms.ConnectionFactory;
-
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.ActiveMQPrefetchPolicy;
import org.apache.activemq.broker.BrokerService;
@@ -48,7 +46,7 @@ public class ActiveMQMailQueueTest implements DelayedManageableMailQueueContract
static final boolean USE_BLOB = true;
- ActiveMQMailQueue mailQueue;
+ ActiveMQCacheableMailQueue mailQueue;
@BeforeEach
public void setUp(BrokerService broker, MailQueueMetricExtension.MailQueueMetricTestSystem metricTestSystem) {
@@ -60,7 +58,7 @@ public class ActiveMQMailQueueTest implements DelayedManageableMailQueueContract
MetricFactory metricFactory = metricTestSystem.getMetricFactory();
GaugeRegistry gaugeRegistry = metricTestSystem.getSpyGaugeRegistry();
String queueName = BrokerExtension.generateRandomQueueName(broker);
- mailQueue = new ActiveMQMailQueue(connectionFactory, mailQueueItemDecoratorFactory, queueName, !USE_BLOB, metricFactory, gaugeRegistry);
+ mailQueue = new ActiveMQCacheableMailQueue(connectionFactory, mailQueueItemDecoratorFactory, queueName, !USE_BLOB, metricFactory, gaugeRegistry);
}
@AfterEach
diff --git a/server/queue/queue-api/src/main/java/org/apache/james/queue/api/MailQueue.java b/server/queue/queue-api/src/main/java/org/apache/james/queue/api/MailQueue.java
index 559d998..d4e9c3a 100644
--- a/server/queue/queue-api/src/main/java/org/apache/james/queue/api/MailQueue.java
+++ b/server/queue/queue-api/src/main/java/org/apache/james/queue/api/MailQueue.java
@@ -19,6 +19,7 @@
package org.apache.james.queue.api;
+import java.io.Closeable;
import java.time.Duration;
import java.util.concurrent.TimeUnit;
@@ -58,7 +59,7 @@ import org.threeten.extra.Temporals;
* </ul>
* </p>
*/
-public interface MailQueue {
+public interface MailQueue extends Closeable {
String ENQUEUED_METRIC_NAME_PREFIX = "enqueuedMail:";
String DEQUEUED_METRIC_NAME_PREFIX = "dequeuedMail:";
diff --git a/server/queue/queue-file/src/main/java/org/apache/james/queue/file/FileMailQueue.java b/server/queue/queue-file/src/main/java/org/apache/james/queue/file/FileCacheableMailQueue.java
similarity index 97%
rename from server/queue/queue-file/src/main/java/org/apache/james/queue/file/FileMailQueue.java
rename to server/queue/queue-file/src/main/java/org/apache/james/queue/file/FileCacheableMailQueue.java
index 4f288b3..6142be8 100644
--- a/server/queue/queue-file/src/main/java/org/apache/james/queue/file/FileMailQueue.java
+++ b/server/queue/queue-file/src/main/java/org/apache/james/queue/file/FileCacheableMailQueue.java
@@ -71,15 +71,15 @@ import reactor.core.publisher.Mono;
/**
* {@link ManageableMailQueue} implementation which use the fs to store {@link Mail}'s
* <p/>
- * On create of the {@link FileMailQueue} the {@link #init()} will get called. This takes care of
+ * On create of the {@link FileCacheableMailQueue} the {@link #init()} will get called. This takes care of
* loading the needed meta-data into memory for fast access.
*
* @deprecated FileMailQueue implementation is unmaintained, incomplete and not thread safe
* We recommend using embedded ActiveMQMailQueue implementation instead
*/
@Deprecated
-public class FileMailQueue implements ManageableMailQueue {
- private static final Logger LOGGER = LoggerFactory.getLogger(FileMailQueue.class);
+public class FileCacheableMailQueue implements ManageableMailQueue {
+ private static final Logger LOGGER = LoggerFactory.getLogger(FileCacheableMailQueue.class);
private final Map<String, FileItem> keyMappings = Collections.synchronizedMap(new LinkedHashMap<>());
private final BlockingQueue<String> inmemoryQueue = new LinkedBlockingQueue<>();
@@ -98,7 +98,7 @@ public class FileMailQueue implements ManageableMailQueue {
private final String queueName;
private final Flux<MailQueueItem> flux;
- public FileMailQueue(MailQueueItemDecoratorFactory mailQueueItemDecoratorFactory, File parentDir, String queuename, boolean sync) throws IOException {
+ public FileCacheableMailQueue(MailQueueItemDecoratorFactory mailQueueItemDecoratorFactory, File parentDir, String queuename, boolean sync) throws IOException {
this.mailQueueItemDecoratorFactory = mailQueueItemDecoratorFactory;
this.sync = sync;
this.queueName = queuename;
@@ -111,6 +111,11 @@ public class FileMailQueue implements ManageableMailQueue {
}
@Override
+ public void close() {
+ //There's no resource to free
+ }
+
+ @Override
public String getName() {
return queueName;
}
diff --git a/server/queue/queue-file/src/main/java/org/apache/james/queue/file/FileMailQueueFactory.java b/server/queue/queue-file/src/main/java/org/apache/james/queue/file/FileMailQueueFactory.java
index 87d88a8..623e1e0 100644
--- a/server/queue/queue-file/src/main/java/org/apache/james/queue/file/FileMailQueueFactory.java
+++ b/server/queue/queue-file/src/main/java/org/apache/james/queue/file/FileMailQueueFactory.java
@@ -23,6 +23,7 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import javax.inject.Inject;
@@ -34,7 +35,7 @@ import org.apache.james.queue.api.ManageableMailQueue;
import com.google.common.collect.ImmutableSet;
/**
- * {@link MailQueueFactory} implementation which returns {@link FileMailQueue} instances
+ * {@link MailQueueFactory} implementation which returns {@link FileCacheableMailQueue} instances
*
* @deprecated FileMailQueue implementation is unmaintained, incomplete and not thread safe
* We recommend using embedded ActiveMQMailQueue implementation instead
@@ -42,7 +43,7 @@ import com.google.common.collect.ImmutableSet;
@Deprecated
public class FileMailQueueFactory implements MailQueueFactory<ManageableMailQueue> {
- private final Map<String, ManageableMailQueue> queues = new HashMap<>();
+ private final Map<String, ManageableMailQueue> queues = new ConcurrentHashMap<>();
private MailQueueItemDecoratorFactory mailQueueActionItemDecoratorFactory;
private FileSystem fs;
private boolean sync = true;
@@ -59,7 +60,7 @@ public class FileMailQueueFactory implements MailQueueFactory<ManageableMailQueu
}
/**
- * If <code>true</code> the later created {@link FileMailQueue} will call <code>fsync</code> after each message {@link FileMailQueue#enQueue(org.apache.mailet.Mail)} call. This
+ * If <code>true</code> the later created {@link FileCacheableMailQueue} will call <code>fsync</code> after each message {@link FileCacheableMailQueue#enQueue(org.apache.mailet.Mail)} call. This
* is needed to be fully RFC conform but gives a performance penalty. If you are brave enough you man set it to <code>false</code>
* <p/>
* The default is <code>true</code>
@@ -77,19 +78,13 @@ public class FileMailQueueFactory implements MailQueueFactory<ManageableMailQueu
@Override
public ManageableMailQueue createQueue(String name) {
- return getQueue(name).orElseGet(() -> createAndRegisterQueue(name));
- }
-
- private ManageableMailQueue createAndRegisterQueue(String name) {
- synchronized (queues) {
+ return queues.computeIfAbsent(name, mailQueueName -> {
try {
- FileMailQueue queue = new FileMailQueue(mailQueueActionItemDecoratorFactory, fs.getFile("file://var/store/queue"), name, sync);
- queues.put(name, queue);
- return queue;
+ return new FileCacheableMailQueue(mailQueueActionItemDecoratorFactory, fs.getFile("file://var/store/queue"), mailQueueName, sync);
} catch (IOException e) {
- throw new RuntimeException("Unable to access queue " + name, e);
+ throw new RuntimeException("Unable to access queue " + mailQueueName, e);
}
- }
+ });
}
}
diff --git a/server/queue/queue-file/src/test/java/org/apache/james/queue/file/FileMailQueueFactoryTest.java b/server/queue/queue-file/src/test/java/org/apache/james/queue/file/FileCacheableMailQueueFactoryTest.java
similarity index 94%
rename from server/queue/queue-file/src/test/java/org/apache/james/queue/file/FileMailQueueFactoryTest.java
rename to server/queue/queue-file/src/test/java/org/apache/james/queue/file/FileCacheableMailQueueFactoryTest.java
index 926092f..923a8db 100644
--- a/server/queue/queue-file/src/test/java/org/apache/james/queue/file/FileMailQueueFactoryTest.java
+++ b/server/queue/queue-file/src/test/java/org/apache/james/queue/file/FileCacheableMailQueueFactoryTest.java
@@ -35,7 +35,7 @@ import org.junit.jupiter.api.Disabled;
" - JAMES-2954 Incomplete browse implementation" +
" - JAMES-2544 Mixing concurrent operation might lead to a deadlock and missing fields" +
" - JAMES-2979 dequeue is not thread safe")
-public class FileMailQueueFactoryTest implements MailQueueFactoryContract<ManageableMailQueue>, ManageableMailQueueFactoryContract {
+public class FileCacheableMailQueueFactoryTest implements MailQueueFactoryContract<ManageableMailQueue>, ManageableMailQueueFactoryContract {
private FileMailQueueFactory mailQueueFactory;
private MockFileSystem fileSystem;
diff --git a/server/queue/queue-file/src/test/java/org/apache/james/queue/file/FileMailQueueTest.java b/server/queue/queue-file/src/test/java/org/apache/james/queue/file/FileCacheableMailQueueTest.java
similarity index 90%
rename from server/queue/queue-file/src/test/java/org/apache/james/queue/file/FileMailQueueTest.java
rename to server/queue/queue-file/src/test/java/org/apache/james/queue/file/FileCacheableMailQueueTest.java
index f2659d6..2b479e9 100644
--- a/server/queue/queue-file/src/test/java/org/apache/james/queue/file/FileMailQueueTest.java
+++ b/server/queue/queue-file/src/test/java/org/apache/james/queue/file/FileCacheableMailQueueTest.java
@@ -34,16 +34,16 @@ import org.junit.rules.TemporaryFolder;
" - JAMES-2954 Incomplete browse implementation" +
" - JAMES-2544 Mixing concurrent operation might lead to a deadlock and missing fields" +
" - JAMES-2979 dequeue is not thread safe")
-public class FileMailQueueTest implements DelayedManageableMailQueueContract {
+public class FileCacheableMailQueueTest implements DelayedManageableMailQueueContract {
private static final boolean SYNC = true;
private TemporaryFolder temporaryFolder = new TemporaryFolder();
- private FileMailQueue mailQueue;
+ private FileCacheableMailQueue mailQueue;
@BeforeEach
public void setUp() throws Exception {
temporaryFolder.create();
- mailQueue = new FileMailQueue(new RawMailQueueItemDecoratorFactory(), temporaryFolder.newFolder(), "test", SYNC);
+ mailQueue = new FileCacheableMailQueue(new RawMailQueueItemDecoratorFactory(), temporaryFolder.newFolder(), "test", SYNC);
}
@AfterEach
diff --git a/server/queue/queue-jms/src/main/java/org/apache/james/queue/jms/JMSMailQueue.java b/server/queue/queue-jms/src/main/java/org/apache/james/queue/jms/JMSCacheableMailQueue.java
similarity index 97%
rename from server/queue/queue-jms/src/main/java/org/apache/james/queue/jms/JMSMailQueue.java
rename to server/queue/queue-jms/src/main/java/org/apache/james/queue/jms/JMSCacheableMailQueue.java
index 2511f49..5171cae 100644
--- a/server/queue/queue-jms/src/main/java/org/apache/james/queue/jms/JMSMailQueue.java
+++ b/server/queue/queue-jms/src/main/java/org/apache/james/queue/jms/JMSCacheableMailQueue.java
@@ -99,7 +99,7 @@ import reactor.core.publisher.Mono;
* {@link Mail} objects.
* </p>
*/
-public class JMSMailQueue implements ManageableMailQueue, JMSSupport, MailPrioritySupport, Disposable {
+public class JMSCacheableMailQueue implements ManageableMailQueue, JMSSupport, MailPrioritySupport, Disposable {
private final Flux<MailQueueItem> flux;
@@ -153,7 +153,7 @@ public class JMSMailQueue implements ManageableMailQueue, JMSSupport, MailPriori
}
}
- private static final Logger LOGGER = LoggerFactory.getLogger(JMSMailQueue.class);
+ private static final Logger LOGGER = LoggerFactory.getLogger(JMSCacheableMailQueue.class);
public static final String FORCE_DELIVERY = "FORCE_DELIVERY";
@@ -172,9 +172,9 @@ public class JMSMailQueue implements ManageableMailQueue, JMSSupport, MailPriori
private final Joiner joiner;
private final Splitter splitter;
- public JMSMailQueue(ConnectionFactory connectionFactory, MailQueueItemDecoratorFactory mailQueueItemDecoratorFactory,
- String queueName, MetricFactory metricFactory,
- GaugeRegistry gaugeRegistry) {
+ public JMSCacheableMailQueue(ConnectionFactory connectionFactory, MailQueueItemDecoratorFactory mailQueueItemDecoratorFactory,
+ String queueName, MetricFactory metricFactory,
+ GaugeRegistry gaugeRegistry) {
try {
connection = connectionFactory.createConnection();
connection.start();
@@ -204,6 +204,14 @@ public class JMSMailQueue implements ManageableMailQueue, JMSSupport, MailPriori
flux = Mono.defer(this::deQueueOneItem).repeat();
}
+ /**
+ * To allow connection reuse (the queue is cacheable), we don't close the queue
+ * on close(), use {@link JMSCacheableMailQueue#dispose} to release resources
+ */
+ @Override
+ public void close() {
+ }
+
@Override
public String getName() {
return queueName;
diff --git a/server/queue/queue-jms/src/main/java/org/apache/james/queue/jms/JMSMailQueueFactory.java b/server/queue/queue-jms/src/main/java/org/apache/james/queue/jms/JMSMailQueueFactory.java
index 051e3c0..701e0b3 100644
--- a/server/queue/queue-jms/src/main/java/org/apache/james/queue/jms/JMSMailQueueFactory.java
+++ b/server/queue/queue-jms/src/main/java/org/apache/james/queue/jms/JMSMailQueueFactory.java
@@ -48,8 +48,8 @@ public class JMSMailQueueFactory extends AbstractMailQueueFactory<ManageableMail
}
@Override
- protected ManageableMailQueue createMailQueue(String name) {
- return new JMSMailQueue(connectionFactory, mailQueueItemDecoratorFactory, name, metricFactory, gaugeRegistry);
+ protected ManageableMailQueue createCacheableMailQueue(String name) {
+ return new JMSCacheableMailQueue(connectionFactory, mailQueueItemDecoratorFactory, name, metricFactory, gaugeRegistry);
}
-
+
}
diff --git a/server/queue/queue-jms/src/main/java/org/apache/james/queue/jms/JMSMailQueueItem.java b/server/queue/queue-jms/src/main/java/org/apache/james/queue/jms/JMSMailQueueItem.java
index ad10836..d20c642 100644
--- a/server/queue/queue-jms/src/main/java/org/apache/james/queue/jms/JMSMailQueueItem.java
+++ b/server/queue/queue-jms/src/main/java/org/apache/james/queue/jms/JMSMailQueueItem.java
@@ -48,13 +48,13 @@ public class JMSMailQueueItem implements MailQueueItem {
if (success) {
session.commit();
} else {
- JMSMailQueue.rollback(session);
+ JMSCacheableMailQueue.rollback(session);
}
} catch (JMSException ex) {
throw new MailQueueException("Unable to commit dequeue operation for mail " + mail.getName(), ex);
} finally {
- JMSMailQueue.closeConsumer(consumer);
- JMSMailQueue.closeSession(session);
+ JMSCacheableMailQueue.closeConsumer(consumer);
+ JMSCacheableMailQueue.closeSession(session);
}
}
diff --git a/server/queue/queue-jms/src/main/java/org/apache/james/queue/library/AbstractMailQueueFactory.java b/server/queue/queue-jms/src/main/java/org/apache/james/queue/library/AbstractMailQueueFactory.java
index e1a29dd..fdd4b86 100644
--- a/server/queue/queue-jms/src/main/java/org/apache/james/queue/library/AbstractMailQueueFactory.java
+++ b/server/queue/queue-jms/src/main/java/org/apache/james/queue/library/AbstractMailQueueFactory.java
@@ -102,7 +102,7 @@ public abstract class AbstractMailQueueFactory<T extends MailQueue> implements M
}
private T createAndRegisterQueue(String name) {
- T queue = createMailQueue(name);
+ T queue = createCacheableMailQueue(name);
if (useJMX) {
registerMBean(name, queue);
}
@@ -111,12 +111,10 @@ public abstract class AbstractMailQueueFactory<T extends MailQueue> implements M
}
/**
- * Create a {@link MailQueue} for the given name
- *
- * @param name
- * @return queue
+ * Create a {@link MailQueue} for the given name that happens to do nothing on close()
+ * to be able to cache the instance
*/
- protected abstract T createMailQueue(String name);
+ protected abstract T createCacheableMailQueue(String name);
protected synchronized void registerMBean(String queuename, MailQueue queue) {
diff --git a/server/queue/queue-jms/src/test/java/org/apache/james/queue/jms/JMSMailQueueFactoryTest.java b/server/queue/queue-jms/src/test/java/org/apache/james/queue/jms/JMSCacheableMailQueueFactoryTest.java
similarity index 95%
rename from server/queue/queue-jms/src/test/java/org/apache/james/queue/jms/JMSMailQueueFactoryTest.java
rename to server/queue/queue-jms/src/test/java/org/apache/james/queue/jms/JMSCacheableMailQueueFactoryTest.java
index f98cf9d..5a62538 100644
--- a/server/queue/queue-jms/src/test/java/org/apache/james/queue/jms/JMSMailQueueFactoryTest.java
+++ b/server/queue/queue-jms/src/test/java/org/apache/james/queue/jms/JMSCacheableMailQueueFactoryTest.java
@@ -37,7 +37,7 @@ import org.junit.jupiter.api.extension.ExtendWith;
@ExtendWith(BrokerExtension.class)
-public class JMSMailQueueFactoryTest implements MailQueueFactoryContract<ManageableMailQueue>, ManageableMailQueueFactoryContract {
+public class JMSCacheableMailQueueFactoryTest implements MailQueueFactoryContract<ManageableMailQueue>, ManageableMailQueueFactoryContract {
private JMSMailQueueFactory mailQueueFactory;
diff --git a/server/queue/queue-jms/src/test/java/org/apache/james/queue/jms/JMSMailQueueTest.java b/server/queue/queue-jms/src/test/java/org/apache/james/queue/jms/JMSCacheableMailQueueTest.java
similarity index 92%
rename from server/queue/queue-jms/src/test/java/org/apache/james/queue/jms/JMSMailQueueTest.java
rename to server/queue/queue-jms/src/test/java/org/apache/james/queue/jms/JMSCacheableMailQueueTest.java
index f436a1d..061218a 100644
--- a/server/queue/queue-jms/src/test/java/org/apache/james/queue/jms/JMSMailQueueTest.java
+++ b/server/queue/queue-jms/src/test/java/org/apache/james/queue/jms/JMSCacheableMailQueueTest.java
@@ -19,8 +19,6 @@
package org.apache.james.queue.jms;
-import javax.jms.ConnectionFactory;
-
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.ActiveMQPrefetchPolicy;
import org.apache.activemq.broker.BrokerService;
@@ -41,10 +39,10 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
@ExtendWith(BrokerExtension.class)
-public class JMSMailQueueTest implements DelayedManageableMailQueueContract, PriorityManageableMailQueueContract, DelayedPriorityMailQueueContract,
+public class JMSCacheableMailQueueTest implements DelayedManageableMailQueueContract, PriorityManageableMailQueueContract, DelayedPriorityMailQueueContract,
MailQueueMetricContract {
- private JMSMailQueue mailQueue;
+ private JMSCacheableMailQueue mailQueue;
@BeforeEach
void setUp(BrokerService broker, MailQueueMetricExtension.MailQueueMetricTestSystem metricTestSystem) {
@@ -56,7 +54,7 @@ public class JMSMailQueueTest implements DelayedManageableMailQueueContract, Pri
MetricFactory metricFactory = metricTestSystem.getMetricFactory();
GaugeRegistry gaugeRegistry = metricTestSystem.getSpyGaugeRegistry();
String queueName = BrokerExtension.generateRandomQueueName(broker);
- mailQueue = new JMSMailQueue(connectionFactory, mailQueueItemDecoratorFactory, queueName, metricFactory, gaugeRegistry);
+ mailQueue = new JMSCacheableMailQueue(connectionFactory, mailQueueItemDecoratorFactory, queueName, metricFactory, gaugeRegistry);
}
@AfterEach
diff --git a/server/queue/queue-jms/src/test/java/org/apache/james/queue/library/AbstractMailQueueFactoryTest.java b/server/queue/queue-jms/src/test/java/org/apache/james/queue/library/AbstractMailQueueFactoryTest.java
index 2ae2c50..2f25606 100644
--- a/server/queue/queue-jms/src/test/java/org/apache/james/queue/library/AbstractMailQueueFactoryTest.java
+++ b/server/queue/queue-jms/src/test/java/org/apache/james/queue/library/AbstractMailQueueFactoryTest.java
@@ -46,7 +46,7 @@ public class AbstractMailQueueFactoryTest {
mBeanServer = mock(MBeanServer.class);
abstractMailQueueFactory = new AbstractMailQueueFactory<ManageableMailQueue>() {
@Override
- protected ManageableMailQueue createMailQueue(String name) {
+ protected ManageableMailQueue createCacheableMailQueue(String name) {
return mock(ManageableMailQueue.class);
}
};
diff --git a/server/queue/queue-memory/src/main/java/org/apache/james/queue/memory/MemoryMailQueueFactory.java b/server/queue/queue-memory/src/main/java/org/apache/james/queue/memory/MemoryMailQueueFactory.java
index 25b80b6..4aa75ec 100644
--- a/server/queue/queue-memory/src/main/java/org/apache/james/queue/memory/MemoryMailQueueFactory.java
+++ b/server/queue/queue-memory/src/main/java/org/apache/james/queue/memory/MemoryMailQueueFactory.java
@@ -19,6 +19,7 @@
package org.apache.james.queue.memory;
+import java.io.IOException;
import java.time.DateTimeException;
import java.time.Duration;
import java.time.Instant;
@@ -60,7 +61,7 @@ import reactor.core.scheduler.Schedulers;
public class MemoryMailQueueFactory implements MailQueueFactory<ManageableMailQueue> {
- private final ConcurrentHashMap<String, MemoryMailQueueFactory.MemoryMailQueue> mailQueues;
+ private final ConcurrentHashMap<String, MemoryCacheableMailQueue> mailQueues;
private final MailQueueItemDecoratorFactory mailQueueItemDecoratorFactory;
@Inject
@@ -80,19 +81,17 @@ public class MemoryMailQueueFactory implements MailQueueFactory<ManageableMailQu
}
@Override
- public MemoryMailQueueFactory.MemoryMailQueue createQueue(String name) {
- MemoryMailQueueFactory.MemoryMailQueue newMailQueue = new MemoryMailQueue(name, mailQueueItemDecoratorFactory);
- return Optional.ofNullable(mailQueues.putIfAbsent(name, newMailQueue))
- .orElse(newMailQueue);
+ public MemoryCacheableMailQueue createQueue(String name) {
+ return mailQueues.computeIfAbsent(name, mailQueueName -> new MemoryCacheableMailQueue(mailQueueName, mailQueueItemDecoratorFactory));
}
- public static class MemoryMailQueue implements ManageableMailQueue {
+ public static class MemoryCacheableMailQueue implements ManageableMailQueue {
private final DelayQueue<MemoryMailQueueItem> mailItems;
private final LinkedBlockingDeque<MemoryMailQueueItem> inProcessingMailItems;
private final String name;
private final Flux<MailQueueItem> flux;
- public MemoryMailQueue(String name, MailQueueItemDecoratorFactory mailQueueItemDecoratorFactory) {
+ public MemoryCacheableMailQueue(String name, MailQueueItemDecoratorFactory mailQueueItemDecoratorFactory) {
this.mailItems = new DelayQueue<>();
this.inProcessingMailItems = new LinkedBlockingDeque<>();
this.name = name;
@@ -105,6 +104,11 @@ public class MemoryMailQueueFactory implements MailQueueFactory<ManageableMailQu
}
@Override
+ public void close() {
+ //There's no resource to free
+ }
+
+ @Override
public String getName() {
return name;
}
@@ -244,7 +248,7 @@ public class MemoryMailQueueFactory implements MailQueueFactory<ManageableMailQu
return false;
}
- MemoryMailQueue that = (MemoryMailQueue) o;
+ MemoryCacheableMailQueue that = (MemoryCacheableMailQueue) o;
return Objects.equal(this.name, that.name);
}
@@ -257,10 +261,10 @@ public class MemoryMailQueueFactory implements MailQueueFactory<ManageableMailQu
public static class MemoryMailQueueItem implements MailQueue.MailQueueItem, Delayed {
private final Mail mail;
- private final MemoryMailQueue queue;
+ private final MemoryCacheableMailQueue queue;
private final ZonedDateTime delivery;
- public MemoryMailQueueItem(Mail mail, MemoryMailQueue queue, ZonedDateTime delivery) {
+ public MemoryMailQueueItem(Mail mail, MemoryCacheableMailQueue queue, ZonedDateTime delivery) {
this.mail = mail;
this.queue = queue;
this.delivery = delivery;
diff --git a/server/queue/queue-memory/src/test/java/org/apache/james/queue/memory/MemoryMailQueueFactoryTest.java b/server/queue/queue-memory/src/test/java/org/apache/james/queue/memory/MemoryCacheableMailQueueFactoryTest.java
similarity index 93%
rename from server/queue/queue-memory/src/test/java/org/apache/james/queue/memory/MemoryMailQueueFactoryTest.java
rename to server/queue/queue-memory/src/test/java/org/apache/james/queue/memory/MemoryCacheableMailQueueFactoryTest.java
index 28002ca..aa2e294 100644
--- a/server/queue/queue-memory/src/test/java/org/apache/james/queue/memory/MemoryMailQueueFactoryTest.java
+++ b/server/queue/queue-memory/src/test/java/org/apache/james/queue/memory/MemoryCacheableMailQueueFactoryTest.java
@@ -26,7 +26,7 @@ import org.apache.james.queue.api.ManageableMailQueueFactoryContract;
import org.apache.james.queue.api.RawMailQueueItemDecoratorFactory;
import org.junit.jupiter.api.BeforeEach;
-class MemoryMailQueueFactoryTest implements MailQueueFactoryContract<ManageableMailQueue>, ManageableMailQueueFactoryContract {
+class MemoryCacheableMailQueueFactoryTest implements MailQueueFactoryContract<ManageableMailQueue>, ManageableMailQueueFactoryContract {
MemoryMailQueueFactory memoryMailQueueFactory;
diff --git a/server/queue/queue-memory/src/test/java/org/apache/james/queue/memory/MemoryMailQueueTest.java b/server/queue/queue-memory/src/test/java/org/apache/james/queue/memory/MemoryCacheableMailQueueTest.java
similarity index 91%
rename from server/queue/queue-memory/src/test/java/org/apache/james/queue/memory/MemoryMailQueueTest.java
rename to server/queue/queue-memory/src/test/java/org/apache/james/queue/memory/MemoryCacheableMailQueueTest.java
index 085e565..5868995 100644
--- a/server/queue/queue-memory/src/test/java/org/apache/james/queue/memory/MemoryMailQueueTest.java
+++ b/server/queue/queue-memory/src/test/java/org/apache/james/queue/memory/MemoryCacheableMailQueueTest.java
@@ -29,13 +29,13 @@ import org.apache.james.queue.api.RawMailQueueItemDecoratorFactory;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-public class MemoryMailQueueTest implements DelayedManageableMailQueueContract {
+public class MemoryCacheableMailQueueTest implements DelayedManageableMailQueueContract {
- private MemoryMailQueueFactory.MemoryMailQueue mailQueue;
+ private MemoryMailQueueFactory.MemoryCacheableMailQueue mailQueue;
@BeforeEach
public void setUp() {
- mailQueue = new MemoryMailQueueFactory.MemoryMailQueue("test", new RawMailQueueItemDecoratorFactory());
+ mailQueue = new MemoryMailQueueFactory.MemoryCacheableMailQueue("test", new RawMailQueueItemDecoratorFactory());
}
@Override
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/Dequeuer.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/Dequeuer.java
index ece7018..64c0c93 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/Dequeuer.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/Dequeuer.java
@@ -21,6 +21,7 @@ package org.apache.james.queue.rabbitmq;
import static org.apache.james.queue.api.MailQueue.DEQUEUED_METRIC_NAME_PREFIX;
+import java.io.Closeable;
import java.io.IOException;
import java.util.function.Consumer;
import java.util.function.Function;
@@ -40,13 +41,14 @@ import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.rabbitmq.AcknowledgableDelivery;
import reactor.rabbitmq.ConsumeOptions;
+import reactor.rabbitmq.Receiver;
-class Dequeuer {
+class Dequeuer implements Closeable {
private static final boolean REQUEUE = true;
private static final int EXECUTION_RATE = 5;
- private final Flux<AcknowledgableDelivery> flux;
private static class RabbitMQMailQueueItem implements MailQueue.MailQueueItem {
+
private final Consumer<Boolean> ack;
private final EnqueueId enqueueId;
private final Mail mail;
@@ -70,12 +72,15 @@ class Dequeuer {
public void done(boolean success) {
ack.accept(success);
}
+
}
private final Function<MailReferenceDTO, MailWithEnqueueId> mailLoader;
private final Metric dequeueMetric;
private final MailReferenceSerializer mailReferenceSerializer;
private final MailQueueView mailQueueView;
+ private final Receiver receiver;
+ private final Flux<AcknowledgableDelivery> flux;
Dequeuer(MailQueueName name, ReceiverProvider receiverProvider, Function<MailReferenceDTO, MailWithEnqueueId> mailLoader,
MailReferenceSerializer serializer, MetricFactory metricFactory,
@@ -84,11 +89,17 @@ class Dequeuer {
this.mailReferenceSerializer = serializer;
this.mailQueueView = mailQueueView;
this.dequeueMetric = metricFactory.generate(DEQUEUED_METRIC_NAME_PREFIX + name.asString());
- this.flux = receiverProvider.createReceiver()
+ this.receiver = receiverProvider.createReceiver();
+ this.flux = this.receiver
.consumeManualAck(name.toWorkQueueName().asString(), new ConsumeOptions().qos(EXECUTION_RATE))
.filter(getResponse -> getResponse.getBody() != null);
}
+ @Override
+ public void close() {
+ receiver.close();
+ }
+
Flux<? extends MailQueue.MailQueueItem> deQueue() {
return flux.concatMap(this::loadItem)
.concatMap(this::filterIfDeleted);
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueue.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueue.java
index 2dc7663..b09fe30 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueue.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueue.java
@@ -58,6 +58,11 @@ public class RabbitMQMailQueue implements ManageableMailQueue {
}
@Override
+ public void close() {
+ dequeuer.close();
+ }
+
+ @Override
public String getName() {
return name.asString();
}
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueFactory.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueFactory.java
index f771118..57bde2f 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueFactory.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueFactory.java
@@ -29,7 +29,6 @@ import static org.apache.james.queue.api.MailQueue.QUEUE_SIZE_METRIC_NAME_PREFIX
import java.time.Clock;
import java.util.Optional;
import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Function;
import javax.inject.Inject;
@@ -46,6 +45,7 @@ import org.apache.james.queue.api.MailQueueFactory;
import org.apache.james.queue.api.MailQueueItemDecoratorFactory;
import org.apache.james.queue.rabbitmq.view.RabbitMQMailQueueConfiguration;
import org.apache.james.queue.rabbitmq.view.api.MailQueueView;
+import org.apache.james.util.OptionalUtils;
import com.github.fge.lambdas.Throwing;
import com.google.common.annotations.VisibleForTesting;
@@ -119,26 +119,8 @@ public class RabbitMQMailQueueFactory implements MailQueueFactory<RabbitMQMailQu
}
}
- /**
- * RabbitMQMailQueue should have a single instance in a given JVM for a given MailQueueName.
- * This class helps at keeping track of previously instantiated MailQueues.
- */
- private class RabbitMQMailQueueObjectPool {
-
- private final ConcurrentHashMap<MailQueueName, RabbitMQMailQueue> instantiatedQueues;
-
- RabbitMQMailQueueObjectPool() {
- this.instantiatedQueues = new ConcurrentHashMap<>();
- }
-
- RabbitMQMailQueue retrieveInstanceFor(MailQueueName name) {
- return instantiatedQueues.computeIfAbsent(name, privateFactory::create);
- }
- }
-
private final RabbitMQMailQueueManagement mqManagementApi;
private final PrivateFactory privateFactory;
- private final RabbitMQMailQueueObjectPool mailQueueObjectPool;
private final Sender sender;
@VisibleForTesting
@@ -149,7 +131,6 @@ public class RabbitMQMailQueueFactory implements MailQueueFactory<RabbitMQMailQu
this.sender = sender;
this.mqManagementApi = mqManagementApi;
this.privateFactory = privateFactory;
- this.mailQueueObjectPool = new RabbitMQMailQueueObjectPool();
}
@Override
@@ -166,8 +147,9 @@ public class RabbitMQMailQueueFactory implements MailQueueFactory<RabbitMQMailQu
@Override
public Set<RabbitMQMailQueue> listCreatedMailQueues() {
+ //TODO: it creates connections and leak them
return mqManagementApi.listCreatedMailQueueNames()
- .map(mailQueueObjectPool::retrieveInstanceFor)
+ .flatMap(name -> OptionalUtils.toStream(getQueue(name.asString())))
.collect(ImmutableSet.toImmutableSet());
}
@@ -188,13 +170,13 @@ public class RabbitMQMailQueueFactory implements MailQueueFactory<RabbitMQMailQu
.routingKey(EMPTY_ROUTING_KEY)))
.then()
.block();
- return mailQueueObjectPool.retrieveInstanceFor(mailQueueName);
+ return privateFactory.create(mailQueueName);
}
private Optional<RabbitMQMailQueue> getQueueFromRabbitServer(MailQueueName name) {
return mqManagementApi.listCreatedMailQueueNames()
.filter(name::equals)
- .map(mailQueueObjectPool::retrieveInstanceFor)
+ .map(privateFactory::create)
.findFirst();
}
}
diff --git a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMqMailQueueFactoryTest.java b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMqMailQueueFactoryTest.java
index a717476..5651d02 100644
--- a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMqMailQueueFactoryTest.java
+++ b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMqMailQueueFactoryTest.java
@@ -91,21 +91,4 @@ class RabbitMqMailQueueFactoryTest implements MailQueueFactoryContract<RabbitMQM
return mailQueueFactory;
}
- @Test
- void createQueueShouldReturnTheSameInstanceWhenParallelCreateSameQueueName() throws Exception {
- Set<RabbitMQMailQueue> createdRabbitMQMailQueues = ConcurrentHashMap.newKeySet();
-
- ConcurrentTestRunner.builder()
- .operation((threadNumber, operationNumber) ->
- createdRabbitMQMailQueues.add(mailQueueFactory.createQueue("spool")))
- .threadCount(100)
- .operationCount(10)
- .runSuccessfullyWithin(Duration.ofMinutes(10));
-
- assertThat(mailQueueFactory.listCreatedMailQueues())
- .hasSize(1)
- .isEqualTo(createdRabbitMQMailQueues)
- .extracting(RabbitMQMailQueue::getName)
- .hasOnlyOneElementSatisfying(queueName -> assertThat(queueName).isEqualTo("spool"));
- }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 03/04: JAMES-3070 strong type MailQueueName
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
matthieu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 850da1e5370066acb5691753a6ecd77f4eb7a457
Author: Matthieu Baechler <ma...@apache.org>
AuthorDate: Thu Feb 20 16:51:38 2020 +0100
JAMES-3070 strong type MailQueueName
---
.../mailets/MailReprocessingIntegrationTest.java | 8 +-
.../delivery/RemoteDeliveryConfiguration.java | 11 +-
.../delivery/RemoteDeliveryConfigurationTest.java | 2 +-
.../remote/delivery/RemoteDeliveryRunningTest.java | 5 +-
.../remote/delivery/RemoteDeliveryTest.java | 2 +-
.../SetMessagesOutboxFlagUpdateTest.java | 9 +-
.../apache/james/webadmin/dto/MailQueueDTO.java | 2 +-
.../james/webadmin/routes/MailQueueRoutes.java | 47 ++++++---
.../james/webadmin/service/ClearMailQueueTask.java | 27 +++--
...ClearMailQueueTaskAdditionalInformationDTO.java | 3 +-
.../webadmin/service/ClearMailQueueTaskDTO.java | 5 +-
.../service/DeleteMailsFromMailQueueTask.java | 40 +++++---
...sFromMailQueueTaskAdditionalInformationDTO.java | 3 +-
.../service/DeleteMailsFromMailQueueTaskDTO.java | 5 +-
.../james/webadmin/routes/MailQueueRoutesTest.java | 112 +++++++++++----------
.../webadmin/service/ClearMailQueueTaskTest.java | 9 +-
.../service/DeleteMailsFromMailQueueTaskTest.java | 8 +-
.../webadmin/routes/MailRepositoriesRoutes.java | 11 +-
.../webadmin/service/ReprocessingAllMailsTask.java | 11 +-
...essingAllMailsTaskAdditionalInformationDTO.java | 3 +-
.../service/ReprocessingAllMailsTaskDTO.java | 5 +-
.../webadmin/service/ReprocessingOneMailTask.java | 11 +-
...cessingOneMailTaskAdditionalInformationDTO.java | 3 +-
.../service/ReprocessingOneMailTaskDTO.java | 5 +-
.../webadmin/service/ReprocessingService.java | 9 +-
.../routes/MailRepositoriesRoutesTest.java | 35 +++----
.../service/ReprocessingAllMailsTaskTest.java | 3 +-
.../service/ReprocessingOneMailTaskTest.java | 3 +-
.../webadmin/service/ReprocessingServiceTest.java | 3 +-
.../queue/activemq/ActiveMQCacheableMailQueue.java | 9 +-
.../queue/activemq/ActiveMQMailQueueFactory.java | 3 +-
.../queue/activemq/ActiveMQMailQueueBlobTest.java | 3 +-
.../queue/activemq/ActiveMQMailQueueTest.java | 3 +-
.../java/org/apache/james/queue/api/MailQueue.java | 2 +-
.../apache/james/queue/api/MailQueueFactory.java | 8 +-
.../org/apache/james/queue/api/MailQueueName.java} | 48 ++++++---
.../james/queue/api/MailQueueFactoryContract.java | 4 +-
.../apache/james/queue/api/MailQueueNameTest.java} | 90 ++++++++---------
.../api/ManageableMailQueueFactoryContract.java | 2 +-
.../james/queue/file/FileCacheableMailQueue.java | 9 +-
.../james/queue/file/FileMailQueueFactory.java | 11 +-
.../queue/file/FileCacheableMailQueueTest.java | 3 +-
.../james/queue/jms/JMSCacheableMailQueue.java | 31 +++---
.../james/queue/jms/JMSMailQueueFactory.java | 3 +-
.../queue/library/AbstractMailQueueFactory.java | 17 ++--
.../apache/james/queue/jms/BrokerExtension.java | 5 +-
.../james/queue/jms/JMSCacheableMailQueueTest.java | 3 +-
.../library/AbstractMailQueueFactoryTest.java | 29 +++---
.../james/queue/memory/MemoryMailQueueFactory.java | 15 +--
.../queue/memory/MemoryCacheableMailQueueTest.java | 3 +-
.../james/queue/rabbitmq/RabbitMQMailQueue.java | 4 +-
.../queue/rabbitmq/RabbitMQMailQueueFactory.java | 13 +--
.../RabbitMQMailQueueConfigurationChangeTest.java | 3 +-
.../queue/rabbitmq/RabbitMQMailQueueTest.java | 6 +-
54 files changed, 414 insertions(+), 313 deletions(-)
diff --git a/server/mailet/integration-testing/src/test/java/org/apache/james/transport/mailets/MailReprocessingIntegrationTest.java b/server/mailet/integration-testing/src/test/java/org/apache/james/transport/mailets/MailReprocessingIntegrationTest.java
index cc6fc22..f7ba937 100644
--- a/server/mailet/integration-testing/src/test/java/org/apache/james/transport/mailets/MailReprocessingIntegrationTest.java
+++ b/server/mailet/integration-testing/src/test/java/org/apache/james/transport/mailets/MailReprocessingIntegrationTest.java
@@ -123,7 +123,7 @@ public class MailReprocessingIntegrationTest {
given()
.spec(specification)
.param("action", "reprocess")
- .param("queue", MailQueueFactory.SPOOL)
+ .param("queue", MailQueueFactory.SPOOL.asString())
.param("processor", TRANSPORT_PROCESSOR)
.patch("/mailRepositories/" + REPOSITORY_B.getPath().urlEncoded() + "/mails");
@@ -151,7 +151,7 @@ public class MailReprocessingIntegrationTest {
given()
.spec(specification)
.param("action", "reprocess")
- .param("queue", MailQueueFactory.SPOOL)
+ .param("queue", MailQueueFactory.SPOOL.asString())
.param("processor", TRANSPORT_PROCESSOR)
.patch("/mailRepositories/" + REPOSITORY_B.getPath().urlEncoded() + "/mails");
@@ -162,7 +162,7 @@ public class MailReprocessingIntegrationTest {
String taskId = given()
.spec(specification)
.param("action", "reprocess")
- .param("queue", MailQueueFactory.SPOOL)
+ .param("queue", MailQueueFactory.SPOOL.asString())
.patch("/mailRepositories/" + REPOSITORY_A.getPath().urlEncoded() + "/mails")
.jsonPath()
.get("taskId");
@@ -195,7 +195,7 @@ public class MailReprocessingIntegrationTest {
given()
.spec(specification)
.param("action", "reprocess")
- .param("queue", MailQueueFactory.SPOOL)
+ .param("queue", MailQueueFactory.SPOOL.asString())
.param("processor", "unknown")
.patch("/mailRepositories/" + REPOSITORY_B.getPath().urlEncoded() + "/mails");
diff --git a/server/mailet/mailets/src/main/java/org/apache/james/transport/mailets/remote/delivery/RemoteDeliveryConfiguration.java b/server/mailet/mailets/src/main/java/org/apache/james/transport/mailets/remote/delivery/RemoteDeliveryConfiguration.java
index fbe60fa..68ef607 100644
--- a/server/mailet/mailets/src/main/java/org/apache/james/transport/mailets/remote/delivery/RemoteDeliveryConfiguration.java
+++ b/server/mailet/mailets/src/main/java/org/apache/james/transport/mailets/remote/delivery/RemoteDeliveryConfiguration.java
@@ -27,6 +27,7 @@ import java.util.Properties;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.james.domainlist.api.DomainList;
+import org.apache.james.queue.api.MailQueueName;
import org.apache.mailet.MailetConfig;
import org.apache.mailet.base.MailetUtil;
import org.slf4j.Logger;
@@ -62,7 +63,7 @@ public class RemoteDeliveryConfiguration {
public static final String DELAY_TIME = "delayTime";
public static final String DEBUG = "debug";
public static final int DEFAULT_SMTP_TIMEOUT = 180000;
- public static final String DEFAULT_OUTGOING_QUEUE_NAME = "outgoing";
+ public static final MailQueueName DEFAULT_OUTGOING_QUEUE_NAME = MailQueueName.of("outgoing");
public static final int DEFAULT_CONNECTION_TIMEOUT = 60000;
public static final int DEFAULT_DNS_RETRY_PROBLEM = 0;
public static final int DEFAULT_MAX_RETRY = 5;
@@ -80,7 +81,7 @@ public class RemoteDeliveryConfiguration {
private final int connectionTimeout;
private final List<Duration> delayTimes;
private final HeloNameProvider heloNameProvider;
- private final String outGoingQueueName;
+ private final MailQueueName outGoingQueueName;
private final String bindAddress;
private final String bounceProcessor;
private final Collection<String> gatewayServer;
@@ -94,7 +95,9 @@ public class RemoteDeliveryConfiguration {
isSSLEnable = MailetUtil.getInitParameter(mailetConfig, SSL_ENABLE).orElse(false);
usePriority = MailetUtil.getInitParameter(mailetConfig, USE_PRIORITY).orElse(false);
sendPartial = MailetUtil.getInitParameter(mailetConfig, SENDPARTIAL).orElse(false);
- outGoingQueueName = Optional.ofNullable(mailetConfig.getInitParameter(OUTGOING)).orElse(DEFAULT_OUTGOING_QUEUE_NAME);
+ outGoingQueueName = Optional.ofNullable(mailetConfig.getInitParameter(OUTGOING))
+ .map(MailQueueName::of)
+ .orElse(DEFAULT_OUTGOING_QUEUE_NAME);
bounceProcessor = mailetConfig.getInitParameter(BOUNCE_PROCESSOR);
bindAddress = mailetConfig.getInitParameter(BIND);
@@ -293,7 +296,7 @@ public class RemoteDeliveryConfiguration {
return heloNameProvider;
}
- public String getOutGoingQueueName() {
+ public MailQueueName getOutGoingQueueName() {
return outGoingQueueName;
}
diff --git a/server/mailet/mailets/src/test/java/org/apache/james/transport/mailets/remote/delivery/RemoteDeliveryConfigurationTest.java b/server/mailet/mailets/src/test/java/org/apache/james/transport/mailets/remote/delivery/RemoteDeliveryConfigurationTest.java
index cd70d4c..64e38e9 100644
--- a/server/mailet/mailets/src/test/java/org/apache/james/transport/mailets/remote/delivery/RemoteDeliveryConfigurationTest.java
+++ b/server/mailet/mailets/src/test/java/org/apache/james/transport/mailets/remote/delivery/RemoteDeliveryConfigurationTest.java
@@ -140,7 +140,7 @@ public class RemoteDeliveryConfigurationTest {
.setProperty(RemoteDeliveryConfiguration.OUTGOING, value)
.build();
- assertThat(new RemoteDeliveryConfiguration(mailetConfig, mock(DomainList.class)).getOutGoingQueueName())
+ assertThat(new RemoteDeliveryConfiguration(mailetConfig, mock(DomainList.class)).getOutGoingQueueName().asString())
.isEqualTo(value);
}
diff --git a/server/mailet/mailets/src/test/java/org/apache/james/transport/mailets/remote/delivery/RemoteDeliveryRunningTest.java b/server/mailet/mailets/src/test/java/org/apache/james/transport/mailets/remote/delivery/RemoteDeliveryRunningTest.java
index 596eb13..b059efa 100644
--- a/server/mailet/mailets/src/test/java/org/apache/james/transport/mailets/remote/delivery/RemoteDeliveryRunningTest.java
+++ b/server/mailet/mailets/src/test/java/org/apache/james/transport/mailets/remote/delivery/RemoteDeliveryRunningTest.java
@@ -31,6 +31,7 @@ import org.apache.james.domainlist.api.DomainList;
import org.apache.james.metrics.api.MetricFactory;
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.transport.mailets.RemoteDelivery;
import org.apache.mailet.base.test.FakeMailetConfig;
import org.junit.After;
@@ -40,7 +41,7 @@ import org.junit.Test;
import reactor.core.publisher.Flux;
public class RemoteDeliveryRunningTest {
- private static final String QUEUE_NAME = "queueName";
+ private static final MailQueueName QUEUE_NAME = MailQueueName.of("queueName");
private RemoteDelivery remoteDelivery;
private MailQueue mailQueue;
@@ -66,7 +67,7 @@ public class RemoteDeliveryRunningTest {
return Flux.never();
});
remoteDelivery.init(FakeMailetConfig.builder()
- .setProperty(RemoteDeliveryConfiguration.OUTGOING, QUEUE_NAME)
+ .setProperty(RemoteDeliveryConfiguration.OUTGOING, QUEUE_NAME.asString())
.setProperty(RemoteDeliveryConfiguration.HELO_NAME, "Hello_name")
.build());
diff --git a/server/mailet/mailets/src/test/java/org/apache/james/transport/mailets/remote/delivery/RemoteDeliveryTest.java b/server/mailet/mailets/src/test/java/org/apache/james/transport/mailets/remote/delivery/RemoteDeliveryTest.java
index 4c7a87e..1e62160 100644
--- a/server/mailet/mailets/src/test/java/org/apache/james/transport/mailets/remote/delivery/RemoteDeliveryTest.java
+++ b/server/mailet/mailets/src/test/java/org/apache/james/transport/mailets/remote/delivery/RemoteDeliveryTest.java
@@ -100,7 +100,7 @@ public class RemoteDeliveryTest {
@Before
public void setUp() throws ConfigurationException {
MailQueueFactory<ManageableMailQueue> queueFactory = new MemoryMailQueueFactory(new RawMailQueueItemDecoratorFactory());
- mailQueue = queueFactory.createQueue(RemoteDeliveryConfiguration.OUTGOING);
+ mailQueue = queueFactory.createQueue(RemoteDeliveryConfiguration.DEFAULT_OUTGOING_QUEUE_NAME);
DNSService dnsService = mock(DNSService.class);
MemoryDomainList domainList = new MemoryDomainList(dnsService);
domainList.configure(DomainListConfiguration.builder().defaultDomain(JAMES_APACHE_ORG_DOMAIN));
diff --git a/server/protocols/jmap-draft-integration-testing/jmap-draft-integration-testing-common/src/test/java/org/apache/james/jmap/draft/methods/integration/SetMessagesOutboxFlagUpdateTest.java b/server/protocols/jmap-draft-integration-testing/jmap-draft-integration-testing-common/src/test/java/org/apache/james/jmap/draft/methods/integration/SetMessagesOutboxFlagUpdateTest.java
index 00dcf07..0f82362 100644
--- a/server/protocols/jmap-draft-integration-testing/jmap-draft-integration-testing-common/src/test/java/org/apache/james/jmap/draft/methods/integration/SetMessagesOutboxFlagUpdateTest.java
+++ b/server/protocols/jmap-draft-integration-testing/jmap-draft-integration-testing-common/src/test/java/org/apache/james/jmap/draft/methods/integration/SetMessagesOutboxFlagUpdateTest.java
@@ -49,6 +49,7 @@ import org.apache.james.modules.MailboxProbeImpl;
import org.apache.james.probe.DataProbe;
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.utils.DataProbeImpl;
import org.apache.mailet.Mail;
import org.junit.After;
@@ -72,19 +73,19 @@ public abstract class SetMessagesOutboxFlagUpdateTest {
protected MailQueueFactory<MailQueue> noopMailQueueFactory = new MailQueueFactory<MailQueue>() {
@Override
- public Optional<MailQueue> getQueue(String name) {
+ public Optional<MailQueue> getQueue(MailQueueName name) {
return Optional.of(createQueue(name));
}
@Override
- public MailQueue createQueue(String name) {
+ public MailQueue createQueue(MailQueueName name) {
return new MailQueue() {
@Override
public void close() throws IOException {
}
@Override
- public String getName() {
+ public MailQueueName getName() {
return name;
}
@@ -105,7 +106,7 @@ public abstract class SetMessagesOutboxFlagUpdateTest {
}
@Override
- public Set<String> listCreatedMailQueues() {
+ public Set<MailQueueName> listCreatedMailQueues() {
throw new NotImplementedException("Minimalistic implementation. Please do not list queues");
}
};
diff --git a/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/dto/MailQueueDTO.java b/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/dto/MailQueueDTO.java
index 39198d0..c2301dc 100644
--- a/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/dto/MailQueueDTO.java
+++ b/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/dto/MailQueueDTO.java
@@ -33,7 +33,7 @@ public class MailQueueDTO {
public static MailQueueDTO from(ManageableMailQueue mailQueue) throws MailQueueException {
return builder()
- .name(mailQueue.getName())
+ .name(mailQueue.getName().asString())
.size(mailQueue.getSize())
.build();
}
diff --git a/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/routes/MailQueueRoutes.java b/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/routes/MailQueueRoutes.java
index 1764b6a..1bb138a 100644
--- a/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/routes/MailQueueRoutes.java
+++ b/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/routes/MailQueueRoutes.java
@@ -21,6 +21,7 @@ package org.apache.james.webadmin.routes;
import static org.apache.james.webadmin.Constants.SEPARATOR;
+import java.io.Closeable;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.time.ZonedDateTime;
@@ -37,6 +38,7 @@ import org.apache.james.core.MailAddress;
import org.apache.james.queue.api.MailQueue;
import org.apache.james.queue.api.MailQueue.MailQueueException;
import org.apache.james.queue.api.MailQueueFactory;
+import org.apache.james.queue.api.MailQueueName;
import org.apache.james.queue.api.ManageableMailQueue;
import org.apache.james.task.Task;
import org.apache.james.task.TaskManager;
@@ -135,7 +137,10 @@ public class MailQueueRoutes implements Routes {
})
public void defineListQueues(Service service) {
service.get(BASE_URL,
- (request, response) -> mailQueueFactory.listCreatedMailQueues(),
+ (request, response) -> mailQueueFactory.listCreatedMailQueues()
+ .stream()
+ .map(MailQueueName::asString)
+ .collect(Guavate.toImmutableList()),
jsonTransformer);
}
@@ -160,7 +165,7 @@ public class MailQueueRoutes implements Routes {
}
private MailQueueDTO getMailQueue(Request request) {
- String mailQueueName = request.params(MAIL_QUEUE_NAME);
+ MailQueueName mailQueueName = MailQueueName.of(request.params(MAIL_QUEUE_NAME));
return mailQueueFactory.getQueue(mailQueueName).map(this::toDTO)
.orElseThrow(
() -> ErrorResponder.builder()
@@ -219,7 +224,7 @@ public class MailQueueRoutes implements Routes {
}
private List<MailQueueItemDTO> listMails(Request request) {
- String mailQueueName = request.params(MAIL_QUEUE_NAME);
+ MailQueueName mailQueueName = MailQueueName.of(request.params(MAIL_QUEUE_NAME));
return mailQueueFactory.getQueue(mailQueueName)
.map(queue -> listMails(queue, isDelayed(request.queryParams(DELAYED_QUERY_PARAM)), ParametersExtractor.extractLimit(request)))
.orElseThrow(
@@ -303,7 +308,8 @@ public class MailQueueRoutes implements Routes {
}
private Task deleteMails(Request request) {
- String mailQueueName = request.params(MAIL_QUEUE_NAME);
+ MailQueueName mailQueueName = MailQueueName.of(request.params(MAIL_QUEUE_NAME));
+ checkQueueExists(mailQueueName);
return deleteMailsTask(mailQueueName,
sender(request.queryParams(SENDER_QUERY_PARAM)),
name(request.queryParams(NAME_QUERY_PARAM)),
@@ -385,7 +391,7 @@ public class MailQueueRoutes implements Routes {
}
private ManageableMailQueue assertMailQueueExists(Request request) {
- String mailQueueName = request.params(MAIL_QUEUE_NAME);
+ MailQueueName mailQueueName = MailQueueName.of(request.params(MAIL_QUEUE_NAME));
return mailQueueFactory.getQueue(mailQueueName)
.orElseThrow(() -> ErrorResponder.builder()
.message("%s can not be found", mailQueueName)
@@ -415,13 +421,13 @@ public class MailQueueRoutes implements Routes {
}
}
- private Task deleteMailsTask(String queueName, Optional<MailAddress> maybeSender, Optional<String> maybeName, Optional<MailAddress> maybeRecipient) {
+ private Task deleteMailsTask(MailQueueName queueName, Optional<MailAddress> maybeSender, Optional<String> maybeName, Optional<MailAddress> maybeRecipient) {
int paramCount = Booleans.countTrue(maybeSender.isPresent(), maybeName.isPresent(), maybeRecipient.isPresent());
switch (paramCount) {
case 0:
- return new ClearMailQueueTask(queueName, this::getQueueOrThrow);
+ return new ClearMailQueueTask(queueName, this::getQueue);
case 1:
- return new DeleteMailsFromMailQueueTask(queueName, this::getQueueOrThrow, maybeSender, maybeName, maybeRecipient);
+ return new DeleteMailsFromMailQueueTask(queueName, this::getQueue, maybeSender, maybeName, maybeRecipient);
default:
throw ErrorResponder.builder()
.statusCode(HttpStatus.BAD_REQUEST_400)
@@ -432,14 +438,23 @@ public class MailQueueRoutes implements Routes {
}
}
- private ManageableMailQueue getQueueOrThrow(String queueName) {
- return mailQueueFactory.getQueue(queueName)
- .orElseThrow(
- () -> ErrorResponder.builder()
- .message("%s can not be found", queueName)
- .statusCode(HttpStatus.NOT_FOUND_404)
- .type(ErrorResponder.ErrorType.NOT_FOUND)
- .haltError());
+ private ManageableMailQueue getQueue(MailQueueName queueName) throws MailQueueException {
+ return mailQueueFactory.getQueue(queueName).orElseThrow(() -> new MailQueueException("unable to find queue " + queueName.asString()));
+ }
+
+ private ManageableMailQueue checkQueueExists(MailQueueName queueName) {
+ Optional<? extends ManageableMailQueue> queue = mailQueueFactory.getQueue(queueName);
+ try {
+ return queue
+ .orElseThrow(
+ () -> ErrorResponder.builder()
+ .message("%s can not be found", queueName)
+ .statusCode(HttpStatus.NOT_FOUND_404)
+ .type(ErrorResponder.ErrorType.NOT_FOUND)
+ .haltError());
+ } finally {
+ queue.ifPresent(Throwing.consumer(Closeable::close));
+ }
}
private void assertDelayedParamIsTrue(Request request) {
diff --git a/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/ClearMailQueueTask.java b/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/ClearMailQueueTask.java
index ab9e25b..96a60be 100644
--- a/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/ClearMailQueueTask.java
+++ b/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/ClearMailQueueTask.java
@@ -25,10 +25,12 @@ import java.time.Instant;
import java.util.Optional;
import org.apache.james.queue.api.MailQueue;
+import org.apache.james.queue.api.MailQueueName;
import org.apache.james.queue.api.ManageableMailQueue;
import org.apache.james.task.Task;
import org.apache.james.task.TaskExecutionDetails;
import org.apache.james.task.TaskType;
+import org.apache.james.util.OptionalUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -38,12 +40,12 @@ public class ClearMailQueueTask implements Task {
public static class AdditionalInformation implements TaskExecutionDetails.AdditionalInformation {
- private final String mailQueueName;
+ private final MailQueueName mailQueueName;
private final long initialCount;
private final long remainingCount;
private final Instant timestamp;
- public AdditionalInformation(String mailQueueName, long initialCount, long remainingCount, Instant timestamp) {
+ public AdditionalInformation(MailQueueName mailQueueName, long initialCount, long remainingCount, Instant timestamp) {
this.mailQueueName = mailQueueName;
this.initialCount = initialCount;
this.remainingCount = remainingCount;
@@ -51,7 +53,7 @@ public class ClearMailQueueTask implements Task {
}
public String getMailQueueName() {
- return mailQueueName;
+ return mailQueueName.asString();
}
public long getInitialCount() {
@@ -78,22 +80,24 @@ public class ClearMailQueueTask implements Task {
@FunctionalInterface
public interface MailQueueFactory {
- ManageableMailQueue create(String mailQueueName);
+ ManageableMailQueue create(MailQueueName mailQueueName) throws MailQueue.MailQueueException;
}
private static final Logger LOGGER = LoggerFactory.getLogger(ClearMailQueueTask.class);
public static final TaskType TYPE = TaskType.of("clear-mail-queue");
- private final String queueName;
+ private final MailQueueName queueName;
private final ClearMailQueueTask.MailQueueFactory factory;
private Optional<Long> initialCount;
private Optional<ManageableMailQueue> queue;
+ private Optional<TaskExecutionDetails.AdditionalInformation> lastAdditionalInformation;
- public ClearMailQueueTask(String queueName, ClearMailQueueTask.MailQueueFactory factory) {
+ public ClearMailQueueTask(MailQueueName queueName, ClearMailQueueTask.MailQueueFactory factory) {
this.queueName = queueName;
this.factory = factory;
this.initialCount = Optional.empty();
this.queue = Optional.empty();
+ this.lastAdditionalInformation = Optional.empty();
}
@Override
@@ -102,10 +106,12 @@ public class ClearMailQueueTask implements Task {
this.initialCount = Optional.of(getRemainingSize(queue));
this.queue = Optional.of(queue);
queue.clear();
- this.queue = Optional.empty();
+ this.lastAdditionalInformation = details();
} catch (MailQueue.MailQueueException | IOException e) {
LOGGER.error("Clear MailQueue got an exception", e);
return Result.PARTIAL;
+ } finally {
+ this.queue = Optional.empty();
}
return Result.COMPLETED;
@@ -118,10 +124,13 @@ public class ClearMailQueueTask implements Task {
@Override
public Optional<TaskExecutionDetails.AdditionalInformation> details() {
- return queue.map(q -> new AdditionalInformation(queueName, initialCount.get(), getRemainingSize(q), Clock.systemUTC().instant()));
+ return
+ OptionalUtils.orSuppliers(
+ () -> lastAdditionalInformation,
+ () -> queue.map(q -> new AdditionalInformation(queueName, initialCount.get(), getRemainingSize(q), Clock.systemUTC().instant())));
}
- String getQueueName() {
+ MailQueueName getQueueName() {
return queueName;
}
diff --git a/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/ClearMailQueueTaskAdditionalInformationDTO.java b/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/ClearMailQueueTaskAdditionalInformationDTO.java
index 94763a1..9cbc1e9 100644
--- a/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/ClearMailQueueTaskAdditionalInformationDTO.java
+++ b/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/ClearMailQueueTaskAdditionalInformationDTO.java
@@ -21,6 +21,7 @@ package org.apache.james.webadmin.service;
import java.time.Instant;
import org.apache.james.json.DTOModule;
+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;
@@ -32,7 +33,7 @@ public class ClearMailQueueTaskAdditionalInformationDTO implements AdditionalInf
DTOModule.forDomainObject(ClearMailQueueTask.AdditionalInformation.class)
.convertToDTO(ClearMailQueueTaskAdditionalInformationDTO.class)
.toDomainObjectConverter(dto -> new ClearMailQueueTask.AdditionalInformation(
- dto.mailQueueName,
+ MailQueueName.of(dto.mailQueueName),
dto.initialCount,
dto.remainingCount,
dto.timestamp
diff --git a/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/ClearMailQueueTaskDTO.java b/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/ClearMailQueueTaskDTO.java
index f4af298..295a0a0 100644
--- a/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/ClearMailQueueTaskDTO.java
+++ b/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/ClearMailQueueTaskDTO.java
@@ -20,6 +20,7 @@ package org.apache.james.webadmin.service;
import org.apache.james.json.DTOModule;
import org.apache.james.queue.api.MailQueueFactory;
+import org.apache.james.queue.api.MailQueueName;
import org.apache.james.queue.api.ManageableMailQueue;
import org.apache.james.server.task.json.dto.TaskDTO;
import org.apache.james.server.task.json.dto.TaskDTOModule;
@@ -39,7 +40,7 @@ public class ClearMailQueueTaskDTO implements TaskDTO {
}
public static ClearMailQueueTaskDTO toDTO(ClearMailQueueTask domainObject, String typeName) {
- return new ClearMailQueueTaskDTO(typeName, domainObject.getQueueName());
+ return new ClearMailQueueTaskDTO(typeName, domainObject.getQueueName().asString());
}
private final String type;
@@ -51,7 +52,7 @@ public class ClearMailQueueTaskDTO implements TaskDTO {
}
public ClearMailQueueTask fromDTO(MailQueueFactory<? extends ManageableMailQueue> mailQueueFactory) {
- return new ClearMailQueueTask(queueName,
+ return new ClearMailQueueTask(MailQueueName.of(queueName),
name -> mailQueueFactory
.getQueue(name)
.orElseThrow(() -> new ClearMailQueueTask.UnknownSerializedQueue(queueName)));
diff --git a/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/DeleteMailsFromMailQueueTask.java b/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/DeleteMailsFromMailQueueTask.java
index 0a1a184..a5abf75 100644
--- a/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/DeleteMailsFromMailQueueTask.java
+++ b/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/DeleteMailsFromMailQueueTask.java
@@ -20,17 +20,18 @@
package org.apache.james.webadmin.service;
import java.io.IOException;
-import java.io.UncheckedIOException;
import java.time.Clock;
import java.time.Instant;
import java.util.Optional;
import org.apache.james.core.MailAddress;
import org.apache.james.queue.api.MailQueue;
+import org.apache.james.queue.api.MailQueueName;
import org.apache.james.queue.api.ManageableMailQueue;
import org.apache.james.task.Task;
import org.apache.james.task.TaskExecutionDetails;
import org.apache.james.task.TaskType;
+import org.apache.james.util.OptionalUtils;
import com.github.fge.lambdas.Throwing;
import com.google.common.base.Preconditions;
@@ -39,7 +40,7 @@ import com.google.common.primitives.Booleans;
public class DeleteMailsFromMailQueueTask implements Task {
public static class AdditionalInformation implements TaskExecutionDetails.AdditionalInformation {
- private final String mailQueueName;
+ private final MailQueueName mailQueueName;
private final long remainingCount;
private final long initialCount;
@@ -48,7 +49,7 @@ public class DeleteMailsFromMailQueueTask implements Task {
private final Optional<String> recipient;
private final Instant timestamp;
- public AdditionalInformation(String mailQueueName, long initialCount, long remainingCount,
+ public AdditionalInformation(MailQueueName mailQueueName, long initialCount, long remainingCount,
Optional<MailAddress> maybeSender, Optional<String> maybeName,
Optional<MailAddress> maybeRecipient, Instant timestamp) {
this.mailQueueName = mailQueueName;
@@ -62,7 +63,7 @@ public class DeleteMailsFromMailQueueTask implements Task {
}
public String getMailQueueName() {
- return mailQueueName;
+ return mailQueueName.asString();
}
public long getRemainingCount() {
@@ -99,7 +100,7 @@ public class DeleteMailsFromMailQueueTask implements Task {
@FunctionalInterface
public interface MailQueueFactory {
- ManageableMailQueue create(String mailQueueName);
+ ManageableMailQueue create(MailQueueName mailQueueName) throws MailQueue.MailQueueException;
}
public static final TaskType TYPE = TaskType.of("delete-mails-from-mail-queue");
@@ -108,11 +109,12 @@ public class DeleteMailsFromMailQueueTask implements Task {
private final Optional<String> maybeName;
private final Optional<MailAddress> maybeRecipient;
private final MailQueueFactory factory;
- private final String queueName;
+ private final MailQueueName queueName;
private Optional<Long> initialCount;
private Optional<ManageableMailQueue> queue;
+ private Optional<TaskExecutionDetails.AdditionalInformation> lastAdditionalInformation;
- public DeleteMailsFromMailQueueTask(String queueName, MailQueueFactory factory,
+ public DeleteMailsFromMailQueueTask(MailQueueName queueName, MailQueueFactory factory,
Optional<MailAddress> maybeSender,
Optional<String> maybeName, Optional<MailAddress> maybeRecipient) {
this.factory = factory;
@@ -126,6 +128,7 @@ public class DeleteMailsFromMailQueueTask implements Task {
this.maybeRecipient = maybeRecipient;
this.initialCount = Optional.empty();
this.queue = Optional.empty();
+ this.lastAdditionalInformation = Optional.empty();
}
@@ -141,15 +144,18 @@ public class DeleteMailsFromMailQueueTask implements Task {
maybeRecipient.ifPresent(Throwing.consumer(
(MailAddress recipient) -> queue.remove(ManageableMailQueue.Type.Recipient, recipient.asString())));
- this.queue = Optional.empty();
+ this.lastAdditionalInformation = details();
return Result.COMPLETED;
- } catch (IOException e) {
- throw new UncheckedIOException(e);
+ } catch (IOException | MailQueue.MailQueueException e) {
+ LOGGER.error("Delete mails from MailQueue got an exception", e);
+ return Result.PARTIAL;
+ } finally {
+ this.queue = Optional.empty();
}
}
- public String getQueueName() {
+ public MailQueueName getQueueName() {
return queueName;
}
@@ -172,10 +178,14 @@ public class DeleteMailsFromMailQueueTask implements Task {
@Override
public Optional<TaskExecutionDetails.AdditionalInformation> details() {
- return this.queue.map(queue -> new AdditionalInformation(queueName,
- initialCount.get(),
- getRemainingSize(queue), maybeSender,
- maybeName, maybeRecipient, Clock.systemUTC().instant()));
+ return OptionalUtils.orSuppliers(
+ () -> this.lastAdditionalInformation,
+ () -> this.queue.map(queue ->
+ new AdditionalInformation(
+ queueName,
+ initialCount.get(),
+ getRemainingSize(queue), maybeSender,
+ maybeName, maybeRecipient, Clock.systemUTC().instant())));
}
public long getRemainingSize(ManageableMailQueue queue) {
diff --git a/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/DeleteMailsFromMailQueueTaskAdditionalInformationDTO.java b/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/DeleteMailsFromMailQueueTaskAdditionalInformationDTO.java
index f4c2928..a35c65f 100644
--- a/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/DeleteMailsFromMailQueueTaskAdditionalInformationDTO.java
+++ b/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/DeleteMailsFromMailQueueTaskAdditionalInformationDTO.java
@@ -5,6 +5,7 @@ import java.util.Optional;
import org.apache.james.core.MailAddress;
import org.apache.james.json.DTOModule;
+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;
@@ -36,7 +37,7 @@ public class DeleteMailsFromMailQueueTaskAdditionalInformationDTO implements Add
private static DeleteMailsFromMailQueueTask.AdditionalInformation fromDTO(DeleteMailsFromMailQueueTaskAdditionalInformationDTO dto) {
return new DeleteMailsFromMailQueueTask.AdditionalInformation(
- dto.getQueue(),
+ MailQueueName.of(dto.getQueue()),
dto.getInitialCount(),
dto.getRemainingCount(),
dto.sender.map(Throwing.<String, MailAddress>function(MailAddress::new).sneakyThrow()),
diff --git a/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/DeleteMailsFromMailQueueTaskDTO.java b/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/DeleteMailsFromMailQueueTaskDTO.java
index 12f3485..fc30ab9 100644
--- a/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/DeleteMailsFromMailQueueTaskDTO.java
+++ b/server/protocols/webadmin/webadmin-mailqueue/src/main/java/org/apache/james/webadmin/service/DeleteMailsFromMailQueueTaskDTO.java
@@ -5,6 +5,7 @@ import java.util.Optional;
import org.apache.james.core.MailAddress;
import org.apache.james.json.DTOModule;
import org.apache.james.queue.api.MailQueueFactory;
+import org.apache.james.queue.api.MailQueueName;
import org.apache.james.queue.api.ManageableMailQueue;
import org.apache.james.server.task.json.dto.TaskDTO;
import org.apache.james.server.task.json.dto.TaskDTOModule;
@@ -27,7 +28,7 @@ public class DeleteMailsFromMailQueueTaskDTO implements TaskDTO {
public static DeleteMailsFromMailQueueTaskDTO toDTO(DeleteMailsFromMailQueueTask domainObject, String typeName) {
return new DeleteMailsFromMailQueueTaskDTO(
typeName,
- domainObject.getQueueName(),
+ domainObject.getQueueName().asString(),
domainObject.getMaybeSender().map(MailAddress::asString),
domainObject.getMaybeName(),
domainObject.getMaybeRecipient().map(MailAddress::asString)
@@ -54,7 +55,7 @@ public class DeleteMailsFromMailQueueTaskDTO implements TaskDTO {
public DeleteMailsFromMailQueueTask fromDTO(MailQueueFactory<? extends ManageableMailQueue> mailQueueFactory) {
return new DeleteMailsFromMailQueueTask(
- queueName, name -> mailQueueFactory.getQueue(name).orElseThrow(() -> new DeleteMailsFromMailQueueTask.UnknownSerializedQueue(queueName)),
+ MailQueueName.of(queueName), name -> mailQueueFactory.getQueue(name).orElseThrow(() -> new DeleteMailsFromMailQueueTask.UnknownSerializedQueue(queueName)),
sender.map(Throwing.<String, MailAddress>function(MailAddress::new).sneakyThrow()),
name,
recipient.map(Throwing.<String, MailAddress>function(MailAddress::new).sneakyThrow())
diff --git a/server/protocols/webadmin/webadmin-mailqueue/src/test/java/org/apache/james/webadmin/routes/MailQueueRoutesTest.java b/server/protocols/webadmin/webadmin-mailqueue/src/test/java/org/apache/james/webadmin/routes/MailQueueRoutesTest.java
index f40f1ac..6ba4b07 100644
--- a/server/protocols/webadmin/webadmin-mailqueue/src/test/java/org/apache/james/webadmin/routes/MailQueueRoutesTest.java
+++ b/server/protocols/webadmin/webadmin-mailqueue/src/test/java/org/apache/james/webadmin/routes/MailQueueRoutesTest.java
@@ -38,6 +38,7 @@ import java.util.List;
import java.util.concurrent.TimeUnit;
import org.apache.james.core.MailAddress;
+import org.apache.james.queue.api.MailQueueName;
import org.apache.james.queue.api.Mails;
import org.apache.james.queue.api.ManageableMailQueue;
import org.apache.james.queue.api.RawMailQueueItemDecoratorFactory;
@@ -67,10 +68,10 @@ import io.restassured.specification.RequestSpecification;
class MailQueueRoutesTest {
- static final String FIRST_QUEUE = "first one";
- static final String SECOND_QUEUE = "second one";
- static final String THIRD_QUEUE = "third one";
- static final String FOURTH_QUEUE = "fourth one";
+ static final MailQueueName FIRST_QUEUE = MailQueueName.of("first one");
+ static final MailQueueName SECOND_QUEUE = MailQueueName.of("second one");
+ static final MailQueueName THIRD_QUEUE = MailQueueName.of("third one");
+ static final MailQueueName FOURTH_QUEUE = MailQueueName.of("fourth one");
static final String SENDER_1_JAMES_ORG = "sender1@james.org";
static final String SENDER_2_JAMES_ORG = "sender2@james.org";
static final String RECIPIENT_JAMES_ORG = "recipient@james.org";
@@ -129,7 +130,7 @@ class MailQueueRoutesTest {
given()
.param("limit", "-1")
.when()
- .get(FIRST_QUEUE + "/mails")
+ .get(FIRST_QUEUE.asString() + "/mails")
.then()
.statusCode(HttpStatus.BAD_REQUEST_400);
}
@@ -141,7 +142,7 @@ class MailQueueRoutesTest {
given()
.param("limit", "0")
.when()
- .get(FIRST_QUEUE + "/mails")
+ .get(FIRST_QUEUE.asString() + "/mails")
.then()
.statusCode(HttpStatus.BAD_REQUEST_400);
}
@@ -153,7 +154,7 @@ class MailQueueRoutesTest {
given()
.param("limit", "abc")
.when()
- .get(FIRST_QUEUE + "/mails")
+ .get(FIRST_QUEUE.asString() + "/mails")
.then()
.statusCode(HttpStatus.BAD_REQUEST_400);
}
@@ -191,7 +192,7 @@ class MailQueueRoutesTest {
.jsonPath()
.getList(".");
- assertThat(actual).containsOnly(FIRST_QUEUE);
+ assertThat(actual).containsOnly(FIRST_QUEUE.asString());
}
@Test
@@ -211,7 +212,12 @@ class MailQueueRoutesTest {
.jsonPath()
.getList(".");
- assertThat(actual).containsOnly(FIRST_QUEUE, SECOND_QUEUE, THIRD_QUEUE, FOURTH_QUEUE);
+ assertThat(actual)
+ .containsOnly(
+ FIRST_QUEUE.asString(),
+ SECOND_QUEUE.asString(),
+ THIRD_QUEUE.asString(),
+ FOURTH_QUEUE.asString());
}
@Test
@@ -219,7 +225,7 @@ class MailQueueRoutesTest {
mailQueueFactory.createQueue(FIRST_QUEUE);
when()
- .get(FIRST_QUEUE + "/mails")
+ .get(FIRST_QUEUE.asString() + "/mails")
.then()
.statusCode(HttpStatus.OK_200)
.contentType(ContentType.JSON)
@@ -233,7 +239,7 @@ class MailQueueRoutesTest {
queue.enQueue(Mails.defaultMail().name("name").build());
when()
- .get(FIRST_QUEUE + "/mails")
+ .get(FIRST_QUEUE.asString() + "/mails")
.then()
.statusCode(HttpStatus.OK_200)
.contentType(ContentType.JSON)
@@ -252,7 +258,7 @@ class MailQueueRoutesTest {
.collect(Guavate.toImmutableList());
when()
- .get(FIRST_QUEUE + "/mails")
+ .get(FIRST_QUEUE.asString() + "/mails")
.then()
.statusCode(HttpStatus.OK_200)
.contentType(ContentType.JSON)
@@ -271,7 +277,7 @@ class MailQueueRoutesTest {
given()
.param("delayed", "true")
.when()
- .get(FIRST_QUEUE + "/mails")
+ .get(FIRST_QUEUE.asString() + "/mails")
.then()
.statusCode(HttpStatus.OK_200)
.contentType(ContentType.JSON)
@@ -287,7 +293,7 @@ class MailQueueRoutesTest {
given()
.param("delayed", "false")
.when()
- .get(FIRST_QUEUE + "/mails")
+ .get(FIRST_QUEUE.asString() + "/mails")
.then()
.statusCode(HttpStatus.OK_200)
.contentType(ContentType.JSON)
@@ -303,7 +309,7 @@ class MailQueueRoutesTest {
given()
.param("delayed", "true")
.when()
- .get(FIRST_QUEUE + "/mails")
+ .get(FIRST_QUEUE.asString() + "/mails")
.then()
.statusCode(HttpStatus.OK_200)
.contentType(ContentType.JSON)
@@ -321,7 +327,7 @@ class MailQueueRoutesTest {
given()
.param("limit", "1")
.when()
- .get(FIRST_QUEUE + "/mails")
+ .get(FIRST_QUEUE.asString() + "/mails")
.then()
.statusCode(HttpStatus.OK_200)
.contentType(ContentType.JSON)
@@ -339,17 +345,17 @@ class MailQueueRoutesTest {
queue.enQueue(Mails.defaultMail().name("name").build());
when()
- .get(FIRST_QUEUE)
+ .get(FIRST_QUEUE.asString())
.then()
.statusCode(HttpStatus.OK_200)
- .body("name", equalTo(FIRST_QUEUE))
+ .body("name", equalTo(FIRST_QUEUE.asString()))
.body("size", equalTo(1));
}
@Test
void getMailQueueShouldReturnNotFoundWhenMailQueueDoesntExist() {
when()
- .get(FIRST_QUEUE)
+ .get(FIRST_QUEUE.asString())
.then()
.statusCode(HttpStatus.NOT_FOUND_404);
}
@@ -369,7 +375,7 @@ class MailQueueRoutesTest {
.queryParam("delayed", "true")
.body("{\"delayed\": \"false\"}")
.when()
- .patch(FIRST_QUEUE + "/mails")
+ .patch(FIRST_QUEUE.asString() + "/mails")
.then()
.statusCode(HttpStatus.NO_CONTENT_204);
}
@@ -392,7 +398,7 @@ class MailQueueRoutesTest {
given()
.body("{\"delayed\": \"false\"}")
.when()
- .patch(FIRST_QUEUE + "/mails")
+ .patch(FIRST_QUEUE.asString() + "/mails")
.then()
.statusCode(HttpStatus.BAD_REQUEST_400);
}
@@ -405,7 +411,7 @@ class MailQueueRoutesTest {
.queryParam("delayed", "false")
.body("{\"delayed\": \"false\"}")
.when()
- .patch(FIRST_QUEUE + "/mails")
+ .patch(FIRST_QUEUE.asString() + "/mails")
.then()
.statusCode(HttpStatus.BAD_REQUEST_400);
}
@@ -418,7 +424,7 @@ class MailQueueRoutesTest {
.queryParam("delayed", "wrong")
.body("{\"delayed\": \"false\"}")
.when()
- .patch(FIRST_QUEUE + "/mails")
+ .patch(FIRST_QUEUE.asString() + "/mails")
.then()
.statusCode(HttpStatus.BAD_REQUEST_400);
}
@@ -430,7 +436,7 @@ class MailQueueRoutesTest {
given()
.queryParam("delayed", "true")
.when()
- .patch(FIRST_QUEUE + "/mails")
+ .patch(FIRST_QUEUE.asString() + "/mails")
.then()
.statusCode(HttpStatus.BAD_REQUEST_400);
}
@@ -443,7 +449,7 @@ class MailQueueRoutesTest {
.queryParam("delayed", "true")
.body("{\"xx\": \"false\"}")
.when()
- .patch(FIRST_QUEUE + "/mails")
+ .patch(FIRST_QUEUE.asString() + "/mails")
.then()
.statusCode(HttpStatus.BAD_REQUEST_400);
}
@@ -459,7 +465,7 @@ class MailQueueRoutesTest {
"\"delayed\": \"false\"" +
"}")
.when()
- .patch(FIRST_QUEUE + "/mails")
+ .patch(FIRST_QUEUE.asString() + "/mails")
.then()
.statusCode(HttpStatus.BAD_REQUEST_400);
}
@@ -472,7 +478,7 @@ class MailQueueRoutesTest {
.queryParam("delayed", "true")
.body("{\"xx\":")
.when()
- .patch(FIRST_QUEUE + "/mails")
+ .patch(FIRST_QUEUE.asString() + "/mails")
.then()
.statusCode(HttpStatus.BAD_REQUEST_400);
}
@@ -485,7 +491,7 @@ class MailQueueRoutesTest {
.queryParam("delayed", "false")
.body("{\"delayed\": \"true\"}")
.when()
- .patch(FIRST_QUEUE + "/mails")
+ .patch(FIRST_QUEUE.asString() + "/mails")
.then()
.statusCode(HttpStatus.BAD_REQUEST_400);
}
@@ -498,7 +504,7 @@ class MailQueueRoutesTest {
.queryParam("delayed", "false")
.body("{\"delayed\": \"string\"}")
.when()
- .patch(FIRST_QUEUE + "/mails")
+ .patch(FIRST_QUEUE.asString() + "/mails")
.then()
.statusCode(HttpStatus.BAD_REQUEST_400);
}
@@ -518,7 +524,7 @@ class MailQueueRoutesTest {
.queryParam("delayed", "true")
.body("{\"delayed\": \"false\"}")
.when()
- .patch(FIRST_QUEUE + "/mails");
+ .patch(FIRST_QUEUE.asString() + "/mails");
assertThat(queue.browse())
.toIterable()
@@ -541,7 +547,7 @@ class MailQueueRoutesTest {
@Test
void deleteMailsShouldReturnNotFoundWhenMailQueueDoesntExist() {
when()
- .delete(FIRST_QUEUE + "/mails")
+ .delete(FIRST_QUEUE.asString() + "/mails")
.then()
.statusCode(HttpStatus.NOT_FOUND_404);
}
@@ -553,7 +559,7 @@ class MailQueueRoutesTest {
given()
.param("sender", "123")
.when()
- .delete(FIRST_QUEUE + "/mails")
+ .delete(FIRST_QUEUE.asString() + "/mails")
.then()
.statusCode(HttpStatus.BAD_REQUEST_400);
}
@@ -565,7 +571,7 @@ class MailQueueRoutesTest {
given()
.param("recipient", "123")
.when()
- .delete(FIRST_QUEUE + "/mails")
+ .delete(FIRST_QUEUE.asString() + "/mails")
.then()
.statusCode(HttpStatus.BAD_REQUEST_400);
}
@@ -578,7 +584,7 @@ class MailQueueRoutesTest {
.param("name", "mailName")
.param("recipient", "recipient@james.org")
.when()
- .delete(FIRST_QUEUE + "/mails")
+ .delete(FIRST_QUEUE.asString() + "/mails")
.then()
.statusCode(HttpStatus.BAD_REQUEST_400);
}
@@ -590,7 +596,7 @@ class MailQueueRoutesTest {
.param("sender", "sender@james.org")
.param("name", "mailName")
.when()
- .delete(FIRST_QUEUE + "/mails")
+ .delete(FIRST_QUEUE.asString() + "/mails")
.then()
.statusCode(HttpStatus.BAD_REQUEST_400);
}
@@ -605,7 +611,7 @@ class MailQueueRoutesTest {
String taskId = with()
.param("sender", SENDER_1_JAMES_ORG)
- .delete(FIRST_QUEUE + "/mails")
+ .delete(FIRST_QUEUE.asString() + "/mails")
.jsonPath()
.getString("taskId");
@@ -623,7 +629,7 @@ class MailQueueRoutesTest {
String taskId = with()
.param("name", "mailName")
- .delete(FIRST_QUEUE + "/mails")
+ .delete(FIRST_QUEUE.asString() + "/mails")
.jsonPath()
.getString("taskId");
@@ -641,7 +647,7 @@ class MailQueueRoutesTest {
String taskId = with()
.param("recipient", RECIPIENT_JAMES_ORG)
- .delete(FIRST_QUEUE + "/mails")
+ .delete(FIRST_QUEUE.asString() + "/mails")
.jsonPath()
.getString("taskId");
@@ -669,7 +675,7 @@ class MailQueueRoutesTest {
String taskId = with()
.param("sender", SENDER_1_JAMES_ORG)
- .delete(FIRST_QUEUE + "/mails")
+ .delete(FIRST_QUEUE.asString() + "/mails")
.jsonPath()
.getString("taskId");
@@ -681,7 +687,7 @@ class MailQueueRoutesTest {
.body("status", is("completed"))
.body("taskId", is(notNullValue()))
.body("type", is(DeleteMailsFromMailQueueTask.TYPE.asString()))
- .body("additionalInformation.mailQueueName", is(FIRST_QUEUE))
+ .body("additionalInformation.mailQueueName", is(FIRST_QUEUE.asString()))
.body("additionalInformation.initialCount", is(2))
.body("additionalInformation.remainingCount", is(1))
.body("additionalInformation.sender", is(SENDER_1_JAMES_ORG))
@@ -704,7 +710,7 @@ class MailQueueRoutesTest {
String taskId = with()
.param("name", FAKE_MAIL_NAME_1)
- .delete(FIRST_QUEUE + "/mails")
+ .delete(FIRST_QUEUE.asString() + "/mails")
.jsonPath()
.getString("taskId");
@@ -716,7 +722,7 @@ class MailQueueRoutesTest {
.body("status", is("completed"))
.body("taskId", is(notNullValue()))
.body("type", is(DeleteMailsFromMailQueueTask.TYPE.asString()))
- .body("additionalInformation.mailQueueName", is(FIRST_QUEUE))
+ .body("additionalInformation.mailQueueName", is(FIRST_QUEUE.asString()))
.body("additionalInformation.initialCount", is(2))
.body("additionalInformation.remainingCount", is(1))
.body("additionalInformation.name", is(FAKE_MAIL_NAME_1))
@@ -746,7 +752,7 @@ class MailQueueRoutesTest {
String taskId = with()
.param("recipient", RECIPIENT_JAMES_ORG)
- .delete(FIRST_QUEUE + "/mails")
+ .delete(FIRST_QUEUE.asString() + "/mails")
.jsonPath()
.getString("taskId");
@@ -758,7 +764,7 @@ class MailQueueRoutesTest {
.body("status", is("completed"))
.body("taskId", is(notNullValue()))
.body("type", is(DeleteMailsFromMailQueueTask.TYPE.asString()))
- .body("additionalInformation.mailQueueName", is(FIRST_QUEUE))
+ .body("additionalInformation.mailQueueName", is(FIRST_QUEUE.asString()))
.body("additionalInformation.initialCount", is(3))
.body("additionalInformation.remainingCount", is(1))
.body("additionalInformation.recipient", is(RECIPIENT_JAMES_ORG))
@@ -787,7 +793,7 @@ class MailQueueRoutesTest {
String taskId = with()
.param("sender", SENDER_1_JAMES_ORG)
- .delete(FIRST_QUEUE + "/mails")
+ .delete(FIRST_QUEUE.asString() + "/mails")
.jsonPath()
.getString("taskId");
@@ -819,7 +825,7 @@ class MailQueueRoutesTest {
String taskId = with()
.param("name", FAKE_MAIL_NAME_1)
- .delete(FIRST_QUEUE + "/mails")
+ .delete(FIRST_QUEUE.asString() + "/mails")
.jsonPath()
.getString("taskId");
@@ -858,7 +864,7 @@ class MailQueueRoutesTest {
String taskId = with()
.param("recipient", RECIPIENT_JAMES_ORG)
- .delete(FIRST_QUEUE + "/mails")
+ .delete(FIRST_QUEUE.asString() + "/mails")
.jsonPath()
.getString("taskId");
@@ -890,7 +896,7 @@ class MailQueueRoutesTest {
String taskId = with()
.param("recipient", recipient)
- .delete(FIRST_QUEUE + "/mails")
+ .delete(FIRST_QUEUE.asString() + "/mails")
.jsonPath()
.getString("taskId");
@@ -920,7 +926,7 @@ class MailQueueRoutesTest {
mailQueueFactory.createQueue(FIRST_QUEUE);
when()
- .delete(SECOND_QUEUE + "/mails")
+ .delete(SECOND_QUEUE.asString() + "/mails")
.then()
.statusCode(HttpStatus.NOT_FOUND_404);
}
@@ -930,7 +936,7 @@ class MailQueueRoutesTest {
mailQueueFactory.createQueue(FIRST_QUEUE);
String taskId = with()
- .delete(FIRST_QUEUE + "/mails")
+ .delete(FIRST_QUEUE.asString() + "/mails")
.jsonPath()
.getString("taskId");
@@ -959,7 +965,7 @@ class MailQueueRoutesTest {
.build());
String taskId = with()
- .delete(FIRST_QUEUE + "/mails")
+ .delete(FIRST_QUEUE.asString() + "/mails")
.jsonPath()
.getString("taskId");
@@ -971,7 +977,7 @@ class MailQueueRoutesTest {
.body("status", is("completed"))
.body("taskId", is(notNullValue()))
.body("type", is(ClearMailQueueTask.TYPE.asString()))
- .body("additionalInformation.mailQueueName", is(FIRST_QUEUE))
+ .body("additionalInformation.mailQueueName", is(FIRST_QUEUE.asString()))
.body("additionalInformation.initialCount", is(3))
.body("additionalInformation.remainingCount", is(0))
.body("startedDate", is(notNullValue()))
@@ -996,7 +1002,7 @@ class MailQueueRoutesTest {
.build());
String taskId = with()
- .delete(FIRST_QUEUE + "/mails")
+ .delete(FIRST_QUEUE.asString() + "/mails")
.jsonPath()
.getString("taskId");
diff --git a/server/protocols/webadmin/webadmin-mailqueue/src/test/java/org/apache/james/webadmin/service/ClearMailQueueTaskTest.java b/server/protocols/webadmin/webadmin-mailqueue/src/test/java/org/apache/james/webadmin/service/ClearMailQueueTaskTest.java
index 94b4432..c78a4c4 100644
--- a/server/protocols/webadmin/webadmin-mailqueue/src/test/java/org/apache/james/webadmin/service/ClearMailQueueTaskTest.java
+++ b/server/protocols/webadmin/webadmin-mailqueue/src/test/java/org/apache/james/webadmin/service/ClearMailQueueTaskTest.java
@@ -19,7 +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.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -28,6 +28,7 @@ import java.util.Optional;
import org.apache.james.JsonSerializationVerifier;
import org.apache.james.queue.api.MailQueueFactory;
+import org.apache.james.queue.api.MailQueueName;
import org.apache.james.queue.api.ManageableMailQueue;
import org.apache.james.server.task.json.JsonTaskSerializer;
import org.junit.jupiter.api.Test;
@@ -35,7 +36,7 @@ import org.junit.jupiter.api.Test;
class ClearMailQueueTaskTest {
private static final Instant TIMESTAMP = Instant.parse("2018-11-13T12:00:55Z");
private static final String SERIALIZED = "{\"type\": \"clear-mail-queue\", \"queue\": \"anyQueue\"}";
- private static final String QUEUE_NAME = "anyQueue";
+ private static final MailQueueName QUEUE_NAME = MailQueueName.of("anyQueue");
private static final long INITIAL_COUNT = 0L;
private static final long REMAINING_COUNT = 10L;
private static final String SERIALIZED_TASK_ADDITIONAL_INFORMATION = "{\"type\": \"clear-mail-queue\", \"mailQueueName\":\"anyQueue\",\"initialCount\":0,\"remainingCount\":10, \"timestamp\":\"2018-11-13T12:00:55Z\"}";
@@ -45,7 +46,7 @@ class ClearMailQueueTaskTest {
MailQueueFactory<ManageableMailQueue> mailQueueFactory = mock(MailQueueFactory.class);
ManageableMailQueue mockedQueue = mock(ManageableMailQueue.class);
when(mockedQueue.getName()).thenReturn(QUEUE_NAME);
- when(mailQueueFactory.getQueue(anyString())).thenAnswer(arg -> Optional.of(mockedQueue));
+ when(mailQueueFactory.getQueue(any(MailQueueName.class))).thenAnswer(arg -> Optional.of(mockedQueue));
ClearMailQueueTask.MailQueueFactory factory = queueName -> mailQueueFactory.getQueue(queueName).orElseThrow(RuntimeException::new);
ClearMailQueueTask task = new ClearMailQueueTask(QUEUE_NAME, factory);
@@ -60,7 +61,7 @@ class ClearMailQueueTaskTest {
@Test
void taskShouldThrowWhenRunOnAnUnknownQueue() {
MailQueueFactory<ManageableMailQueue> mailQueueFactory = mock(MailQueueFactory.class);
- when(mailQueueFactory.getQueue(anyString())).thenReturn(Optional.empty());
+ when(mailQueueFactory.getQueue(any(MailQueueName.class))).thenReturn(Optional.empty());
JsonTaskSerializer testee = JsonTaskSerializer.of(ClearMailQueueTaskDTO.module(mailQueueFactory));
String serializedJson = "{\"type\": \"clear-mail-queue\", \"queue\": \"anyQueue\"}";
diff --git a/server/protocols/webadmin/webadmin-mailqueue/src/test/java/org/apache/james/webadmin/service/DeleteMailsFromMailQueueTaskTest.java b/server/protocols/webadmin/webadmin-mailqueue/src/test/java/org/apache/james/webadmin/service/DeleteMailsFromMailQueueTaskTest.java
index 6e44829..eb2b79c 100644
--- a/server/protocols/webadmin/webadmin-mailqueue/src/test/java/org/apache/james/webadmin/service/DeleteMailsFromMailQueueTaskTest.java
+++ b/server/protocols/webadmin/webadmin-mailqueue/src/test/java/org/apache/james/webadmin/service/DeleteMailsFromMailQueueTaskTest.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.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -31,6 +32,7 @@ import javax.mail.internet.AddressException;
import org.apache.james.JsonSerializationVerifier;
import org.apache.james.core.MailAddress;
import org.apache.james.queue.api.MailQueueFactory;
+import org.apache.james.queue.api.MailQueueName;
import org.apache.james.queue.api.ManageableMailQueue;
import org.apache.james.server.task.json.JsonTaskAdditionalInformationSerializer;
import org.apache.james.server.task.json.JsonTaskSerializer;
@@ -42,14 +44,14 @@ class DeleteMailsFromMailQueueTaskTest {
private MailQueueFactory<ManageableMailQueue> mailQueueFactory;
private ManageableMailQueue mockedQueue;
- private final static String queueName = "anyQueue";
+ private final static MailQueueName queueName = MailQueueName.of("anyQueue");
@BeforeEach
void setUp() {
mailQueueFactory = mock(MailQueueFactory.class);
mockedQueue = mock(ManageableMailQueue.class);
when(mockedQueue.getName()).thenReturn(queueName);
- when(mailQueueFactory.getQueue(anyString())).thenAnswer(arg -> Optional.of(mockedQueue));
+ when(mailQueueFactory.getQueue(any(MailQueueName.class))).thenAnswer(arg -> Optional.of(mockedQueue));
}
@Test
@@ -75,7 +77,7 @@ class DeleteMailsFromMailQueueTaskTest {
@Test
void taskShouldThrowWhenRunOnAnUnknownQueue() {
MailQueueFactory<ManageableMailQueue> mailQueueFactory = mock(MailQueueFactory.class);
- when(mailQueueFactory.getQueue(anyString())).thenReturn(Optional.empty());
+ when(mailQueueFactory.getQueue(any(MailQueueName.class))).thenReturn(Optional.empty());
JsonTaskSerializer testee = JsonTaskSerializer.of(DeleteMailsFromMailQueueTaskDTO.module(mailQueueFactory));
String serializedJson = "{\"type\": \"delete-mails-from-mail-queue\", \"queue\": \"anyQueue\", \"sender\": \"a@b.c\"}";
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 6a070a8..85b3b6d 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
@@ -42,6 +42,7 @@ 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.MailQueueFactory;
+import org.apache.james.queue.api.MailQueueName;
import org.apache.james.task.Task;
import org.apache.james.task.TaskManager;
import org.apache.james.util.streams.Limit;
@@ -453,7 +454,7 @@ public class MailRepositoriesRoutes implements Routes {
private Task reprocessAll(Request request) throws UnsupportedEncodingException, MailRepositoryStore.MailRepositoryStoreException {
MailRepositoryPath path = decodedRepositoryPath(request);
Optional<String> targetProcessor = parseTargetProcessor(request);
- String targetQueue = parseTargetQueue(request);
+ MailQueueName targetQueue = parseTargetQueue(request);
Long repositorySize = repositoryStoreService.size(path).orElse(0L);
return new ReprocessingAllMailsTask(reprocessingService, repositorySize, path, targetQueue, targetProcessor);
@@ -505,7 +506,7 @@ public class MailRepositoriesRoutes implements Routes {
MailKey key = new MailKey(request.params("key"));
Optional<String> targetProcessor = parseTargetProcessor(request);
- String targetQueue = parseTargetQueue(request);
+ MailQueueName targetQueue = parseTargetQueue(request);
return new ReprocessingOneMailTask(reprocessingService, path, targetQueue, key, targetProcessor, Clock.systemUTC());
}
@@ -525,8 +526,10 @@ public class MailRepositoriesRoutes implements Routes {
return Optional.ofNullable(request.queryParams("processor"));
}
- private String parseTargetQueue(Request request) {
- return Optional.ofNullable(request.queryParams("queue")).orElse(MailQueueFactory.SPOOL);
+ private MailQueueName parseTargetQueue(Request request) {
+ return Optional.ofNullable(request.queryParams("queue"))
+ .map(MailQueueName::of)
+ .orElse(MailQueueFactory.SPOOL);
}
private MailRepositoryPath decodedRepositoryPath(Request request) throws UnsupportedEncodingException {
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 1fa8d43..5d8ea39 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,6 +29,7 @@ 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;
@@ -45,9 +46,9 @@ public class ReprocessingAllMailsTask implements Task {
private final long remainingCount;
private final Instant timestamp;
- public AdditionalInformation(MailRepositoryPath repositoryPath, String targetQueue, Optional<String> targetProcessor, long initialCount, long remainingCount, Instant timestamp) {
+ public AdditionalInformation(MailRepositoryPath repositoryPath, MailQueueName targetQueue, Optional<String> targetProcessor, long initialCount, long remainingCount, Instant timestamp) {
this.repositoryPath = repositoryPath;
- this.targetQueue = targetQueue;
+ this.targetQueue = targetQueue.asString();
this.targetProcessor = targetProcessor;
this.initialCount = initialCount;
this.remainingCount = remainingCount;
@@ -96,13 +97,13 @@ public class ReprocessingAllMailsTask implements Task {
private final ReprocessingService reprocessingService;
private final MailRepositoryPath repositoryPath;
- private final String targetQueue;
+ private final MailQueueName targetQueue;
private final Optional<String> targetProcessor;
private final long repositorySize;
private final AtomicLong processedCount;
public ReprocessingAllMailsTask(ReprocessingService reprocessingService, long repositorySize,
- MailRepositoryPath repositoryPath, String targetQueue, Optional<String> targetProcessor) {
+ MailRepositoryPath repositoryPath, MailQueueName targetQueue, Optional<String> targetProcessor) {
this.reprocessingService = reprocessingService;
this.repositoryPath = repositoryPath;
this.targetQueue = targetQueue;
@@ -138,7 +139,7 @@ public class ReprocessingAllMailsTask implements Task {
return targetProcessor;
}
- String getTargetQueue() {
+ MailQueueName getTargetQueue() {
return targetQueue;
}
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 15b8cee..ee821a5 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
@@ -23,6 +23,7 @@ import java.util.Optional;
import org.apache.james.json.DTOModule;
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;
@@ -35,7 +36,7 @@ public class ReprocessingAllMailsTaskAdditionalInformationDTO implements Additio
.convertToDTO(ReprocessingAllMailsTaskAdditionalInformationDTO.class)
.toDomainObjectConverter(dto -> new ReprocessingAllMailsTask.AdditionalInformation(
MailRepositoryPath.from(dto.repositoryPath),
- dto.targetQueue,
+ MailQueueName.of(dto.targetQueue),
dto.targetProcessor,
dto.initialCount,
dto.remainingCount,
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 bc04212..d81f2a8 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
@@ -22,6 +22,7 @@ import java.util.Optional;
import org.apache.james.json.DTOModule;
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;
@@ -45,7 +46,7 @@ public class ReprocessingAllMailsTaskDTO implements TaskDTO {
typeName,
domainObject.getRepositorySize(),
domainObject.getRepositoryPath().urlEncoded(),
- domainObject.getTargetQueue(),
+ domainObject.getTargetQueue().asString(),
domainObject.getTargetProcessor()
);
} catch (Exception e) {
@@ -77,7 +78,7 @@ public class ReprocessingAllMailsTaskDTO implements TaskDTO {
reprocessingService,
repositorySize,
MailRepositoryPath.fromEncoded(repositoryPath),
- targetQueue,
+ MailQueueName.of(targetQueue),
targetProcessor
);
} catch (Exception e) {
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 0225abb..2777ce2 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,6 +28,7 @@ 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;
@@ -43,9 +44,9 @@ public class ReprocessingOneMailTask implements Task {
private final Optional<String> targetProcessor;
private final Instant timestamp;
- public AdditionalInformation(MailRepositoryPath repositoryPath, String targetQueue, MailKey mailKey, Optional<String> targetProcessor, Instant timestamp) {
+ public AdditionalInformation(MailRepositoryPath repositoryPath, MailQueueName targetQueue, MailKey mailKey, Optional<String> targetProcessor, Instant timestamp) {
this.repositoryPath = repositoryPath;
- this.targetQueue = targetQueue;
+ this.targetQueue = targetQueue.asString();
this.mailKey = mailKey;
this.targetProcessor = targetProcessor;
this.timestamp = timestamp;
@@ -89,14 +90,14 @@ public class ReprocessingOneMailTask implements Task {
private final ReprocessingService reprocessingService;
private final MailRepositoryPath repositoryPath;
- private final String targetQueue;
+ private final MailQueueName targetQueue;
private final MailKey mailKey;
private final Optional<String> targetProcessor;
private final AdditionalInformation additionalInformation;
public ReprocessingOneMailTask(ReprocessingService reprocessingService,
MailRepositoryPath repositoryPath,
- String targetQueue,
+ MailQueueName targetQueue,
MailKey mailKey,
Optional<String> targetProcessor,
Clock clock) {
@@ -128,7 +129,7 @@ public class ReprocessingOneMailTask implements Task {
return repositoryPath;
}
- String getTargetQueue() {
+ MailQueueName getTargetQueue() {
return targetQueue;
}
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 528caf5..89dd1d8 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
@@ -24,6 +24,7 @@ import java.util.Optional;
import org.apache.james.json.DTOModule;
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.dto.AdditionalInformationDTO;
import org.apache.james.server.task.json.dto.AdditionalInformationDTOModule;
@@ -36,7 +37,7 @@ public class ReprocessingOneMailTaskAdditionalInformationDTO implements Addition
.convertToDTO(ReprocessingOneMailTaskAdditionalInformationDTO.class)
.toDomainObjectConverter(dto -> new ReprocessingOneMailTask.AdditionalInformation(
MailRepositoryPath.from(dto.repositoryPath),
- dto.targetQueue,
+ MailQueueName.of(dto.targetQueue),
new MailKey(dto.mailKey),
dto.targetProcessor,
dto.timestamp
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 462e598..ea09c18 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
@@ -24,6 +24,7 @@ import java.util.Optional;
import org.apache.james.json.DTOModule;
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.dto.TaskDTO;
import org.apache.james.server.task.json.dto.TaskDTOModule;
@@ -46,7 +47,7 @@ public class ReprocessingOneMailTaskDTO implements TaskDTO {
return new ReprocessingOneMailTaskDTO(
typeName,
domainObject.getRepositoryPath().urlEncoded(),
- domainObject.getTargetQueue(),
+ domainObject.getTargetQueue().asString(),
domainObject.getMailKey().asString(),
domainObject.getTargetProcessor()
);
@@ -77,7 +78,7 @@ public class ReprocessingOneMailTaskDTO implements TaskDTO {
return new ReprocessingOneMailTask(
reprocessingService,
getMailRepositoryPath(),
- targetQueue,
+ MailQueueName.of(targetQueue),
new MailKey(mailKey),
targetProcessor,
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 6171b00..b93ec2c 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
@@ -34,6 +34,7 @@ import org.apache.james.mailrepository.api.MailRepositoryPath;
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.util.OptionalUtils;
import org.apache.james.util.streams.Iterators;
import org.apache.mailet.Mail;
@@ -91,7 +92,7 @@ public class ReprocessingService {
this.mailRepositoryStoreService = mailRepositoryStoreService;
}
- public void reprocessAll(MailRepositoryPath path, Optional<String> targetProcessor, String targetQueue, Consumer<MailKey> keyListener) throws MailRepositoryStore.MailRepositoryStoreException, MessagingException {
+ 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)) {
mailRepositoryStoreService
.getRepositories(path)
@@ -104,7 +105,7 @@ public class ReprocessingService {
}
}
- public void reprocess(MailRepositoryPath path, MailKey key, Optional<String> targetProcessor, String targetQueue) throws MailRepositoryStore.MailRepositoryStoreException, MessagingException {
+ public void reprocess(MailRepositoryPath path, MailKey key, Optional<String> targetProcessor, MailQueueName targetQueue) throws MailRepositoryStore.MailRepositoryStoreException, MessagingException {
try (Reprocessor reprocessor = new Reprocessor(getMailQueue(targetQueue), targetProcessor)) {
Pair<MailRepository, Mail> mailPair = mailRepositoryStoreService
.getRepositories(path)
@@ -118,8 +119,8 @@ public class ReprocessingService {
}
}
- private MailQueue getMailQueue(String targetQueue) {
+ private MailQueue getMailQueue(MailQueueName targetQueue) {
return mailQueueFactory.getQueue(targetQueue)
- .orElseThrow(() -> new RuntimeException("Can not find queue " + targetQueue));
+ .orElseThrow(() -> new RuntimeException("Can not find queue " + targetQueue.asString()));
}
}
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 eb6998a..613c506 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
@@ -60,6 +60,7 @@ import org.apache.james.mailrepository.memory.MemoryMailRepositoryStore;
import org.apache.james.mailrepository.memory.MemoryMailRepositoryUrlStore;
import org.apache.james.mailrepository.memory.TestingMailRepositoryLoader;
import org.apache.james.queue.api.MailQueueFactory;
+import org.apache.james.queue.api.MailQueueName;
import org.apache.james.queue.api.ManageableMailQueue;
import org.apache.james.queue.api.RawMailQueueItemDecoratorFactory;
import org.apache.james.queue.memory.MemoryMailQueueFactory;
@@ -99,7 +100,7 @@ public class MailRepositoriesRoutesTest {
private static final MailRepositoryPath PATH_MY_REPO = MailRepositoryPath.from("myRepo");
private static final String PATH_ESCAPED_MY_REPO = "myRepo";
private static final String MY_REPO_MAILS = "myRepo/mails";
- private static final String CUSTOM_QUEUE = "customQueue";
+ private static final MailQueueName CUSTOM_QUEUE = MailQueueName.of("customQueue");
private static final String NAME_1 = "name1";
private static final String NAME_2 = "name2";
private WebAdminServer webAdminServer;
@@ -1119,7 +1120,7 @@ public class MailRepositoriesRoutesTest {
.body("additionalInformation.initialCount", is(2))
.body("additionalInformation.remainingCount", is(0))
.body("additionalInformation.targetProcessor", isEmptyOrNullString())
- .body("additionalInformation.targetQueue", is(MailQueueFactory.SPOOL))
+ .body("additionalInformation.targetQueue", is(MailQueueFactory.SPOOL.asString()))
.body("startedDate", is(notNullValue()))
.body("submitDate", is(notNullValue()))
.body("completedDate", is(notNullValue()));
@@ -1140,7 +1141,7 @@ public class MailRepositoriesRoutesTest {
String transport = "transport";
String taskId = with()
.param("action", "reprocess")
- .param("queue", CUSTOM_QUEUE)
+ .param("queue", CUSTOM_QUEUE.asString())
.param("processor", transport)
.patch(PATH_ESCAPED_MY_REPO + "/mails")
.jsonPath()
@@ -1158,7 +1159,7 @@ public class MailRepositoriesRoutesTest {
.body("additionalInformation.initialCount", is(2))
.body("additionalInformation.remainingCount", is(0))
.body("additionalInformation.targetProcessor", is(transport))
- .body("additionalInformation.targetQueue", is(CUSTOM_QUEUE))
+ .body("additionalInformation.targetQueue", is(CUSTOM_QUEUE.asString()))
.body("startedDate", is(notNullValue()))
.body("submitDate", is(notNullValue()))
.body("completedDate", is(notNullValue()));
@@ -1180,7 +1181,7 @@ public class MailRepositoriesRoutesTest {
String transport = "transport";
String taskId = with()
.param("action", "reprocess")
- .param("queue", CUSTOM_QUEUE)
+ .param("queue", CUSTOM_QUEUE.asString())
.param("processor", transport)
.patch(PATH_ESCAPED_MY_REPO + "/mails")
.jsonPath()
@@ -1209,7 +1210,7 @@ public class MailRepositoriesRoutesTest {
String transport = "transport";
String taskId = with()
.param("action", "reprocess")
- .param("queue", CUSTOM_QUEUE)
+ .param("queue", CUSTOM_QUEUE.asString())
.param("processor", transport)
.patch(PATH_ESCAPED_MY_REPO + "/mails")
.jsonPath()
@@ -1240,7 +1241,7 @@ public class MailRepositoriesRoutesTest {
String transport = "transport";
String taskId = with()
.param("action", "reprocess")
- .param("queue", CUSTOM_QUEUE)
+ .param("queue", CUSTOM_QUEUE.asString())
.param("processor", transport)
.patch(PATH_ESCAPED_MY_REPO + "/mails")
.jsonPath()
@@ -1387,7 +1388,7 @@ public class MailRepositoriesRoutesTest {
String taskId = with()
.param("action", "reprocess")
- .param("queue", CUSTOM_QUEUE)
+ .param("queue", CUSTOM_QUEUE.asString())
.patch(PATH_ESCAPED_MY_REPO + "/mails")
.jsonPath()
.get("taskId");
@@ -1485,7 +1486,7 @@ public class MailRepositoriesRoutesTest {
.body("additionalInformation.repositoryPath", is(PATH_MY_REPO.asString()))
.body("additionalInformation.mailKey", is(NAME_1))
.body("additionalInformation.targetProcessor", isEmptyOrNullString())
- .body("additionalInformation.targetQueue", is(MailQueueFactory.SPOOL))
+ .body("additionalInformation.targetQueue", is(MailQueueFactory.SPOOL.asString()))
.body("startedDate", is(notNullValue()))
.body("submitDate", is(notNullValue()))
.body("completedDate", is(notNullValue()));
@@ -1506,7 +1507,7 @@ public class MailRepositoriesRoutesTest {
String transport = "transport";
String taskId = with()
.param("action", "reprocess")
- .param("queue", CUSTOM_QUEUE)
+ .param("queue", CUSTOM_QUEUE.asString())
.param("processor", transport)
.patch(PATH_ESCAPED_MY_REPO + "/mails/" + NAME_1)
.jsonPath()
@@ -1523,7 +1524,7 @@ public class MailRepositoriesRoutesTest {
.body("additionalInformation.repositoryPath", is(PATH_MY_REPO.asString()))
.body("additionalInformation.mailKey", is(NAME_1))
.body("additionalInformation.targetProcessor", is(transport))
- .body("additionalInformation.targetQueue", is(CUSTOM_QUEUE))
+ .body("additionalInformation.targetQueue", is(CUSTOM_QUEUE.asString()))
.body("startedDate", is(notNullValue()))
.body("submitDate", is(notNullValue()))
.body("completedDate", is(notNullValue()));
@@ -1545,7 +1546,7 @@ public class MailRepositoriesRoutesTest {
String transport = "transport";
String taskId = with()
.param("action", "reprocess")
- .param("queue", CUSTOM_QUEUE)
+ .param("queue", CUSTOM_QUEUE.asString())
.param("processor", transport)
.patch(PATH_ESCAPED_MY_REPO + "/mails/" + NAME_1)
.jsonPath()
@@ -1574,7 +1575,7 @@ public class MailRepositoriesRoutesTest {
String transport = "transport";
String taskId = with()
.param("action", "reprocess")
- .param("queue", CUSTOM_QUEUE)
+ .param("queue", CUSTOM_QUEUE.asString())
.param("processor", transport)
.patch(PATH_ESCAPED_MY_REPO + "/mails/" + NAME_1)
.jsonPath()
@@ -1692,7 +1693,7 @@ public class MailRepositoriesRoutesTest {
String taskId = with()
.param("action", "reprocess")
- .param("queue", CUSTOM_QUEUE)
+ .param("queue", CUSTOM_QUEUE.asString())
.patch(PATH_ESCAPED_MY_REPO + "/mails/" + NAME_1)
.jsonPath()
.get("taskId");
@@ -1720,7 +1721,7 @@ public class MailRepositoriesRoutesTest {
String taskId = with()
.param("action", "reprocess")
- .param("queue", CUSTOM_QUEUE)
+ .param("queue", CUSTOM_QUEUE.asString())
.patch(PATH_ESCAPED_MY_REPO + "/mails/" + "unknown")
.jsonPath()
.get("taskId");
@@ -1746,7 +1747,7 @@ public class MailRepositoriesRoutesTest {
String taskId = with()
.param("action", "reprocess")
- .param("queue", CUSTOM_QUEUE)
+ .param("queue", CUSTOM_QUEUE.asString())
.patch(PATH_ESCAPED_MY_REPO + "/mails/" + "unknown")
.jsonPath()
.get("taskId");
@@ -1771,7 +1772,7 @@ public class MailRepositoriesRoutesTest {
String taskId = with()
.param("action", "reprocess")
- .param("queue", CUSTOM_QUEUE)
+ .param("queue", CUSTOM_QUEUE.asString())
.patch(PATH_ESCAPED_MY_REPO + "/mails/" + "unknown")
.jsonPath()
.get("taskId");
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 df0fe9e..7ad6a8b 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
@@ -27,6 +27,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.JsonTaskSerializer;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
@@ -37,7 +38,7 @@ class ReprocessingAllMailsTaskTest {
private static final ReprocessingService REPROCESSING_SERVICE = mock(ReprocessingService.class);
private static final long REPOSITORY_SIZE = 5L;
private static final MailRepositoryPath REPOSITORY_PATH = MailRepositoryPath.from("a");
- private static final String TARGET_QUEUE = "queue";
+ 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\"}";
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 5927ee0..b61ed7e 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
@@ -31,6 +31,7 @@ import java.util.Optional;
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.JsonTaskSerializer;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
@@ -43,7 +44,7 @@ class ReprocessingOneMailTaskTest {
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 MailRepositoryPath REPOSITORY_PATH = MailRepositoryPath.from("a");
- private static final String TARGET_QUEUE = "queue";
+ 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");
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 4c6dca1..994c7aa 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
@@ -37,6 +37,7 @@ import org.apache.james.mailrepository.memory.MemoryMailRepositoryStore;
import org.apache.james.mailrepository.memory.MemoryMailRepositoryUrlStore;
import org.apache.james.mailrepository.memory.TestingMailRepositoryLoader;
import org.apache.james.queue.api.MailQueueFactory;
+import org.apache.james.queue.api.MailQueueName;
import org.apache.james.queue.api.ManageableMailQueue;
import org.apache.james.queue.api.RawMailQueueItemDecoratorFactory;
import org.apache.james.queue.memory.MemoryMailQueueFactory;
@@ -57,7 +58,7 @@ public class ReprocessingServiceTest {
private static final MailKey KEY_1 = new MailKey(NAME_1);
private static final MailKey KEY_2 = new MailKey(NAME_2);
private static final MailKey KEY_3 = new MailKey(NAME_3);
- private static final String SPOOL = "spool";
+ private static final MailQueueName SPOOL = MailQueueName.of("spool");
private static final Consumer<MailKey> NOOP_CONSUMER = key -> { };
private static final Optional<String> NO_TARGET_PROCESSOR = Optional.empty();
diff --git a/server/queue/queue-activemq/src/main/java/org/apache/james/queue/activemq/ActiveMQCacheableMailQueue.java b/server/queue/queue-activemq/src/main/java/org/apache/james/queue/activemq/ActiveMQCacheableMailQueue.java
index 145acd5..e798251 100644
--- a/server/queue/queue-activemq/src/main/java/org/apache/james/queue/activemq/ActiveMQCacheableMailQueue.java
+++ b/server/queue/queue-activemq/src/main/java/org/apache/james/queue/activemq/ActiveMQCacheableMailQueue.java
@@ -44,6 +44,7 @@ import org.apache.james.metrics.api.GaugeRegistry;
import org.apache.james.metrics.api.MetricFactory;
import org.apache.james.queue.api.MailQueue;
import org.apache.james.queue.api.MailQueueItemDecoratorFactory;
+import org.apache.james.queue.api.MailQueueName;
import org.apache.james.queue.jms.JMSCacheableMailQueue;
import org.apache.james.server.core.MailImpl;
import org.apache.james.server.core.MimeMessageCopyOnWriteProxy;
@@ -98,7 +99,7 @@ public class ActiveMQCacheableMailQueue extends JMSCacheableMailQueue implements
* Construct a {@link ActiveMQCacheableMailQueue} which only use {@link BlobMessage}
*
*/
- public ActiveMQCacheableMailQueue(ConnectionFactory connectionFactory, MailQueueItemDecoratorFactory mailQueueItemDecoratorFactory, String queuename, MetricFactory metricFactory,
+ public ActiveMQCacheableMailQueue(ConnectionFactory connectionFactory, MailQueueItemDecoratorFactory mailQueueItemDecoratorFactory, MailQueueName queuename, MetricFactory metricFactory,
GaugeRegistry gaugeRegistry) {
this(connectionFactory, mailQueueItemDecoratorFactory, queuename, true, metricFactory, gaugeRegistry);
}
@@ -110,7 +111,7 @@ public class ActiveMQCacheableMailQueue extends JMSCacheableMailQueue implements
* @param queuename
* @param useBlob
*/
- public ActiveMQCacheableMailQueue(ConnectionFactory connectionFactory, MailQueueItemDecoratorFactory mailQueueItemDecoratorFactory, String queuename, boolean useBlob, MetricFactory metricFactory,
+ public ActiveMQCacheableMailQueue(ConnectionFactory connectionFactory, MailQueueItemDecoratorFactory mailQueueItemDecoratorFactory, MailQueueName queuename, boolean useBlob, MetricFactory metricFactory,
GaugeRegistry gaugeRegistry) {
super(connectionFactory, mailQueueItemDecoratorFactory, queuename, metricFactory, gaugeRegistry);
this.useBlob = useBlob;
@@ -124,7 +125,7 @@ public class ActiveMQCacheableMailQueue extends JMSCacheableMailQueue implements
try {
// store URL and queueName for later usage
builder.addAttribute(new Attribute(JAMES_BLOB_URL, AttributeValue.of(blobMessage.getURL())));
- builder.addAttribute(new Attribute(JAMES_QUEUE_NAME, AttributeValue.of(queueName)));
+ builder.addAttribute(new Attribute(JAMES_QUEUE_NAME, AttributeValue.of(queueName.asString())));
} catch (MalformedURLException e) {
// Ignore on error
LOGGER.debug("Unable to get url from blobmessage for mail");
@@ -280,7 +281,7 @@ public class ActiveMQCacheableMailQueue extends JMSCacheableMailQueue implements
replyTo = session.createTemporaryQueue();
consumer = session.createConsumer(replyTo);
- Queue myQueue = session.createQueue(queueName);
+ Queue myQueue = session.createQueue(queueName.asString());
producer = session.createProducer(null);
String queueName = "ActiveMQ.Statistics.Destination." + myQueue.getQueueName();
diff --git a/server/queue/queue-activemq/src/main/java/org/apache/james/queue/activemq/ActiveMQMailQueueFactory.java b/server/queue/queue-activemq/src/main/java/org/apache/james/queue/activemq/ActiveMQMailQueueFactory.java
index 5b61ca5..dcfe273 100644
--- a/server/queue/queue-activemq/src/main/java/org/apache/james/queue/activemq/ActiveMQMailQueueFactory.java
+++ b/server/queue/queue-activemq/src/main/java/org/apache/james/queue/activemq/ActiveMQMailQueueFactory.java
@@ -25,6 +25,7 @@ import org.apache.james.metrics.api.GaugeRegistry;
import org.apache.james.metrics.api.MetricFactory;
import org.apache.james.queue.api.MailQueueFactory;
import org.apache.james.queue.api.MailQueueItemDecoratorFactory;
+import org.apache.james.queue.api.MailQueueName;
import org.apache.james.queue.api.ManageableMailQueue;
import org.apache.james.queue.jms.JMSMailQueueFactory;
@@ -51,7 +52,7 @@ public class ActiveMQMailQueueFactory extends JMSMailQueueFactory {
}
@Override
- protected ManageableMailQueue createCacheableMailQueue(String name) {
+ protected ManageableMailQueue createCacheableMailQueue(MailQueueName name) {
return new ActiveMQCacheableMailQueue(connectionFactory, mailQueueItemDecoratorFactory, name, useBlob, metricFactory, gaugeRegistry);
}
}
diff --git a/server/queue/queue-activemq/src/test/java/org/apache/james/queue/activemq/ActiveMQMailQueueBlobTest.java b/server/queue/queue-activemq/src/test/java/org/apache/james/queue/activemq/ActiveMQMailQueueBlobTest.java
index da22eb7..1a68b89 100644
--- a/server/queue/queue-activemq/src/test/java/org/apache/james/queue/activemq/ActiveMQMailQueueBlobTest.java
+++ b/server/queue/queue-activemq/src/test/java/org/apache/james/queue/activemq/ActiveMQMailQueueBlobTest.java
@@ -38,6 +38,7 @@ import org.apache.james.queue.api.DelayedPriorityMailQueueContract;
import org.apache.james.queue.api.MailQueue;
import org.apache.james.queue.api.MailQueueMetricContract;
import org.apache.james.queue.api.MailQueueMetricExtension;
+import org.apache.james.queue.api.MailQueueName;
import org.apache.james.queue.api.ManageableMailQueue;
import org.apache.james.queue.api.PriorityManageableMailQueueContract;
import org.apache.james.queue.api.RawMailQueueItemDecoratorFactory;
@@ -77,7 +78,7 @@ public class ActiveMQMailQueueBlobTest implements DelayedManageableMailQueueCont
RawMailQueueItemDecoratorFactory mailQueueItemDecoratorFactory = new RawMailQueueItemDecoratorFactory();
MetricFactory metricFactory = metricTestSystem.getMetricFactory();
GaugeRegistry gaugeRegistry = metricTestSystem.getSpyGaugeRegistry();
- String queueName = BrokerExtension.generateRandomQueueName(broker);
+ MailQueueName queueName = BrokerExtension.generateRandomQueueName(broker);
mailQueue = new ActiveMQCacheableMailQueue(connectionFactory, mailQueueItemDecoratorFactory, queueName, USE_BLOB, metricFactory, gaugeRegistry);
}
diff --git a/server/queue/queue-activemq/src/test/java/org/apache/james/queue/activemq/ActiveMQMailQueueTest.java b/server/queue/queue-activemq/src/test/java/org/apache/james/queue/activemq/ActiveMQMailQueueTest.java
index e7a08dd..b90c1a1 100644
--- a/server/queue/queue-activemq/src/test/java/org/apache/james/queue/activemq/ActiveMQMailQueueTest.java
+++ b/server/queue/queue-activemq/src/test/java/org/apache/james/queue/activemq/ActiveMQMailQueueTest.java
@@ -28,6 +28,7 @@ import org.apache.james.queue.api.DelayedPriorityMailQueueContract;
import org.apache.james.queue.api.MailQueue;
import org.apache.james.queue.api.MailQueueMetricContract;
import org.apache.james.queue.api.MailQueueMetricExtension;
+import org.apache.james.queue.api.MailQueueName;
import org.apache.james.queue.api.ManageableMailQueue;
import org.apache.james.queue.api.PriorityManageableMailQueueContract;
import org.apache.james.queue.api.RawMailQueueItemDecoratorFactory;
@@ -57,7 +58,7 @@ public class ActiveMQMailQueueTest implements DelayedManageableMailQueueContract
RawMailQueueItemDecoratorFactory mailQueueItemDecoratorFactory = new RawMailQueueItemDecoratorFactory();
MetricFactory metricFactory = metricTestSystem.getMetricFactory();
GaugeRegistry gaugeRegistry = metricTestSystem.getSpyGaugeRegistry();
- String queueName = BrokerExtension.generateRandomQueueName(broker);
+ MailQueueName queueName = BrokerExtension.generateRandomQueueName(broker);
mailQueue = new ActiveMQCacheableMailQueue(connectionFactory, mailQueueItemDecoratorFactory, queueName, !USE_BLOB, metricFactory, gaugeRegistry);
}
diff --git a/server/queue/queue-api/src/main/java/org/apache/james/queue/api/MailQueue.java b/server/queue/queue-api/src/main/java/org/apache/james/queue/api/MailQueue.java
index d4e9c3a..38b8c3f 100644
--- a/server/queue/queue-api/src/main/java/org/apache/james/queue/api/MailQueue.java
+++ b/server/queue/queue-api/src/main/java/org/apache/james/queue/api/MailQueue.java
@@ -71,7 +71,7 @@ public interface MailQueue extends Closeable {
*/
int NO_DELAY = -1;
- String getName();
+ MailQueueName getName();
/**
* Enqueue the Mail to the queue. The given delay and unit are used to
diff --git a/server/queue/queue-api/src/main/java/org/apache/james/queue/api/MailQueueFactory.java b/server/queue/queue-api/src/main/java/org/apache/james/queue/api/MailQueueFactory.java
index 09850b3..3f0a5dc 100644
--- a/server/queue/queue-api/src/main/java/org/apache/james/queue/api/MailQueueFactory.java
+++ b/server/queue/queue-api/src/main/java/org/apache/james/queue/api/MailQueueFactory.java
@@ -30,7 +30,7 @@ public interface MailQueueFactory<T extends MailQueue> {
/**
* {@link MailQueue} which is used for spooling the messages
*/
- String SPOOL = "spool";
+ MailQueueName SPOOL = MailQueueName.of("spool");
/**
* Return the {@link MailQueue} for the name.
@@ -38,9 +38,9 @@ public interface MailQueueFactory<T extends MailQueue> {
* @param name
* @return queue
*/
- Optional<T> getQueue(String name);
+ Optional<T> getQueue(MailQueueName name);
- T createQueue(String name);
+ T createQueue(MailQueueName name);
- Set<String> listCreatedMailQueues();
+ Set<MailQueueName> listCreatedMailQueues();
}
diff --git a/server/queue/queue-api/src/test/java/org/apache/james/queue/api/ManageableMailQueueFactoryContract.java b/server/queue/queue-api/src/main/java/org/apache/james/queue/api/MailQueueName.java
similarity index 55%
copy from server/queue/queue-api/src/test/java/org/apache/james/queue/api/ManageableMailQueueFactoryContract.java
copy to server/queue/queue-api/src/main/java/org/apache/james/queue/api/MailQueueName.java
index 439e7d0..8ad937c 100644
--- a/server/queue/queue-api/src/test/java/org/apache/james/queue/api/ManageableMailQueueFactoryContract.java
+++ b/server/queue/queue-api/src/main/java/org/apache/james/queue/api/MailQueueName.java
@@ -19,27 +19,47 @@
package org.apache.james.queue.api;
-import static org.assertj.core.api.Assertions.assertThat;
+import java.util.Objects;
-import javax.mail.MessagingException;
+import com.google.common.base.MoreObjects;
+import com.google.common.base.Preconditions;
-import org.junit.jupiter.api.Test;
+public final class MailQueueName {
-public interface ManageableMailQueueFactoryContract {
+ public static MailQueueName of(String value) {
+ Preconditions.checkNotNull(value);
+ Preconditions.checkArgument(!value.isEmpty());
+ return new MailQueueName(value);
+ }
- String NAME_1 = "name1";
+ private final String value;
- MailQueueFactory<ManageableMailQueue> getMailQueueFactory();
+ private MailQueueName(String value) {
+ this.value = value;
+ }
- @Test
- default void createMailQueueShouldNotConflictIfAlreadyExists() throws MessagingException {
- MailQueueFactory<ManageableMailQueue> mailQueueFactory = getMailQueueFactory();
- MailQueue firstCreation = mailQueueFactory.createQueue(NAME_1);
+ public String asString() {
+ return value;
+ }
- firstCreation.enQueue(Mails.defaultMail().name("name").build());
+ @Override
+ public boolean equals(Object o) {
+ if (o instanceof MailQueueName) {
+ MailQueueName that = (MailQueueName) o;
+ return Objects.equals(value, that.value);
+ }
+ return false;
+ }
- ManageableMailQueue secondCreation = mailQueueFactory.createQueue(NAME_1);
- assertThat(secondCreation.getSize()).isEqualTo(1);
+ @Override
+ public int hashCode() {
+ return Objects.hash(value);
}
-}
\ No newline at end of file
+ @Override
+ public String toString() {
+ return MoreObjects.toStringHelper(this)
+ .add("value", value)
+ .toString();
+ }
+}
diff --git a/server/queue/queue-api/src/test/java/org/apache/james/queue/api/MailQueueFactoryContract.java b/server/queue/queue-api/src/test/java/org/apache/james/queue/api/MailQueueFactoryContract.java
index 20d99a6..34801b3 100644
--- a/server/queue/queue-api/src/test/java/org/apache/james/queue/api/MailQueueFactoryContract.java
+++ b/server/queue/queue-api/src/test/java/org/apache/james/queue/api/MailQueueFactoryContract.java
@@ -25,8 +25,8 @@ import org.junit.jupiter.api.Test;
public interface MailQueueFactoryContract<T extends MailQueue> {
- String NAME_1 = "name1";
- String NAME_2 = "name2";
+ MailQueueName NAME_1 = MailQueueName.of("name1");
+ MailQueueName NAME_2 = MailQueueName.of("name2");
MailQueueFactory<T> getMailQueueFactory();
diff --git a/server/queue/queue-api/src/main/java/org/apache/james/queue/api/MailQueueFactory.java b/server/queue/queue-api/src/test/java/org/apache/james/queue/api/MailQueueNameTest.java
similarity index 67%
copy from server/queue/queue-api/src/main/java/org/apache/james/queue/api/MailQueueFactory.java
copy to server/queue/queue-api/src/test/java/org/apache/james/queue/api/MailQueueNameTest.java
index 09850b3..1ed2714 100644
--- a/server/queue/queue-api/src/main/java/org/apache/james/queue/api/MailQueueFactory.java
+++ b/server/queue/queue-api/src/test/java/org/apache/james/queue/api/MailQueueNameTest.java
@@ -1,46 +1,44 @@
-/****************************************************************
- * 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.queue.api;
-
-import java.util.Optional;
-import java.util.Set;
-
-/**
- * Factory for {@link MailQueue}
- */
-public interface MailQueueFactory<T extends MailQueue> {
-
- /**
- * {@link MailQueue} which is used for spooling the messages
- */
- String SPOOL = "spool";
-
- /**
- * Return the {@link MailQueue} for the name.
- *
- * @param name
- * @return queue
- */
- Optional<T> getQueue(String name);
-
- T createQueue(String name);
-
- Set<String> listCreatedMailQueues();
-}
+/****************************************************************
+ * 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.queue.api;
+
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import nl.jqno.equalsverifier.EqualsVerifier;
+
+class MailQueueNameTest {
+
+ @Test
+ void shouldMatchBeanContract() {
+ EqualsVerifier.forClass(MailQueueName.class).verify();
+ }
+
+ @Test
+ void shouldNotAllowNullName() {
+ Assertions.assertThatThrownBy(() -> MailQueueName.of(null)).isInstanceOf(NullPointerException.class);
+ }
+
+ @Test
+ void shouldNotAllowEmptyName() {
+ Assertions.assertThatThrownBy(() -> MailQueueName.of("")).isInstanceOf(IllegalArgumentException.class);
+ }
+
+}
\ No newline at end of file
diff --git a/server/queue/queue-api/src/test/java/org/apache/james/queue/api/ManageableMailQueueFactoryContract.java b/server/queue/queue-api/src/test/java/org/apache/james/queue/api/ManageableMailQueueFactoryContract.java
index 439e7d0..5ae065f 100644
--- a/server/queue/queue-api/src/test/java/org/apache/james/queue/api/ManageableMailQueueFactoryContract.java
+++ b/server/queue/queue-api/src/test/java/org/apache/james/queue/api/ManageableMailQueueFactoryContract.java
@@ -27,7 +27,7 @@ import org.junit.jupiter.api.Test;
public interface ManageableMailQueueFactoryContract {
- String NAME_1 = "name1";
+ MailQueueName NAME_1 = MailQueueName.of("name1");
MailQueueFactory<ManageableMailQueue> getMailQueueFactory();
diff --git a/server/queue/queue-file/src/main/java/org/apache/james/queue/file/FileCacheableMailQueue.java b/server/queue/queue-file/src/main/java/org/apache/james/queue/file/FileCacheableMailQueue.java
index 6142be8..041e13d 100644
--- a/server/queue/queue-file/src/main/java/org/apache/james/queue/file/FileCacheableMailQueue.java
+++ b/server/queue/queue-file/src/main/java/org/apache/james/queue/file/FileCacheableMailQueue.java
@@ -51,6 +51,7 @@ import org.apache.commons.io.FileUtils;
import org.apache.james.lifecycle.api.Disposable;
import org.apache.james.lifecycle.api.LifecycleUtil;
import org.apache.james.queue.api.MailQueueItemDecoratorFactory;
+import org.apache.james.queue.api.MailQueueName;
import org.apache.james.queue.api.ManageableMailQueue;
import org.apache.james.server.core.MimeMessageCopyOnWriteProxy;
import org.apache.james.server.core.MimeMessageSource;
@@ -95,14 +96,14 @@ public class FileCacheableMailQueue implements ManageableMailQueue {
private static final AttributeName NEXT_DELIVERY = AttributeName.of("FileQueueNextDelivery");
private static final int SPLITCOUNT = 10;
private static final SecureRandom RANDOM = new SecureRandom();
- private final String queueName;
+ private final MailQueueName queueName;
private final Flux<MailQueueItem> flux;
- public FileCacheableMailQueue(MailQueueItemDecoratorFactory mailQueueItemDecoratorFactory, File parentDir, String queuename, boolean sync) throws IOException {
+ public FileCacheableMailQueue(MailQueueItemDecoratorFactory mailQueueItemDecoratorFactory, File parentDir, MailQueueName queuename, boolean sync) throws IOException {
this.mailQueueItemDecoratorFactory = mailQueueItemDecoratorFactory;
this.sync = sync;
this.queueName = queuename;
- this.queueDir = new File(parentDir, queueName);
+ this.queueDir = new File(parentDir, queueName.asString());
this.queueDirName = queueDir.getAbsolutePath();
init();
this.flux = Mono.defer(this::deQueueOneItem)
@@ -116,7 +117,7 @@ public class FileCacheableMailQueue implements ManageableMailQueue {
}
@Override
- public String getName() {
+ public MailQueueName getName() {
return queueName;
}
diff --git a/server/queue/queue-file/src/main/java/org/apache/james/queue/file/FileMailQueueFactory.java b/server/queue/queue-file/src/main/java/org/apache/james/queue/file/FileMailQueueFactory.java
index 663272a..0587620 100644
--- a/server/queue/queue-file/src/main/java/org/apache/james/queue/file/FileMailQueueFactory.java
+++ b/server/queue/queue-file/src/main/java/org/apache/james/queue/file/FileMailQueueFactory.java
@@ -30,6 +30,7 @@ import org.apache.james.filesystem.api.FileSystem;
import org.apache.james.queue.api.MailQueue;
import org.apache.james.queue.api.MailQueueFactory;
import org.apache.james.queue.api.MailQueueItemDecoratorFactory;
+import org.apache.james.queue.api.MailQueueName;
import org.apache.james.queue.api.ManageableMailQueue;
import com.github.steveash.guavate.Guavate;
@@ -43,7 +44,7 @@ import com.github.steveash.guavate.Guavate;
@Deprecated
public class FileMailQueueFactory implements MailQueueFactory<ManageableMailQueue> {
- private final Map<String, ManageableMailQueue> queues = new ConcurrentHashMap<>();
+ private final Map<MailQueueName, ManageableMailQueue> queues = new ConcurrentHashMap<>();
private MailQueueItemDecoratorFactory mailQueueActionItemDecoratorFactory;
private FileSystem fs;
private boolean sync = true;
@@ -55,7 +56,7 @@ public class FileMailQueueFactory implements MailQueueFactory<ManageableMailQueu
}
@Override
- public Set<String> listCreatedMailQueues() {
+ public Set<MailQueueName> listCreatedMailQueues() {
return queues.values()
.stream()
.map(MailQueue::getName)
@@ -75,17 +76,17 @@ public class FileMailQueueFactory implements MailQueueFactory<ManageableMailQueu
}
@Override
- public Optional<ManageableMailQueue> getQueue(String name) {
+ public Optional<ManageableMailQueue> getQueue(MailQueueName name) {
return Optional.ofNullable(queues.get(name));
}
@Override
- public ManageableMailQueue createQueue(String name) {
+ public ManageableMailQueue createQueue(MailQueueName name) {
return queues.computeIfAbsent(name, mailQueueName -> {
try {
return new FileCacheableMailQueue(mailQueueActionItemDecoratorFactory, fs.getFile("file://var/store/queue"), mailQueueName, sync);
} catch (IOException e) {
- throw new RuntimeException("Unable to access queue " + mailQueueName, e);
+ throw new RuntimeException("Unable to access queue " + mailQueueName.asString(), e);
}
});
}
diff --git a/server/queue/queue-file/src/test/java/org/apache/james/queue/file/FileCacheableMailQueueTest.java b/server/queue/queue-file/src/test/java/org/apache/james/queue/file/FileCacheableMailQueueTest.java
index 2b479e9..8fe07cc 100644
--- a/server/queue/queue-file/src/test/java/org/apache/james/queue/file/FileCacheableMailQueueTest.java
+++ b/server/queue/queue-file/src/test/java/org/apache/james/queue/file/FileCacheableMailQueueTest.java
@@ -21,6 +21,7 @@ package org.apache.james.queue.file;
import org.apache.james.queue.api.DelayedManageableMailQueueContract;
import org.apache.james.queue.api.MailQueue;
+import org.apache.james.queue.api.MailQueueName;
import org.apache.james.queue.api.ManageableMailQueue;
import org.apache.james.queue.api.RawMailQueueItemDecoratorFactory;
import org.junit.jupiter.api.AfterEach;
@@ -43,7 +44,7 @@ public class FileCacheableMailQueueTest implements DelayedManageableMailQueueCon
@BeforeEach
public void setUp() throws Exception {
temporaryFolder.create();
- mailQueue = new FileCacheableMailQueue(new RawMailQueueItemDecoratorFactory(), temporaryFolder.newFolder(), "test", SYNC);
+ mailQueue = new FileCacheableMailQueue(new RawMailQueueItemDecoratorFactory(), temporaryFolder.newFolder(), MailQueueName.of("test"), SYNC);
}
@AfterEach
diff --git a/server/queue/queue-jms/src/main/java/org/apache/james/queue/jms/JMSCacheableMailQueue.java b/server/queue/queue-jms/src/main/java/org/apache/james/queue/jms/JMSCacheableMailQueue.java
index 5171cae..5c3bb17 100644
--- a/server/queue/queue-jms/src/main/java/org/apache/james/queue/jms/JMSCacheableMailQueue.java
+++ b/server/queue/queue-jms/src/main/java/org/apache/james/queue/jms/JMSCacheableMailQueue.java
@@ -65,6 +65,7 @@ import org.apache.james.metrics.api.TimeMetric;
import org.apache.james.queue.api.MailPrioritySupport;
import org.apache.james.queue.api.MailQueue;
import org.apache.james.queue.api.MailQueueItemDecoratorFactory;
+import org.apache.james.queue.api.MailQueueName;
import org.apache.james.queue.api.ManageableMailQueue;
import org.apache.james.server.core.MailImpl;
import org.apache.james.server.core.MimeMessageCopyOnWriteProxy;
@@ -157,7 +158,7 @@ public class JMSCacheableMailQueue implements ManageableMailQueue, JMSSupport, M
public static final String FORCE_DELIVERY = "FORCE_DELIVERY";
- protected final String queueName;
+ protected final MailQueueName queueName;
protected final Connection connection;
protected final MailQueueItemDecoratorFactory mailQueueItemDecoratorFactory;
protected final Metric enqueuedMailsMetric;
@@ -173,7 +174,7 @@ public class JMSCacheableMailQueue implements ManageableMailQueue, JMSSupport, M
private final Splitter splitter;
public JMSCacheableMailQueue(ConnectionFactory connectionFactory, MailQueueItemDecoratorFactory mailQueueItemDecoratorFactory,
- String queueName, MetricFactory metricFactory,
+ MailQueueName queueName, MetricFactory metricFactory,
GaugeRegistry gaugeRegistry) {
try {
connection = connectionFactory.createConnection();
@@ -184,11 +185,11 @@ public class JMSCacheableMailQueue implements ManageableMailQueue, JMSSupport, M
this.mailQueueItemDecoratorFactory = mailQueueItemDecoratorFactory;
this.queueName = queueName;
this.metricFactory = metricFactory;
- this.enqueuedMailsMetric = metricFactory.generate(ENQUEUED_METRIC_NAME_PREFIX + queueName);
- this.dequeuedMailsMetric = metricFactory.generate(DEQUEUED_METRIC_NAME_PREFIX + queueName);
+ this.enqueuedMailsMetric = metricFactory.generate(ENQUEUED_METRIC_NAME_PREFIX + queueName.asString());
+ this.dequeuedMailsMetric = metricFactory.generate(DEQUEUED_METRIC_NAME_PREFIX + queueName.asString());
this.gaugeRegistry = gaugeRegistry;
- this.gaugeRegistry.register(QUEUE_SIZE_METRIC_NAME_PREFIX + queueName, queueSizeGauge());
+ this.gaugeRegistry.register(QUEUE_SIZE_METRIC_NAME_PREFIX + queueName.asString(), queueSizeGauge());
this.joiner = Joiner.on(JAMES_MAIL_SEPARATOR).skipNulls();
this.splitter = Splitter.on(JAMES_MAIL_SEPARATOR)
@@ -196,7 +197,7 @@ public class JMSCacheableMailQueue implements ManageableMailQueue, JMSSupport, M
.trimResults();
try {
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
- queue = session.createQueue(queueName);
+ queue = session.createQueue(queueName.asString());
producer = session.createProducer(queue);
} catch (JMSException e) {
throw new RuntimeException(e);
@@ -213,7 +214,7 @@ public class JMSCacheableMailQueue implements ManageableMailQueue, JMSSupport, M
}
@Override
- public String getName() {
+ public MailQueueName getName() {
return queueName;
}
@@ -239,7 +240,7 @@ public class JMSCacheableMailQueue implements ManageableMailQueue, JMSSupport, M
MessageConsumer consumer = null;
try {
session = connection.createSession(true, Session.SESSION_TRANSACTED);
- Queue queue = session.createQueue(queueName);
+ Queue queue = session.createQueue(queueName.asString());
consumer = session.createConsumer(queue, getMessageSelector());
Message message = consumer.receive(10000);
@@ -264,7 +265,7 @@ public class JMSCacheableMailQueue implements ManageableMailQueue, JMSSupport, M
@Override
public void enQueue(Mail mail, Duration delay) throws MailQueueException {
- TimeMetric timeMetric = metricFactory.timer(ENQUEUED_TIMER_METRIC_NAME_PREFIX + queueName);
+ TimeMetric timeMetric = metricFactory.timer(ENQUEUED_TIMER_METRIC_NAME_PREFIX + queueName.asString());
long nextDeliveryTimestamp = computeNextDeliveryTimestamp(delay);
@@ -465,7 +466,7 @@ public class JMSCacheableMailQueue implements ManageableMailQueue, JMSSupport, M
@Override
public String toString() {
- return "MailQueue:" + queueName;
+ return "MailQueue:" + queueName.asString();
}
/**
@@ -495,7 +496,7 @@ public class JMSCacheableMailQueue implements ManageableMailQueue, JMSSupport, M
return Iterators.size(new EnumerationIterator(enumeration));
} catch (Exception e) {
LOGGER.error("Unable to get size of queue {}", queueName, e);
- throw new MailQueueException("Unable to get size of queue " + queueName, e);
+ throw new MailQueueException("Unable to get size of queue " + queueName.asString(), e);
}
}
@@ -504,7 +505,7 @@ public class JMSCacheableMailQueue implements ManageableMailQueue, JMSSupport, M
boolean first = true;
long count = 0;
try (Session session = connection.createSession(true, Session.SESSION_TRANSACTED)) {
- Queue queue = session.createQueue(queueName);
+ Queue queue = session.createQueue(queueName.asString());
try (MessageConsumer consumer = session.createConsumer(queue)) {
try (MessageProducer producer = session.createProducer(queue)) {
@@ -531,7 +532,7 @@ public class JMSCacheableMailQueue implements ManageableMailQueue, JMSSupport, M
}
} catch (Exception e) {
LOGGER.error("Unable to flush mail", e);
- throw new MailQueueException("Unable to get size of queue " + queueName, e);
+ throw new MailQueueException("Unable to get size of queue " + queueName.asString(), e);
}
}
@@ -560,7 +561,7 @@ public class JMSCacheableMailQueue implements ManageableMailQueue, JMSSupport, M
try {
try (Session session = connection.createSession(true, Session.SESSION_TRANSACTED)) {
- Queue queue = session.createQueue(queueName);
+ Queue queue = session.createQueue(queueName.asString());
try (MessageConsumer consumer = session.createConsumer(queue, selector)) {
Message message = null;
while (first || message != null) {
@@ -675,7 +676,7 @@ public class JMSCacheableMailQueue implements ManageableMailQueue, JMSSupport, M
closeBrowser(browser);
LOGGER.error("Unable to browse queue {}", queueName, e);
- throw new MailQueueException("Unable to browse queue " + queueName, e);
+ throw new MailQueueException("Unable to browse queue " + queueName.asString(), e);
}
}
diff --git a/server/queue/queue-jms/src/main/java/org/apache/james/queue/jms/JMSMailQueueFactory.java b/server/queue/queue-jms/src/main/java/org/apache/james/queue/jms/JMSMailQueueFactory.java
index 701e0b3..dd2e12e 100644
--- a/server/queue/queue-jms/src/main/java/org/apache/james/queue/jms/JMSMailQueueFactory.java
+++ b/server/queue/queue-jms/src/main/java/org/apache/james/queue/jms/JMSMailQueueFactory.java
@@ -25,6 +25,7 @@ import org.apache.james.metrics.api.GaugeRegistry;
import org.apache.james.metrics.api.MetricFactory;
import org.apache.james.queue.api.MailQueueFactory;
import org.apache.james.queue.api.MailQueueItemDecoratorFactory;
+import org.apache.james.queue.api.MailQueueName;
import org.apache.james.queue.api.ManageableMailQueue;
import org.apache.james.queue.library.AbstractMailQueueFactory;
@@ -48,7 +49,7 @@ public class JMSMailQueueFactory extends AbstractMailQueueFactory<ManageableMail
}
@Override
- protected ManageableMailQueue createCacheableMailQueue(String name) {
+ protected ManageableMailQueue createCacheableMailQueue(MailQueueName name) {
return new JMSCacheableMailQueue(connectionFactory, mailQueueItemDecoratorFactory, name, metricFactory, gaugeRegistry);
}
diff --git a/server/queue/queue-jms/src/main/java/org/apache/james/queue/library/AbstractMailQueueFactory.java b/server/queue/queue-jms/src/main/java/org/apache/james/queue/library/AbstractMailQueueFactory.java
index 7d41b75..5fa22ef 100644
--- a/server/queue/queue-jms/src/main/java/org/apache/james/queue/library/AbstractMailQueueFactory.java
+++ b/server/queue/queue-jms/src/main/java/org/apache/james/queue/library/AbstractMailQueueFactory.java
@@ -35,6 +35,7 @@ import org.apache.james.lifecycle.api.LifecycleUtil;
import org.apache.james.queue.api.MailQueue;
import org.apache.james.queue.api.MailQueueFactory;
import org.apache.james.queue.api.MailQueueManagementMBean;
+import org.apache.james.queue.api.MailQueueName;
import org.apache.james.queue.api.ManageableMailQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -51,7 +52,7 @@ public abstract class AbstractMailQueueFactory<T extends MailQueue> implements M
public static final String MBEAN_NAME_QUEUE_PREFIX = "org.apache.james:type=component,name=queue,queue=";
- protected final Map<String, T> queues = new HashMap<>();
+ protected final Map<MailQueueName, T> queues = new HashMap<>();
private boolean useJMX = true;
private MBeanServer mbeanServer;
private final List<String> mbeans = new ArrayList<>();
@@ -71,7 +72,7 @@ public abstract class AbstractMailQueueFactory<T extends MailQueue> implements M
}
@Override
- public Set<String> listCreatedMailQueues() {
+ public Set<MailQueueName> listCreatedMailQueues() {
return queues.values()
.stream()
.map(MailQueue::getName)
@@ -95,16 +96,16 @@ public abstract class AbstractMailQueueFactory<T extends MailQueue> implements M
}
@Override
- public final synchronized Optional<T> getQueue(String name) {
+ public final synchronized Optional<T> getQueue(MailQueueName name) {
return Optional.ofNullable(queues.get(name));
}
@Override
- public synchronized T createQueue(String name) {
+ public synchronized T createQueue(MailQueueName name) {
return getQueue(name).orElseGet(() -> createAndRegisterQueue(name));
}
- private T createAndRegisterQueue(String name) {
+ private T createAndRegisterQueue(MailQueueName name) {
T queue = createCacheableMailQueue(name);
if (useJMX) {
registerMBean(name, queue);
@@ -117,11 +118,11 @@ public abstract class AbstractMailQueueFactory<T extends MailQueue> implements M
* Create a {@link MailQueue} for the given name that happens to do nothing on close()
* to be able to cache the instance
*/
- protected abstract T createCacheableMailQueue(String name);
+ protected abstract T createCacheableMailQueue(MailQueueName name);
- protected synchronized void registerMBean(String queuename, MailQueue queue) {
+ protected synchronized void registerMBean(MailQueueName queuename, MailQueue queue) {
- String mbeanName = MBEAN_NAME_QUEUE_PREFIX + queuename;
+ String mbeanName = MBEAN_NAME_QUEUE_PREFIX + queuename.asString();
try {
MailQueueManagementMBean mbean = null;
if (queue instanceof ManageableMailQueue) {
diff --git a/server/queue/queue-jms/src/test/java/org/apache/james/queue/jms/BrokerExtension.java b/server/queue/queue-jms/src/test/java/org/apache/james/queue/jms/BrokerExtension.java
index 99a6868..f1740a2 100644
--- a/server/queue/queue-jms/src/test/java/org/apache/james/queue/jms/BrokerExtension.java
+++ b/server/queue/queue-jms/src/test/java/org/apache/james/queue/jms/BrokerExtension.java
@@ -25,6 +25,7 @@ import org.apache.activemq.broker.region.policy.PolicyEntry;
import org.apache.activemq.broker.region.policy.PolicyMap;
import org.apache.activemq.plugin.StatisticsBrokerPlugin;
import org.apache.commons.text.RandomStringGenerator;
+import org.apache.james.queue.api.MailQueueName;
import org.junit.jupiter.api.extension.AfterAllCallback;
import org.junit.jupiter.api.extension.BeforeAllCallback;
import org.junit.jupiter.api.extension.ExtensionContext;
@@ -38,10 +39,10 @@ public class BrokerExtension implements ParameterResolver, BeforeAllCallback, A
public static final String STATISTICS = "Statistics";
- public static String generateRandomQueueName(BrokerService broker) {
+ public static MailQueueName generateRandomQueueName(BrokerService broker) {
String queueName = new RandomStringGenerator.Builder().withinRange('a', 'z').build().generate(10);
BrokerExtension.enablePrioritySupport(broker, queueName);
- return queueName;
+ return MailQueueName.of(queueName);
}
private static void enablePrioritySupport(BrokerService aBroker, String queueName) {
diff --git a/server/queue/queue-jms/src/test/java/org/apache/james/queue/jms/JMSCacheableMailQueueTest.java b/server/queue/queue-jms/src/test/java/org/apache/james/queue/jms/JMSCacheableMailQueueTest.java
index 061218a..fb0e189 100644
--- a/server/queue/queue-jms/src/test/java/org/apache/james/queue/jms/JMSCacheableMailQueueTest.java
+++ b/server/queue/queue-jms/src/test/java/org/apache/james/queue/jms/JMSCacheableMailQueueTest.java
@@ -29,6 +29,7 @@ import org.apache.james.queue.api.DelayedPriorityMailQueueContract;
import org.apache.james.queue.api.MailQueue;
import org.apache.james.queue.api.MailQueueMetricContract;
import org.apache.james.queue.api.MailQueueMetricExtension;
+import org.apache.james.queue.api.MailQueueName;
import org.apache.james.queue.api.ManageableMailQueue;
import org.apache.james.queue.api.PriorityManageableMailQueueContract;
import org.apache.james.queue.api.RawMailQueueItemDecoratorFactory;
@@ -53,7 +54,7 @@ public class JMSCacheableMailQueueTest implements DelayedManageableMailQueueCont
RawMailQueueItemDecoratorFactory mailQueueItemDecoratorFactory = new RawMailQueueItemDecoratorFactory();
MetricFactory metricFactory = metricTestSystem.getMetricFactory();
GaugeRegistry gaugeRegistry = metricTestSystem.getSpyGaugeRegistry();
- String queueName = BrokerExtension.generateRandomQueueName(broker);
+ MailQueueName queueName = BrokerExtension.generateRandomQueueName(broker);
mailQueue = new JMSCacheableMailQueue(connectionFactory, mailQueueItemDecoratorFactory, queueName, metricFactory, gaugeRegistry);
}
diff --git a/server/queue/queue-jms/src/test/java/org/apache/james/queue/library/AbstractMailQueueFactoryTest.java b/server/queue/queue-jms/src/test/java/org/apache/james/queue/library/AbstractMailQueueFactoryTest.java
index 2f25606..ff6f8fc 100644
--- a/server/queue/queue-jms/src/test/java/org/apache/james/queue/library/AbstractMailQueueFactoryTest.java
+++ b/server/queue/queue-jms/src/test/java/org/apache/james/queue/library/AbstractMailQueueFactoryTest.java
@@ -29,14 +29,15 @@ import javax.management.InstanceNotFoundException;
import javax.management.MBeanServer;
import javax.management.ObjectName;
+import org.apache.james.queue.api.MailQueueName;
import org.apache.james.queue.api.ManageableMailQueue;
import org.junit.Before;
import org.junit.Test;
public class AbstractMailQueueFactoryTest {
- private static final String QUEUE_1 = "queue1";
- private static final String QUEUE_2 = "queue2";
- private static final String QUEUE_3 = "queue3";
+ private static final MailQueueName QUEUE_1 = MailQueueName.of("queue1");
+ private static final MailQueueName QUEUE_2 = MailQueueName.of("queue2");
+ private static final MailQueueName QUEUE_3 = MailQueueName.of("queue3");
private AbstractMailQueueFactory<?> abstractMailQueueFactory;
private MBeanServer mBeanServer;
@@ -46,7 +47,7 @@ public class AbstractMailQueueFactoryTest {
mBeanServer = mock(MBeanServer.class);
abstractMailQueueFactory = new AbstractMailQueueFactory<ManageableMailQueue>() {
@Override
- protected ManageableMailQueue createCacheableMailQueue(String name) {
+ protected ManageableMailQueue createCacheableMailQueue(MailQueueName name) {
return mock(ManageableMailQueue.class);
}
};
@@ -56,7 +57,7 @@ public class AbstractMailQueueFactoryTest {
@Test
public void destroyShouldRegisterManageableQueues() throws Exception {
abstractMailQueueFactory.createQueue(QUEUE_1);
- verify(mBeanServer).registerMBean(any(MailQueueManagement.class), eq(new ObjectName(AbstractMailQueueFactory.MBEAN_NAME_QUEUE_PREFIX + QUEUE_1)));
+ verify(mBeanServer).registerMBean(any(MailQueueManagement.class), eq(new ObjectName(AbstractMailQueueFactory.MBEAN_NAME_QUEUE_PREFIX + QUEUE_1.asString())));
}
@Test
@@ -65,16 +66,16 @@ public class AbstractMailQueueFactoryTest {
abstractMailQueueFactory.createQueue(QUEUE_2);
abstractMailQueueFactory.createQueue(QUEUE_3);
abstractMailQueueFactory.destroy();
- verify(mBeanServer).unregisterMBean(eq(new ObjectName(AbstractMailQueueFactory.MBEAN_NAME_QUEUE_PREFIX + QUEUE_1)));
- verify(mBeanServer).unregisterMBean(eq(new ObjectName(AbstractMailQueueFactory.MBEAN_NAME_QUEUE_PREFIX + QUEUE_2)));
- verify(mBeanServer).unregisterMBean(eq(new ObjectName(AbstractMailQueueFactory.MBEAN_NAME_QUEUE_PREFIX + QUEUE_3)));
+ verify(mBeanServer).unregisterMBean(eq(new ObjectName(AbstractMailQueueFactory.MBEAN_NAME_QUEUE_PREFIX + QUEUE_1.asString())));
+ verify(mBeanServer).unregisterMBean(eq(new ObjectName(AbstractMailQueueFactory.MBEAN_NAME_QUEUE_PREFIX + QUEUE_2.asString())));
+ verify(mBeanServer).unregisterMBean(eq(new ObjectName(AbstractMailQueueFactory.MBEAN_NAME_QUEUE_PREFIX + QUEUE_3.asString())));
}
@Test
public void unregisterMBeanShouldWork() throws Exception {
abstractMailQueueFactory.createQueue(QUEUE_1);
- abstractMailQueueFactory.unregisterMBean(AbstractMailQueueFactory.MBEAN_NAME_QUEUE_PREFIX + QUEUE_1);
- verify(mBeanServer).unregisterMBean(eq(new ObjectName(AbstractMailQueueFactory.MBEAN_NAME_QUEUE_PREFIX + QUEUE_1)));
+ abstractMailQueueFactory.unregisterMBean(AbstractMailQueueFactory.MBEAN_NAME_QUEUE_PREFIX + QUEUE_1.asString());
+ verify(mBeanServer).unregisterMBean(eq(new ObjectName(AbstractMailQueueFactory.MBEAN_NAME_QUEUE_PREFIX + QUEUE_1.asString())));
}
@Test
@@ -85,11 +86,11 @@ public class AbstractMailQueueFactoryTest {
doThrow(InstanceNotFoundException.class)
.doNothing()
.when(mBeanServer)
- .unregisterMBean(new ObjectName(AbstractMailQueueFactory.MBEAN_NAME_QUEUE_PREFIX + QUEUE_1));
+ .unregisterMBean(new ObjectName(AbstractMailQueueFactory.MBEAN_NAME_QUEUE_PREFIX + QUEUE_1.asString()));
abstractMailQueueFactory.destroy();
- verify(mBeanServer).unregisterMBean(eq(new ObjectName(AbstractMailQueueFactory.MBEAN_NAME_QUEUE_PREFIX + QUEUE_1)));
- verify(mBeanServer).unregisterMBean(eq(new ObjectName(AbstractMailQueueFactory.MBEAN_NAME_QUEUE_PREFIX + QUEUE_2)));
- verify(mBeanServer).unregisterMBean(eq(new ObjectName(AbstractMailQueueFactory.MBEAN_NAME_QUEUE_PREFIX + QUEUE_3)));
+ verify(mBeanServer).unregisterMBean(eq(new ObjectName(AbstractMailQueueFactory.MBEAN_NAME_QUEUE_PREFIX + QUEUE_1.asString())));
+ verify(mBeanServer).unregisterMBean(eq(new ObjectName(AbstractMailQueueFactory.MBEAN_NAME_QUEUE_PREFIX + QUEUE_2.asString())));
+ verify(mBeanServer).unregisterMBean(eq(new ObjectName(AbstractMailQueueFactory.MBEAN_NAME_QUEUE_PREFIX + QUEUE_3.asString())));
}
}
diff --git a/server/queue/queue-memory/src/main/java/org/apache/james/queue/memory/MemoryMailQueueFactory.java b/server/queue/queue-memory/src/main/java/org/apache/james/queue/memory/MemoryMailQueueFactory.java
index 6c3d8f3..fa949e8 100644
--- a/server/queue/queue-memory/src/main/java/org/apache/james/queue/memory/MemoryMailQueueFactory.java
+++ b/server/queue/queue-memory/src/main/java/org/apache/james/queue/memory/MemoryMailQueueFactory.java
@@ -42,6 +42,7 @@ import org.apache.james.core.MailAddress;
import org.apache.james.queue.api.MailQueue;
import org.apache.james.queue.api.MailQueueFactory;
import org.apache.james.queue.api.MailQueueItemDecoratorFactory;
+import org.apache.james.queue.api.MailQueueName;
import org.apache.james.queue.api.ManageableMailQueue;
import org.apache.james.server.core.MailImpl;
import org.apache.mailet.Mail;
@@ -59,7 +60,7 @@ import reactor.core.scheduler.Schedulers;
public class MemoryMailQueueFactory implements MailQueueFactory<ManageableMailQueue> {
- private final ConcurrentHashMap<String, MemoryCacheableMailQueue> mailQueues;
+ private final ConcurrentHashMap<MailQueueName, MemoryCacheableMailQueue> mailQueues;
private final MailQueueItemDecoratorFactory mailQueueItemDecoratorFactory;
@Inject
@@ -69,7 +70,7 @@ public class MemoryMailQueueFactory implements MailQueueFactory<ManageableMailQu
}
@Override
- public Set<String> listCreatedMailQueues() {
+ public Set<MailQueueName> listCreatedMailQueues() {
return mailQueues.values()
.stream()
.map(MemoryCacheableMailQueue::getName)
@@ -77,22 +78,22 @@ public class MemoryMailQueueFactory implements MailQueueFactory<ManageableMailQu
}
@Override
- public Optional<ManageableMailQueue> getQueue(String name) {
+ public Optional<ManageableMailQueue> getQueue(MailQueueName name) {
return Optional.ofNullable(mailQueues.get(name));
}
@Override
- public MemoryCacheableMailQueue createQueue(String name) {
+ public MemoryCacheableMailQueue createQueue(MailQueueName name) {
return mailQueues.computeIfAbsent(name, mailQueueName -> new MemoryCacheableMailQueue(mailQueueName, mailQueueItemDecoratorFactory));
}
public static class MemoryCacheableMailQueue implements ManageableMailQueue {
private final DelayQueue<MemoryMailQueueItem> mailItems;
private final LinkedBlockingDeque<MemoryMailQueueItem> inProcessingMailItems;
- private final String name;
+ private final MailQueueName name;
private final Flux<MailQueueItem> flux;
- public MemoryCacheableMailQueue(String name, MailQueueItemDecoratorFactory mailQueueItemDecoratorFactory) {
+ public MemoryCacheableMailQueue(MailQueueName name, MailQueueItemDecoratorFactory mailQueueItemDecoratorFactory) {
this.mailItems = new DelayQueue<>();
this.inProcessingMailItems = new LinkedBlockingDeque<>();
this.name = name;
@@ -110,7 +111,7 @@ public class MemoryMailQueueFactory implements MailQueueFactory<ManageableMailQu
}
@Override
- public String getName() {
+ public MailQueueName getName() {
return name;
}
diff --git a/server/queue/queue-memory/src/test/java/org/apache/james/queue/memory/MemoryCacheableMailQueueTest.java b/server/queue/queue-memory/src/test/java/org/apache/james/queue/memory/MemoryCacheableMailQueueTest.java
index 5868995..f8b4566 100644
--- a/server/queue/queue-memory/src/test/java/org/apache/james/queue/memory/MemoryCacheableMailQueueTest.java
+++ b/server/queue/queue-memory/src/test/java/org/apache/james/queue/memory/MemoryCacheableMailQueueTest.java
@@ -24,6 +24,7 @@ import static org.assertj.core.api.Assertions.assertThat;
import org.apache.james.queue.api.DelayedManageableMailQueueContract;
import org.apache.james.queue.api.MailQueue;
+import org.apache.james.queue.api.MailQueueName;
import org.apache.james.queue.api.ManageableMailQueue;
import org.apache.james.queue.api.RawMailQueueItemDecoratorFactory;
import org.junit.jupiter.api.BeforeEach;
@@ -35,7 +36,7 @@ public class MemoryCacheableMailQueueTest implements DelayedManageableMailQueueC
@BeforeEach
public void setUp() {
- mailQueue = new MemoryMailQueueFactory.MemoryCacheableMailQueue("test", new RawMailQueueItemDecoratorFactory());
+ mailQueue = new MemoryMailQueueFactory.MemoryCacheableMailQueue(MailQueueName.of("test"), new RawMailQueueItemDecoratorFactory());
}
@Override
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueue.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueue.java
index b09fe30..822fc5a 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueue.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueue.java
@@ -63,8 +63,8 @@ public class RabbitMQMailQueue implements ManageableMailQueue {
}
@Override
- public String getName() {
- return name.asString();
+ public org.apache.james.queue.api.MailQueueName getName() {
+ return org.apache.james.queue.api.MailQueueName.of(name.asString());
}
@Override
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueFactory.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueFactory.java
index cc328c0..3805e74 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueFactory.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueFactory.java
@@ -113,7 +113,7 @@ public class RabbitMQMailQueueFactory implements MailQueueFactory<RabbitMQMailQu
private void registerGaugeFor(RabbitMQMailQueue rabbitMQMailQueue) {
if (configuration.isSizeMetricsEnabled()) {
- this.gaugeRegistry.register(QUEUE_SIZE_METRIC_NAME_PREFIX + rabbitMQMailQueue.getName(), rabbitMQMailQueue::getSize);
+ this.gaugeRegistry.register(QUEUE_SIZE_METRIC_NAME_PREFIX + rabbitMQMailQueue.getName().asString(), rabbitMQMailQueue::getSize);
}
}
}
@@ -133,21 +133,22 @@ public class RabbitMQMailQueueFactory implements MailQueueFactory<RabbitMQMailQu
}
@Override
- public Optional<RabbitMQMailQueue> getQueue(String name) {
- return getQueueFromRabbitServer(MailQueueName.fromString(name));
+ public Optional<RabbitMQMailQueue> getQueue(org.apache.james.queue.api.MailQueueName name) {
+ return getQueueFromRabbitServer(MailQueueName.fromString(name.asString()));
}
@Override
- public RabbitMQMailQueue createQueue(String name) {
- MailQueueName mailQueueName = MailQueueName.fromString(name);
+ public RabbitMQMailQueue createQueue(org.apache.james.queue.api.MailQueueName name) {
+ MailQueueName mailQueueName = MailQueueName.fromString(name.asString());
return getQueueFromRabbitServer(mailQueueName)
.orElseGet(() -> createQueueIntoRabbitServer(mailQueueName));
}
@Override
- public Set<String> listCreatedMailQueues() {
+ public Set<org.apache.james.queue.api.MailQueueName> listCreatedMailQueues() {
return mqManagementApi.listCreatedMailQueueNames()
.map(MailQueueName::asString)
+ .map(org.apache.james.queue.api.MailQueueName::of)
.collect(ImmutableSet.toImmutableSet());
}
diff --git a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueConfigurationChangeTest.java b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueConfigurationChangeTest.java
index 35792e7..f79622e 100644
--- a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueConfigurationChangeTest.java
+++ b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueConfigurationChangeTest.java
@@ -44,6 +44,7 @@ import org.apache.james.eventsourcing.eventstore.cassandra.CassandraEventStoreMo
import org.apache.james.metrics.api.NoopGaugeRegistry;
import org.apache.james.metrics.tests.RecordingMetricFactory;
import org.apache.james.queue.api.MailQueue;
+import org.apache.james.queue.api.MailQueueName;
import org.apache.james.queue.api.ManageableMailQueue;
import org.apache.james.queue.api.RawMailQueueItemDecoratorFactory;
import org.apache.james.queue.rabbitmq.view.RabbitMQMailQueueConfiguration;
@@ -71,7 +72,7 @@ class RabbitMQMailQueueConfigurationChangeTest {
.updateBrowseStartPace(UPDATE_BROWSE_START_PACE)
.sliceWindow(ONE_HOUR_SLICE_WINDOW)
.build();
- private static final String SPOOL = "spool";
+ private static final MailQueueName SPOOL = MailQueueName.of("spool");
private static final Instant IN_SLICE_1 = Instant.parse("2007-12-03T10:15:30.00Z");
private static final Instant IN_SLICE_2 = IN_SLICE_1.plus(1, HOURS);
private static final Instant IN_SLICE_3 = IN_SLICE_1.plus(2, HOURS);
diff --git a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java
index 57783d2..1b7e610 100644
--- a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java
+++ b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java
@@ -79,7 +79,7 @@ class RabbitMQMailQueueTest {
private static final int THREE_BUCKET_COUNT = 3;
private static final int UPDATE_BROWSE_START_PACE = 2;
private static final Duration ONE_HOUR_SLICE_WINDOW = Duration.ofHours(1);
- private static final String SPOOL = "spool";
+ private static final org.apache.james.queue.api.MailQueueName SPOOL = org.apache.james.queue.api.MailQueueName.of("spool");
private static final Instant IN_SLICE_1 = Instant.parse("2007-12-03T10:15:30.00Z");
private static final Instant IN_SLICE_2 = IN_SLICE_1.plus(1, HOURS);
private static final Instant IN_SLICE_3 = IN_SLICE_1.plus(2, HOURS);
@@ -198,10 +198,10 @@ class RabbitMQMailQueueTest {
@Test
void mailQueueShouldBeInitializedWhenCreating(CassandraCluster cassandra) {
- String name = "myQueue";
+ org.apache.james.queue.api.MailQueueName name = org.apache.james.queue.api.MailQueueName.of("myQueue");
mailQueueFactory.createQueue(name);
- boolean initialized = CassandraMailQueueViewTestFactory.isInitialized(cassandra.getConf(), MailQueueName.fromString(name));
+ boolean initialized = CassandraMailQueueViewTestFactory.isInitialized(cassandra.getConf(), MailQueueName.fromString(name.asString()));
assertThat(initialized).isTrue();
}
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org