You are viewing a plain text version of this content. The canonical link for it is here.
Posted to server-dev@james.apache.org by bt...@apache.org on 2019/06/26 03:28:40 UTC
[james-project] branch master updated (3d39f3c -> 581b64b)
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git.
from 3d39f3c JAMES-2294 s/STATE_TRANSPORT/TRANSPORT_PROCESSOR
new 98f9ed4 JAMES-2794 RabbitMQ should not dequeue deleted elements
new a54fe44 JAMES-2794 Solve an IntelliJ warning
new 49939ce JAMES-2794 Demonstrate reprocessing was broken on top of RabbitMQ mailQueue
new 3e6f9f3 JAMES-2794 Create an EnqueueId POJO in RabbitMQMailQueue
new b29d2e6 JAMES-2794 MailQueue should deliver multiple time the same mail
new 961ef33 JAMES-2794 RabbitMQ MailQueue projections should rely on EnqueueId
new 7f9a7cb JAMES-2794 Refactor DeleteCondition to implement WithEnqueueId
new e4d5c50 JAMES-2794 Reprocessing works on top of the distributed RabbitMQ JAMES
new a0c2fe2 JAMES-2794 Upgrade instructions
new aad4442 JAMES-2794 Dequeuer should rely on concatMap and not flatMap
new 9e9e935 JAMES-2794 Avoid calling *block* in doOnNext
new 9d6feef JAMES-2794 Avoid relying on a DTO in RabbitMQ MailQueue logic
new 4deed8b JAMES-2794 s/EnQueueId/EnqueueId
new 4627075 JAMES-2794 Create an EnqueueId POJO in RabbitMQMailQueue
new 92f9e65 JAMES-2794 Solve some "channel closed" exceptions on topof RabbitMQ MailQueue
new 1785ab5 JAMES-2794 Avoid relying on a DTO in RabbitMQ MailQueue logic
new eeaa8fe JAMES-2794 RabbitMQ should not dequeue deleted elements
new 16617ab JAMES-2794 Replace a pair by MailWithEnqueueId
new edd3280 JAMES-2794 ack should update the view using an EnqueueId reference
new e329898 JAMES-2803 allow configuration of ES request timeout
new a7b2cf7 JAMES-2803 ElasticSearch cluster name is no longer useful when using REST API
new be0233b JAMES-2803 Always close ES connections
new a867be6 JAMES-2803 Finally shorten ReindexingWithEventDeadLettersTest by configuring request timeout
new 2be4ed2 JAMES-2659 dispatch result should always be waited for success
new 6e53870 JAMES-2659 don't expect too much from eventBus stop
new 45d2f06 JAMES-2774 Remove useless comment & rename a test in RabbitMQEventBusTest
new 6884456 JAMES-2774 Remove an useless counter field in EventDispatcher
new 581b64b JAMES-2774 log when fail to dispatch in EventDispatcher
The 28 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:
.../apache/james/backends/es/ClientProvider.java | 68 +++++++-
.../james/backends/es/ClientProviderImpl.java | 80 ---------
.../backends/es/ElasticSearchConfiguration.java | 42 ++---
.../es/ClientProviderImplConnectionTest.java | 30 ++--
.../james/backends/es/ClientProviderImplTest.java | 142 ----------------
...{AliasNameTest.java => ClientProviderTest.java} | 13 +-
.../james/backends/es/DockerElasticSearch.java | 11 +-
.../james/backends/es/DockerElasticSearchRule.java | 2 +-
.../es/ElasticSearchConfigurationTest.java | 46 ------
.../backends/es/ElasticSearchIndexerTest.java | 90 +++++-----
.../backends/es/IndexCreationFactoryTest.java | 17 +-
.../james/backends/es/NodeMappingFactoryTest.java | 23 ++-
.../backends/es/search/ScrolledSearchTest.java | 181 ++++++++++-----------
.../ElasticSearchIntegrationTest.java | 13 +-
.../james/mailbox/events/EventDispatcher.java | 4 +-
.../james/mailbox/events/RabbitMQEventBusTest.java | 25 ++-
...lasticSearchQuotaSearchTestSystemExtension.java | 13 +-
.../ElasticSearchQuotaMailboxListenerTest.java | 6 +
.../host/ElasticSearchHostSystem.java | 7 +-
.../modules/mailbox/ElasticSearchClientModule.java | 40 +----
.../apache/james/DockerElasticSearchExtension.java | 11 ++
.../modules/TestDockerElasticSearchModule.java | 16 +-
.../james/RabbitMQJamesServerReprocessingTest.java | 93 +++++++++++
.../ReindexingWithEventDeadLettersTest.java | 2 +-
.../routes/ElasticSearchQuotaSearchExtension.java | 9 +-
.../queue/activemq/ActiveMQMailQueueBlobTest.java | 7 +
.../queue/activemq/ActiveMQMailQueueTest.java | 7 +
.../apache/james/queue/api/MailQueueContract.java | 25 +++
.../queue/api/ManageableMailQueueContract.java | 9 +-
.../apache/james/queue/jms/JMSMailQueueTest.java | 7 +
.../org/apache/james/queue/rabbitmq/Dequeuer.java | 45 +++--
.../org/apache/james/queue/rabbitmq/EnqueueId.java | 49 +++---
.../apache/james/queue/rabbitmq/EnqueuedItem.java | 37 +++--
.../org/apache/james/queue/rabbitmq/Enqueuer.java | 18 +-
.../apache/james/queue/rabbitmq/MailLoader.java | 15 +-
.../james/queue/rabbitmq/MailReference.java} | 33 ++--
.../james/queue/rabbitmq/MailReferenceDTO.java | 41 +++--
.../james/queue/rabbitmq/MailWithEnqueueId.java} | 25 +--
.../queue/rabbitmq/RabbitMQMailQueueFactory.java | 3 +-
.../queue/rabbitmq/view/api/DeleteCondition.java | 59 +++++--
.../queue/rabbitmq/view/api/MailQueueView.java | 4 +-
.../view/cassandra/CassandraMailQueueBrowser.java | 3 +-
.../cassandra/CassandraMailQueueMailDelete.java | 19 +--
.../view/cassandra/CassandraMailQueueView.java | 21 ++-
.../cassandra/CassandraMailQueueViewModule.java | 15 +-
.../rabbitmq/view/cassandra/DeletedMailsDAO.java | 21 +--
.../rabbitmq/view/cassandra/EnqueuedMailsDAO.java | 9 +-
.../view/cassandra/EnqueuedMailsDaoUtil.java | 8 +-
.../rabbitmq/view/cassandra/model/MailKey.java | 65 --------
.../james/queue/rabbitmq/EnqueuedItemTest.java | 17 ++
.../apache/james/queue/rabbitmq/MailDTOTest.java | 23 ++-
.../queue/rabbitmq/RabbitMQMailQueueTest.java | 8 -
.../rabbitmq/view/api/DeleteConditionTest.java | 111 ++++++++++---
.../view/cassandra/DeletedMailsDAOTest.java | 35 ++--
.../view/cassandra/EnqueuedMailsDaoTest.java | 17 +-
.../view/cassandra/model/EnqueueIdTest.java} | 37 +++--
.../model/EnqueuedItemWithSlicingContextTest.java | 5 +-
.../rabbitmq/view/cassandra/model/MailKeyTest.java | 33 ----
.../src/test/resources/json/mail1.json | 1 +
.../src/test/resources/json/mail_min.json | 1 +
upgrade-instructions.md | 35 ++++
61 files changed, 957 insertions(+), 895 deletions(-)
delete mode 100644 backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ClientProviderImpl.java
delete mode 100644 backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ClientProviderImplTest.java
copy backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/{AliasNameTest.java => ClientProviderTest.java} (81%)
create mode 100644 server/container/guice/cassandra-rabbitmq-guice/src/test/java/org/apache/james/RabbitMQJamesServerReprocessingTest.java
copy mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/EventBusId.java => server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/EnqueueId.java (65%)
copy server/queue/{queue-api/src/main/java/org/apache/james/queue/api/RawMailQueueItem.java => queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailReference.java} (68%)
copy server/{mailet/integration-testing/src/test/java/org/apache/james/transport/mailets/ErrorMatcher.java => queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailWithEnqueueId.java} (76%)
delete mode 100644 server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/model/MailKey.java
copy server/{container/util/src/test/java/org/apache/james/util/streams/OffsetTest.java => queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/model/EnqueueIdTest.java} (62%)
delete mode 100644 server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/model/MailKeyTest.java
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 15/28: JAMES-2794 Solve some "channel closed"
exceptions on topof RabbitMQ MailQueue
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 92f9e6543c09bc3fdd818b5598afff1039d1d73f
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Wed Jun 19 14:41:45 2019 +0700
JAMES-2794 Solve some "channel closed" exceptions on topof RabbitMQ MailQueue
Doing the ack operations out of the stream caused some 'channel closed' exception
to arise.
---
.../org/apache/james/queue/api/ManageableMailQueueContract.java | 9 +++++++--
1 file changed, 7 insertions(+), 2 deletions(-)
diff --git a/server/queue/queue-api/src/test/java/org/apache/james/queue/api/ManageableMailQueueContract.java b/server/queue/queue-api/src/test/java/org/apache/james/queue/api/ManageableMailQueueContract.java
index c911e64..521bd44 100644
--- a/server/queue/queue-api/src/test/java/org/apache/james/queue/api/ManageableMailQueueContract.java
+++ b/server/queue/queue-api/src/test/java/org/apache/james/queue/api/ManageableMailQueueContract.java
@@ -40,6 +40,7 @@ import org.apache.mailet.Mail;
import org.apache.mailet.base.MailAddressFixture;
import org.junit.jupiter.api.Test;
+import com.github.fge.lambdas.Throwing;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Iterators;
@@ -79,7 +80,9 @@ public interface ManageableMailQueueContract extends MailQueueContract {
default void dequeueShouldDecreaseQueueSize() throws Exception {
enQueue(defaultMail().name("name").build());
- Flux.from(getManageableMailQueue().deQueue()).blockFirst().done(true);
+ Flux.from(getManageableMailQueue().deQueue())
+ .doOnNext(Throwing.consumer(item -> item.done(true)))
+ .blockFirst();
long size = getManageableMailQueue().getSize();
@@ -90,7 +93,9 @@ public interface ManageableMailQueueContract extends MailQueueContract {
default void noAckShouldNotDecreaseSize() throws Exception {
enQueue(defaultMail().name("name").build());
- Flux.from(getManageableMailQueue().deQueue()).blockFirst().done(false);
+ Flux.from(getManageableMailQueue().deQueue())
+ .doOnNext(Throwing.consumer(item -> item.done(false)))
+ .blockFirst();
long size = getManageableMailQueue().getSize();
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 03/28: JAMES-2794 Demonstrate reprocessing was
broken on top of RabbitMQ mailQueue
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 49939ce407d60e24b04efea3da430af3972a7789
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Mon Jun 17 13:32:02 2019 +0700
JAMES-2794 Demonstrate reprocessing was broken on top of RabbitMQ mailQueue
Because the reprocessed mail had the same name, it was assumed to had been already pr
---
.../james/RabbitMQJamesServerReprocessingTest.java | 94 ++++++++++++++++++++++
1 file changed, 94 insertions(+)
diff --git a/server/container/guice/cassandra-rabbitmq-guice/src/test/java/org/apache/james/RabbitMQJamesServerReprocessingTest.java b/server/container/guice/cassandra-rabbitmq-guice/src/test/java/org/apache/james/RabbitMQJamesServerReprocessingTest.java
new file mode 100644
index 0000000..bc62128
--- /dev/null
+++ b/server/container/guice/cassandra-rabbitmq-guice/src/test/java/org/apache/james/RabbitMQJamesServerReprocessingTest.java
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.james;
+
+import static io.restassured.config.ParamConfig.UpdateStrategy.REPLACE;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.apache.james.mailrepository.api.MailRepositoryUrl;
+import org.apache.james.modules.RabbitMQExtension;
+import org.apache.james.modules.TestJMAPServerModule;
+import org.apache.james.modules.protocols.SmtpGuiceProbe;
+import org.apache.james.utils.MailRepositoryProbeImpl;
+import org.apache.james.utils.SMTPMessageSender;
+import org.apache.james.utils.WebAdminGuiceProbe;
+import org.apache.james.webadmin.WebAdminConfiguration;
+import org.apache.james.webadmin.WebAdminUtils;
+import org.awaitility.Awaitility;
+import org.awaitility.Duration;
+import org.awaitility.core.ConditionFactory;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.restassured.RestAssured;
+import io.restassured.config.ParamConfig;
+import io.restassured.parsing.Parser;
+import io.restassured.specification.RequestSpecification;
+
+class RabbitMQJamesServerReprocessingTest {
+ private static final ConditionFactory AWAIT = Awaitility.await().atMost(Duration.ONE_MINUTE);
+ private static final MailRepositoryUrl SENDER_DENIED = MailRepositoryUrl.from("cassandra://var/mail/sender-denied/");
+ private RabbitMQExtension rabbitMQExtension = new RabbitMQExtension();
+ private RequestSpecification webAdminApi;
+
+ @RegisterExtension
+ JamesServerExtension jamesServerExtension = CassandraRabbitMQJamesServerFixture
+ .baseExtensionBuilder(rabbitMQExtension)
+ .server(configuration -> GuiceJamesServer
+ .forConfiguration(configuration)
+ .combineWith(CassandraRabbitMQJamesServerMain.MODULES)
+ .overrideWith(new TestJMAPServerModule(10))
+ .overrideWith(binder -> binder.bind(WebAdminConfiguration.class).toInstance(WebAdminConfiguration.TEST_CONFIGURATION))
+ .overrideWith(JmapJamesServerContract.DOMAIN_LIST_CONFIGURATION_MODULE))
+ .build();
+
+ @BeforeEach
+ void setUp(GuiceJamesServer server) {
+ RestAssured.defaultParser = Parser.JSON;
+ webAdminApi = WebAdminUtils.spec(server.getProbe(WebAdminGuiceProbe.class).getWebAdminPort())
+ .config(WebAdminUtils.defaultConfig()
+ .paramConfig(new ParamConfig(REPLACE, REPLACE, REPLACE)));
+ }
+
+ @Disabled("JAMES-2733 Reprocessing is broken for RabbitMQ mailQueue - the reprocessed mail name is preserved and" +
+ " is thus considered deleted.")
+ @Test
+ void reprocessingADeniedMailShouldNotLooseIt(GuiceJamesServer server) throws Exception {
+ new SMTPMessageSender("other.com")
+ .connect("127.0.0.1", server.getProbe(SmtpGuiceProbe.class).getSmtpPort())
+ .sendMessage("denied@other.com", "any@domain.tld");
+
+ MailRepositoryProbeImpl mailRepositoryProbe = server.getProbe(MailRepositoryProbeImpl.class);
+ AWAIT.until(() -> mailRepositoryProbe.listMailKeys(SENDER_DENIED).size() == 1);
+
+ String taskId = webAdminApi
+ .param("action", "reprocess")
+ .patch("/mailRepositories/var%2Fmail%2Fsender-denied/mails")
+ .jsonPath()
+ .get("taskId");
+
+ webAdminApi.get("/tasks/" + taskId + "/await");
+
+ AWAIT.until(() -> mailRepositoryProbe.listMailKeys(SENDER_DENIED).size() == 1);
+ assertThat(mailRepositoryProbe.listMailKeys(SENDER_DENIED)).hasSize(1);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 25/28: JAMES-2659 don't expect too much from
eventBus stop
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 6e53870f093046bfa6f9c12af7df1177c8c13b99
Author: Matthieu Baechler <ma...@apache.org>
AuthorDate: Fri Jun 21 14:31:23 2019 +0200
JAMES-2659 don't expect too much from eventBus stop
---
.../org/apache/james/mailbox/events/RabbitMQEventBusTest.java | 11 +++++------
1 file changed, 5 insertions(+), 6 deletions(-)
diff --git a/mailbox/event/event-rabbitmq/src/test/java/org/apache/james/mailbox/events/RabbitMQEventBusTest.java b/mailbox/event/event-rabbitmq/src/test/java/org/apache/james/mailbox/events/RabbitMQEventBusTest.java
index 50df7d4..dac6986 100644
--- a/mailbox/event/event-rabbitmq/src/test/java/org/apache/james/mailbox/events/RabbitMQEventBusTest.java
+++ b/mailbox/event/event-rabbitmq/src/test/java/org/apache/james/mailbox/events/RabbitMQEventBusTest.java
@@ -65,6 +65,7 @@ import org.apache.james.mailbox.model.TestMessageId;
import org.apache.james.mailbox.util.EventCollector;
import org.apache.james.metrics.api.NoopMetricFactory;
import org.apache.james.util.concurrency.ConcurrentTestRunner;
+import org.assertj.core.data.Percentage;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Disabled;
@@ -559,7 +560,7 @@ class RabbitMQEventBusTest implements GroupContract.SingleEventBusGroupContract,
}
@Test
- void registrationsShouldNotHandleEventsAfterStop() throws Exception {
+ void dispatchShouldStopDeliveringEventsShortlyAfterStopIsCalled() throws Exception {
eventBus.start();
MailboxListenerCountingSuccessfulExecution listener = new MailboxListenerCountingSuccessfulExecution();
@@ -579,8 +580,7 @@ class RabbitMQEventBusTest implements GroupContract.SingleEventBusGroupContract,
TimeUnit.SECONDS.sleep(1);
assertThat(listener.numberOfEventCalls())
- .isEqualTo(callsAfterStop)
- .isLessThanOrEqualTo(MAX_EVENT_DISPATCHED_COUNT);
+ .isCloseTo(callsAfterStop, Percentage.withPercentage(2));
}
}
}
@@ -650,7 +650,7 @@ class RabbitMQEventBusTest implements GroupContract.SingleEventBusGroupContract,
}
@Test
- void registrationsShouldNotHandleEventsAfterStop() throws Exception {
+ void dispatchShouldStopDeliveringEventsShortlyAfterStopIsCalled() throws Exception {
eventBus.start();
eventBus2.start();
@@ -672,8 +672,7 @@ class RabbitMQEventBusTest implements GroupContract.SingleEventBusGroupContract,
TimeUnit.SECONDS.sleep(1);
assertThat(listener.numberOfEventCalls())
- .isEqualTo(callsAfterStop)
- .isLessThanOrEqualTo(MAX_EVENT_DISPATCHED_COUNT);
+ .isCloseTo(callsAfterStop, Percentage.withPercentage(2));
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 05/28: JAMES-2794 MailQueue should deliver multiple
time the same mail
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit b29d2e6b01f75be43713572f2d99dcf636190997
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Tue Jun 18 11:48:55 2019 +0700
JAMES-2794 MailQueue should deliver multiple time the same mail
Multiple features are based on this assumption, not yet tested within the MailQueueContract:
- Reprocessing
- RRT rewrites to a distant server
- Some remoteDelivery bouncing options
---
.../queue/activemq/ActiveMQMailQueueBlobTest.java | 7 ++++++
.../queue/activemq/ActiveMQMailQueueTest.java | 7 ++++++
.../apache/james/queue/api/MailQueueContract.java | 25 ++++++++++++++++++++++
.../apache/james/queue/jms/JMSMailQueueTest.java | 7 ++++++
.../queue/rabbitmq/RabbitMQMailQueueTest.java | 7 ++++++
5 files changed, 53 insertions(+)
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 f6baa55..1b021d0 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
@@ -131,6 +131,13 @@ public class ActiveMQMailQueueBlobTest implements DelayedManageableMailQueueCont
}
@Test
+ @Override
+ @Disabled("JAMES-2794 This test never finishes")
+ public void enQueueShouldAcceptMailWithDuplicatedNames() {
+
+ }
+
+ @Test
void computeNextDeliveryTimestampShouldReturnLongMaxWhenOverflow() {
long deliveryTimestamp = mailQueue.computeNextDeliveryTimestamp(ChronoUnit.FOREVER.getDuration());
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 a4a7512..acfdbfb 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
@@ -127,6 +127,13 @@ public class ActiveMQMailQueueTest implements DelayedManageableMailQueueContract
@Test
@Override
+ @Disabled("JAMES-2794 This test never finishes")
+ public void enQueueShouldAcceptMailWithDuplicatedNames() {
+
+ }
+
+ @Test
+ @Override
@Disabled("JAMES-2544 Mixing concurrent ack/nack might lead to a deadlock")
public void concurrentEnqueueDequeueWithAckNackShouldNotFail() {
diff --git a/server/queue/queue-api/src/test/java/org/apache/james/queue/api/MailQueueContract.java b/server/queue/queue-api/src/test/java/org/apache/james/queue/api/MailQueueContract.java
index c09d292..a6492cc 100644
--- a/server/queue/queue-api/src/test/java/org/apache/james/queue/api/MailQueueContract.java
+++ b/server/queue/queue-api/src/test/java/org/apache/james/queue/api/MailQueueContract.java
@@ -37,6 +37,7 @@ import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
import javax.mail.internet.MimeMessage;
@@ -53,6 +54,7 @@ import org.junit.jupiter.api.Test;
import com.github.fge.lambdas.Throwing;
import com.google.common.base.MoreObjects;
import com.google.common.base.Strings;
+
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
@@ -110,6 +112,29 @@ public interface MailQueueContract {
}
@Test
+ default void enQueueShouldAcceptMailWithDuplicatedNames() throws Exception {
+ String name = "name";
+ FakeMail mail = FakeMail.builder()
+ .name(name)
+ .mimeMessage(createMimeMessage())
+ .recipients(RECIPIENT1, RECIPIENT2)
+ .sender(MailAddress.nullSender())
+ .lastUpdated(new Date())
+ .build();
+
+ enQueue(mail);
+ enQueue(mail);
+
+ Stream<String> dequeuedItemNames = Flux.from(getMailQueue().deQueue())
+ .take(2)
+ .map(MailQueue.MailQueueItem::getMail)
+ .map(Mail::getName)
+ .toStream();
+
+ assertThat(dequeuedItemNames).hasSize(2).containsOnly(name);
+ }
+
+ @Test
default void queueShouldHandleNoSender() throws Exception {
enQueue(FakeMail.builder()
.name("name")
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/JMSMailQueueTest.java
index a31ed09..8f92851 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/JMSMailQueueTest.java
@@ -113,4 +113,11 @@ public class JMSMailQueueTest implements DelayedManageableMailQueueContract, Pri
public void concurrentEnqueueDequeueWithAckNackShouldNotFail() {
}
+
+ @Test
+ @Override
+ @Disabled("JAMES-2794 This test never finishes")
+ public void enQueueShouldAcceptMailWithDuplicatedNames() {
+
+ }
}
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 f6d57c6..03455ff 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
@@ -231,6 +231,13 @@ public class RabbitMQMailQueueTest implements ManageableMailQueueContract, MailQ
}
+ @Test
+ @Override
+ @Disabled("JAMES-2794 This test never finishes")
+ public void enQueueShouldAcceptMailWithDuplicatedNames() {
+
+ }
+
private void enqueueSomeMails(Function<Integer, String> namePattern, int emailCount) {
IntStream.rangeClosed(1, emailCount)
.forEach(Throwing.intConsumer(i -> enQueue(defaultMail()
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 08/28: JAMES-2794 Reprocessing works on top of the
distributed RabbitMQ JAMES
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit e4d5c50ea3dd3d8e1f3c28d97e29a1cdc8b27c83
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Tue Jun 18 12:20:27 2019 +0700
JAMES-2794 Reprocessing works on top of the distributed RabbitMQ JAMES
---
.../java/org/apache/james/RabbitMQJamesServerReprocessingTest.java | 5 ++---
1 file changed, 2 insertions(+), 3 deletions(-)
diff --git a/server/container/guice/cassandra-rabbitmq-guice/src/test/java/org/apache/james/RabbitMQJamesServerReprocessingTest.java b/server/container/guice/cassandra-rabbitmq-guice/src/test/java/org/apache/james/RabbitMQJamesServerReprocessingTest.java
index bc62128..31ec83d 100644
--- a/server/container/guice/cassandra-rabbitmq-guice/src/test/java/org/apache/james/RabbitMQJamesServerReprocessingTest.java
+++ b/server/container/guice/cassandra-rabbitmq-guice/src/test/java/org/apache/james/RabbitMQJamesServerReprocessingTest.java
@@ -35,7 +35,6 @@ import org.awaitility.Awaitility;
import org.awaitility.Duration;
import org.awaitility.core.ConditionFactory;
import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
@@ -69,8 +68,6 @@ class RabbitMQJamesServerReprocessingTest {
.paramConfig(new ParamConfig(REPLACE, REPLACE, REPLACE)));
}
- @Disabled("JAMES-2733 Reprocessing is broken for RabbitMQ mailQueue - the reprocessed mail name is preserved and" +
- " is thus considered deleted.")
@Test
void reprocessingADeniedMailShouldNotLooseIt(GuiceJamesServer server) throws Exception {
new SMTPMessageSender("other.com")
@@ -86,8 +83,10 @@ class RabbitMQJamesServerReprocessingTest {
.jsonPath()
.get("taskId");
+ // Awaiting the task ensure the reprocessdid start and that the repository was emptied
webAdminApi.get("/tasks/" + taskId + "/await");
+ // Awaiting that an other mail is present in the mail repository ensures that the reprocessing successfully finished
AWAIT.until(() -> mailRepositoryProbe.listMailKeys(SENDER_DENIED).size() == 1);
assertThat(mailRepositoryProbe.listMailKeys(SENDER_DENIED)).hasSize(1);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 07/28: JAMES-2794 Refactor DeleteCondition to
implement WithEnqueueId
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 7f9a7cb7b701c7da260e1c64c1620d615bfb971c
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Tue Jun 18 12:12:25 2019 +0700
JAMES-2794 Refactor DeleteCondition to implement WithEnqueueId
We need to apply the predicates against EnqueuedItem POJO to gain genericity
---
.../queue/rabbitmq/view/api/DeleteCondition.java | 44 ++++----
.../view/cassandra/CassandraMailQueueView.java | 2 +-
.../rabbitmq/view/api/DeleteConditionTest.java | 111 +++++++++++++++++----
3 files changed, 118 insertions(+), 39 deletions(-)
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/api/DeleteCondition.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/api/DeleteCondition.java
index 4b640fa..aa3973a 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/api/DeleteCondition.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/api/DeleteCondition.java
@@ -24,17 +24,17 @@ import java.util.Objects;
import org.apache.commons.lang3.NotImplementedException;
import org.apache.james.queue.api.ManageableMailQueue;
import org.apache.james.queue.rabbitmq.EnQueueId;
-import org.apache.mailet.Mail;
+import org.apache.james.queue.rabbitmq.EnqueuedItem;
import com.google.common.base.Preconditions;
public interface DeleteCondition {
- boolean shouldBeDeleted(Mail mail);
+ boolean shouldBeDeleted(EnqueuedItem enqueuedItem);
class All implements DeleteCondition {
@Override
- public boolean shouldBeDeleted(Mail mail) {
- Preconditions.checkNotNull(mail);
+ public boolean shouldBeDeleted(EnqueuedItem enqueuedItem) {
+ Preconditions.checkNotNull(enqueuedItem);
return true;
}
@@ -57,9 +57,10 @@ public interface DeleteCondition {
}
@Override
- public boolean shouldBeDeleted(Mail mail) {
- Preconditions.checkNotNull(mail);
- return mail.getMaybeSender()
+ public boolean shouldBeDeleted(EnqueuedItem enqueuedItem) {
+ Preconditions.checkNotNull(enqueuedItem);
+ return enqueuedItem.getMail()
+ .getMaybeSender()
.asString()
.equals(senderAsString);
}
@@ -88,9 +89,11 @@ public interface DeleteCondition {
}
@Override
- public boolean shouldBeDeleted(Mail mail) {
- Preconditions.checkNotNull(mail);
- return mail.getName().equals(name);
+ public boolean shouldBeDeleted(EnqueuedItem enqueuedItem) {
+ Preconditions.checkNotNull(enqueuedItem);
+ return enqueuedItem.getMail()
+ .getName()
+ .equals(name);
}
@Override
@@ -112,8 +115,7 @@ public interface DeleteCondition {
class WithEnqueueId implements DeleteCondition {
private final EnQueueId enQueueId;
-
- public WithEnqueueId(EnQueueId enQueueId) {
+ WithEnqueueId(EnQueueId enQueueId) {
this.enQueueId = enQueueId;
}
@@ -122,8 +124,9 @@ public interface DeleteCondition {
}
@Override
- public boolean shouldBeDeleted(Mail mail) {
- throw new NotImplementedException("EnQueueId is not carried as a Mail property");
+ public boolean shouldBeDeleted(EnqueuedItem enqueuedItem) {
+ Preconditions.checkNotNull(enqueuedItem);
+ return enqueuedItem.getEnQueueId().equals(enQueueId);
}
}
@@ -135,9 +138,11 @@ public interface DeleteCondition {
}
@Override
- public boolean shouldBeDeleted(Mail mail) {
- Preconditions.checkNotNull(mail);
- return mail.getRecipients()
+ public boolean shouldBeDeleted(EnqueuedItem enqueuedItem) {
+ Preconditions.checkNotNull(enqueuedItem);
+ return enqueuedItem
+ .getMail()
+ .getRecipients()
.stream()
.anyMatch(mailAddress -> mailAddress.asString().equals(recipientAsString));
}
@@ -186,6 +191,11 @@ public interface DeleteCondition {
return new WithName(value);
}
+ static WithEnqueueId withEnqueueId(EnQueueId value) {
+ Preconditions.checkNotNull(value);
+ return new WithEnqueueId(value);
+ }
+
static DeleteCondition all() {
return new All();
}
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueView.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueView.java
index 15fd0fc..a2cabda 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueView.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueView.java
@@ -111,7 +111,7 @@ public class CassandraMailQueueView implements MailQueueView {
private long browseThenDelete(DeleteCondition deleteCondition) {
return cassandraMailQueueBrowser.browseReferences(mailQueueName)
.map(EnqueuedItemWithSlicingContext::getEnqueuedItem)
- .filter(mailReference -> deleteCondition.shouldBeDeleted(mailReference.getMail()))
+ .filter(deleteCondition::shouldBeDeleted)
.flatMap(mailReference -> cassandraMailQueueMailDelete.considerDeleted(mailReference.getEnQueueId(), mailQueueName))
.count()
.doOnNext(ignored -> cassandraMailQueueMailDelete.updateBrowseStart(mailQueueName))
diff --git a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/api/DeleteConditionTest.java b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/api/DeleteConditionTest.java
index b559aa7..d7d34fa 100644
--- a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/api/DeleteConditionTest.java
+++ b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/api/DeleteConditionTest.java
@@ -22,8 +22,15 @@ package org.apache.james.queue.rabbitmq.view.api;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import java.time.Instant;
+
+import org.apache.james.blob.api.HashBlobId;
+import org.apache.james.blob.mail.MimeMessagePartsId;
import org.apache.james.core.MailAddress;
import org.apache.james.queue.api.ManageableMailQueue;
+import org.apache.james.queue.rabbitmq.EnQueueId;
+import org.apache.james.queue.rabbitmq.EnqueuedItem;
+import org.apache.james.queue.rabbitmq.MailQueueName;
import org.apache.mailet.base.test.FakeMail;
import org.junit.jupiter.api.Nested;
import org.junit.jupiter.api.Test;
@@ -35,6 +42,14 @@ class DeleteConditionTest {
private static final String ADDRESS_2 = "any2@toto.com";
private static final String NAME = "name";
private static final String VALUE = "value";
+ private static final EnQueueId EN_QUEUE_ID_1 = EnQueueId.ofSerialized("110e8400-e29b-11d4-a716-446655440000");
+ private static final EnQueueId EN_QUEUE_ID_2 = EnQueueId.ofSerialized("464765a0-e4e7-11e4-aba4-710c1de3782b");
+ private static final MailQueueName OUT_GOING_1 = MailQueueName.fromString("OUT_GOING_1");
+ private static final Instant ENQUEUE_TIME = Instant.now();
+ private static final MimeMessagePartsId MESSAGE_PARTS_ID = MimeMessagePartsId.builder()
+ .headerBlobId(new HashBlobId.Factory().from("headerBlobId"))
+ .bodyBlobId(new HashBlobId.Factory().from("bodyBlobId"))
+ .build();
@Nested
class AllTest {
@@ -42,7 +57,7 @@ class DeleteConditionTest {
void allShouldReturnTrue() throws Exception {
assertThat(
DeleteCondition.all()
- .shouldBeDeleted(FakeMail.builder().name("name").build()))
+ .shouldBeDeleted(enQueueItemForMail(FakeMail.builder().name("name").build())))
.isTrue();
}
@@ -61,6 +76,50 @@ class DeleteConditionTest {
}
@Nested
+ class WithEnQueueIdTest {
+ @Test
+ void withSenderShouldThrowOnNullCondition() {
+ assertThatThrownBy(() ->
+ DeleteCondition.withEnqueueId(null))
+ .isInstanceOf(NullPointerException.class);
+ }
+
+ @Test
+ void shouldBeDeletedShouldReturnTrueWhenSameId() throws Exception {
+ EnqueuedItem enqueuedItem = EnqueuedItem.builder()
+ .enQueueId(EN_QUEUE_ID_1)
+ .mailQueueName(OUT_GOING_1)
+ .mail(FakeMail.builder()
+ .name("name")
+ .sender(ADDRESS)
+ .build())
+ .enqueuedTime(ENQUEUE_TIME)
+ .mimeMessagePartsId(MESSAGE_PARTS_ID)
+ .build();
+
+ assertThat(DeleteCondition.withEnqueueId(EN_QUEUE_ID_1).shouldBeDeleted(enqueuedItem))
+ .isTrue();
+ }
+
+ @Test
+ void shouldBeDeletedShouldReturnFalseWhenDifferentId() throws Exception {
+ EnqueuedItem enqueuedItem = EnqueuedItem.builder()
+ .enQueueId(EN_QUEUE_ID_2)
+ .mailQueueName(OUT_GOING_1)
+ .mail(FakeMail.builder()
+ .name("name")
+ .sender(ADDRESS)
+ .build())
+ .enqueuedTime(ENQUEUE_TIME)
+ .mimeMessagePartsId(MESSAGE_PARTS_ID)
+ .build();
+
+ assertThat(DeleteCondition.withEnqueueId(EN_QUEUE_ID_1).shouldBeDeleted(enqueuedItem))
+ .isFalse();
+ }
+ }
+
+ @Nested
class WithSenderTest {
@Test
void withSenderShouldThrowOnNullCondition() {
@@ -81,10 +140,10 @@ class DeleteConditionTest {
void withSenderShouldReturnTrueWhenSameAddress() throws Exception {
assertThat(
DeleteCondition.withSender(ADDRESS)
- .shouldBeDeleted(FakeMail.builder()
+ .shouldBeDeleted(enQueueItemForMail(FakeMail.builder()
.name("name")
.sender(ADDRESS)
- .build()))
+ .build())))
.isTrue();
}
@@ -92,11 +151,11 @@ class DeleteConditionTest {
void withSenderShouldReturnFalseWhenDifferentAddress() throws Exception {
assertThat(
DeleteCondition.withSender(ADDRESS)
- .shouldBeDeleted(FakeMail.builder()
+ .shouldBeDeleted(enQueueItemForMail(FakeMail.builder()
.name("name")
.sender(ADDRESS_2)
.recipient(ADDRESS)
- .build()))
+ .build())))
.isFalse();
}
@@ -104,10 +163,10 @@ class DeleteConditionTest {
void withSenderShouldNotThrowOnNullSender() throws Exception {
assertThat(
DeleteCondition.withSender(ADDRESS)
- .shouldBeDeleted(FakeMail.builder()
+ .shouldBeDeleted(enQueueItemForMail(FakeMail.builder()
.name("name")
.sender(MailAddress.nullSender())
- .build()))
+ .build())))
.isFalse();
}
@@ -115,10 +174,10 @@ class DeleteConditionTest {
void withSenderShouldAllowNullSenderMatchingNullSender() throws Exception {
assertThat(
DeleteCondition.withSender(MailAddress.NULL_SENDER_AS_STRING)
- .shouldBeDeleted(FakeMail.builder()
+ .shouldBeDeleted(enQueueItemForMail(FakeMail.builder()
.name("name")
.sender(MailAddress.nullSender())
- .build()))
+ .build())))
.isTrue();
}
@@ -149,9 +208,9 @@ class DeleteConditionTest {
void withNameShouldReturnTrueWhenSameName() throws Exception {
assertThat(
DeleteCondition.withName(NAME)
- .shouldBeDeleted(FakeMail.builder()
+ .shouldBeDeleted(enQueueItemForMail(FakeMail.builder()
.name(NAME)
- .build()))
+ .build())))
.isTrue();
}
@@ -159,9 +218,9 @@ class DeleteConditionTest {
void withSenderShouldReturnFalseWhenDifferentAddress() throws Exception {
assertThat(
DeleteCondition.withName(NAME)
- .shouldBeDeleted(FakeMail.builder()
+ .shouldBeDeleted(enQueueItemForMail(FakeMail.builder()
.name("other")
- .build()))
+ .build())))
.isFalse();
}
@@ -192,10 +251,10 @@ class DeleteConditionTest {
void withRecipientShouldReturnTrueWhenSameAddress() throws Exception {
assertThat(
DeleteCondition.withRecipient(ADDRESS)
- .shouldBeDeleted(FakeMail.builder()
+ .shouldBeDeleted(enQueueItemForMail(FakeMail.builder()
.name("name")
.recipient(ADDRESS)
- .build()))
+ .build())))
.isTrue();
}
@@ -203,10 +262,10 @@ class DeleteConditionTest {
void withRecipientShouldReturnTrueWhenAtListOneMatches() throws Exception {
assertThat(
DeleteCondition.withRecipient(ADDRESS)
- .shouldBeDeleted(FakeMail.builder()
+ .shouldBeDeleted(enQueueItemForMail(FakeMail.builder()
.name("name")
.recipients(ADDRESS, ADDRESS_2)
- .build()))
+ .build())))
.isTrue();
}
@@ -214,11 +273,11 @@ class DeleteConditionTest {
void withRecipientShouldReturnFalseWhenDifferentAddress() throws Exception {
assertThat(
DeleteCondition.withRecipient(ADDRESS)
- .shouldBeDeleted(FakeMail.builder()
+ .shouldBeDeleted(enQueueItemForMail(FakeMail.builder()
.name("name")
.sender(ADDRESS)
.recipient(ADDRESS_2)
- .build()))
+ .build())))
.isFalse();
}
@@ -226,9 +285,9 @@ class DeleteConditionTest {
void withRecipientShouldReturnFalseWhenNoRecipient() throws Exception {
assertThat(
DeleteCondition.withRecipient(ADDRESS)
- .shouldBeDeleted(FakeMail.builder()
+ .shouldBeDeleted(enQueueItemForMail(FakeMail.builder()
.name("name")
- .build()))
+ .build())))
.isFalse();
}
@@ -258,4 +317,14 @@ class DeleteConditionTest {
.isEqualTo(DeleteCondition.withName(VALUE));
}
}
+
+ private EnqueuedItem enQueueItemForMail(FakeMail mail) {
+ return EnqueuedItem.builder()
+ .enQueueId(EN_QUEUE_ID_1)
+ .mailQueueName(OUT_GOING_1)
+ .mail(mail)
+ .enqueuedTime(ENQUEUE_TIME)
+ .mimeMessagePartsId(MESSAGE_PARTS_ID)
+ .build();
+ }
}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 26/28: JAMES-2774 Remove useless comment & rename a
test in RabbitMQEventBusTest
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 45d2f0619f0cb70d849f377c26131a82e2c236a2
Author: Tran Tien Duc <dt...@linagora.com>
AuthorDate: Mon Jun 17 11:05:05 2019 +0700
JAMES-2774 Remove useless comment & rename a test in RabbitMQEventBusTest
---
.../org/apache/james/mailbox/events/RabbitMQEventBusTest.java | 8 +++-----
1 file changed, 3 insertions(+), 5 deletions(-)
diff --git a/mailbox/event/event-rabbitmq/src/test/java/org/apache/james/mailbox/events/RabbitMQEventBusTest.java b/mailbox/event/event-rabbitmq/src/test/java/org/apache/james/mailbox/events/RabbitMQEventBusTest.java
index dac6986..49cfe24 100644
--- a/mailbox/event/event-rabbitmq/src/test/java/org/apache/james/mailbox/events/RabbitMQEventBusTest.java
+++ b/mailbox/event/event-rabbitmq/src/test/java/org/apache/james/mailbox/events/RabbitMQEventBusTest.java
@@ -186,11 +186,10 @@ class RabbitMQEventBusTest implements GroupContract.SingleEventBusGroupContract,
EventBusConcurrentTestContract.SingleEventBusConcurrentContract {
@Test
- void rabbitMQEventBusCannotHandleHugeDispatchingOperations() throws Exception {
+ void rabbitMQEventBusShouldHandleBulksGracefully() throws Exception {
EventBusTestFixture.MailboxListenerCountingSuccessfulExecution countingListener1 = newCountingListener();
-
eventBus().register(countingListener1, new EventBusTestFixture.GroupA());
- int totalGlobalRegistrations = 1;
+ int totalGlobalRegistrations = 1; // GroupA
int threadCount = 10;
int operationCount = 10000;
@@ -200,9 +199,8 @@ class RabbitMQEventBusTest implements GroupContract.SingleEventBusGroupContract,
.operation((threadNumber, operationNumber) -> eventBus.dispatch(EVENT, NO_KEYS).block())
.threadCount(threadCount)
.operationCount(operationCount)
- .runSuccessfullyWithin(Duration.ofMinutes(10));
+ .runSuccessfullyWithin(Duration.ofMinutes(3));
- // there is a moment when RabbitMQ EventBus consumed amount of messages, then it will stop to consume more
await()
.pollInterval(com.jayway.awaitility.Duration.FIVE_SECONDS)
.timeout(com.jayway.awaitility.Duration.TEN_MINUTES).until(() ->
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 02/28: JAMES-2794 Solve an IntelliJ warning
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit a54fe44f4e20069415f38ecffe2949e64cf52287
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Fri Jun 14 11:22:27 2019 +0700
JAMES-2794 Solve an IntelliJ warning
---
.../src/main/java/org/apache/james/queue/rabbitmq/Dequeuer.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
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 a2e7a7e..ee59815 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
@@ -100,14 +100,14 @@ class Dequeuer {
private Mono<RabbitMQMailQueueItem> loadItem(AcknowledgableDelivery response) {
try {
Mail mail = loadMail(response);
- ThrowingConsumer<Boolean> ack = ack(response, response.getEnvelope().getDeliveryTag(), mail);
+ ThrowingConsumer<Boolean> ack = ack(response, mail);
return Mono.just(new RabbitMQMailQueueItem(ack, mail));
} catch (MailQueue.MailQueueException e) {
return Mono.error(e);
}
}
- private ThrowingConsumer<Boolean> ack(AcknowledgableDelivery response, long deliveryTag, Mail mail) {
+ private ThrowingConsumer<Boolean> ack(AcknowledgableDelivery response, Mail mail) {
return success -> {
if (success) {
dequeueMetric.increment();
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 20/28: JAMES-2803 allow configuration of ES request
timeout
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit e3298980db6c4fa3eadc64cdce958c325e4a71ae
Author: Matthieu Baechler <ma...@apache.org>
AuthorDate: Wed Jun 19 15:17:25 2019 +0200
JAMES-2803 allow configuration of ES request timeout
---
.../james/backends/es/ClientProviderImpl.java | 47 +++------
.../backends/es/ElasticSearchConfiguration.java | 25 ++++-
.../es/ClientProviderImplConnectionTest.java | 30 +++---
.../james/backends/es/ClientProviderImplTest.java | 109 +--------------------
.../james/backends/es/DockerElasticSearch.java | 8 +-
.../modules/mailbox/ElasticSearchClientModule.java | 2 +-
.../apache/james/DockerElasticSearchExtension.java | 11 +++
7 files changed, 71 insertions(+), 161 deletions(-)
diff --git a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ClientProviderImpl.java b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ClientProviderImpl.java
index 3ba32f0..7816e8d 100644
--- a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ClientProviderImpl.java
+++ b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ClientProviderImpl.java
@@ -18,63 +18,48 @@
****************************************************************/
package org.apache.james.backends.es;
-import java.util.Optional;
-
import org.apache.http.HttpHost;
-import org.apache.james.util.Host;
import org.elasticsearch.client.RestClient;
+import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.settings.Settings;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
-import com.google.common.collect.ImmutableList;
public class ClientProviderImpl implements ClientProvider {
- public static ClientProviderImpl forHost(String address, Integer port, Optional<String> clusterName) {
- return new ClientProviderImpl(ImmutableList.of(Host.from(address, port)), clusterName);
- }
-
- public static ClientProviderImpl fromHostsString(String hostsString, Optional<String> clusterName) {
- Preconditions.checkNotNull(hostsString, "HostString should not be null");
- return new ClientProviderImpl(Host.parseHosts(hostsString), clusterName);
- }
-
- public static ClientProviderImpl fromHosts(ImmutableList<Host> hosts, Optional<String> clusterName) {
- Preconditions.checkNotNull(hosts, "Hosts should not be null");
- return new ClientProviderImpl(hosts, clusterName);
+ public static ClientProviderImpl fromConfiguration(ElasticSearchConfiguration configuration) {
+ Preconditions.checkNotNull(configuration);
+ return new ClientProviderImpl(configuration);
}
private static final String CLUSTER_NAME_SETTING = "cluster.name";
private static final String HTTP_HOST_SCHEME = "http";
+ private final ElasticSearchConfiguration configuration;
- private final ImmutableList<Host> hosts;
- private final Optional<String> clusterName;
-
- private ClientProviderImpl(ImmutableList<Host> hosts, Optional<String> clusterName) {
- Preconditions.checkArgument(!hosts.isEmpty(), "You should provide at least one host");
- this.hosts = hosts;
- this.clusterName = clusterName;
+ private ClientProviderImpl(ElasticSearchConfiguration configuration) {
+ this.configuration = configuration;
}
private HttpHost[] hostsToHttpHosts() {
- return hosts.stream()
+ return configuration.getHosts().stream()
.map(host -> new HttpHost(host.getHostName(), host.getPort(), HTTP_HOST_SCHEME))
.toArray(HttpHost[]::new);
}
@Override
public RestHighLevelClient get() {
- return new RestHighLevelClient(RestClient.builder(hostsToHttpHosts()));
+ RestClientBuilder restClient = RestClient.builder(hostsToHttpHosts())
+ .setMaxRetryTimeoutMillis(Math.toIntExact(configuration.getRequestTimeout().toMillis()));
+ return new RestHighLevelClient(restClient);
}
@VisibleForTesting Settings settings() {
- if (clusterName.isPresent()) {
- return Settings.builder()
- .put(CLUSTER_NAME_SETTING, clusterName.get())
- .build();
- }
- return Settings.EMPTY;
+ return configuration.getClusterName()
+ .map(clusterName -> Settings.builder()
+ .put(CLUSTER_NAME_SETTING, clusterName)
+ .build())
+ .orElse(Settings.EMPTY);
}
}
diff --git a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ElasticSearchConfiguration.java b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ElasticSearchConfiguration.java
index a1dec4d..ed3199d 100644
--- a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ElasticSearchConfiguration.java
+++ b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ElasticSearchConfiguration.java
@@ -19,6 +19,7 @@
package org.apache.james.backends.es;
+import java.time.Duration;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
@@ -46,6 +47,7 @@ public class ElasticSearchConfiguration {
private Optional<Integer> nbReplica;
private Optional<Integer> minDelay;
private Optional<Integer> maxRetries;
+ private Optional<Duration> requestTimeout;
public Builder() {
hosts = ImmutableList.builder();
@@ -54,6 +56,7 @@ public class ElasticSearchConfiguration {
nbReplica = Optional.empty();
minDelay = Optional.empty();
maxRetries = Optional.empty();
+ requestTimeout = Optional.empty();
}
public Builder addHost(Host host) {
@@ -93,6 +96,11 @@ public class ElasticSearchConfiguration {
return this;
}
+ public Builder requestTimeout(Optional<Duration> requestTimeout) {
+ this.requestTimeout = requestTimeout;
+ return this;
+ }
+
public ElasticSearchConfiguration build() {
ImmutableList<Host> hosts = this.hosts.build();
Preconditions.checkState(!hosts.isEmpty(), "You need to specify ElasticSearch host");
@@ -102,7 +110,8 @@ public class ElasticSearchConfiguration {
nbShards.orElse(DEFAULT_NB_SHARDS),
nbReplica.orElse(DEFAULT_NB_REPLICA),
minDelay.orElse(DEFAULT_CONNECTION_MIN_DELAY),
- maxRetries.orElse(DEFAULT_CONNECTION_MAX_RETRIES));
+ maxRetries.orElse(DEFAULT_CONNECTION_MAX_RETRIES),
+ requestTimeout.orElse(DEFAULT_REQUEST_TIMEOUT));
}
}
@@ -121,6 +130,7 @@ public class ElasticSearchConfiguration {
public static final int DEFAULT_CONNECTION_MAX_RETRIES = 7;
public static final int DEFAULT_CONNECTION_MIN_DELAY = 3000;
+ public static final Duration DEFAULT_REQUEST_TIMEOUT = Duration.ofSeconds(30);
public static final int DEFAULT_NB_SHARDS = 5;
public static final int DEFAULT_NB_REPLICA = 1;
public static final int DEFAULT_PORT = 9200;
@@ -184,14 +194,16 @@ public class ElasticSearchConfiguration {
private final int nbReplica;
private final int minDelay;
private final int maxRetries;
+ private final Duration requestTimeout;
- private ElasticSearchConfiguration(ImmutableList<Host> hosts, Optional<String> clusterName, int nbShards, int nbReplica, int minDelay, int maxRetries) {
+ private ElasticSearchConfiguration(ImmutableList<Host> hosts, Optional<String> clusterName, int nbShards, int nbReplica, int minDelay, int maxRetries, Duration requestTimeout) {
this.hosts = hosts;
this.clusterName = clusterName;
this.nbShards = nbShards;
this.nbReplica = nbReplica;
this.minDelay = minDelay;
this.maxRetries = maxRetries;
+ this.requestTimeout = requestTimeout;
}
public ImmutableList<Host> getHosts() {
@@ -218,6 +230,10 @@ public class ElasticSearchConfiguration {
return maxRetries;
}
+ public Duration getRequestTimeout() {
+ return requestTimeout;
+ }
+
@Override
public final boolean equals(Object o) {
if (o instanceof ElasticSearchConfiguration) {
@@ -228,13 +244,14 @@ public class ElasticSearchConfiguration {
&& Objects.equals(this.nbReplica, that.nbReplica)
&& Objects.equals(this.minDelay, that.minDelay)
&& Objects.equals(this.maxRetries, that.maxRetries)
- && Objects.equals(this.hosts, that.hosts);
+ && Objects.equals(this.hosts, that.hosts)
+ && Objects.equals(this.requestTimeout, that.requestTimeout);
}
return false;
}
@Override
public final int hashCode() {
- return Objects.hash(hosts, clusterName, nbShards, nbReplica, minDelay, maxRetries);
+ return Objects.hash(hosts, clusterName, nbShards, nbReplica, minDelay, maxRetries, requestTimeout);
}
}
diff --git a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ClientProviderImplConnectionTest.java b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ClientProviderImplConnectionTest.java
index 149aa07..72ef84a 100644
--- a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ClientProviderImplConnectionTest.java
+++ b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ClientProviderImplConnectionTest.java
@@ -19,9 +19,9 @@
package org.apache.james.backends.es;
-import java.util.Optional;
import java.util.concurrent.TimeUnit;
+import org.apache.james.util.Host;
import org.apache.james.util.docker.DockerGenericContainer;
import org.apache.james.util.docker.Images;
import org.awaitility.Awaitility;
@@ -53,22 +53,27 @@ public class ClientProviderImplConnectionTest {
@Test
public void connectingASingleServerShouldWork() {
+ ElasticSearchConfiguration configuration = ElasticSearchConfiguration.builder()
+ .addHost(Host.from(es1.getContainerIp(), ES_APPLICATIVE_PORT))
+ .build();
+
Awaitility.await()
.atMost(1, TimeUnit.MINUTES)
.pollInterval(5, TimeUnit.SECONDS)
- .until(() -> isConnected(ClientProviderImpl.forHost(es1.getContainerIp(), ES_APPLICATIVE_PORT, Optional.empty())));
+ .until(() -> isConnected(ClientProviderImpl.fromConfiguration(configuration)));
}
@Test
public void connectingAClusterShouldWork() {
+ ElasticSearchConfiguration configuration = ElasticSearchConfiguration.builder()
+ .addHost(Host.from(es1.getContainerIp(), ES_APPLICATIVE_PORT))
+ .addHost(Host.from(es2.getContainerIp(), ES_APPLICATIVE_PORT))
+ .build();
+
Awaitility.await()
.atMost(1, TimeUnit.MINUTES)
.pollInterval(5, TimeUnit.SECONDS)
- .until(() -> isConnected(
- ClientProviderImpl.fromHostsString(
- es1.getContainerIp() + ":" + ES_APPLICATIVE_PORT + ","
- + es2.getContainerIp() + ":" + ES_APPLICATIVE_PORT,
- Optional.empty())));
+ .until(() -> isConnected(ClientProviderImpl.fromConfiguration(configuration)));
}
@Test
@@ -77,14 +82,15 @@ public class ClientProviderImplConnectionTest {
String es2Ip = es2.getContainerIp();
es2.stop();
+ ElasticSearchConfiguration configuration = ElasticSearchConfiguration.builder()
+ .addHost(Host.from(es1Ip, ES_APPLICATIVE_PORT))
+ .addHost(Host.from(es2Ip, ES_APPLICATIVE_PORT))
+ .build();
+
Awaitility.await()
.atMost(1, TimeUnit.MINUTES)
.pollInterval(5, TimeUnit.SECONDS)
- .until(() -> isConnected(
- ClientProviderImpl.fromHostsString(
- es1Ip + ":" + ES_APPLICATIVE_PORT + ","
- + es2Ip + ":" + ES_APPLICATIVE_PORT,
- Optional.empty())));
+ .until(() -> isConnected(ClientProviderImpl.fromConfiguration(configuration)));
}
private boolean isConnected(ClientProvider clientProvider) {
diff --git a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ClientProviderImplTest.java b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ClientProviderImplTest.java
index dac6ce3..949ad74 100644
--- a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ClientProviderImplTest.java
+++ b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ClientProviderImplTest.java
@@ -30,113 +30,8 @@ import org.junit.Test;
public class ClientProviderImplTest {
@Test
- public void fromHostsStringShouldThrowOnNullString() {
- assertThatThrownBy(() -> ClientProviderImpl.fromHostsString(null, Optional.empty()))
+ public void fromConfigurationShouldThrowOnNull() {
+ assertThatThrownBy(() -> ClientProviderImpl.fromConfiguration(null))
.isInstanceOf(NullPointerException.class);
}
-
- @Test
- public void fromHostsStringShouldThrowOnEmptyString() {
- assertThatThrownBy(() -> ClientProviderImpl.fromHostsString("", Optional.empty()))
- .isInstanceOf(IllegalArgumentException.class);
- }
-
- @Test
- public void forHostShouldThrowOnNullHost() {
- assertThatThrownBy(() -> ClientProviderImpl.forHost(null, 9200, Optional.empty()))
- .isInstanceOf(NullPointerException.class);
- }
-
- @Test
- public void forHostShouldThrowOnEmptyHost() {
- assertThatThrownBy(() -> ClientProviderImpl.forHost("", 9200, Optional.empty()))
- .isInstanceOf(IllegalArgumentException.class);
- }
-
- @Test
- public void forHostShouldThrowOnNegativePort() {
- assertThatThrownBy(() -> ClientProviderImpl.forHost("localhost", -1, Optional.empty()))
- .isInstanceOf(IllegalArgumentException.class);
- }
-
- @Test
- public void forHostShouldThrowOnZeroPort() {
- assertThatThrownBy(() -> ClientProviderImpl.forHost("localhost", 0, Optional.empty()))
- .isInstanceOf(IllegalArgumentException.class);
- }
-
- @Test
- public void forHostShouldThrowOnTooBigPort() {
- assertThatThrownBy(() -> ClientProviderImpl.forHost("localhost", 65536, Optional.empty()))
- .isInstanceOf(IllegalArgumentException.class);
- }
-
- @Test
- public void fromHostsStringShouldEmptyAddress() {
- assertThatThrownBy(() -> ClientProviderImpl.fromHostsString(":9200", Optional.empty()))
- .isInstanceOf(IllegalArgumentException.class);
- }
-
- @Test
- public void fromHostsStringShouldThrowOnAbsentPort() {
- assertThatThrownBy(() -> ClientProviderImpl.fromHostsString("localhost", Optional.empty()))
- .isInstanceOf(IllegalArgumentException.class);
- }
-
- @Test
- public void fromHostsStringShouldThrowWhenTooMuchParts() {
- assertThatThrownBy(() -> ClientProviderImpl.fromHostsString("localhost:9200:9200", Optional.empty()))
- .isInstanceOf(IllegalArgumentException.class);
- }
-
- @Test
- public void fromHostsStringShouldThrowOnEmptyPort() {
- assertThatThrownBy(() -> ClientProviderImpl.fromHostsString("localhost:", Optional.empty()))
- .isInstanceOf(NumberFormatException.class);
- }
-
- @Test
- public void fromHostsStringShouldThrowOnInvalidPort() {
- assertThatThrownBy(() -> ClientProviderImpl.fromHostsString("localhost:invalid", Optional.empty()))
- .isInstanceOf(NumberFormatException.class);
- }
-
- @Test
- public void fromHostsStringShouldThrowOnNegativePort() {
- assertThatThrownBy(() -> ClientProviderImpl.fromHostsString("localhost:-1", Optional.empty()))
- .isInstanceOf(IllegalArgumentException.class);
- }
-
- @Test
- public void fromHostsStringShouldThrowOnZeroPort() {
- assertThatThrownBy(() -> ClientProviderImpl.fromHostsString("localhost:0", Optional.empty()))
- .isInstanceOf(IllegalArgumentException.class);
- }
-
- @Test
- public void fromHostsStringShouldThrowOnTooBigPort() {
- assertThatThrownBy(() -> ClientProviderImpl.fromHostsString("localhost:65536", Optional.empty()))
- .isInstanceOf(IllegalArgumentException.class);
- }
-
- @Test
- public void fromHostsStringShouldThrowIfOneHostIsInvalid() {
- assertThatThrownBy(() -> ClientProviderImpl.fromHostsString("localhost:9200,localhost", Optional.empty()))
- .isInstanceOf(IllegalArgumentException.class);
- }
-
- @Test
- public void settingsShouldBeEmptyWhenClusterNameIsEmpty() {
- ClientProviderImpl clientProvider = ClientProviderImpl.fromHostsString("localhost:9200", Optional.empty());
-
- assertThat(clientProvider.settings()).isEqualTo(Settings.EMPTY);
- }
-
- @Test
- public void settingsShouldContainClusterNameSettingWhenClusterNameIsGiven() {
- String clusterName = "myClusterName";
- ClientProviderImpl clientProvider = ClientProviderImpl.fromHostsString("localhost:9200", Optional.of(clusterName));
-
- assertThat(clientProvider.settings().get("cluster.name")).isEqualTo(clusterName);
- }
}
diff --git a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/DockerElasticSearch.java b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/DockerElasticSearch.java
index e2d5f78..82eb83b 100644
--- a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/DockerElasticSearch.java
+++ b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/DockerElasticSearch.java
@@ -21,8 +21,6 @@ package org.apache.james.backends.es;
import static org.assertj.core.api.Assertions.assertThat;
-import java.util.Optional;
-
import org.apache.http.HttpStatus;
import org.apache.james.util.Host;
import org.apache.james.util.docker.DockerGenericContainer;
@@ -30,9 +28,7 @@ import org.apache.james.util.docker.Images;
import org.apache.james.util.docker.RateLimiters;
import org.testcontainers.containers.wait.strategy.HostPortWaitStrategy;
-import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
-
import feign.Feign;
import feign.Logger;
import feign.RequestLine;
@@ -117,8 +113,8 @@ public class DockerElasticSearch {
}
public ClientProvider clientProvider() {
- Optional<String> noClusterName = Optional.empty();
- return ClientProviderImpl.fromHosts(ImmutableList.of(getHttpHost()), noClusterName);
+ ElasticSearchConfiguration configuration = ElasticSearchConfiguration.builder().addHost(getHttpHost()).build();
+ return ClientProviderImpl.fromConfiguration(configuration);
}
private ElasticSearchAPI esAPI() {
diff --git a/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchClientModule.java b/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchClientModule.java
index e485fee..020a365 100644
--- a/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchClientModule.java
+++ b/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchClientModule.java
@@ -62,7 +62,7 @@ public class ElasticSearchClientModule extends AbstractModule {
private RestHighLevelClient connectToCluster(ElasticSearchConfiguration configuration) throws IOException {
LOGGER.info("Trying to connect to ElasticSearch service at {}", LocalDateTime.now());
- return ClientProviderImpl.fromHosts(configuration.getHosts(), configuration.getClusterName())
+ return ClientProviderImpl.fromConfiguration(configuration)
.get();
}
}
diff --git a/server/container/guice/cassandra-guice/src/test/java/org/apache/james/DockerElasticSearchExtension.java b/server/container/guice/cassandra-guice/src/test/java/org/apache/james/DockerElasticSearchExtension.java
index 531bf13..7ff07f4 100644
--- a/server/container/guice/cassandra-guice/src/test/java/org/apache/james/DockerElasticSearchExtension.java
+++ b/server/container/guice/cassandra-guice/src/test/java/org/apache/james/DockerElasticSearchExtension.java
@@ -19,6 +19,9 @@
package org.apache.james;
+import java.time.Duration;
+import java.util.Optional;
+
import org.apache.james.backends.es.DockerElasticSearch;
import org.apache.james.backends.es.DockerElasticSearchSingleton;
import org.apache.james.backends.es.ElasticSearchConfiguration;
@@ -29,13 +32,20 @@ import com.google.inject.Module;
public class DockerElasticSearchExtension implements GuiceModuleTestExtension {
private final DockerElasticSearch dockerElasticSearch;
+ private Optional<Duration> requestTimeout;
public DockerElasticSearchExtension() {
this(DockerElasticSearchSingleton.INSTANCE);
}
+ public DockerElasticSearchExtension withRequestTimeout(Duration requestTimeout) {
+ this.requestTimeout = Optional.of(requestTimeout);
+ return this;
+ }
+
public DockerElasticSearchExtension(DockerElasticSearch dockerElasticSearch) {
this.dockerElasticSearch = dockerElasticSearch;
+ requestTimeout = Optional.empty();
}
@Override
@@ -61,6 +71,7 @@ public class DockerElasticSearchExtension implements GuiceModuleTestExtension {
private ElasticSearchConfiguration getElasticSearchConfigurationForDocker() {
return ElasticSearchConfiguration.builder()
.addHost(getDockerES().getHttpHost())
+ .requestTimeout(requestTimeout)
.build();
}
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 17/28: JAMES-2794 RabbitMQ should not dequeue
deleted elements
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit eeaa8fe0678c02f26fd480714df84ee75d3dda62
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Tue Jun 25 08:15:06 2019 +0700
JAMES-2794 RabbitMQ should not dequeue deleted elements
---
.../java/org/apache/james/queue/rabbitmq/Dequeuer.java | 16 +++++++++-------
1 file changed, 9 insertions(+), 7 deletions(-)
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 db5352c..01db228 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
@@ -95,13 +95,15 @@ class Dequeuer {
private Mono<RabbitMQMailQueueItem> filterIfDeleted(RabbitMQMailQueueItem item) {
return mailQueueView.isPresent(item.getEnqueueId())
- .flatMap(isPresent -> {
- if (isPresent) {
- return Mono.just(item);
- }
- item.done(true);
- return Mono.empty();
- });
+ .flatMap(isPresent -> keepWhenPresent(item, isPresent));
+ }
+
+ private Mono<? extends RabbitMQMailQueueItem> keepWhenPresent(RabbitMQMailQueueItem item, Boolean isPresent) {
+ if (isPresent) {
+ return Mono.just(item);
+ }
+ item.done(true);
+ return Mono.empty();
}
private Mono<RabbitMQMailQueueItem> loadItem(AcknowledgableDelivery response) {
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 24/28: JAMES-2659 dispatch result should always be
waited for success
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 2be4ed2270df8c5b278084e2549a2f127ae4013a
Author: Matthieu Baechler <ma...@apache.org>
AuthorDate: Fri Jun 21 14:31:05 2019 +0200
JAMES-2659 dispatch result should always be waited for success
---
.../java/org/apache/james/mailbox/events/RabbitMQEventBusTest.java | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
diff --git a/mailbox/event/event-rabbitmq/src/test/java/org/apache/james/mailbox/events/RabbitMQEventBusTest.java b/mailbox/event/event-rabbitmq/src/test/java/org/apache/james/mailbox/events/RabbitMQEventBusTest.java
index d160031..50df7d4 100644
--- a/mailbox/event/event-rabbitmq/src/test/java/org/apache/james/mailbox/events/RabbitMQEventBusTest.java
+++ b/mailbox/event/event-rabbitmq/src/test/java/org/apache/james/mailbox/events/RabbitMQEventBusTest.java
@@ -297,7 +297,7 @@ class RabbitMQEventBusTest implements GroupContract.SingleEventBusGroupContract,
@Test
void dispatchShouldPublishSerializedEventToRabbitMQWhenNotBlocking() {
- eventBus.dispatch(EVENT, NO_KEYS);
+ eventBus.dispatch(EVENT, NO_KEYS).block();
assertThat(dequeueEvent()).isEqualTo(EVENT);
}
@@ -566,7 +566,7 @@ class RabbitMQEventBusTest implements GroupContract.SingleEventBusGroupContract,
eventBus.register(listener, GROUP_A);
try (Closeable closeable = ConcurrentTestRunner.builder()
- .operation((threadNumber, step) -> eventBus.dispatch(EVENT, KEY_1))
+ .operation((threadNumber, step) -> eventBus.dispatch(EVENT, KEY_1).block())
.threadCount(THREAD_COUNT)
.operationCount(OPERATION_COUNT)
.run()) {
@@ -659,7 +659,7 @@ class RabbitMQEventBusTest implements GroupContract.SingleEventBusGroupContract,
eventBus2.register(listener, GROUP_A);
try (Closeable closeable = ConcurrentTestRunner.builder()
- .operation((threadNumber, step) -> eventBus.dispatch(EVENT, KEY_1))
+ .operation((threadNumber, step) -> eventBus.dispatch(EVENT, KEY_1).block())
.threadCount(THREAD_COUNT)
.operationCount(OPERATION_COUNT)
.run()) {
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 27/28: JAMES-2774 Remove an useless counter field
in EventDispatcher
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 6884456e2e5374b2f6a7c4c1c17ddffeea2bc264
Author: Tran Tien Duc <dt...@linagora.com>
AuthorDate: Mon Jun 17 11:08:49 2019 +0700
JAMES-2774 Remove an useless counter field in EventDispatcher
---
.../src/main/java/org/apache/james/mailbox/events/EventDispatcher.java | 3 ---
1 file changed, 3 deletions(-)
diff --git a/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/EventDispatcher.java b/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/EventDispatcher.java
index 8f31102..311d6ed 100644
--- a/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/EventDispatcher.java
+++ b/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/EventDispatcher.java
@@ -26,7 +26,6 @@ import static org.apache.james.mailbox.events.RabbitMQEventBus.MAILBOX_EVENT_EXC
import java.nio.charset.StandardCharsets;
import java.util.Set;
-import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Stream;
import org.apache.james.event.json.EventSerializer;
@@ -58,7 +57,6 @@ class EventDispatcher {
private final LocalListenerRegistry localListenerRegistry;
private final AMQP.BasicProperties basicProperties;
private final MailboxListenerExecutor mailboxListenerExecutor;
- final AtomicInteger dispatchCount = new AtomicInteger();
EventDispatcher(EventBusId eventBusId, EventSerializer eventSerializer, Sender sender, LocalListenerRegistry localListenerRegistry, MailboxListenerExecutor mailboxListenerExecutor) {
this.eventSerializer = eventSerializer;
@@ -84,7 +82,6 @@ class EventDispatcher {
dispatchToRemoteListeners(serializeEvent(event), keys))
.subscribeOn(Schedulers.elastic())
.then()
- .doOnSuccess(any -> dispatchCount.incrementAndGet())
.subscribeWith(MonoProcessor.create());
}
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 19/28: JAMES-2794 ack should update the view using
an EnqueueId reference
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit edd328029485c64a63ef8dbc883f3db881c6197f
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Tue Jun 25 08:29:30 2019 +0700
JAMES-2794 ack should update the view using an EnqueueId reference
---
.../src/main/java/org/apache/james/queue/rabbitmq/Dequeuer.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
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 94b803c..ae9829e 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
@@ -120,7 +120,7 @@ class Dequeuer {
if (success) {
dequeueMetric.increment();
response.ack();
- mailQueueView.delete(DeleteCondition.withName(mailWithEnqueueId.getMail().getName()));
+ mailQueueView.delete(DeleteCondition.withEnqueueId(mailWithEnqueueId.getEnqueueId()));
} else {
response.nack(REQUEUE);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 12/28: JAMES-2794 Avoid relying on a DTO in
RabbitMQ MailQueue logic
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 9d6feef74feefbfdcc06b7544f7aa43ea6abaed2
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Wed Jun 19 13:31:20 2019 +0700
JAMES-2794 Avoid relying on a DTO in RabbitMQ MailQueue logic
---
.../org/apache/james/queue/rabbitmq/Dequeuer.java | 8 ++--
.../org/apache/james/queue/rabbitmq/Enqueuer.java | 17 ++++----
.../apache/james/queue/rabbitmq/MailLoader.java | 16 ++++----
.../{MailLoader.java => MailReference.java} | 46 +++++++++-------------
.../james/queue/rabbitmq/MailReferenceDTO.java | 25 ++++++++----
.../queue/rabbitmq/RabbitMQMailQueueFactory.java | 3 +-
.../apache/james/queue/rabbitmq/MailDTOTest.java | 22 ++++++-----
7 files changed, 68 insertions(+), 69 deletions(-)
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 7ed8d3e..e697cc2 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
@@ -71,12 +71,12 @@ class Dequeuer {
}
}
- private final Function<MailReferenceDTO, Mail> mailLoader;
+ private final Function<MailReferenceDTO, Pair<EnQueueId, Mail>> mailLoader;
private final Metric dequeueMetric;
private final MailReferenceSerializer mailReferenceSerializer;
private final MailQueueView mailQueueView;
- Dequeuer(MailQueueName name, RabbitClient rabbitClient, Function<MailReferenceDTO, Mail> mailLoader,
+ Dequeuer(MailQueueName name, RabbitClient rabbitClient, Function<MailReferenceDTO, Pair<EnQueueId, Mail>> mailLoader,
MailReferenceSerializer serializer, MetricFactory metricFactory,
MailQueueView mailQueueView) {
this.mailLoader = mailLoader;
@@ -129,9 +129,7 @@ class Dequeuer {
private Pair<EnQueueId, Mail> loadMail(Delivery response) throws MailQueue.MailQueueException {
MailReferenceDTO mailDTO = toMailReference(response);
- EnQueueId enQueueId = mailDTO.retrieveEnqueueId();
- Mail mail = mailLoader.apply(mailDTO);
- return Pair.of(enQueueId, mail);
+ return mailLoader.apply(mailDTO);
}
private MailReferenceDTO toMailReference(Delivery getResponse) throws MailQueue.MailQueueException {
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/Enqueuer.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/Enqueuer.java
index 9081acc..8f0c0fd 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/Enqueuer.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/Enqueuer.java
@@ -63,7 +63,8 @@ class Enqueuer {
void enQueue(Mail mail) throws MailQueue.MailQueueException {
EnQueueId enQueueId = EnQueueId.generate();
saveMail(mail)
- .map(Throwing.<MimeMessagePartsId, EnqueuedItem>function(partsId -> publishReferenceToRabbit(enQueueId, mail, partsId)).sneakyThrow())
+ .map(partIds -> new MailReference(enQueueId, mail, partIds))
+ .map(Throwing.function(this::publishReferenceToRabbit).sneakyThrow())
.flatMap(mailQueueView::storeMail)
.thenEmpty(Mono.fromRunnable(enqueueMetric::increment))
.block();
@@ -77,21 +78,21 @@ class Enqueuer {
}
}
- private EnqueuedItem publishReferenceToRabbit(EnQueueId enQueueId, Mail mail, MimeMessagePartsId partsId) throws MailQueue.MailQueueException {
- rabbitClient.publish(name, getMailReferenceBytes(enQueueId, mail, partsId));
+ private EnqueuedItem publishReferenceToRabbit(MailReference mailReference) throws MailQueue.MailQueueException {
+ rabbitClient.publish(name, getMailReferenceBytes(mailReference));
return EnqueuedItem.builder()
- .enQueueId(enQueueId)
+ .enQueueId(mailReference.getEnQueueId())
.mailQueueName(name)
- .mail(mail)
+ .mail(mailReference.getMail())
.enqueuedTime(clock.instant())
- .mimeMessagePartsId(partsId)
+ .mimeMessagePartsId(mailReference.getPartsId())
.build();
}
- private byte[] getMailReferenceBytes(EnQueueId enQueueId, Mail mail, MimeMessagePartsId partsId) throws MailQueue.MailQueueException {
+ private byte[] getMailReferenceBytes(MailReference mailReference) throws MailQueue.MailQueueException {
try {
- MailReferenceDTO mailDTO = MailReferenceDTO.fromMail(enQueueId, mail, partsId);
+ MailReferenceDTO mailDTO = MailReferenceDTO.fromMailReference(mailReference);
return mailReferenceSerializer.write(mailDTO);
} catch (JsonProcessingException e) {
throw new MailQueue.MailQueueException("Unable to serialize message", e);
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailLoader.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailLoader.java
index f69c533..ca7df53 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailLoader.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailLoader.java
@@ -23,6 +23,7 @@ import javax.mail.MessagingException;
import javax.mail.internet.AddressException;
import javax.mail.internet.MimeMessage;
+import org.apache.commons.lang3.tuple.Pair;
import org.apache.james.blob.api.BlobId;
import org.apache.james.blob.api.Store;
import org.apache.james.blob.mail.MimeMessagePartsId;
@@ -30,7 +31,6 @@ import org.apache.james.queue.api.MailQueue;
import org.apache.mailet.Mail;
class MailLoader {
-
private final Store<MimeMessage, MimeMessagePartsId> mimeMessageStore;
private final BlobId.Factory blobIdFactory;
@@ -39,16 +39,14 @@ class MailLoader {
this.blobIdFactory = blobIdFactory;
}
- Mail load(MailReferenceDTO dto) throws MailQueue.MailQueueException {
+ Pair<EnQueueId, Mail> load(MailReferenceDTO dto) throws MailQueue.MailQueueException {
try {
- MimeMessage mimeMessage = mimeMessageStore.read(
- MimeMessagePartsId.builder()
- .headerBlobId(blobIdFactory.from(dto.getHeaderBlobId()))
- .bodyBlobId(blobIdFactory.from(dto.getBodyBlobId()))
- .build())
- .block();
+ MailReference mailReference = dto.toMailReference(blobIdFactory);
- return dto.toMailWithMimeMessage(mimeMessage);
+ Mail mail = mailReference.getMail();
+ MimeMessage mimeMessage = mimeMessageStore.read(mailReference.getPartsId()).block();
+ mail.setMessage(mimeMessage);
+ return Pair.of(mailReference.getEnQueueId(), mail);
} catch (AddressException e) {
throw new MailQueue.MailQueueException("Failed to parse mail address", e);
} catch (MessagingException e) {
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailLoader.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailReference.java
similarity index 50%
copy from server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailLoader.java
copy to server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailReference.java
index f69c533..031d97b 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailLoader.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailReference.java
@@ -19,40 +19,30 @@
package org.apache.james.queue.rabbitmq;
-import javax.mail.MessagingException;
-import javax.mail.internet.AddressException;
-import javax.mail.internet.MimeMessage;
-
-import org.apache.james.blob.api.BlobId;
-import org.apache.james.blob.api.Store;
import org.apache.james.blob.mail.MimeMessagePartsId;
-import org.apache.james.queue.api.MailQueue;
import org.apache.mailet.Mail;
-class MailLoader {
+public class MailReference {
+
+ private final EnQueueId enQueueId;
+ private final Mail mail;
+ private final MimeMessagePartsId partsId;
- private final Store<MimeMessage, MimeMessagePartsId> mimeMessageStore;
- private final BlobId.Factory blobIdFactory;
+ public MailReference(EnQueueId enQueueId, Mail mail, MimeMessagePartsId partsId) {
+ this.enQueueId = enQueueId;
+ this.mail = mail;
+ this.partsId = partsId;
+ }
+
+ public EnQueueId getEnQueueId() {
+ return enQueueId;
+ }
- MailLoader(Store<MimeMessage, MimeMessagePartsId> mimeMessageStore, BlobId.Factory blobIdFactory) {
- this.mimeMessageStore = mimeMessageStore;
- this.blobIdFactory = blobIdFactory;
+ public Mail getMail() {
+ return mail;
}
- Mail load(MailReferenceDTO dto) throws MailQueue.MailQueueException {
- try {
- MimeMessage mimeMessage = mimeMessageStore.read(
- MimeMessagePartsId.builder()
- .headerBlobId(blobIdFactory.from(dto.getHeaderBlobId()))
- .bodyBlobId(blobIdFactory.from(dto.getBodyBlobId()))
- .build())
- .block();
-
- return dto.toMailWithMimeMessage(mimeMessage);
- } catch (AddressException e) {
- throw new MailQueue.MailQueueException("Failed to parse mail address", e);
- } catch (MessagingException e) {
- throw new MailQueue.MailQueueException("Failed to generate mime message", e);
- }
+ public MimeMessagePartsId getPartsId() {
+ return partsId;
}
}
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailReferenceDTO.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailReferenceDTO.java
index f119325..f63458f 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailReferenceDTO.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailReferenceDTO.java
@@ -28,10 +28,8 @@ import java.util.Optional;
import java.util.function.Function;
import java.util.stream.Stream;
-import javax.mail.MessagingException;
-import javax.mail.internet.MimeMessage;
-
import org.apache.commons.lang3.tuple.Pair;
+import org.apache.james.blob.api.BlobId;
import org.apache.james.blob.mail.MimeMessagePartsId;
import org.apache.james.core.MailAddress;
import org.apache.james.core.MaybeSender;
@@ -52,9 +50,12 @@ import com.google.common.collect.ImmutableMap;
class MailReferenceDTO {
- static MailReferenceDTO fromMail(EnQueueId enQueueId, Mail mail, MimeMessagePartsId partsId) {
+ static MailReferenceDTO fromMailReference(MailReference mailReference) {
+ Mail mail = mailReference.getMail();
+ MimeMessagePartsId partsId = mailReference.getPartsId();
+
return new MailReferenceDTO(
- enQueueId.serialize(),
+ mailReference.getEnQueueId().serialize(),
Optional.ofNullable(mail.getRecipients()).map(Collection::stream)
.orElse(Stream.empty())
.map(MailAddress::asString)
@@ -202,14 +203,22 @@ class MailReferenceDTO {
return EnQueueId.ofSerialized(enQueueId);
}
- MailImpl toMailWithMimeMessage(MimeMessage mimeMessage) throws MessagingException {
+ MailReference toMailReference(BlobId.Factory blobIdFactory) {
+ MimeMessagePartsId messagePartsId = MimeMessagePartsId.builder()
+ .headerBlobId(blobIdFactory.from(headerBlobId))
+ .bodyBlobId(blobIdFactory.from(bodyBlobId))
+ .build();
+
+ return new MailReference(EnQueueId.ofSerialized(enQueueId), mailMetadata(), messagePartsId);
+ }
+
+ private MailImpl mailMetadata() {
MailImpl.Builder builder = MailImpl.builder()
.name(name)
.sender(sender.map(MaybeSender::getMailSender).orElse(MaybeSender.nullSender()))
.addRecipients(recipients.stream()
.map(Throwing.<String, MailAddress>function(MailAddress::new).sneakyThrow())
- .toArray(MailAddress[]::new))
- .mimeMessage(mimeMessage)
+ .collect(Guavate.toImmutableList()))
.errorMessage(errorMessage)
.remoteAddr(remoteAddr)
.remoteHost(remoteHost)
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 57d5181..3660d12 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
@@ -30,6 +30,7 @@ import java.util.function.Function;
import javax.inject.Inject;
import javax.mail.internet.MimeMessage;
+import org.apache.commons.lang3.tuple.Pair;
import org.apache.james.blob.api.BlobId;
import org.apache.james.blob.api.Store;
import org.apache.james.blob.mail.MimeMessagePartsId;
@@ -53,7 +54,7 @@ public class RabbitMQMailQueueFactory implements MailQueueFactory<RabbitMQMailQu
private final RabbitClient rabbitClient;
private final Store<MimeMessage, MimeMessagePartsId> mimeMessageStore;
private final MailReferenceSerializer mailReferenceSerializer;
- private final Function<MailReferenceDTO, Mail> mailLoader;
+ private final Function<MailReferenceDTO, Pair<EnQueueId, Mail>> mailLoader;
private final MailQueueView.Factory mailQueueViewFactory;
private final Clock clock;
private final MailQueueItemDecoratorFactory decoratorFactory;
diff --git a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/MailDTOTest.java b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/MailDTOTest.java
index 2c98b1d..2b110fa 100644
--- a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/MailDTOTest.java
+++ b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/MailDTOTest.java
@@ -85,7 +85,8 @@ class MailDTOTest {
}
private MailReferenceDTO mailDTO1() throws MessagingException {
- return MailReferenceDTO.fromMail(
+ return MailReferenceDTO.fromMailReference(
+ new MailReference(
EN_QUEUE_ID,
FakeMail.builder()
.name("mail-name-558")
@@ -109,21 +110,22 @@ class MailDTOTest {
MimeMessagePartsId.builder()
.headerBlobId(BLOB_ID_FACTORY.from("210e7136-ede3-44eb-9495-3ed816d6e23b"))
.bodyBlobId(BLOB_ID_FACTORY.from("ef46c026-7819-4048-b562-3a37469191ed"))
- .build());
+ .build()));
}
- private MailReferenceDTO mailDTOMin() throws MessagingException {
+ private MailReferenceDTO mailDTOMin() {
MailImpl mail = MailImpl.builder()
.name("mail-name-558")
.build();
mail.setState(null);
mail.setLastUpdated(null);
- return MailReferenceDTO.fromMail(
- EN_QUEUE_ID,
- mail,
- MimeMessagePartsId.builder()
- .headerBlobId(BLOB_ID_FACTORY.from("210e7136-ede3-44eb-9495-3ed816d6e23b"))
- .bodyBlobId(BLOB_ID_FACTORY.from("ef46c026-7819-4048-b562-3a37469191ed"))
- .build());
+ return MailReferenceDTO.fromMailReference(
+ new MailReference(
+ EN_QUEUE_ID,
+ mail,
+ MimeMessagePartsId.builder()
+ .headerBlobId(BLOB_ID_FACTORY.from("210e7136-ede3-44eb-9495-3ed816d6e23b"))
+ .bodyBlobId(BLOB_ID_FACTORY.from("ef46c026-7819-4048-b562-3a37469191ed"))
+ .build()));
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 10/28: JAMES-2794 Dequeuer should rely on concatMap
and not flatMap
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit aad4442b6cb902ea216434fd644d6615661ead78
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Tue Jun 18 14:16:33 2019 +0700
JAMES-2794 Dequeuer should rely on concatMap and not flatMap
flatMap can lead to event reordering (8% occurrence) which causes
browseAndDequeueShouldCombineWellWhenDifferentSlices to be flacky.
Note that filtering out deleted emails actually increased the inconsistency
window
---
.../src/main/java/org/apache/james/queue/rabbitmq/Dequeuer.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
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 95f13ee..7ed8d3e 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
@@ -89,8 +89,8 @@ class Dequeuer {
}
Flux<? extends MailQueue.MailQueueItem> deQueue() {
- return flux.flatMap(this::loadItem)
- .flatMap(this::filterIfDeleted);
+ return flux.concatMap(this::loadItem)
+ .concatMap(this::filterIfDeleted);
}
private Mono<RabbitMQMailQueueItem> filterIfDeleted(RabbitMQMailQueueItem item) {
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 09/28: JAMES-2794 Upgrade instructions
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit a0c2fe2df0215d3ad45b00829edd875604f6e64b
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Tue Jun 18 14:03:15 2019 +0700
JAMES-2794 Upgrade instructions
---
upgrade-instructions.md | 35 +++++++++++++++++++++++++++++++++++
1 file changed, 35 insertions(+)
diff --git a/upgrade-instructions.md b/upgrade-instructions.md
index 95dbef6..6229179 100644
--- a/upgrade-instructions.md
+++ b/upgrade-instructions.md
@@ -17,6 +17,41 @@ Changes to apply between 3.3.x and 3.4.x will be reported here.
Change list:
- [Upgrade to ElasticSearch 6.3](#upgrade-to-elasticsearch-6.3)
+
+#### Enqueuing several times a mail with the same name
+
+Date: XX/06/2019
+
+SHA-1: XXXXX
+
+JIRA: https://issues.apache.org/jira/browse/JAMES-2794
+
+Concerned products: (experimental) RabbitMQ MailQueue
+
+RabbitMQ mail queue combines RabbitMQ with projections in Cassandra to offer advanced management capabilities expected from
+a mail queue (browse, delete, size, clear). In these projections, the mails are identified by there name. Thus enqueuing a
+mail that had already been processed will lead the given email to be considered already deleted and it will be discarded
+and lost.
+
+This is an issue, as several other components build features around submitting a mail several time with the name.
+
+For instance:
+
+ - MailRepository reprocessing
+ - RemoteDelivery bouncing under some configurations
+ - RecipientRewriteTable rewriting to a remote server
+
+We thus changed the table structure of RabbitMQ mail queue projections to be built around an EnqueueId. This additional
+level of indirection allows several enqueues with the same name.
+
+Upgrade to the newest James server needs to be performed with an empty MailQueue.
+
+To do so:
+
+ - Given a distributed Guice James server
+ - Stop incoming traffic
+ - Monitor the mailQueues and wait them to be empty (spool & outgoing)
+ - Once empty, upgrade James server and re-enable incoming traffic.
#### Upgrade to ElasticSearch 6.3
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 01/28: JAMES-2794 RabbitMQ should not dequeue
deleted elements
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 98f9ed49ac5c2d0b97af592ca1d914e52ce108ac
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Fri Jun 14 11:06:28 2019 +0700
JAMES-2794 RabbitMQ should not dequeue deleted elements
---
.../java/org/apache/james/queue/rabbitmq/Dequeuer.java | 17 +++++++++++++++--
.../james/queue/rabbitmq/RabbitMQMailQueueTest.java | 8 --------
2 files changed, 15 insertions(+), 10 deletions(-)
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 d053096..a2e7a7e 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
@@ -34,6 +34,7 @@ import org.apache.mailet.Mail;
import com.github.fge.lambdas.consumers.ThrowingConsumer;
import com.rabbitmq.client.Delivery;
+
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.rabbitmq.AcknowledgableDelivery;
@@ -80,8 +81,20 @@ class Dequeuer {
.filter(getResponse -> getResponse.getBody() != null);
}
- Flux<MailQueue.MailQueueItem> deQueue() {
- return flux.flatMap(this::loadItem);
+ Flux<? extends MailQueue.MailQueueItem> deQueue() {
+ return flux.flatMap(this::loadItem)
+ .flatMap(this::filterIfDeleted);
+ }
+
+ private Mono<RabbitMQMailQueueItem> filterIfDeleted(RabbitMQMailQueueItem item) {
+ return mailQueueView.isPresent(item.getMail())
+ .flatMap(isPresent -> {
+ if (isPresent) {
+ return Mono.just(item);
+ }
+ item.done(true);
+ return Mono.empty();
+ });
}
private Mono<RabbitMQMailQueueItem> loadItem(AcknowledgableDelivery response) {
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 65a6f20..f6d57c6 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
@@ -231,14 +231,6 @@ public class RabbitMQMailQueueTest implements ManageableMailQueueContract, MailQ
}
- @Disabled("JAMES-2733 Deleted elements are still dequeued")
- @Test
- @Override
- public void deletedElementsShouldNotBeDequeued() {
-
- }
-
-
private void enqueueSomeMails(Function<Integer, String> namePattern, int emailCount) {
IntStream.rangeClosed(1, emailCount)
.forEach(Throwing.intConsumer(i -> enQueue(defaultMail()
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 23/28: JAMES-2803 Finally shorten
ReindexingWithEventDeadLettersTest by configuring request timeout
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit a867be65daec6d7a493e9b16a4cc7fd57c4a03ad
Author: Matthieu Baechler <ma...@apache.org>
AuthorDate: Wed Jun 19 17:27:47 2019 +0200
JAMES-2803 Finally shorten ReindexingWithEventDeadLettersTest by configuring request timeout
---
.../apache/james/jmap/rabbitmq/ReindexingWithEventDeadLettersTest.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/server/protocols/jmap-integration-testing/rabbitmq-jmap-integration-testing/src/test/java/org/apache/james/jmap/rabbitmq/ReindexingWithEventDeadLettersTest.java b/server/protocols/jmap-integration-testing/rabbitmq-jmap-integration-testing/src/test/java/org/apache/james/jmap/rabbitmq/ReindexingWithEventDeadLettersTest.java
index c4385af..84169f3 100644
--- a/server/protocols/jmap-integration-testing/rabbitmq-jmap-integration-testing/src/test/java/org/apache/james/jmap/rabbitmq/ReindexingWithEventDeadLettersTest.java
+++ b/server/protocols/jmap-integration-testing/rabbitmq-jmap-integration-testing/src/test/java/org/apache/james/jmap/rabbitmq/ReindexingWithEventDeadLettersTest.java
@@ -71,7 +71,7 @@ class ReindexingWithEventDeadLettersTest {
.atMost(Duration.ONE_MINUTE)
.await();
- private static final DockerElasticSearchExtension dockerElasticSearch = new DockerElasticSearchExtension();
+ private static final DockerElasticSearchExtension dockerElasticSearch = new DockerElasticSearchExtension().withRequestTimeout(java.time.Duration.ofSeconds(1));
private static final JamesServerBuilder.ServerProvider CONFIGURATION_BUILDER = configuration -> GuiceJamesServer
.forConfiguration(configuration)
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 13/28: JAMES-2794 s/EnQueueId/EnqueueId
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 4deed8b3dc7850cb89ccbbbc7572e870853a5e39
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Wed Jun 19 13:38:16 2019 +0700
JAMES-2794 s/EnQueueId/EnqueueId
---
.../org/apache/james/queue/rabbitmq/Dequeuer.java | 20 +++++------
.../rabbitmq/{EnQueueId.java => EnqueueId.java} | 18 +++++-----
.../apache/james/queue/rabbitmq/EnqueuedItem.java | 28 +++++++--------
.../org/apache/james/queue/rabbitmq/Enqueuer.java | 6 ++--
.../apache/james/queue/rabbitmq/MailLoader.java | 4 +--
.../apache/james/queue/rabbitmq/MailReference.java | 10 +++---
.../james/queue/rabbitmq/MailReferenceDTO.java | 24 ++++++-------
.../queue/rabbitmq/RabbitMQMailQueueFactory.java | 2 +-
.../queue/rabbitmq/view/api/DeleteCondition.java | 16 ++++-----
.../queue/rabbitmq/view/api/MailQueueView.java | 4 +--
.../view/cassandra/CassandraMailQueueBrowser.java | 2 +-
.../cassandra/CassandraMailQueueMailDelete.java | 10 +++---
.../view/cassandra/CassandraMailQueueView.java | 14 ++++----
.../rabbitmq/view/cassandra/DeletedMailsDAO.java | 14 ++++----
.../rabbitmq/view/cassandra/EnqueuedMailsDAO.java | 2 +-
.../view/cassandra/EnqueuedMailsDaoUtil.java | 6 ++--
.../james/queue/rabbitmq/EnqueuedItemTest.java | 10 +++---
.../apache/james/queue/rabbitmq/MailDTOTest.java | 2 +-
.../rabbitmq/view/api/DeleteConditionTest.java | 42 +++++++++++-----------
.../view/cassandra/DeletedMailsDAOTest.java | 28 +++++++--------
.../view/cassandra/EnqueuedMailsDaoTest.java | 12 +++----
.../{EnQueueIdTest.java => EnqueueIdTest.java} | 20 +++++------
.../model/EnqueuedItemWithSlicingContextTest.java | 4 +--
.../src/test/resources/json/mail1.json | 2 +-
.../src/test/resources/json/mail_min.json | 2 +-
25 files changed, 151 insertions(+), 151 deletions(-)
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 e697cc2..db5352c 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
@@ -47,12 +47,12 @@ class Dequeuer {
private static class RabbitMQMailQueueItem implements MailQueue.MailQueueItem {
private final Consumer<Boolean> ack;
- private final EnQueueId enQueueId;
+ private final EnqueueId enqueueId;
private final Mail mail;
- private RabbitMQMailQueueItem(Consumer<Boolean> ack, EnQueueId enQueueId, Mail mail) {
+ private RabbitMQMailQueueItem(Consumer<Boolean> ack, EnqueueId enqueueId, Mail mail) {
this.ack = ack;
- this.enQueueId = enQueueId;
+ this.enqueueId = enqueueId;
this.mail = mail;
}
@@ -61,8 +61,8 @@ class Dequeuer {
return mail;
}
- public EnQueueId getEnQueueId() {
- return enQueueId;
+ public EnqueueId getEnqueueId() {
+ return enqueueId;
}
@Override
@@ -71,12 +71,12 @@ class Dequeuer {
}
}
- private final Function<MailReferenceDTO, Pair<EnQueueId, Mail>> mailLoader;
+ private final Function<MailReferenceDTO, Pair<EnqueueId, Mail>> mailLoader;
private final Metric dequeueMetric;
private final MailReferenceSerializer mailReferenceSerializer;
private final MailQueueView mailQueueView;
- Dequeuer(MailQueueName name, RabbitClient rabbitClient, Function<MailReferenceDTO, Pair<EnQueueId, Mail>> mailLoader,
+ Dequeuer(MailQueueName name, RabbitClient rabbitClient, Function<MailReferenceDTO, Pair<EnqueueId, Mail>> mailLoader,
MailReferenceSerializer serializer, MetricFactory metricFactory,
MailQueueView mailQueueView) {
this.mailLoader = mailLoader;
@@ -94,7 +94,7 @@ class Dequeuer {
}
private Mono<RabbitMQMailQueueItem> filterIfDeleted(RabbitMQMailQueueItem item) {
- return mailQueueView.isPresent(item.getEnQueueId())
+ return mailQueueView.isPresent(item.getEnqueueId())
.flatMap(isPresent -> {
if (isPresent) {
return Mono.just(item);
@@ -106,7 +106,7 @@ class Dequeuer {
private Mono<RabbitMQMailQueueItem> loadItem(AcknowledgableDelivery response) {
try {
- Pair<EnQueueId, Mail> idAndMail = loadMail(response);
+ Pair<EnqueueId, Mail> idAndMail = loadMail(response);
Mail mail = idAndMail.getRight();
ThrowingConsumer<Boolean> ack = ack(response, mail);
return Mono.just(new RabbitMQMailQueueItem(ack, idAndMail.getLeft(), mail));
@@ -127,7 +127,7 @@ class Dequeuer {
};
}
- private Pair<EnQueueId, Mail> loadMail(Delivery response) throws MailQueue.MailQueueException {
+ private Pair<EnqueueId, Mail> loadMail(Delivery response) throws MailQueue.MailQueueException {
MailReferenceDTO mailDTO = toMailReference(response);
return mailLoader.apply(mailDTO);
}
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/EnQueueId.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/EnqueueId.java
similarity index 83%
rename from server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/EnQueueId.java
rename to server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/EnqueueId.java
index 133172d..e2057ac 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/EnQueueId.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/EnqueueId.java
@@ -25,25 +25,25 @@ import java.util.UUID;
import com.datastax.driver.core.utils.UUIDs;
import com.google.common.base.Preconditions;
-public class EnQueueId {
+public class EnqueueId {
- public static EnQueueId generate() {
+ public static EnqueueId generate() {
return of(UUIDs.timeBased());
}
- public static EnQueueId of(UUID uuid) {
+ public static EnqueueId of(UUID uuid) {
Preconditions.checkNotNull(uuid);
- return new EnQueueId(uuid);
+ return new EnqueueId(uuid);
}
- public static EnQueueId ofSerialized(String serialized) {
+ public static EnqueueId ofSerialized(String serialized) {
Preconditions.checkNotNull(serialized);
return of(UUID.fromString(serialized));
}
private final UUID id;
- private EnQueueId(UUID id) {
+ private EnqueueId(UUID id) {
this.id = id;
}
@@ -57,10 +57,10 @@ public class EnQueueId {
@Override
public final boolean equals(Object o) {
- if (o instanceof EnQueueId) {
- EnQueueId enQueueId = (EnQueueId) o;
+ if (o instanceof EnqueueId) {
+ EnqueueId enqueueId = (EnqueueId) o;
- return Objects.equals(this.id, enQueueId.id);
+ return Objects.equals(this.id, enqueueId.id);
}
return false;
}
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/EnqueuedItem.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/EnqueuedItem.java
index d693906..4bdc4a9 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/EnqueuedItem.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/EnqueuedItem.java
@@ -33,7 +33,7 @@ public class EnqueuedItem {
@FunctionalInterface
interface RequireEnqueueId {
- RequireMailQueueName enQueueId(EnQueueId id);
+ RequireMailQueueName enqueueId(EnqueueId id);
}
@FunctionalInterface
@@ -57,20 +57,20 @@ public class EnqueuedItem {
}
class ReadyToBuild {
- private final EnQueueId enQueueId;
+ private final EnqueueId enqueueId;
private final MailQueueName mailQueueName;
private final Mail mail;
private final Instant enqueuedTime;
private final MimeMessagePartsId partsId;
- ReadyToBuild(EnQueueId enQueueId, MailQueueName mailQueueName, Mail mail, Instant enqueuedTime, MimeMessagePartsId partsId) {
- Preconditions.checkNotNull(enQueueId, "'enQueueId' is mandatory");
+ ReadyToBuild(EnqueueId enqueueId, MailQueueName mailQueueName, Mail mail, Instant enqueuedTime, MimeMessagePartsId partsId) {
+ Preconditions.checkNotNull(enqueueId, "'enqueueId' is mandatory");
Preconditions.checkNotNull(mailQueueName, "'mailQueueName' is mandatory");
Preconditions.checkNotNull(mail, "'mail' is mandatory");
Preconditions.checkNotNull(enqueuedTime, "'enqueuedTime' is mandatory");
Preconditions.checkNotNull(partsId, "'partsId' is mandatory");
- this.enQueueId = enQueueId;
+ this.enqueueId = enqueueId;
this.mailQueueName = mailQueueName;
this.mail = mail;
this.enqueuedTime = enqueuedTime;
@@ -78,31 +78,31 @@ public class EnqueuedItem {
}
public EnqueuedItem build() {
- return new EnqueuedItem(enQueueId, mailQueueName, mail, enqueuedTime, partsId);
+ return new EnqueuedItem(enqueueId, mailQueueName, mail, enqueuedTime, partsId);
}
}
}
public static Builder.RequireEnqueueId builder() {
- return enQueueId -> queueName -> mail -> enqueuedTime -> partsId -> new Builder.ReadyToBuild(enQueueId, queueName, mail, enqueuedTime, partsId);
+ return enqueueId -> queueName -> mail -> enqueuedTime -> partsId -> new Builder.ReadyToBuild(enqueueId, queueName, mail, enqueuedTime, partsId);
}
- private final EnQueueId enQueueId;
+ private final EnqueueId enqueueId;
private final MailQueueName mailQueueName;
private final Mail mail;
private final Instant enqueuedTime;
private final MimeMessagePartsId partsId;
- EnqueuedItem(EnQueueId enQueueId, MailQueueName mailQueueName, Mail mail, Instant enqueuedTime, MimeMessagePartsId partsId) {
- this.enQueueId = enQueueId;
+ EnqueuedItem(EnqueueId enqueueId, MailQueueName mailQueueName, Mail mail, Instant enqueuedTime, MimeMessagePartsId partsId) {
+ this.enqueueId = enqueueId;
this.mailQueueName = mailQueueName;
this.mail = mail;
this.enqueuedTime = enqueuedTime;
this.partsId = partsId;
}
- public EnQueueId getEnQueueId() {
- return enQueueId;
+ public EnqueueId getEnqueueId() {
+ return enqueueId;
}
public MailQueueName getMailQueueName() {
@@ -126,7 +126,7 @@ public class EnqueuedItem {
if (o instanceof EnqueuedItem) {
EnqueuedItem that = (EnqueuedItem) o;
- return Objects.equals(this.enQueueId, that.enQueueId)
+ return Objects.equals(this.enqueueId, that.enqueueId)
&& Objects.equals(this.mailQueueName, that.mailQueueName)
&& Objects.equals(this.mail, that.mail)
&& Objects.equals(this.enqueuedTime, that.enqueuedTime)
@@ -137,6 +137,6 @@ public class EnqueuedItem {
@Override
public final int hashCode() {
- return Objects.hash(enQueueId, mailQueueName, mail, enqueuedTime, partsId);
+ return Objects.hash(enqueueId, mailQueueName, mail, enqueuedTime, partsId);
}
}
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/Enqueuer.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/Enqueuer.java
index 8f0c0fd..776a739 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/Enqueuer.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/Enqueuer.java
@@ -61,9 +61,9 @@ class Enqueuer {
}
void enQueue(Mail mail) throws MailQueue.MailQueueException {
- EnQueueId enQueueId = EnQueueId.generate();
+ EnqueueId enqueueId = EnqueueId.generate();
saveMail(mail)
- .map(partIds -> new MailReference(enQueueId, mail, partIds))
+ .map(partIds -> new MailReference(enqueueId, mail, partIds))
.map(Throwing.function(this::publishReferenceToRabbit).sneakyThrow())
.flatMap(mailQueueView::storeMail)
.thenEmpty(Mono.fromRunnable(enqueueMetric::increment))
@@ -82,7 +82,7 @@ class Enqueuer {
rabbitClient.publish(name, getMailReferenceBytes(mailReference));
return EnqueuedItem.builder()
- .enQueueId(mailReference.getEnQueueId())
+ .enqueueId(mailReference.getEnqueueId())
.mailQueueName(name)
.mail(mailReference.getMail())
.enqueuedTime(clock.instant())
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailLoader.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailLoader.java
index ca7df53..98fdc5c 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailLoader.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailLoader.java
@@ -39,14 +39,14 @@ class MailLoader {
this.blobIdFactory = blobIdFactory;
}
- Pair<EnQueueId, Mail> load(MailReferenceDTO dto) throws MailQueue.MailQueueException {
+ Pair<EnqueueId, Mail> load(MailReferenceDTO dto) throws MailQueue.MailQueueException {
try {
MailReference mailReference = dto.toMailReference(blobIdFactory);
Mail mail = mailReference.getMail();
MimeMessage mimeMessage = mimeMessageStore.read(mailReference.getPartsId()).block();
mail.setMessage(mimeMessage);
- return Pair.of(mailReference.getEnQueueId(), mail);
+ return Pair.of(mailReference.getEnqueueId(), mail);
} catch (AddressException e) {
throw new MailQueue.MailQueueException("Failed to parse mail address", e);
} catch (MessagingException e) {
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailReference.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailReference.java
index 031d97b..16165d1 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailReference.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailReference.java
@@ -24,18 +24,18 @@ import org.apache.mailet.Mail;
public class MailReference {
- private final EnQueueId enQueueId;
+ private final EnqueueId enqueueId;
private final Mail mail;
private final MimeMessagePartsId partsId;
- public MailReference(EnQueueId enQueueId, Mail mail, MimeMessagePartsId partsId) {
- this.enQueueId = enQueueId;
+ public MailReference(EnqueueId enqueueId, Mail mail, MimeMessagePartsId partsId) {
+ this.enqueueId = enqueueId;
this.mail = mail;
this.partsId = partsId;
}
- public EnQueueId getEnQueueId() {
- return enQueueId;
+ public EnqueueId getEnqueueId() {
+ return enqueueId;
}
public Mail getMail() {
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailReferenceDTO.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailReferenceDTO.java
index f63458f..4d88292 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailReferenceDTO.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailReferenceDTO.java
@@ -55,7 +55,7 @@ class MailReferenceDTO {
MimeMessagePartsId partsId = mailReference.getPartsId();
return new MailReferenceDTO(
- mailReference.getEnQueueId().serialize(),
+ mailReference.getEnqueueId().serialize(),
Optional.ofNullable(mail.getRecipients()).map(Collection::stream)
.orElse(Stream.empty())
.map(MailAddress::asString)
@@ -91,7 +91,7 @@ class MailReferenceDTO {
.collect(Guavate.toImmutableMap(name, value));
}
- private final String enQueueId;
+ private final String enqueueId;
private final ImmutableList<String> recipients;
private final String name;
private final Optional<String> sender;
@@ -106,7 +106,7 @@ class MailReferenceDTO {
private final String bodyBlobId;
@JsonCreator
- private MailReferenceDTO(@JsonProperty("enQueueId") String enQueueId,
+ private MailReferenceDTO(@JsonProperty("enqueueId") String enqueueId,
@JsonProperty("recipients") ImmutableList<String> recipients,
@JsonProperty("name") String name,
@JsonProperty("sender") Optional<String> sender,
@@ -119,7 +119,7 @@ class MailReferenceDTO {
@JsonProperty("perRecipientHeaders") Map<String, HeadersDto> perRecipientHeaders,
@JsonProperty("headerBlobId") String headerBlobId,
@JsonProperty("bodyBlobId") String bodyBlobId) {
- this.enQueueId = enQueueId;
+ this.enqueueId = enqueueId;
this.recipients = recipients;
this.name = name;
this.sender = sender;
@@ -134,9 +134,9 @@ class MailReferenceDTO {
this.bodyBlobId = bodyBlobId;
}
- @JsonProperty("enQueueId")
- public String getEnQueueId() {
- return enQueueId;
+ @JsonProperty("enqueueId")
+ public String getEnqueueId() {
+ return enqueueId;
}
@JsonProperty("recipients")
@@ -199,8 +199,8 @@ class MailReferenceDTO {
return bodyBlobId;
}
- EnQueueId retrieveEnqueueId() {
- return EnQueueId.ofSerialized(enQueueId);
+ EnqueueId retrieveEnqueueId() {
+ return EnqueueId.ofSerialized(enQueueId);
}
MailReference toMailReference(BlobId.Factory blobIdFactory) {
@@ -209,7 +209,7 @@ class MailReferenceDTO {
.bodyBlobId(blobIdFactory.from(bodyBlobId))
.build();
- return new MailReference(EnQueueId.ofSerialized(enQueueId), mailMetadata(), messagePartsId);
+ return new MailReference(EnqueueId.ofSerialized(enqueueId), mailMetadata(), messagePartsId);
}
private MailImpl mailMetadata() {
@@ -255,7 +255,7 @@ class MailReferenceDTO {
if (o instanceof MailReferenceDTO) {
MailReferenceDTO mailDTO = (MailReferenceDTO) o;
- return Objects.equals(this.enQueueId, mailDTO.enQueueId)
+ return Objects.equals(this.enqueueId, mailDTO.enqueueId)
&& Objects.equals(this.recipients, mailDTO.recipients)
&& Objects.equals(this.name, mailDTO.name)
&& Objects.equals(this.sender, mailDTO.sender)
@@ -274,6 +274,6 @@ class MailReferenceDTO {
@Override
public final int hashCode() {
- return Objects.hash(enQueueId, recipients, name, sender, state, errorMessage, lastUpdated, attributes, remoteAddr, remoteHost, perRecipientHeaders, headerBlobId, bodyBlobId);
+ return Objects.hash(enqueueId, recipients, name, sender, state, errorMessage, lastUpdated, attributes, remoteAddr, remoteHost, perRecipientHeaders, headerBlobId, bodyBlobId);
}
}
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 3660d12..fdf24a1 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
@@ -54,7 +54,7 @@ public class RabbitMQMailQueueFactory implements MailQueueFactory<RabbitMQMailQu
private final RabbitClient rabbitClient;
private final Store<MimeMessage, MimeMessagePartsId> mimeMessageStore;
private final MailReferenceSerializer mailReferenceSerializer;
- private final Function<MailReferenceDTO, Pair<EnQueueId, Mail>> mailLoader;
+ private final Function<MailReferenceDTO, Pair<EnqueueId, Mail>> mailLoader;
private final MailQueueView.Factory mailQueueViewFactory;
private final Clock clock;
private final MailQueueItemDecoratorFactory decoratorFactory;
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/api/DeleteCondition.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/api/DeleteCondition.java
index aa3973a..12aba7b 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/api/DeleteCondition.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/api/DeleteCondition.java
@@ -23,7 +23,7 @@ import java.util.Objects;
import org.apache.commons.lang3.NotImplementedException;
import org.apache.james.queue.api.ManageableMailQueue;
-import org.apache.james.queue.rabbitmq.EnQueueId;
+import org.apache.james.queue.rabbitmq.EnqueueId;
import org.apache.james.queue.rabbitmq.EnqueuedItem;
import com.google.common.base.Preconditions;
@@ -113,20 +113,20 @@ public interface DeleteCondition {
}
class WithEnqueueId implements DeleteCondition {
- private final EnQueueId enQueueId;
+ private final EnqueueId enqueueId;
- WithEnqueueId(EnQueueId enQueueId) {
- this.enQueueId = enQueueId;
+ WithEnqueueId(EnqueueId enqueueId) {
+ this.enqueueId = enqueueId;
}
- public EnQueueId getEnQueueId() {
- return enQueueId;
+ public EnqueueId getEnqueueId() {
+ return enqueueId;
}
@Override
public boolean shouldBeDeleted(EnqueuedItem enqueuedItem) {
Preconditions.checkNotNull(enqueuedItem);
- return enqueuedItem.getEnQueueId().equals(enQueueId);
+ return enqueuedItem.getEnqueueId().equals(enqueueId);
}
}
@@ -191,7 +191,7 @@ public interface DeleteCondition {
return new WithName(value);
}
- static WithEnqueueId withEnqueueId(EnQueueId value) {
+ static WithEnqueueId withEnqueueId(EnqueueId value) {
Preconditions.checkNotNull(value);
return new WithEnqueueId(value);
}
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/api/MailQueueView.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/api/MailQueueView.java
index a641484..33eec80 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/api/MailQueueView.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/api/MailQueueView.java
@@ -20,7 +20,7 @@
package org.apache.james.queue.rabbitmq.view.api;
import org.apache.james.queue.api.ManageableMailQueue;
-import org.apache.james.queue.rabbitmq.EnQueueId;
+import org.apache.james.queue.rabbitmq.EnqueueId;
import org.apache.james.queue.rabbitmq.EnqueuedItem;
import org.apache.james.queue.rabbitmq.MailQueueName;
@@ -38,7 +38,7 @@ public interface MailQueueView {
long delete(DeleteCondition deleteCondition);
- Mono<Boolean> isPresent(EnQueueId id);
+ Mono<Boolean> isPresent(EnqueueId id);
ManageableMailQueue.MailQueueIterator browse();
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueBrowser.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueBrowser.java
index baf19a3..b5a5d7f 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueBrowser.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueBrowser.java
@@ -139,7 +139,7 @@ public class CassandraMailQueueBrowser {
private Flux<EnqueuedItemWithSlicingContext> browseBucket(MailQueueName queueName, Slice slice, BucketId bucketId) {
return enqueuedMailsDao.selectEnqueuedMails(queueName, slice, bucketId)
- .filterWhen(mailReference -> deletedMailsDao.isStillEnqueued(queueName, mailReference.getEnqueuedItem().getEnQueueId()));
+ .filterWhen(mailReference -> deletedMailsDao.isStillEnqueued(queueName, mailReference.getEnqueuedItem().getEnqueueId()));
}
private Flux<Slice> allSlicesStartingAt(Instant browseStart) {
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueMailDelete.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueMailDelete.java
index b326eb4..a08318e 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueMailDelete.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueMailDelete.java
@@ -24,7 +24,7 @@ import java.util.concurrent.ThreadLocalRandom;
import javax.inject.Inject;
-import org.apache.james.queue.rabbitmq.EnQueueId;
+import org.apache.james.queue.rabbitmq.EnqueueId;
import org.apache.james.queue.rabbitmq.MailQueueName;
import org.apache.james.queue.rabbitmq.view.cassandra.configuration.CassandraMailQueueViewConfiguration;
@@ -49,14 +49,14 @@ public class CassandraMailQueueMailDelete {
this.configuration = configuration;
}
- Mono<Void> considerDeleted(EnQueueId enQueueId, MailQueueName mailQueueName) {
+ Mono<Void> considerDeleted(EnqueueId enqueueId, MailQueueName mailQueueName) {
return deletedMailsDao
- .markAsDeleted(mailQueueName, enQueueId)
+ .markAsDeleted(mailQueueName, enqueueId)
.doOnNext(ignored -> maybeUpdateBrowseStart(mailQueueName));
}
- Mono<Boolean> isDeleted(EnQueueId enQueueId, MailQueueName mailQueueName) {
- return deletedMailsDao.isDeleted(mailQueueName, enQueueId);
+ Mono<Boolean> isDeleted(EnqueueId enqueueId, MailQueueName mailQueueName) {
+ return deletedMailsDao.isDeleted(mailQueueName, enqueueId);
}
void updateBrowseStart(MailQueueName mailQueueName) {
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueView.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueView.java
index a2cabda..0c5af4c 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueView.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueView.java
@@ -22,7 +22,7 @@ package org.apache.james.queue.rabbitmq.view.cassandra;
import javax.inject.Inject;
import org.apache.james.queue.api.ManageableMailQueue;
-import org.apache.james.queue.rabbitmq.EnQueueId;
+import org.apache.james.queue.rabbitmq.EnqueueId;
import org.apache.james.queue.rabbitmq.EnqueuedItem;
import org.apache.james.queue.rabbitmq.MailQueueName;
import org.apache.james.queue.rabbitmq.view.api.DeleteCondition;
@@ -101,8 +101,8 @@ public class CassandraMailQueueView implements MailQueueView {
@Override
public long delete(DeleteCondition deleteCondition) {
if (deleteCondition instanceof DeleteCondition.WithEnqueueId) {
- DeleteCondition.WithEnqueueId enQueueIdCondition = (DeleteCondition.WithEnqueueId) deleteCondition;
- delete(enQueueIdCondition.getEnQueueId()).block();
+ DeleteCondition.WithEnqueueId enqueueIdCondition = (DeleteCondition.WithEnqueueId) deleteCondition;
+ delete(enqueueIdCondition.getEnqueueId()).block();
return 1L;
}
return browseThenDelete(deleteCondition);
@@ -112,18 +112,18 @@ public class CassandraMailQueueView implements MailQueueView {
return cassandraMailQueueBrowser.browseReferences(mailQueueName)
.map(EnqueuedItemWithSlicingContext::getEnqueuedItem)
.filter(deleteCondition::shouldBeDeleted)
- .flatMap(mailReference -> cassandraMailQueueMailDelete.considerDeleted(mailReference.getEnQueueId(), mailQueueName))
+ .flatMap(mailReference -> cassandraMailQueueMailDelete.considerDeleted(mailReference.getEnqueueId(), mailQueueName))
.count()
.doOnNext(ignored -> cassandraMailQueueMailDelete.updateBrowseStart(mailQueueName))
.block();
}
- private Mono<Void> delete(EnQueueId enQueueId) {
- return cassandraMailQueueMailDelete.considerDeleted(enQueueId, mailQueueName);
+ private Mono<Void> delete(EnqueueId enqueueId) {
+ return cassandraMailQueueMailDelete.considerDeleted(enqueueId, mailQueueName);
}
@Override
- public Mono<Boolean> isPresent(EnQueueId id) {
+ public Mono<Boolean> isPresent(EnqueueId id) {
return cassandraMailQueueMailDelete.isDeleted(id, mailQueueName)
.map(bool -> !bool);
}
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/DeletedMailsDAO.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/DeletedMailsDAO.java
index 7f44c6f..a0551e9 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/DeletedMailsDAO.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/DeletedMailsDAO.java
@@ -30,7 +30,7 @@ import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueV
import javax.inject.Inject;
import org.apache.james.backends.cassandra.utils.CassandraAsyncExecutor;
-import org.apache.james.queue.rabbitmq.EnQueueId;
+import org.apache.james.queue.rabbitmq.EnqueueId;
import org.apache.james.queue.rabbitmq.MailQueueName;
import com.datastax.driver.core.PreparedStatement;
@@ -64,21 +64,21 @@ public class DeletedMailsDAO {
.and(eq(ENQUEUE_ID, bindMarker(ENQUEUE_ID))));
}
- Mono<Void> markAsDeleted(MailQueueName mailQueueName, EnQueueId enQueueId) {
+ Mono<Void> markAsDeleted(MailQueueName mailQueueName, EnqueueId enqueueId) {
return executor.executeVoid(insertOne.bind()
.setString(QUEUE_NAME, mailQueueName.asString())
- .setUUID(ENQUEUE_ID, enQueueId.asUUID()));
+ .setUUID(ENQUEUE_ID, enqueueId.asUUID()));
}
- Mono<Boolean> isDeleted(MailQueueName mailQueueName, EnQueueId enQueueId) {
+ Mono<Boolean> isDeleted(MailQueueName mailQueueName, EnqueueId enqueueId) {
return executor.executeReturnExists(
selectOne.bind()
.setString(QUEUE_NAME, mailQueueName.asString())
- .setUUID(ENQUEUE_ID, enQueueId.asUUID()));
+ .setUUID(ENQUEUE_ID, enqueueId.asUUID()));
}
- Mono<Boolean> isStillEnqueued(MailQueueName mailQueueName, EnQueueId enQueueId) {
- return isDeleted(mailQueueName, enQueueId)
+ Mono<Boolean> isStillEnqueued(MailQueueName mailQueueName, EnqueueId enqueueId) {
+ return isDeleted(mailQueueName, enqueueId)
.map(b -> !b);
}
}
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/EnqueuedMailsDAO.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/EnqueuedMailsDAO.java
index 1c8627e..55f0e43 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/EnqueuedMailsDAO.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/EnqueuedMailsDAO.java
@@ -125,7 +125,7 @@ public class EnqueuedMailsDAO {
.setTimestamp(TIME_RANGE_START, Date.from(slicingContext.getTimeRangeStart()))
.setInt(BUCKET_ID, slicingContext.getBucketId().getValue())
.setTimestamp(ENQUEUED_TIME, Date.from(enqueuedItem.getEnqueuedTime()))
- .setUUID(ENQUEUE_ID, enqueuedItem.getEnQueueId().asUUID())
+ .setUUID(ENQUEUE_ID, enqueuedItem.getEnqueueId().asUUID())
.setString(NAME, mail.getName())
.setString(HEADER_BLOB_ID, mimeMessagePartsId.getHeaderBlobId().asString())
.setString(BODY_BLOB_ID, mimeMessagePartsId.getBodyBlobId().asString())
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/EnqueuedMailsDaoUtil.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/EnqueuedMailsDaoUtil.java
index 004e9af..04c6cec 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/EnqueuedMailsDaoUtil.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/EnqueuedMailsDaoUtil.java
@@ -58,7 +58,7 @@ import org.apache.james.backends.cassandra.init.CassandraTypesProvider;
import org.apache.james.blob.api.BlobId;
import org.apache.james.blob.mail.MimeMessagePartsId;
import org.apache.james.core.MailAddress;
-import org.apache.james.queue.rabbitmq.EnQueueId;
+import org.apache.james.queue.rabbitmq.EnqueueId;
import org.apache.james.queue.rabbitmq.EnqueuedItem;
import org.apache.james.queue.rabbitmq.MailQueueName;
import org.apache.james.queue.rabbitmq.view.cassandra.model.BucketedSlices;
@@ -81,7 +81,7 @@ public class EnqueuedMailsDaoUtil {
static EnqueuedItemWithSlicingContext toEnqueuedMail(Row row, BlobId.Factory blobFactory) {
MailQueueName queueName = MailQueueName.fromString(row.getString(QUEUE_NAME));
- EnQueueId enQueueId = EnQueueId.of(row.getUUID(ENQUEUE_ID));
+ EnqueueId enqueueId = EnqueueId.of(row.getUUID(ENQUEUE_ID));
Instant timeRangeStart = row.getTimestamp(TIME_RANGE_START).toInstant();
BucketedSlices.BucketId bucketId = BucketedSlices.BucketId.of(row.getInt(BUCKET_ID));
Instant enqueuedTime = row.getTimestamp(ENQUEUED_TIME).toInstant();
@@ -122,7 +122,7 @@ public class EnqueuedMailsDaoUtil {
.addAttributes(toAttributes(rawAttributes))
.build();
EnqueuedItem enqueuedItem = EnqueuedItem.builder()
- .enQueueId(enQueueId)
+ .enqueueId(enqueueId)
.mailQueueName(queueName)
.mail(mail)
.enqueuedTime(enqueuedTime)
diff --git a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/EnqueuedItemTest.java b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/EnqueuedItemTest.java
index 73c8302..d7b5e66 100644
--- a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/EnqueuedItemTest.java
+++ b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/EnqueuedItemTest.java
@@ -59,7 +59,7 @@ class EnqueuedItemTest {
@Test
void buildShouldThrowWhenEnqueueIdIsNull() {
assertThatThrownBy(() -> EnqueuedItem.builder()
- .enQueueId(null)
+ .enqueueId(null)
.mailQueueName(mailQueueName)
.mail(mail)
.enqueuedTime(enqueuedTime)
@@ -71,7 +71,7 @@ class EnqueuedItemTest {
@Test
void buildShouldThrowWhenMailQueueNameIsNull() {
assertThatThrownBy(() -> EnqueuedItem.builder()
- .enQueueId(EnQueueId.generate())
+ .enqueueId(EnqueueId.generate())
.mailQueueName(null)
.mail(mail)
.enqueuedTime(enqueuedTime)
@@ -83,7 +83,7 @@ class EnqueuedItemTest {
@Test
void buildShouldThrowWhenMailIsNull() {
assertThatThrownBy(() -> EnqueuedItem.builder()
- .enQueueId(EnQueueId.generate())
+ .enqueueId(EnqueueId.generate())
.mailQueueName(mailQueueName)
.mail(null)
.enqueuedTime(enqueuedTime)
@@ -95,7 +95,7 @@ class EnqueuedItemTest {
@Test
void buildShouldThrowWhenEnqueuedTimeIsNull() {
assertThatThrownBy(() -> EnqueuedItem.builder()
- .enQueueId(EnQueueId.generate())
+ .enqueueId(EnqueueId.generate())
.mailQueueName(mailQueueName)
.mail(mail)
.enqueuedTime(null)
@@ -107,7 +107,7 @@ class EnqueuedItemTest {
@Test
void buildShouldThrowWhenMimeMessagePartsIdIsNull() {
assertThatThrownBy(() -> EnqueuedItem.builder()
- .enQueueId(EnQueueId.generate())
+ .enqueueId(EnqueueId.generate())
.mailQueueName(mailQueueName)
.mail(mail)
.enqueuedTime(enqueuedTime)
diff --git a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/MailDTOTest.java b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/MailDTOTest.java
index 2b110fa..c4dd573 100644
--- a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/MailDTOTest.java
+++ b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/MailDTOTest.java
@@ -46,7 +46,7 @@ import com.fasterxml.jackson.datatype.jdk8.Jdk8Module;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
class MailDTOTest {
- static final EnQueueId EN_QUEUE_ID = EnQueueId.ofSerialized("110e8400-e29b-11d4-a716-446655440000");
+ static final EnqueueId EN_QUEUE_ID = EnqueueId.ofSerialized("110e8400-e29b-11d4-a716-446655440000");
static final HashBlobId.Factory BLOB_ID_FACTORY = new HashBlobId.Factory();
static final Date LAST_UPDATED = Date.from(Instant.parse("2016-09-08T14:25:52.000Z"));
diff --git a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/api/DeleteConditionTest.java b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/api/DeleteConditionTest.java
index d7d34fa..ddcd5da 100644
--- a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/api/DeleteConditionTest.java
+++ b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/api/DeleteConditionTest.java
@@ -28,7 +28,7 @@ import org.apache.james.blob.api.HashBlobId;
import org.apache.james.blob.mail.MimeMessagePartsId;
import org.apache.james.core.MailAddress;
import org.apache.james.queue.api.ManageableMailQueue;
-import org.apache.james.queue.rabbitmq.EnQueueId;
+import org.apache.james.queue.rabbitmq.EnqueueId;
import org.apache.james.queue.rabbitmq.EnqueuedItem;
import org.apache.james.queue.rabbitmq.MailQueueName;
import org.apache.mailet.base.test.FakeMail;
@@ -42,8 +42,8 @@ class DeleteConditionTest {
private static final String ADDRESS_2 = "any2@toto.com";
private static final String NAME = "name";
private static final String VALUE = "value";
- private static final EnQueueId EN_QUEUE_ID_1 = EnQueueId.ofSerialized("110e8400-e29b-11d4-a716-446655440000");
- private static final EnQueueId EN_QUEUE_ID_2 = EnQueueId.ofSerialized("464765a0-e4e7-11e4-aba4-710c1de3782b");
+ private static final EnqueueId ENQUEUE_ID_1 = EnqueueId.ofSerialized("110e8400-e29b-11d4-a716-446655440000");
+ private static final EnqueueId ENQUEUE_ID_2 = EnqueueId.ofSerialized("464765a0-e4e7-11e4-aba4-710c1de3782b");
private static final MailQueueName OUT_GOING_1 = MailQueueName.fromString("OUT_GOING_1");
private static final Instant ENQUEUE_TIME = Instant.now();
private static final MimeMessagePartsId MESSAGE_PARTS_ID = MimeMessagePartsId.builder()
@@ -57,7 +57,7 @@ class DeleteConditionTest {
void allShouldReturnTrue() throws Exception {
assertThat(
DeleteCondition.all()
- .shouldBeDeleted(enQueueItemForMail(FakeMail.builder().name("name").build())))
+ .shouldBeDeleted(enqueuedItemForMail(FakeMail.builder().name("name").build())))
.isTrue();
}
@@ -76,7 +76,7 @@ class DeleteConditionTest {
}
@Nested
- class WithEnQueueIdTest {
+ class WithEnqueueIdTest {
@Test
void withSenderShouldThrowOnNullCondition() {
assertThatThrownBy(() ->
@@ -87,7 +87,7 @@ class DeleteConditionTest {
@Test
void shouldBeDeletedShouldReturnTrueWhenSameId() throws Exception {
EnqueuedItem enqueuedItem = EnqueuedItem.builder()
- .enQueueId(EN_QUEUE_ID_1)
+ .enqueueId(ENQUEUE_ID_1)
.mailQueueName(OUT_GOING_1)
.mail(FakeMail.builder()
.name("name")
@@ -97,14 +97,14 @@ class DeleteConditionTest {
.mimeMessagePartsId(MESSAGE_PARTS_ID)
.build();
- assertThat(DeleteCondition.withEnqueueId(EN_QUEUE_ID_1).shouldBeDeleted(enqueuedItem))
+ assertThat(DeleteCondition.withEnqueueId(ENQUEUE_ID_1).shouldBeDeleted(enqueuedItem))
.isTrue();
}
@Test
void shouldBeDeletedShouldReturnFalseWhenDifferentId() throws Exception {
EnqueuedItem enqueuedItem = EnqueuedItem.builder()
- .enQueueId(EN_QUEUE_ID_2)
+ .enqueueId(ENQUEUE_ID_2)
.mailQueueName(OUT_GOING_1)
.mail(FakeMail.builder()
.name("name")
@@ -114,7 +114,7 @@ class DeleteConditionTest {
.mimeMessagePartsId(MESSAGE_PARTS_ID)
.build();
- assertThat(DeleteCondition.withEnqueueId(EN_QUEUE_ID_1).shouldBeDeleted(enqueuedItem))
+ assertThat(DeleteCondition.withEnqueueId(ENQUEUE_ID_1).shouldBeDeleted(enqueuedItem))
.isFalse();
}
}
@@ -140,7 +140,7 @@ class DeleteConditionTest {
void withSenderShouldReturnTrueWhenSameAddress() throws Exception {
assertThat(
DeleteCondition.withSender(ADDRESS)
- .shouldBeDeleted(enQueueItemForMail(FakeMail.builder()
+ .shouldBeDeleted(enqueuedItemForMail(FakeMail.builder()
.name("name")
.sender(ADDRESS)
.build())))
@@ -151,7 +151,7 @@ class DeleteConditionTest {
void withSenderShouldReturnFalseWhenDifferentAddress() throws Exception {
assertThat(
DeleteCondition.withSender(ADDRESS)
- .shouldBeDeleted(enQueueItemForMail(FakeMail.builder()
+ .shouldBeDeleted(enqueuedItemForMail(FakeMail.builder()
.name("name")
.sender(ADDRESS_2)
.recipient(ADDRESS)
@@ -163,7 +163,7 @@ class DeleteConditionTest {
void withSenderShouldNotThrowOnNullSender() throws Exception {
assertThat(
DeleteCondition.withSender(ADDRESS)
- .shouldBeDeleted(enQueueItemForMail(FakeMail.builder()
+ .shouldBeDeleted(enqueuedItemForMail(FakeMail.builder()
.name("name")
.sender(MailAddress.nullSender())
.build())))
@@ -174,7 +174,7 @@ class DeleteConditionTest {
void withSenderShouldAllowNullSenderMatchingNullSender() throws Exception {
assertThat(
DeleteCondition.withSender(MailAddress.NULL_SENDER_AS_STRING)
- .shouldBeDeleted(enQueueItemForMail(FakeMail.builder()
+ .shouldBeDeleted(enqueuedItemForMail(FakeMail.builder()
.name("name")
.sender(MailAddress.nullSender())
.build())))
@@ -208,7 +208,7 @@ class DeleteConditionTest {
void withNameShouldReturnTrueWhenSameName() throws Exception {
assertThat(
DeleteCondition.withName(NAME)
- .shouldBeDeleted(enQueueItemForMail(FakeMail.builder()
+ .shouldBeDeleted(enqueuedItemForMail(FakeMail.builder()
.name(NAME)
.build())))
.isTrue();
@@ -218,7 +218,7 @@ class DeleteConditionTest {
void withSenderShouldReturnFalseWhenDifferentAddress() throws Exception {
assertThat(
DeleteCondition.withName(NAME)
- .shouldBeDeleted(enQueueItemForMail(FakeMail.builder()
+ .shouldBeDeleted(enqueuedItemForMail(FakeMail.builder()
.name("other")
.build())))
.isFalse();
@@ -251,7 +251,7 @@ class DeleteConditionTest {
void withRecipientShouldReturnTrueWhenSameAddress() throws Exception {
assertThat(
DeleteCondition.withRecipient(ADDRESS)
- .shouldBeDeleted(enQueueItemForMail(FakeMail.builder()
+ .shouldBeDeleted(enqueuedItemForMail(FakeMail.builder()
.name("name")
.recipient(ADDRESS)
.build())))
@@ -262,7 +262,7 @@ class DeleteConditionTest {
void withRecipientShouldReturnTrueWhenAtListOneMatches() throws Exception {
assertThat(
DeleteCondition.withRecipient(ADDRESS)
- .shouldBeDeleted(enQueueItemForMail(FakeMail.builder()
+ .shouldBeDeleted(enqueuedItemForMail(FakeMail.builder()
.name("name")
.recipients(ADDRESS, ADDRESS_2)
.build())))
@@ -273,7 +273,7 @@ class DeleteConditionTest {
void withRecipientShouldReturnFalseWhenDifferentAddress() throws Exception {
assertThat(
DeleteCondition.withRecipient(ADDRESS)
- .shouldBeDeleted(enQueueItemForMail(FakeMail.builder()
+ .shouldBeDeleted(enqueuedItemForMail(FakeMail.builder()
.name("name")
.sender(ADDRESS)
.recipient(ADDRESS_2)
@@ -285,7 +285,7 @@ class DeleteConditionTest {
void withRecipientShouldReturnFalseWhenNoRecipient() throws Exception {
assertThat(
DeleteCondition.withRecipient(ADDRESS)
- .shouldBeDeleted(enQueueItemForMail(FakeMail.builder()
+ .shouldBeDeleted(enqueuedItemForMail(FakeMail.builder()
.name("name")
.build())))
.isFalse();
@@ -318,9 +318,9 @@ class DeleteConditionTest {
}
}
- private EnqueuedItem enQueueItemForMail(FakeMail mail) {
+ private EnqueuedItem enqueuedItemForMail(FakeMail mail) {
return EnqueuedItem.builder()
- .enQueueId(EN_QUEUE_ID_1)
+ .enqueueId(ENQUEUE_ID_1)
.mailQueueName(OUT_GOING_1)
.mail(mail)
.enqueuedTime(ENQUEUE_TIME)
diff --git a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/DeletedMailsDAOTest.java b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/DeletedMailsDAOTest.java
index 73f0f27..98c918a 100644
--- a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/DeletedMailsDAOTest.java
+++ b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/DeletedMailsDAOTest.java
@@ -25,7 +25,7 @@ import org.apache.james.backends.cassandra.CassandraCluster;
import org.apache.james.backends.cassandra.CassandraClusterExtension;
import org.apache.james.backends.cassandra.components.CassandraModule;
import org.apache.james.backends.cassandra.versions.CassandraSchemaVersionModule;
-import org.apache.james.queue.rabbitmq.EnQueueId;
+import org.apache.james.queue.rabbitmq.EnqueueId;
import org.apache.james.queue.rabbitmq.MailQueueName;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -35,8 +35,8 @@ class DeletedMailsDAOTest {
private static final MailQueueName OUT_GOING_1 = MailQueueName.fromString("OUT_GOING_1");
private static final MailQueueName OUT_GOING_2 = MailQueueName.fromString("OUT_GOING_2");
- private static final EnQueueId EN_QUEUE_ID_1 = EnQueueId.ofSerialized("110e8400-e29b-11d4-a716-446655440000");
- private static final EnQueueId EN_QUEUE_ID_2 = EnQueueId.ofSerialized("464765a0-e4e7-11e4-aba4-710c1de3782b");
+ private static final EnqueueId ENQUEUE_ID_1 = EnqueueId.ofSerialized("110e8400-e29b-11d4-a716-446655440000");
+ private static final EnqueueId ENQUEUE_ID_2 = EnqueueId.ofSerialized("464765a0-e4e7-11e4-aba4-710c1de3782b");
@RegisterExtension
static CassandraClusterExtension cassandraCluster = new CassandraClusterExtension(
@@ -54,14 +54,14 @@ class DeletedMailsDAOTest {
@Test
void markAsDeletedShouldWork() {
Boolean isDeletedBeforeMark = testee
- .isDeleted(OUT_GOING_1, EN_QUEUE_ID_1)
+ .isDeleted(OUT_GOING_1, ENQUEUE_ID_1)
.block();
assertThat(isDeletedBeforeMark).isFalse();
- testee.markAsDeleted(OUT_GOING_1, EN_QUEUE_ID_1).block();
+ testee.markAsDeleted(OUT_GOING_1, ENQUEUE_ID_1).block();
Boolean isDeletedAfterMark = testee
- .isDeleted(OUT_GOING_1, EN_QUEUE_ID_1)
+ .isDeleted(OUT_GOING_1, ENQUEUE_ID_1)
.block();
assertThat(isDeletedAfterMark).isTrue();
@@ -69,10 +69,10 @@ class DeletedMailsDAOTest {
@Test
void checkDeletedShouldReturnFalseWhenTableDoesntContainBothMailQueueAndMailKey() {
- testee.markAsDeleted(OUT_GOING_2, EN_QUEUE_ID_2).block();
+ testee.markAsDeleted(OUT_GOING_2, ENQUEUE_ID_2).block();
Boolean isDeleted = testee
- .isDeleted(OUT_GOING_1, EN_QUEUE_ID_1)
+ .isDeleted(OUT_GOING_1, ENQUEUE_ID_1)
.block();
assertThat(isDeleted).isFalse();
@@ -80,10 +80,10 @@ class DeletedMailsDAOTest {
@Test
void checkDeletedShouldReturnFalseWhenTableContainsMailQueueButNotMailKey() {
- testee.markAsDeleted(OUT_GOING_1, EN_QUEUE_ID_2).block();
+ testee.markAsDeleted(OUT_GOING_1, ENQUEUE_ID_2).block();
Boolean isDeleted = testee
- .isDeleted(OUT_GOING_1, EN_QUEUE_ID_1)
+ .isDeleted(OUT_GOING_1, ENQUEUE_ID_1)
.block();
assertThat(isDeleted).isFalse();
@@ -91,10 +91,10 @@ class DeletedMailsDAOTest {
@Test
void checkDeletedShouldReturnFalseWhenTableContainsMailKeyButNotMailQueue() {
- testee.markAsDeleted(OUT_GOING_2, EN_QUEUE_ID_1).block();
+ testee.markAsDeleted(OUT_GOING_2, ENQUEUE_ID_1).block();
Boolean isDeleted = testee
- .isDeleted(OUT_GOING_1, EN_QUEUE_ID_1)
+ .isDeleted(OUT_GOING_1, ENQUEUE_ID_1)
.block();
assertThat(isDeleted).isFalse();
@@ -102,10 +102,10 @@ class DeletedMailsDAOTest {
@Test
void checkDeletedShouldReturnTrueWhenTableContainsMailItem() {
- testee.markAsDeleted(OUT_GOING_1, EN_QUEUE_ID_1).block();
+ testee.markAsDeleted(OUT_GOING_1, ENQUEUE_ID_1).block();
Boolean isDeleted = testee
- .isDeleted(OUT_GOING_1, EN_QUEUE_ID_1)
+ .isDeleted(OUT_GOING_1, ENQUEUE_ID_1)
.block();
assertThat(isDeleted).isTrue();
diff --git a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/EnqueuedMailsDaoTest.java b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/EnqueuedMailsDaoTest.java
index 7dce6e5..94145c5 100644
--- a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/EnqueuedMailsDaoTest.java
+++ b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/EnqueuedMailsDaoTest.java
@@ -30,7 +30,7 @@ import org.apache.james.backends.cassandra.CassandraClusterExtension;
import org.apache.james.blob.api.BlobId;
import org.apache.james.blob.api.HashBlobId;
import org.apache.james.blob.mail.MimeMessagePartsId;
-import org.apache.james.queue.rabbitmq.EnQueueId;
+import org.apache.james.queue.rabbitmq.EnqueueId;
import org.apache.james.queue.rabbitmq.EnqueuedItem;
import org.apache.james.queue.rabbitmq.MailQueueName;
import org.apache.james.queue.rabbitmq.view.cassandra.model.BucketedSlices.BucketId;
@@ -44,7 +44,7 @@ import org.junit.jupiter.api.extension.RegisterExtension;
class EnqueuedMailsDaoTest {
private static final MailQueueName OUT_GOING_1 = MailQueueName.fromString("OUT_GOING_1");
- private static final EnQueueId EN_QUEUE_ID = EnQueueId.ofSerialized("110e8400-e29b-11d4-a716-446655440000");
+ private static final EnqueueId ENQUEUE_ID = EnqueueId.ofSerialized("110e8400-e29b-11d4-a716-446655440000");
private static final String NAME = "name";
private static int BUCKET_ID_VALUE = 10;
private static final BucketId BUCKET_ID = BucketId.of(BUCKET_ID_VALUE);
@@ -76,7 +76,7 @@ class EnqueuedMailsDaoTest {
void insertShouldWork() throws Exception {
testee.insert(EnqueuedItemWithSlicingContext.builder()
.enqueuedItem(EnqueuedItem.builder()
- .enQueueId(EN_QUEUE_ID)
+ .enqueueId(ENQUEUE_ID)
.mailQueueName(OUT_GOING_1)
.mail(FakeMail.builder()
.name(NAME)
@@ -99,7 +99,7 @@ class EnqueuedMailsDaoTest {
void selectEnqueuedMailsShouldWork() throws Exception {
testee.insert(EnqueuedItemWithSlicingContext.builder()
.enqueuedItem(EnqueuedItem.builder()
- .enQueueId(EN_QUEUE_ID)
+ .enqueueId(ENQUEUE_ID)
.mailQueueName(OUT_GOING_1)
.mail(FakeMail.builder()
.name(NAME)
@@ -113,7 +113,7 @@ class EnqueuedMailsDaoTest {
testee.insert(EnqueuedItemWithSlicingContext.builder()
.enqueuedItem(EnqueuedItem.builder()
- .enQueueId(EN_QUEUE_ID)
+ .enqueueId(ENQUEUE_ID)
.mailQueueName(OUT_GOING_1)
.mail(FakeMail.builder()
.name(NAME)
@@ -138,7 +138,7 @@ class EnqueuedMailsDaoTest {
softly.assertThat(slicingContext.getTimeRangeStart()).isEqualTo(NOW);
softly.assertThat(enqueuedItem.getMailQueueName()).isEqualTo(OUT_GOING_1);
softly.assertThat(enqueuedItem.getEnqueuedTime()).isEqualTo(NOW);
- softly.assertThat(enqueuedItem.getEnQueueId()).isEqualTo(EN_QUEUE_ID);
+ softly.assertThat(enqueuedItem.getEnqueueId()).isEqualTo(ENQUEUE_ID);
softly.assertThat(enqueuedItem.getMail().getName()).isEqualTo(NAME);
softly.assertThat(enqueuedItem.getPartsId()).isEqualTo(MIME_MESSAGE_PARTS_ID);
});
diff --git a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/model/EnQueueIdTest.java b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/model/EnqueueIdTest.java
similarity index 79%
rename from server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/model/EnQueueIdTest.java
rename to server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/model/EnqueueIdTest.java
index 2f21c34..3ee99ad 100644
--- a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/model/EnQueueIdTest.java
+++ b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/model/EnqueueIdTest.java
@@ -23,39 +23,39 @@ import static org.assertj.core.api.Assertions.assertThat;
import java.util.UUID;
-import org.apache.james.queue.rabbitmq.EnQueueId;
+import org.apache.james.queue.rabbitmq.EnqueueId;
import org.junit.jupiter.api.Test;
import nl.jqno.equalsverifier.EqualsVerifier;
-class EnQueueIdTest {
+class EnqueueIdTest {
private static final UUID UUID_1 = UUID.fromString("110e8400-e29b-11d4-a716-446655440000");
@Test
void shouldMatchBeanContract() {
- EqualsVerifier.forClass(EnQueueId.class)
+ EqualsVerifier.forClass(EnqueueId.class)
.verify();
}
@Test
void ofSerializedShouldDeserializeTheGivenEnqueueId() {
- assertThat(EnQueueId.ofSerialized(UUID_1.toString()))
- .isEqualTo(EnQueueId.of(UUID_1));
+ assertThat(EnqueueId.ofSerialized(UUID_1.toString()))
+ .isEqualTo(EnqueueId.of(UUID_1));
}
@Test
void serializeShouldReturnAStringRepresentation() {
- EnQueueId enQueueId = EnQueueId.of(UUID_1);
+ EnqueueId enqueueId = EnqueueId.of(UUID_1);
- assertThat(enQueueId.serialize())
+ assertThat(enqueueId.serialize())
.isEqualTo(UUID_1.toString());
}
@Test
void ofSerializedShouldRevertSerialize() {
- EnQueueId enQueueId = EnQueueId.of(UUID_1);
+ EnqueueId enqueueId = EnqueueId.of(UUID_1);
- assertThat(EnQueueId.ofSerialized(enQueueId.serialize()))
- .isEqualTo(enQueueId);
+ assertThat(EnqueueId.ofSerialized(enqueueId.serialize()))
+ .isEqualTo(enqueueId);
}
}
\ No newline at end of file
diff --git a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/model/EnqueuedItemWithSlicingContextTest.java b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/model/EnqueuedItemWithSlicingContextTest.java
index f1c2729..990ebe8 100644
--- a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/model/EnqueuedItemWithSlicingContextTest.java
+++ b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/model/EnqueuedItemWithSlicingContextTest.java
@@ -27,7 +27,7 @@ import javax.mail.MessagingException;
import org.apache.james.blob.api.HashBlobId;
import org.apache.james.blob.mail.MimeMessagePartsId;
-import org.apache.james.queue.rabbitmq.EnQueueId;
+import org.apache.james.queue.rabbitmq.EnqueueId;
import org.apache.james.queue.rabbitmq.EnqueuedItem;
import org.apache.james.queue.rabbitmq.MailQueueName;
import org.apache.mailet.base.test.FakeMail;
@@ -42,7 +42,7 @@ class EnqueuedItemWithSlicingContextTest {
private EnqueuedItemWithSlicingContextTest() throws MessagingException {
enqueuedItem = EnqueuedItem.builder()
- .enQueueId(EnQueueId.generate())
+ .enqueueId(EnqueueId.generate())
.mailQueueName(MailQueueName.fromString("mailQueueName"))
.mail(FakeMail.builder()
.name("name")
diff --git a/server/queue/queue-rabbitmq/src/test/resources/json/mail1.json b/server/queue/queue-rabbitmq/src/test/resources/json/mail1.json
index 9bfbee0..04d40d2 100644
--- a/server/queue/queue-rabbitmq/src/test/resources/json/mail1.json
+++ b/server/queue/queue-rabbitmq/src/test/resources/json/mail1.json
@@ -1,5 +1,5 @@
{
- "enQueueId": "110e8400-e29b-11d4-a716-446655440000",
+ "enqueueId": "110e8400-e29b-11d4-a716-446655440000",
"recipients": [
"recipient1@localhost",
"recipient2@localhost"
diff --git a/server/queue/queue-rabbitmq/src/test/resources/json/mail_min.json b/server/queue/queue-rabbitmq/src/test/resources/json/mail_min.json
index 21dcd7c..8f737c6 100644
--- a/server/queue/queue-rabbitmq/src/test/resources/json/mail_min.json
+++ b/server/queue/queue-rabbitmq/src/test/resources/json/mail_min.json
@@ -1,5 +1,5 @@
{
- "enQueueId": "110e8400-e29b-11d4-a716-446655440000",
+ "enqueueId": "110e8400-e29b-11d4-a716-446655440000",
"recipients":[],
"name":"mail-name-558",
"sender":null,
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 06/28: JAMES-2794 RabbitMQ MailQueue projections
should rely on EnqueueId
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 961ef33e62b74982ef3cc17768887288f9b84910
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Tue Jun 18 11:50:30 2019 +0700
JAMES-2794 RabbitMQ MailQueue projections should rely on EnqueueId
This allows insertions of several emails sharing the same name.
---
.../org/apache/james/queue/rabbitmq/Dequeuer.java | 22 +++++++++----
.../apache/james/queue/rabbitmq/EnqueuedItem.java | 37 +++++++++++++---------
.../org/apache/james/queue/rabbitmq/Enqueuer.java | 13 +++++---
.../james/queue/rabbitmq/MailReferenceDTO.java | 24 +++++++++++---
.../queue/rabbitmq/view/api/DeleteCondition.java | 23 +++++++++++---
.../queue/rabbitmq/view/api/MailQueueView.java | 4 +--
.../view/cassandra/CassandraMailQueueBrowser.java | 3 +-
.../cassandra/CassandraMailQueueMailDelete.java | 15 +++------
.../view/cassandra/CassandraMailQueueView.java | 19 ++++++-----
.../cassandra/CassandraMailQueueViewModule.java | 15 +++++----
.../rabbitmq/view/cassandra/DeletedMailsDAO.java | 21 ++++++------
.../rabbitmq/view/cassandra/EnqueuedMailsDAO.java | 9 ++++--
.../view/cassandra/EnqueuedMailsDaoUtil.java | 8 +++--
.../james/queue/rabbitmq/EnqueuedItemTest.java | 17 ++++++++++
.../apache/james/queue/rabbitmq/MailDTOTest.java | 3 ++
.../queue/rabbitmq/RabbitMQMailQueueTest.java | 7 ----
.../view/cassandra/DeletedMailsDAOTest.java | 35 +++++++++++---------
.../view/cassandra/EnqueuedMailsDaoTest.java | 17 ++++++----
.../model/EnqueuedItemWithSlicingContextTest.java | 5 ++-
.../src/test/resources/json/mail1.json | 1 +
.../src/test/resources/json/mail_min.json | 1 +
21 files changed, 191 insertions(+), 108 deletions(-)
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 ee59815..95f13ee 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
@@ -25,6 +25,7 @@ import java.io.IOException;
import java.util.function.Consumer;
import java.util.function.Function;
+import org.apache.commons.lang3.tuple.Pair;
import org.apache.james.metrics.api.Metric;
import org.apache.james.metrics.api.MetricFactory;
import org.apache.james.queue.api.MailQueue;
@@ -46,10 +47,12 @@ class Dequeuer {
private static class RabbitMQMailQueueItem implements MailQueue.MailQueueItem {
private final Consumer<Boolean> ack;
+ private final EnQueueId enQueueId;
private final Mail mail;
- private RabbitMQMailQueueItem(Consumer<Boolean> ack, Mail mail) {
+ private RabbitMQMailQueueItem(Consumer<Boolean> ack, EnQueueId enQueueId, Mail mail) {
this.ack = ack;
+ this.enQueueId = enQueueId;
this.mail = mail;
}
@@ -58,6 +61,10 @@ class Dequeuer {
return mail;
}
+ public EnQueueId getEnQueueId() {
+ return enQueueId;
+ }
+
@Override
public void done(boolean success) {
ack.accept(success);
@@ -87,7 +94,7 @@ class Dequeuer {
}
private Mono<RabbitMQMailQueueItem> filterIfDeleted(RabbitMQMailQueueItem item) {
- return mailQueueView.isPresent(item.getMail())
+ return mailQueueView.isPresent(item.getEnQueueId())
.flatMap(isPresent -> {
if (isPresent) {
return Mono.just(item);
@@ -99,9 +106,10 @@ class Dequeuer {
private Mono<RabbitMQMailQueueItem> loadItem(AcknowledgableDelivery response) {
try {
- Mail mail = loadMail(response);
+ Pair<EnQueueId, Mail> idAndMail = loadMail(response);
+ Mail mail = idAndMail.getRight();
ThrowingConsumer<Boolean> ack = ack(response, mail);
- return Mono.just(new RabbitMQMailQueueItem(ack, mail));
+ return Mono.just(new RabbitMQMailQueueItem(ack, idAndMail.getLeft(), mail));
} catch (MailQueue.MailQueueException e) {
return Mono.error(e);
}
@@ -119,9 +127,11 @@ class Dequeuer {
};
}
- private Mail loadMail(Delivery response) throws MailQueue.MailQueueException {
+ private Pair<EnQueueId, Mail> loadMail(Delivery response) throws MailQueue.MailQueueException {
MailReferenceDTO mailDTO = toMailReference(response);
- return mailLoader.apply(mailDTO);
+ EnQueueId enQueueId = mailDTO.retrieveEnqueueId();
+ Mail mail = mailLoader.apply(mailDTO);
+ return Pair.of(enQueueId, mail);
}
private MailReferenceDTO toMailReference(Delivery getResponse) throws MailQueue.MailQueueException {
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/EnqueuedItem.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/EnqueuedItem.java
index 9991123..d693906 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/EnqueuedItem.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/EnqueuedItem.java
@@ -23,7 +23,6 @@ import java.time.Instant;
import java.util.Objects;
import org.apache.james.blob.mail.MimeMessagePartsId;
-import org.apache.james.queue.rabbitmq.view.cassandra.model.MailKey;
import org.apache.mailet.Mail;
import com.google.common.base.Preconditions;
@@ -33,6 +32,11 @@ public class EnqueuedItem {
interface Builder {
@FunctionalInterface
+ interface RequireEnqueueId {
+ RequireMailQueueName enQueueId(EnQueueId id);
+ }
+
+ @FunctionalInterface
interface RequireMailQueueName {
RequireMail mailQueueName(MailQueueName mailQueueName);
}
@@ -53,16 +57,20 @@ public class EnqueuedItem {
}
class ReadyToBuild {
+ private final EnQueueId enQueueId;
private final MailQueueName mailQueueName;
private final Mail mail;
private final Instant enqueuedTime;
private final MimeMessagePartsId partsId;
- ReadyToBuild(MailQueueName mailQueueName, Mail mail, Instant enqueuedTime, MimeMessagePartsId partsId) {
+ ReadyToBuild(EnQueueId enQueueId, MailQueueName mailQueueName, Mail mail, Instant enqueuedTime, MimeMessagePartsId partsId) {
+ Preconditions.checkNotNull(enQueueId, "'enQueueId' is mandatory");
Preconditions.checkNotNull(mailQueueName, "'mailQueueName' is mandatory");
Preconditions.checkNotNull(mail, "'mail' is mandatory");
Preconditions.checkNotNull(enqueuedTime, "'enqueuedTime' is mandatory");
Preconditions.checkNotNull(partsId, "'partsId' is mandatory");
+
+ this.enQueueId = enQueueId;
this.mailQueueName = mailQueueName;
this.mail = mail;
this.enqueuedTime = enqueuedTime;
@@ -70,28 +78,31 @@ public class EnqueuedItem {
}
public EnqueuedItem build() {
- return new EnqueuedItem(mailQueueName, mail, enqueuedTime, partsId);
+ return new EnqueuedItem(enQueueId, mailQueueName, mail, enqueuedTime, partsId);
}
}
}
- public static Builder.RequireMailQueueName builder() {
- return queueName -> mail -> enqueuedTime -> partsId -> new Builder.ReadyToBuild(queueName, mail, enqueuedTime, partsId);
+ public static Builder.RequireEnqueueId builder() {
+ return enQueueId -> queueName -> mail -> enqueuedTime -> partsId -> new Builder.ReadyToBuild(enQueueId, queueName, mail, enqueuedTime, partsId);
}
+ private final EnQueueId enQueueId;
private final MailQueueName mailQueueName;
private final Mail mail;
- private final MailKey mailKey;
private final Instant enqueuedTime;
private final MimeMessagePartsId partsId;
- EnqueuedItem(MailQueueName mailQueueName, Mail mail, Instant enqueuedTime, MimeMessagePartsId partsId) {
+ EnqueuedItem(EnQueueId enQueueId, MailQueueName mailQueueName, Mail mail, Instant enqueuedTime, MimeMessagePartsId partsId) {
+ this.enQueueId = enQueueId;
this.mailQueueName = mailQueueName;
this.mail = mail;
this.enqueuedTime = enqueuedTime;
this.partsId = partsId;
+ }
- this.mailKey = MailKey.of(mail.getName());
+ public EnQueueId getEnQueueId() {
+ return enQueueId;
}
public MailQueueName getMailQueueName() {
@@ -110,18 +121,14 @@ public class EnqueuedItem {
return partsId;
}
- public MailKey getMailKey() {
- return mailKey;
- }
-
@Override
public final boolean equals(Object o) {
if (o instanceof EnqueuedItem) {
EnqueuedItem that = (EnqueuedItem) o;
- return Objects.equals(this.mailQueueName, that.mailQueueName)
+ return Objects.equals(this.enQueueId, that.enQueueId)
+ && Objects.equals(this.mailQueueName, that.mailQueueName)
&& Objects.equals(this.mail, that.mail)
- && Objects.equals(this.mailKey, that.mailKey)
&& Objects.equals(this.enqueuedTime, that.enqueuedTime)
&& Objects.equals(this.partsId, that.partsId);
}
@@ -130,6 +137,6 @@ public class EnqueuedItem {
@Override
public final int hashCode() {
- return Objects.hash(mailQueueName, mail, mailKey, enqueuedTime, partsId);
+ return Objects.hash(enQueueId, mailQueueName, mail, enqueuedTime, partsId);
}
}
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/Enqueuer.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/Enqueuer.java
index 7ca896d..9081acc 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/Enqueuer.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/Enqueuer.java
@@ -36,6 +36,7 @@ import org.apache.mailet.Mail;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.github.fge.lambdas.Throwing;
+
import reactor.core.publisher.Mono;
class Enqueuer {
@@ -60,8 +61,9 @@ class Enqueuer {
}
void enQueue(Mail mail) throws MailQueue.MailQueueException {
+ EnQueueId enQueueId = EnQueueId.generate();
saveMail(mail)
- .map(Throwing.<MimeMessagePartsId, EnqueuedItem>function(partsId -> publishReferenceToRabbit(mail, partsId)).sneakyThrow())
+ .map(Throwing.<MimeMessagePartsId, EnqueuedItem>function(partsId -> publishReferenceToRabbit(enQueueId, mail, partsId)).sneakyThrow())
.flatMap(mailQueueView::storeMail)
.thenEmpty(Mono.fromRunnable(enqueueMetric::increment))
.block();
@@ -75,10 +77,11 @@ class Enqueuer {
}
}
- private EnqueuedItem publishReferenceToRabbit(Mail mail, MimeMessagePartsId partsId) throws MailQueue.MailQueueException {
- rabbitClient.publish(name, getMailReferenceBytes(mail, partsId));
+ private EnqueuedItem publishReferenceToRabbit(EnQueueId enQueueId, Mail mail, MimeMessagePartsId partsId) throws MailQueue.MailQueueException {
+ rabbitClient.publish(name, getMailReferenceBytes(enQueueId, mail, partsId));
return EnqueuedItem.builder()
+ .enQueueId(enQueueId)
.mailQueueName(name)
.mail(mail)
.enqueuedTime(clock.instant())
@@ -86,9 +89,9 @@ class Enqueuer {
.build();
}
- private byte[] getMailReferenceBytes(Mail mail, MimeMessagePartsId partsId) throws MailQueue.MailQueueException {
+ private byte[] getMailReferenceBytes(EnQueueId enQueueId, Mail mail, MimeMessagePartsId partsId) throws MailQueue.MailQueueException {
try {
- MailReferenceDTO mailDTO = MailReferenceDTO.fromMail(mail, partsId);
+ MailReferenceDTO mailDTO = MailReferenceDTO.fromMail(enQueueId, mail, partsId);
return mailReferenceSerializer.write(mailDTO);
} catch (JsonProcessingException e) {
throw new MailQueue.MailQueueException("Unable to serialize message", e);
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailReferenceDTO.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailReferenceDTO.java
index a80bee1..f119325 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailReferenceDTO.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailReferenceDTO.java
@@ -52,8 +52,9 @@ import com.google.common.collect.ImmutableMap;
class MailReferenceDTO {
- static MailReferenceDTO fromMail(Mail mail, MimeMessagePartsId partsId) {
+ static MailReferenceDTO fromMail(EnQueueId enQueueId, Mail mail, MimeMessagePartsId partsId) {
return new MailReferenceDTO(
+ enQueueId.serialize(),
Optional.ofNullable(mail.getRecipients()).map(Collection::stream)
.orElse(Stream.empty())
.map(MailAddress::asString)
@@ -89,6 +90,7 @@ class MailReferenceDTO {
.collect(Guavate.toImmutableMap(name, value));
}
+ private final String enQueueId;
private final ImmutableList<String> recipients;
private final String name;
private final Optional<String> sender;
@@ -103,7 +105,8 @@ class MailReferenceDTO {
private final String bodyBlobId;
@JsonCreator
- private MailReferenceDTO(@JsonProperty("recipients") ImmutableList<String> recipients,
+ private MailReferenceDTO(@JsonProperty("enQueueId") String enQueueId,
+ @JsonProperty("recipients") ImmutableList<String> recipients,
@JsonProperty("name") String name,
@JsonProperty("sender") Optional<String> sender,
@JsonProperty("state") String state,
@@ -112,9 +115,10 @@ class MailReferenceDTO {
@JsonProperty("attributes") ImmutableMap<String, String> attributes,
@JsonProperty("remoteAddr") String remoteAddr,
@JsonProperty("remoteHost") String remoteHost,
- @JsonProperty("perRecipientHeaders") Map<String, HeadersDto> perRecipientHeaders,
+ @JsonProperty("perRecipientHeaders") Map<String, HeadersDto> perRecipientHeaders,
@JsonProperty("headerBlobId") String headerBlobId,
@JsonProperty("bodyBlobId") String bodyBlobId) {
+ this.enQueueId = enQueueId;
this.recipients = recipients;
this.name = name;
this.sender = sender;
@@ -129,6 +133,11 @@ class MailReferenceDTO {
this.bodyBlobId = bodyBlobId;
}
+ @JsonProperty("enQueueId")
+ public String getEnQueueId() {
+ return enQueueId;
+ }
+
@JsonProperty("recipients")
Collection<String> getRecipients() {
return recipients;
@@ -189,6 +198,10 @@ class MailReferenceDTO {
return bodyBlobId;
}
+ EnQueueId retrieveEnqueueId() {
+ return EnQueueId.ofSerialized(enQueueId);
+ }
+
MailImpl toMailWithMimeMessage(MimeMessage mimeMessage) throws MessagingException {
MailImpl.Builder builder = MailImpl.builder()
.name(name)
@@ -233,7 +246,8 @@ class MailReferenceDTO {
if (o instanceof MailReferenceDTO) {
MailReferenceDTO mailDTO = (MailReferenceDTO) o;
- return Objects.equals(this.recipients, mailDTO.recipients)
+ return Objects.equals(this.enQueueId, mailDTO.enQueueId)
+ && Objects.equals(this.recipients, mailDTO.recipients)
&& Objects.equals(this.name, mailDTO.name)
&& Objects.equals(this.sender, mailDTO.sender)
&& Objects.equals(this.state, mailDTO.state)
@@ -251,6 +265,6 @@ class MailReferenceDTO {
@Override
public final int hashCode() {
- return Objects.hash(recipients, name, sender, state, errorMessage, lastUpdated, attributes, remoteAddr, remoteHost, perRecipientHeaders, headerBlobId, bodyBlobId);
+ return Objects.hash(enQueueId, recipients, name, sender, state, errorMessage, lastUpdated, attributes, remoteAddr, remoteHost, perRecipientHeaders, headerBlobId, bodyBlobId);
}
}
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/api/DeleteCondition.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/api/DeleteCondition.java
index 96060dd..4b640fa 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/api/DeleteCondition.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/api/DeleteCondition.java
@@ -23,6 +23,7 @@ import java.util.Objects;
import org.apache.commons.lang3.NotImplementedException;
import org.apache.james.queue.api.ManageableMailQueue;
+import org.apache.james.queue.rabbitmq.EnQueueId;
import org.apache.mailet.Mail;
import com.google.common.base.Preconditions;
@@ -86,10 +87,6 @@ public interface DeleteCondition {
this.name = name;
}
- public String getName() {
- return name;
- }
-
@Override
public boolean shouldBeDeleted(Mail mail) {
Preconditions.checkNotNull(mail);
@@ -112,6 +109,24 @@ public interface DeleteCondition {
}
}
+ class WithEnqueueId implements DeleteCondition {
+ private final EnQueueId enQueueId;
+
+
+ public WithEnqueueId(EnQueueId enQueueId) {
+ this.enQueueId = enQueueId;
+ }
+
+ public EnQueueId getEnQueueId() {
+ return enQueueId;
+ }
+
+ @Override
+ public boolean shouldBeDeleted(Mail mail) {
+ throw new NotImplementedException("EnQueueId is not carried as a Mail property");
+ }
+ }
+
class WithRecipient implements DeleteCondition {
private final String recipientAsString;
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/api/MailQueueView.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/api/MailQueueView.java
index 93c79e4..a641484 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/api/MailQueueView.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/api/MailQueueView.java
@@ -20,9 +20,9 @@
package org.apache.james.queue.rabbitmq.view.api;
import org.apache.james.queue.api.ManageableMailQueue;
+import org.apache.james.queue.rabbitmq.EnQueueId;
import org.apache.james.queue.rabbitmq.EnqueuedItem;
import org.apache.james.queue.rabbitmq.MailQueueName;
-import org.apache.mailet.Mail;
import reactor.core.publisher.Mono;
@@ -38,7 +38,7 @@ public interface MailQueueView {
long delete(DeleteCondition deleteCondition);
- Mono<Boolean> isPresent(Mail mail);
+ Mono<Boolean> isPresent(EnQueueId id);
ManageableMailQueue.MailQueueIterator browse();
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueBrowser.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueBrowser.java
index 9054e1b..baf19a3 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueBrowser.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueBrowser.java
@@ -44,6 +44,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.base.Preconditions;
+
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
@@ -138,7 +139,7 @@ public class CassandraMailQueueBrowser {
private Flux<EnqueuedItemWithSlicingContext> browseBucket(MailQueueName queueName, Slice slice, BucketId bucketId) {
return enqueuedMailsDao.selectEnqueuedMails(queueName, slice, bucketId)
- .filterWhen(mailReference -> deletedMailsDao.isStillEnqueued(queueName, mailReference.getEnqueuedItem().getMailKey()));
+ .filterWhen(mailReference -> deletedMailsDao.isStillEnqueued(queueName, mailReference.getEnqueuedItem().getEnQueueId()));
}
private Flux<Slice> allSlicesStartingAt(Instant browseStart) {
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueMailDelete.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueMailDelete.java
index 168cc6c..55d563a 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueMailDelete.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueMailDelete.java
@@ -24,10 +24,9 @@ import java.util.concurrent.ThreadLocalRandom;
import javax.inject.Inject;
+import org.apache.james.queue.rabbitmq.EnQueueId;
import org.apache.james.queue.rabbitmq.MailQueueName;
import org.apache.james.queue.rabbitmq.view.cassandra.configuration.CassandraMailQueueViewConfiguration;
-import org.apache.james.queue.rabbitmq.view.cassandra.model.MailKey;
-import org.apache.mailet.Mail;
import reactor.core.publisher.Mono;
@@ -49,18 +48,14 @@ public class CassandraMailQueueMailDelete {
this.configuration = configuration;
}
- Mono<Void> considerDeleted(Mail mail, MailQueueName mailQueueName) {
- return considerDeleted(MailKey.fromMail(mail), mailQueueName);
- }
-
- Mono<Void> considerDeleted(MailKey mailKey, MailQueueName mailQueueName) {
+ Mono<Void> considerDeleted(EnQueueId enQueueId, MailQueueName mailQueueName) {
return deletedMailsDao
- .markAsDeleted(mailQueueName, mailKey)
+ .markAsDeleted(mailQueueName, enQueueId)
.doOnNext(ignored -> maybeUpdateBrowseStart(mailQueueName));
}
- Mono<Boolean> isDeleted(Mail mail, MailQueueName mailQueueName) {
- return deletedMailsDao.isDeleted(mailQueueName, MailKey.fromMail(mail));
+ Mono<Boolean> isDeleted(EnQueueId enQueueId, MailQueueName mailQueueName) {
+ return deletedMailsDao.isDeleted(mailQueueName, enQueueId);
}
void updateBrowseStart(MailQueueName mailQueueName) {
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueView.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueView.java
index 85ae197..15fd0fc 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueView.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueView.java
@@ -22,6 +22,7 @@ package org.apache.james.queue.rabbitmq.view.cassandra;
import javax.inject.Inject;
import org.apache.james.queue.api.ManageableMailQueue;
+import org.apache.james.queue.rabbitmq.EnQueueId;
import org.apache.james.queue.rabbitmq.EnqueuedItem;
import org.apache.james.queue.rabbitmq.MailQueueName;
import org.apache.james.queue.rabbitmq.view.api.DeleteCondition;
@@ -29,8 +30,6 @@ import org.apache.james.queue.rabbitmq.view.api.MailQueueView;
import org.apache.james.queue.rabbitmq.view.cassandra.configuration.CassandraMailQueueViewConfiguration;
import org.apache.james.queue.rabbitmq.view.cassandra.configuration.EventsourcingConfigurationManagement;
import org.apache.james.queue.rabbitmq.view.cassandra.model.EnqueuedItemWithSlicingContext;
-import org.apache.james.queue.rabbitmq.view.cassandra.model.MailKey;
-import org.apache.mailet.Mail;
import reactor.core.publisher.Mono;
@@ -101,9 +100,9 @@ public class CassandraMailQueueView implements MailQueueView {
@Override
public long delete(DeleteCondition deleteCondition) {
- if (deleteCondition instanceof DeleteCondition.WithName) {
- DeleteCondition.WithName nameDeleteCondition = (DeleteCondition.WithName) deleteCondition;
- delete(MailKey.of(nameDeleteCondition.getName())).block();
+ if (deleteCondition instanceof DeleteCondition.WithEnqueueId) {
+ DeleteCondition.WithEnqueueId enQueueIdCondition = (DeleteCondition.WithEnqueueId) deleteCondition;
+ delete(enQueueIdCondition.getEnQueueId()).block();
return 1L;
}
return browseThenDelete(deleteCondition);
@@ -113,19 +112,19 @@ public class CassandraMailQueueView implements MailQueueView {
return cassandraMailQueueBrowser.browseReferences(mailQueueName)
.map(EnqueuedItemWithSlicingContext::getEnqueuedItem)
.filter(mailReference -> deleteCondition.shouldBeDeleted(mailReference.getMail()))
- .flatMap(mailReference -> cassandraMailQueueMailDelete.considerDeleted(mailReference.getMail(), mailQueueName))
+ .flatMap(mailReference -> cassandraMailQueueMailDelete.considerDeleted(mailReference.getEnQueueId(), mailQueueName))
.count()
.doOnNext(ignored -> cassandraMailQueueMailDelete.updateBrowseStart(mailQueueName))
.block();
}
- private Mono<Void> delete(MailKey mailKey) {
- return cassandraMailQueueMailDelete.considerDeleted(mailKey, mailQueueName);
+ private Mono<Void> delete(EnQueueId enQueueId) {
+ return cassandraMailQueueMailDelete.considerDeleted(enQueueId, mailQueueName);
}
@Override
- public Mono<Boolean> isPresent(Mail mail) {
- return cassandraMailQueueMailDelete.isDeleted(mail, mailQueueName)
+ public Mono<Boolean> isPresent(EnQueueId id) {
+ return cassandraMailQueueMailDelete.isDeleted(id, mailQueueName)
.map(bool -> !bool);
}
}
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueViewModule.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueViewModule.java
index 9068f61..a160b9b 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueViewModule.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueViewModule.java
@@ -25,6 +25,7 @@ import static com.datastax.driver.core.DataType.list;
import static com.datastax.driver.core.DataType.map;
import static com.datastax.driver.core.DataType.text;
import static com.datastax.driver.core.DataType.timestamp;
+import static com.datastax.driver.core.DataType.uuid;
import static com.datastax.driver.core.schemabuilder.SchemaBuilder.frozen;
import org.apache.james.backends.cassandra.components.CassandraModule;
@@ -32,14 +33,15 @@ import org.apache.james.backends.cassandra.components.CassandraModule;
public interface CassandraMailQueueViewModule {
interface EnqueuedMailsTable {
- String TABLE_NAME = "enqueuedMails";
+ String TABLE_NAME = "enqueuedMailsV2";
String QUEUE_NAME = "queueName";
String TIME_RANGE_START = "timeRangeStart";
String BUCKET_ID = "bucketId";
String ENQUEUED_TIME = "enqueuedTime";
- String MAIL_KEY = "mailKey";
+ String ENQUEUE_ID = "enqueueId";
+ String NAME = "name";
String HEADER_BLOB_ID = "headerBlobId";
String BODY_BLOB_ID = "bodyBlobId";
String STATE = "state";
@@ -65,10 +67,10 @@ public interface CassandraMailQueueViewModule {
}
interface DeletedMailTable {
- String TABLE_NAME = "deletedMails";
+ String TABLE_NAME = "deletedMailsV2";
String QUEUE_NAME = "queueName";
- String MAIL_KEY = "mailKey";
+ String ENQUEUE_ID = "enqueueId";
}
CassandraModule MODULE = CassandraModule
@@ -86,8 +88,9 @@ public interface CassandraMailQueueViewModule {
.addPartitionKey(EnqueuedMailsTable.QUEUE_NAME, text())
.addPartitionKey(EnqueuedMailsTable.TIME_RANGE_START, timestamp())
.addPartitionKey(EnqueuedMailsTable.BUCKET_ID, cint())
- .addClusteringColumn(EnqueuedMailsTable.MAIL_KEY, text())
+ .addClusteringColumn(EnqueuedMailsTable.ENQUEUE_ID, uuid())
.addColumn(EnqueuedMailsTable.ENQUEUED_TIME, timestamp())
+ .addColumn(EnqueuedMailsTable.NAME, text())
.addColumn(EnqueuedMailsTable.STATE, text())
.addColumn(EnqueuedMailsTable.HEADER_BLOB_ID, text())
.addColumn(EnqueuedMailsTable.BODY_BLOB_ID, text())
@@ -115,7 +118,7 @@ public interface CassandraMailQueueViewModule {
.options(options -> options)
.statement(statement -> statement
.addPartitionKey(DeletedMailTable.QUEUE_NAME, text())
- .addPartitionKey(DeletedMailTable.MAIL_KEY, text()))
+ .addPartitionKey(DeletedMailTable.ENQUEUE_ID, uuid()))
.build();
}
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/DeletedMailsDAO.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/DeletedMailsDAO.java
index eb64990..7f44c6f 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/DeletedMailsDAO.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/DeletedMailsDAO.java
@@ -23,18 +23,19 @@ import static com.datastax.driver.core.querybuilder.QueryBuilder.bindMarker;
import static com.datastax.driver.core.querybuilder.QueryBuilder.eq;
import static com.datastax.driver.core.querybuilder.QueryBuilder.insertInto;
import static com.datastax.driver.core.querybuilder.QueryBuilder.select;
-import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.DeletedMailTable.MAIL_KEY;
+import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.DeletedMailTable.ENQUEUE_ID;
import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.DeletedMailTable.QUEUE_NAME;
import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.DeletedMailTable.TABLE_NAME;
import javax.inject.Inject;
import org.apache.james.backends.cassandra.utils.CassandraAsyncExecutor;
+import org.apache.james.queue.rabbitmq.EnQueueId;
import org.apache.james.queue.rabbitmq.MailQueueName;
-import org.apache.james.queue.rabbitmq.view.cassandra.model.MailKey;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.Session;
+
import reactor.core.publisher.Mono;
public class DeletedMailsDAO {
@@ -53,31 +54,31 @@ public class DeletedMailsDAO {
private PreparedStatement prepareInsert(Session session) {
return session.prepare(insertInto(TABLE_NAME)
.value(QUEUE_NAME, bindMarker(QUEUE_NAME))
- .value(MAIL_KEY, bindMarker(MAIL_KEY)));
+ .value(ENQUEUE_ID, bindMarker(ENQUEUE_ID)));
}
private PreparedStatement prepareSelectExist(Session session) {
return session.prepare(select()
.from(TABLE_NAME)
.where(eq(QUEUE_NAME, bindMarker(QUEUE_NAME)))
- .and(eq(MAIL_KEY, bindMarker(MAIL_KEY))));
+ .and(eq(ENQUEUE_ID, bindMarker(ENQUEUE_ID))));
}
- Mono<Void> markAsDeleted(MailQueueName mailQueueName, MailKey mailKey) {
+ Mono<Void> markAsDeleted(MailQueueName mailQueueName, EnQueueId enQueueId) {
return executor.executeVoid(insertOne.bind()
.setString(QUEUE_NAME, mailQueueName.asString())
- .setString(MAIL_KEY, mailKey.getMailKey()));
+ .setUUID(ENQUEUE_ID, enQueueId.asUUID()));
}
- Mono<Boolean> isDeleted(MailQueueName mailQueueName, MailKey mailKey) {
+ Mono<Boolean> isDeleted(MailQueueName mailQueueName, EnQueueId enQueueId) {
return executor.executeReturnExists(
selectOne.bind()
.setString(QUEUE_NAME, mailQueueName.asString())
- .setString(MAIL_KEY, mailKey.getMailKey()));
+ .setUUID(ENQUEUE_ID, enQueueId.asUUID()));
}
- Mono<Boolean> isStillEnqueued(MailQueueName mailQueueName, MailKey mailKey) {
- return isDeleted(mailQueueName, mailKey)
+ Mono<Boolean> isStillEnqueued(MailQueueName mailQueueName, EnQueueId enQueueId) {
+ return isDeleted(mailQueueName, enQueueId)
.map(b -> !b);
}
}
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/EnqueuedMailsDAO.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/EnqueuedMailsDAO.java
index dde1aa6..1c8627e 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/EnqueuedMailsDAO.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/EnqueuedMailsDAO.java
@@ -27,10 +27,11 @@ import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueV
import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.BODY_BLOB_ID;
import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.BUCKET_ID;
import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.ENQUEUED_TIME;
+import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.ENQUEUE_ID;
import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.ERROR_MESSAGE;
import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.HEADER_BLOB_ID;
import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.LAST_UPDATED;
-import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.MAIL_KEY;
+import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.NAME;
import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.PER_RECIPIENT_SPECIFIC_HEADERS;
import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.QUEUE_NAME;
import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.RECIPIENTS;
@@ -97,7 +98,8 @@ public class EnqueuedMailsDAO {
.value(QUEUE_NAME, bindMarker(QUEUE_NAME))
.value(TIME_RANGE_START, bindMarker(TIME_RANGE_START))
.value(BUCKET_ID, bindMarker(BUCKET_ID))
- .value(MAIL_KEY, bindMarker(MAIL_KEY))
+ .value(ENQUEUE_ID, bindMarker(ENQUEUE_ID))
+ .value(NAME, bindMarker(NAME))
.value(HEADER_BLOB_ID, bindMarker(HEADER_BLOB_ID))
.value(BODY_BLOB_ID, bindMarker(BODY_BLOB_ID))
.value(ENQUEUED_TIME, bindMarker(ENQUEUED_TIME))
@@ -123,7 +125,8 @@ public class EnqueuedMailsDAO {
.setTimestamp(TIME_RANGE_START, Date.from(slicingContext.getTimeRangeStart()))
.setInt(BUCKET_ID, slicingContext.getBucketId().getValue())
.setTimestamp(ENQUEUED_TIME, Date.from(enqueuedItem.getEnqueuedTime()))
- .setString(MAIL_KEY, mail.getName())
+ .setUUID(ENQUEUE_ID, enqueuedItem.getEnQueueId().asUUID())
+ .setString(NAME, mail.getName())
.setString(HEADER_BLOB_ID, mimeMessagePartsId.getHeaderBlobId().asString())
.setString(BODY_BLOB_ID, mimeMessagePartsId.getBodyBlobId().asString())
.setString(STATE, mail.getState())
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/EnqueuedMailsDaoUtil.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/EnqueuedMailsDaoUtil.java
index a4e7eaa..004e9af 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/EnqueuedMailsDaoUtil.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/EnqueuedMailsDaoUtil.java
@@ -23,13 +23,14 @@ import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueV
import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.BODY_BLOB_ID;
import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.BUCKET_ID;
import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.ENQUEUED_TIME;
+import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.ENQUEUE_ID;
import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.ERROR_MESSAGE;
import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.HEADER_BLOB_ID;
import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.HEADER_NAME;
import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.HEADER_TYPE;
import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.HEADER_VALUE;
import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.LAST_UPDATED;
-import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.MAIL_KEY;
+import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.NAME;
import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.PER_RECIPIENT_SPECIFIC_HEADERS;
import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.QUEUE_NAME;
import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.RECIPIENTS;
@@ -57,6 +58,7 @@ import org.apache.james.backends.cassandra.init.CassandraTypesProvider;
import org.apache.james.blob.api.BlobId;
import org.apache.james.blob.mail.MimeMessagePartsId;
import org.apache.james.core.MailAddress;
+import org.apache.james.queue.rabbitmq.EnQueueId;
import org.apache.james.queue.rabbitmq.EnqueuedItem;
import org.apache.james.queue.rabbitmq.MailQueueName;
import org.apache.james.queue.rabbitmq.view.cassandra.model.BucketedSlices;
@@ -79,6 +81,7 @@ public class EnqueuedMailsDaoUtil {
static EnqueuedItemWithSlicingContext toEnqueuedMail(Row row, BlobId.Factory blobFactory) {
MailQueueName queueName = MailQueueName.fromString(row.getString(QUEUE_NAME));
+ EnQueueId enQueueId = EnQueueId.of(row.getUUID(ENQUEUE_ID));
Instant timeRangeStart = row.getTimestamp(TIME_RANGE_START).toInstant();
BucketedSlices.BucketId bucketId = BucketedSlices.BucketId.of(row.getInt(BUCKET_ID));
Instant enqueuedTime = row.getTimestamp(ENQUEUED_TIME).toInstant();
@@ -101,7 +104,7 @@ public class EnqueuedMailsDaoUtil {
String remoteAddr = row.getString(REMOTE_ADDR);
String remoteHost = row.getString(REMOTE_HOST);
String errorMessage = row.getString(ERROR_MESSAGE);
- String name = row.getString(MAIL_KEY);
+ String name = row.getString(NAME);
Date lastUpdated = row.getTimestamp(LAST_UPDATED);
Map<String, ByteBuffer> rawAttributes = row.getMap(ATTRIBUTES, String.class, ByteBuffer.class);
PerRecipientHeaders perRecipientHeaders = fromHeaderMap(row.getMap(PER_RECIPIENT_SPECIFIC_HEADERS, String.class, UDTValue.class));
@@ -119,6 +122,7 @@ public class EnqueuedMailsDaoUtil {
.addAttributes(toAttributes(rawAttributes))
.build();
EnqueuedItem enqueuedItem = EnqueuedItem.builder()
+ .enQueueId(enQueueId)
.mailQueueName(queueName)
.mail(mail)
.enqueuedTime(enqueuedTime)
diff --git a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/EnqueuedItemTest.java b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/EnqueuedItemTest.java
index a3a205a..73c8302 100644
--- a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/EnqueuedItemTest.java
+++ b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/EnqueuedItemTest.java
@@ -22,6 +22,7 @@ package org.apache.james.queue.rabbitmq;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import java.time.Instant;
+
import javax.mail.MessagingException;
import org.apache.james.blob.api.HashBlobId;
@@ -56,8 +57,21 @@ class EnqueuedItemTest {
}
@Test
+ void buildShouldThrowWhenEnqueueIdIsNull() {
+ assertThatThrownBy(() -> EnqueuedItem.builder()
+ .enQueueId(null)
+ .mailQueueName(mailQueueName)
+ .mail(mail)
+ .enqueuedTime(enqueuedTime)
+ .mimeMessagePartsId(partsId)
+ .build())
+ .isInstanceOf(NullPointerException.class);
+ }
+
+ @Test
void buildShouldThrowWhenMailQueueNameIsNull() {
assertThatThrownBy(() -> EnqueuedItem.builder()
+ .enQueueId(EnQueueId.generate())
.mailQueueName(null)
.mail(mail)
.enqueuedTime(enqueuedTime)
@@ -69,6 +83,7 @@ class EnqueuedItemTest {
@Test
void buildShouldThrowWhenMailIsNull() {
assertThatThrownBy(() -> EnqueuedItem.builder()
+ .enQueueId(EnQueueId.generate())
.mailQueueName(mailQueueName)
.mail(null)
.enqueuedTime(enqueuedTime)
@@ -80,6 +95,7 @@ class EnqueuedItemTest {
@Test
void buildShouldThrowWhenEnqueuedTimeIsNull() {
assertThatThrownBy(() -> EnqueuedItem.builder()
+ .enQueueId(EnQueueId.generate())
.mailQueueName(mailQueueName)
.mail(mail)
.enqueuedTime(null)
@@ -91,6 +107,7 @@ class EnqueuedItemTest {
@Test
void buildShouldThrowWhenMimeMessagePartsIdIsNull() {
assertThatThrownBy(() -> EnqueuedItem.builder()
+ .enQueueId(EnQueueId.generate())
.mailQueueName(mailQueueName)
.mail(mail)
.enqueuedTime(enqueuedTime)
diff --git a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/MailDTOTest.java b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/MailDTOTest.java
index 712bcc5..2c98b1d 100644
--- a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/MailDTOTest.java
+++ b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/MailDTOTest.java
@@ -46,6 +46,7 @@ import com.fasterxml.jackson.datatype.jdk8.Jdk8Module;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
class MailDTOTest {
+ static final EnQueueId EN_QUEUE_ID = EnQueueId.ofSerialized("110e8400-e29b-11d4-a716-446655440000");
static final HashBlobId.Factory BLOB_ID_FACTORY = new HashBlobId.Factory();
static final Date LAST_UPDATED = Date.from(Instant.parse("2016-09-08T14:25:52.000Z"));
@@ -85,6 +86,7 @@ class MailDTOTest {
private MailReferenceDTO mailDTO1() throws MessagingException {
return MailReferenceDTO.fromMail(
+ EN_QUEUE_ID,
FakeMail.builder()
.name("mail-name-558")
.recipients(MailAddressFixture.RECIPIENT1, MailAddressFixture.RECIPIENT2)
@@ -117,6 +119,7 @@ class MailDTOTest {
mail.setState(null);
mail.setLastUpdated(null);
return MailReferenceDTO.fromMail(
+ EN_QUEUE_ID,
mail,
MimeMessagePartsId.builder()
.headerBlobId(BLOB_ID_FACTORY.from("210e7136-ede3-44eb-9495-3ed816d6e23b"))
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 03455ff..f6d57c6 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
@@ -231,13 +231,6 @@ public class RabbitMQMailQueueTest implements ManageableMailQueueContract, MailQ
}
- @Test
- @Override
- @Disabled("JAMES-2794 This test never finishes")
- public void enQueueShouldAcceptMailWithDuplicatedNames() {
-
- }
-
private void enqueueSomeMails(Function<Integer, String> namePattern, int emailCount) {
IntStream.rangeClosed(1, emailCount)
.forEach(Throwing.intConsumer(i -> enQueue(defaultMail()
diff --git a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/DeletedMailsDAOTest.java b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/DeletedMailsDAOTest.java
index e21804f..73f0f27 100644
--- a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/DeletedMailsDAOTest.java
+++ b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/DeletedMailsDAOTest.java
@@ -23,8 +23,10 @@ import static org.assertj.core.api.Assertions.assertThat;
import org.apache.james.backends.cassandra.CassandraCluster;
import org.apache.james.backends.cassandra.CassandraClusterExtension;
+import org.apache.james.backends.cassandra.components.CassandraModule;
+import org.apache.james.backends.cassandra.versions.CassandraSchemaVersionModule;
+import org.apache.james.queue.rabbitmq.EnQueueId;
import org.apache.james.queue.rabbitmq.MailQueueName;
-import org.apache.james.queue.rabbitmq.view.cassandra.model.MailKey;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
@@ -33,11 +35,14 @@ class DeletedMailsDAOTest {
private static final MailQueueName OUT_GOING_1 = MailQueueName.fromString("OUT_GOING_1");
private static final MailQueueName OUT_GOING_2 = MailQueueName.fromString("OUT_GOING_2");
- private static final MailKey MAIL_KEY_1 = MailKey.of("mailkey1");
- private static final MailKey MAIL_KEY_2 = MailKey.of("mailkey2");
+ private static final EnQueueId EN_QUEUE_ID_1 = EnQueueId.ofSerialized("110e8400-e29b-11d4-a716-446655440000");
+ private static final EnQueueId EN_QUEUE_ID_2 = EnQueueId.ofSerialized("464765a0-e4e7-11e4-aba4-710c1de3782b");
@RegisterExtension
- static CassandraClusterExtension cassandraCluster = new CassandraClusterExtension(CassandraMailQueueViewModule.MODULE);
+ static CassandraClusterExtension cassandraCluster = new CassandraClusterExtension(
+ CassandraModule.aggregateModules(
+ CassandraSchemaVersionModule.MODULE,
+ CassandraMailQueueViewModule.MODULE));
private DeletedMailsDAO testee;
@@ -49,14 +54,14 @@ class DeletedMailsDAOTest {
@Test
void markAsDeletedShouldWork() {
Boolean isDeletedBeforeMark = testee
- .isDeleted(OUT_GOING_1, MAIL_KEY_1)
+ .isDeleted(OUT_GOING_1, EN_QUEUE_ID_1)
.block();
assertThat(isDeletedBeforeMark).isFalse();
- testee.markAsDeleted(OUT_GOING_1, MAIL_KEY_1).block();
+ testee.markAsDeleted(OUT_GOING_1, EN_QUEUE_ID_1).block();
Boolean isDeletedAfterMark = testee
- .isDeleted(OUT_GOING_1, MAIL_KEY_1)
+ .isDeleted(OUT_GOING_1, EN_QUEUE_ID_1)
.block();
assertThat(isDeletedAfterMark).isTrue();
@@ -64,10 +69,10 @@ class DeletedMailsDAOTest {
@Test
void checkDeletedShouldReturnFalseWhenTableDoesntContainBothMailQueueAndMailKey() {
- testee.markAsDeleted(OUT_GOING_2, MAIL_KEY_2).block();
+ testee.markAsDeleted(OUT_GOING_2, EN_QUEUE_ID_2).block();
Boolean isDeleted = testee
- .isDeleted(OUT_GOING_1, MAIL_KEY_1)
+ .isDeleted(OUT_GOING_1, EN_QUEUE_ID_1)
.block();
assertThat(isDeleted).isFalse();
@@ -75,10 +80,10 @@ class DeletedMailsDAOTest {
@Test
void checkDeletedShouldReturnFalseWhenTableContainsMailQueueButNotMailKey() {
- testee.markAsDeleted(OUT_GOING_1, MAIL_KEY_2).block();
+ testee.markAsDeleted(OUT_GOING_1, EN_QUEUE_ID_2).block();
Boolean isDeleted = testee
- .isDeleted(OUT_GOING_1, MAIL_KEY_1)
+ .isDeleted(OUT_GOING_1, EN_QUEUE_ID_1)
.block();
assertThat(isDeleted).isFalse();
@@ -86,10 +91,10 @@ class DeletedMailsDAOTest {
@Test
void checkDeletedShouldReturnFalseWhenTableContainsMailKeyButNotMailQueue() {
- testee.markAsDeleted(OUT_GOING_2, MAIL_KEY_1).block();
+ testee.markAsDeleted(OUT_GOING_2, EN_QUEUE_ID_1).block();
Boolean isDeleted = testee
- .isDeleted(OUT_GOING_1, MAIL_KEY_1)
+ .isDeleted(OUT_GOING_1, EN_QUEUE_ID_1)
.block();
assertThat(isDeleted).isFalse();
@@ -97,10 +102,10 @@ class DeletedMailsDAOTest {
@Test
void checkDeletedShouldReturnTrueWhenTableContainsMailItem() {
- testee.markAsDeleted(OUT_GOING_1, MAIL_KEY_1).block();
+ testee.markAsDeleted(OUT_GOING_1, EN_QUEUE_ID_1).block();
Boolean isDeleted = testee
- .isDeleted(OUT_GOING_1, MAIL_KEY_1)
+ .isDeleted(OUT_GOING_1, EN_QUEUE_ID_1)
.block();
assertThat(isDeleted).isTrue();
diff --git a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/EnqueuedMailsDaoTest.java b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/EnqueuedMailsDaoTest.java
index d855d80..7dce6e5 100644
--- a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/EnqueuedMailsDaoTest.java
+++ b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/EnqueuedMailsDaoTest.java
@@ -30,12 +30,12 @@ import org.apache.james.backends.cassandra.CassandraClusterExtension;
import org.apache.james.blob.api.BlobId;
import org.apache.james.blob.api.HashBlobId;
import org.apache.james.blob.mail.MimeMessagePartsId;
+import org.apache.james.queue.rabbitmq.EnQueueId;
import org.apache.james.queue.rabbitmq.EnqueuedItem;
import org.apache.james.queue.rabbitmq.MailQueueName;
import org.apache.james.queue.rabbitmq.view.cassandra.model.BucketedSlices.BucketId;
import org.apache.james.queue.rabbitmq.view.cassandra.model.BucketedSlices.Slice;
import org.apache.james.queue.rabbitmq.view.cassandra.model.EnqueuedItemWithSlicingContext;
-import org.apache.james.queue.rabbitmq.view.cassandra.model.MailKey;
import org.apache.mailet.base.test.FakeMail;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -44,7 +44,8 @@ import org.junit.jupiter.api.extension.RegisterExtension;
class EnqueuedMailsDaoTest {
private static final MailQueueName OUT_GOING_1 = MailQueueName.fromString("OUT_GOING_1");
- private static final MailKey MAIL_KEY_1 = MailKey.of("mailkey1");
+ private static final EnQueueId EN_QUEUE_ID = EnQueueId.ofSerialized("110e8400-e29b-11d4-a716-446655440000");
+ private static final String NAME = "name";
private static int BUCKET_ID_VALUE = 10;
private static final BucketId BUCKET_ID = BucketId.of(BUCKET_ID_VALUE);
private static final Instant NOW = Instant.now();
@@ -75,9 +76,10 @@ class EnqueuedMailsDaoTest {
void insertShouldWork() throws Exception {
testee.insert(EnqueuedItemWithSlicingContext.builder()
.enqueuedItem(EnqueuedItem.builder()
+ .enQueueId(EN_QUEUE_ID)
.mailQueueName(OUT_GOING_1)
.mail(FakeMail.builder()
- .name(MAIL_KEY_1.getMailKey())
+ .name(NAME)
.build())
.enqueuedTime(NOW)
.mimeMessagePartsId(MIME_MESSAGE_PARTS_ID)
@@ -97,9 +99,10 @@ class EnqueuedMailsDaoTest {
void selectEnqueuedMailsShouldWork() throws Exception {
testee.insert(EnqueuedItemWithSlicingContext.builder()
.enqueuedItem(EnqueuedItem.builder()
+ .enQueueId(EN_QUEUE_ID)
.mailQueueName(OUT_GOING_1)
.mail(FakeMail.builder()
- .name(MAIL_KEY_1.getMailKey())
+ .name(NAME)
.build())
.enqueuedTime(NOW)
.mimeMessagePartsId(MIME_MESSAGE_PARTS_ID)
@@ -110,9 +113,10 @@ class EnqueuedMailsDaoTest {
testee.insert(EnqueuedItemWithSlicingContext.builder()
.enqueuedItem(EnqueuedItem.builder()
+ .enQueueId(EN_QUEUE_ID)
.mailQueueName(OUT_GOING_1)
.mail(FakeMail.builder()
- .name(MAIL_KEY_1.getMailKey())
+ .name(NAME)
.build())
.enqueuedTime(NOW)
.mimeMessagePartsId(MIME_MESSAGE_PARTS_ID)
@@ -134,7 +138,8 @@ class EnqueuedMailsDaoTest {
softly.assertThat(slicingContext.getTimeRangeStart()).isEqualTo(NOW);
softly.assertThat(enqueuedItem.getMailQueueName()).isEqualTo(OUT_GOING_1);
softly.assertThat(enqueuedItem.getEnqueuedTime()).isEqualTo(NOW);
- softly.assertThat(enqueuedItem.getMailKey()).isEqualTo(MAIL_KEY_1);
+ softly.assertThat(enqueuedItem.getEnQueueId()).isEqualTo(EN_QUEUE_ID);
+ softly.assertThat(enqueuedItem.getMail().getName()).isEqualTo(NAME);
softly.assertThat(enqueuedItem.getPartsId()).isEqualTo(MIME_MESSAGE_PARTS_ID);
});
});
diff --git a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/model/EnqueuedItemWithSlicingContextTest.java b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/model/EnqueuedItemWithSlicingContextTest.java
index 5eb63a1..f1c2729 100644
--- a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/model/EnqueuedItemWithSlicingContextTest.java
+++ b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/model/EnqueuedItemWithSlicingContextTest.java
@@ -22,10 +22,12 @@ package org.apache.james.queue.rabbitmq.view.cassandra.model;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import java.time.Instant;
+
import javax.mail.MessagingException;
import org.apache.james.blob.api.HashBlobId;
import org.apache.james.blob.mail.MimeMessagePartsId;
+import org.apache.james.queue.rabbitmq.EnQueueId;
import org.apache.james.queue.rabbitmq.EnqueuedItem;
import org.apache.james.queue.rabbitmq.MailQueueName;
import org.apache.mailet.base.test.FakeMail;
@@ -40,7 +42,8 @@ class EnqueuedItemWithSlicingContextTest {
private EnqueuedItemWithSlicingContextTest() throws MessagingException {
enqueuedItem = EnqueuedItem.builder()
- .mailQueueName(MailQueueName.fromString("mailQueueName"))
+ .enQueueId(EnQueueId.generate())
+ .mailQueueName(MailQueueName.fromString("mailQueueName"))
.mail(FakeMail.builder()
.name("name")
.build())
diff --git a/server/queue/queue-rabbitmq/src/test/resources/json/mail1.json b/server/queue/queue-rabbitmq/src/test/resources/json/mail1.json
index b1d6d56..9bfbee0 100644
--- a/server/queue/queue-rabbitmq/src/test/resources/json/mail1.json
+++ b/server/queue/queue-rabbitmq/src/test/resources/json/mail1.json
@@ -1,4 +1,5 @@
{
+ "enQueueId": "110e8400-e29b-11d4-a716-446655440000",
"recipients": [
"recipient1@localhost",
"recipient2@localhost"
diff --git a/server/queue/queue-rabbitmq/src/test/resources/json/mail_min.json b/server/queue/queue-rabbitmq/src/test/resources/json/mail_min.json
index b2b8eea..21dcd7c 100644
--- a/server/queue/queue-rabbitmq/src/test/resources/json/mail_min.json
+++ b/server/queue/queue-rabbitmq/src/test/resources/json/mail_min.json
@@ -1,4 +1,5 @@
{
+ "enQueueId": "110e8400-e29b-11d4-a716-446655440000",
"recipients":[],
"name":"mail-name-558",
"sender":null,
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 11/28: JAMES-2794 Avoid calling *block* in doOnNext
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 9e9e93522991e53bd3980e952d2dd8300e9644bc
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Tue Jun 18 14:24:59 2019 +0700
JAMES-2794 Avoid calling *block* in doOnNext
Reactor is complaining that block was called in a parallel thread around 4-8%
of the time. Switching the subscription scheduler did not modify this behaviour.
However, we do not need to actually wait the browse update to be performed
before returning: this can perfectly be a background process.
This actually solves stability issues of browseAndDequeueShouldCombineWellWhenDifferentSlices
---
.../queue/rabbitmq/view/cassandra/CassandraMailQueueMailDelete.java | 4 +++-
1 file changed, 3 insertions(+), 1 deletion(-)
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueMailDelete.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueMailDelete.java
index 55d563a..b326eb4 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueMailDelete.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueMailDelete.java
@@ -29,6 +29,7 @@ import org.apache.james.queue.rabbitmq.MailQueueName;
import org.apache.james.queue.rabbitmq.view.cassandra.configuration.CassandraMailQueueViewConfiguration;
import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
public class CassandraMailQueueMailDelete {
@@ -61,7 +62,8 @@ public class CassandraMailQueueMailDelete {
void updateBrowseStart(MailQueueName mailQueueName) {
findNewBrowseStart(mailQueueName)
.flatMap(newBrowseStart -> updateNewBrowseStart(mailQueueName, newBrowseStart))
- .block();
+ .subscribeOn(Schedulers.elastic())
+ .subscribe();
}
private void maybeUpdateBrowseStart(MailQueueName mailQueueName) {
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 14/28: JAMES-2794 Create an EnqueueId POJO in
RabbitMQMailQueue
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 462707509bb492d2fdac1da955d4d2b787cd6964
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Wed Jun 19 13:40:27 2019 +0700
JAMES-2794 Create an EnqueueId POJO in RabbitMQMailQueue
---
.../src/main/java/org/apache/james/queue/rabbitmq/EnqueueId.java | 3 +--
1 file changed, 1 insertion(+), 2 deletions(-)
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/EnqueueId.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/EnqueueId.java
index e2057ac..f9bc3b4 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/EnqueueId.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/EnqueueId.java
@@ -22,13 +22,12 @@ package org.apache.james.queue.rabbitmq;
import java.util.Objects;
import java.util.UUID;
-import com.datastax.driver.core.utils.UUIDs;
import com.google.common.base.Preconditions;
public class EnqueueId {
public static EnqueueId generate() {
- return of(UUIDs.timeBased());
+ return of(UUID.randomUUID());
}
public static EnqueueId of(UUID uuid) {
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 22/28: JAMES-2803 Always close ES connections
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit be0233b511a8b7ac02265979c0dbba328775f615
Author: Matthieu Baechler <ma...@apache.org>
AuthorDate: Wed Jun 19 16:25:14 2019 +0200
JAMES-2803 Always close ES connections
---
.../apache/james/backends/es/ClientProvider.java | 68 +++++++-
.../james/backends/es/ClientProviderImpl.java | 54 ------
.../es/ClientProviderImplConnectionTest.java | 6 +-
...oviderImplTest.java => ClientProviderTest.java} | 10 +-
.../james/backends/es/DockerElasticSearch.java | 7 +-
.../james/backends/es/DockerElasticSearchRule.java | 2 +-
.../backends/es/ElasticSearchIndexerTest.java | 90 +++++-----
.../backends/es/IndexCreationFactoryTest.java | 17 +-
.../james/backends/es/NodeMappingFactoryTest.java | 23 ++-
.../backends/es/search/ScrolledSearchTest.java | 181 ++++++++++-----------
.../ElasticSearchIntegrationTest.java | 13 +-
...lasticSearchQuotaSearchTestSystemExtension.java | 13 +-
.../ElasticSearchQuotaMailboxListenerTest.java | 6 +
.../host/ElasticSearchHostSystem.java | 7 +-
.../modules/mailbox/ElasticSearchClientModule.java | 40 +----
.../modules/TestDockerElasticSearchModule.java | 16 +-
.../routes/ElasticSearchQuotaSearchExtension.java | 9 +-
17 files changed, 276 insertions(+), 286 deletions(-)
diff --git a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ClientProvider.java b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ClientProvider.java
index 2a3aba1..9fb7f1a 100644
--- a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ClientProvider.java
+++ b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ClientProvider.java
@@ -18,9 +18,73 @@
****************************************************************/
package org.apache.james.backends.es;
+import java.io.IOException;
+import java.time.Duration;
+import java.time.LocalDateTime;
+
+import javax.annotation.PreDestroy;
+import javax.inject.Inject;
+import javax.inject.Provider;
+
+import org.apache.commons.lang3.time.DurationFormatUtils;
+import org.apache.http.HttpHost;
+import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.annotations.VisibleForTesting;
+import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
+
+public class ClientProvider implements Provider<RestHighLevelClient> {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ClientProvider.class);
+
+ private static final String HTTP_HOST_SCHEME = "http";
+ private final ElasticSearchConfiguration configuration;
+ private final RestHighLevelClient client;
+
+ @Inject
+ @VisibleForTesting
+ ClientProvider(ElasticSearchConfiguration configuration) {
+ this.configuration = configuration;
+ this.client = connect(configuration);
+ }
+
+ private RestHighLevelClient connect(ElasticSearchConfiguration configuration) {
+ Duration waitDelay = Duration.ofMillis(configuration.getMinDelay());
+ Duration forever = Duration.ofMillis(Long.MAX_VALUE);
+ boolean suppressLeadingZeroElements = true;
+ boolean suppressTrailingZeroElements = true;
+ return Mono.fromCallable(() -> connectToCluster(configuration))
+ .doOnError(e -> LOGGER.warn("Error establishing ElasticSearch connection. Next retry scheduled in {}",
+ DurationFormatUtils.formatDurationWords(waitDelay.toMillis(), suppressLeadingZeroElements, suppressTrailingZeroElements), e))
+ .retryBackoff(configuration.getMaxRetries(), waitDelay, forever, Schedulers.elastic())
+ .publishOn(Schedulers.elastic())
+ .block();
+ }
+
+ private RestHighLevelClient connectToCluster(ElasticSearchConfiguration configuration) throws IOException {
+ LOGGER.info("Trying to connect to ElasticSearch service at {}", LocalDateTime.now());
+ return new RestHighLevelClient(
+ RestClient.builder(hostsToHttpHosts())
+ .setMaxRetryTimeoutMillis(Math.toIntExact(configuration.getRequestTimeout().toMillis())));
+ }
+
+ private HttpHost[] hostsToHttpHosts() {
+ return configuration.getHosts().stream()
+ .map(host -> new HttpHost(host.getHostName(), host.getPort(), HTTP_HOST_SCHEME))
+ .toArray(HttpHost[]::new);
+ }
-public interface ClientProvider {
+ @Override
+ public RestHighLevelClient get() {
+ return client;
+ }
- RestHighLevelClient get();
+ @PreDestroy
+ public void close() throws IOException {
+ client.close();
+ }
}
diff --git a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ClientProviderImpl.java b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ClientProviderImpl.java
deleted file mode 100644
index 98611df..0000000
--- a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ClientProviderImpl.java
+++ /dev/null
@@ -1,54 +0,0 @@
-/****************************************************************
- * Licensed to the Apache Software Foundation (ASF) under one *
- * or more contributor license agreements. See the NOTICE file *
- * distributed with this work for additional information *
- * regarding copyright ownership. The ASF licenses this file *
- * to you under the Apache License, Version 2.0 (the *
- * "License"); you may not use this file except in compliance *
- * with the License. You may obtain a copy of the License at *
- * *
- * http://www.apache.org/licenses/LICENSE-2.0 *
- * *
- * Unless required by applicable law or agreed to in writing, *
- * software distributed under the License is distributed on an *
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY *
- * KIND, either express or implied. See the License for the *
- * specific language governing permissions and limitations *
- * under the License. *
- ****************************************************************/
-package org.apache.james.backends.es;
-
-import org.apache.http.HttpHost;
-import org.elasticsearch.client.RestClient;
-import org.elasticsearch.client.RestClientBuilder;
-import org.elasticsearch.client.RestHighLevelClient;
-
-import com.google.common.base.Preconditions;
-
-public class ClientProviderImpl implements ClientProvider {
-
- public static ClientProviderImpl fromConfiguration(ElasticSearchConfiguration configuration) {
- Preconditions.checkNotNull(configuration);
- return new ClientProviderImpl(configuration);
- }
-
- private static final String HTTP_HOST_SCHEME = "http";
- private final ElasticSearchConfiguration configuration;
-
- private ClientProviderImpl(ElasticSearchConfiguration configuration) {
- this.configuration = configuration;
- }
-
- private HttpHost[] hostsToHttpHosts() {
- return configuration.getHosts().stream()
- .map(host -> new HttpHost(host.getHostName(), host.getPort(), HTTP_HOST_SCHEME))
- .toArray(HttpHost[]::new);
- }
-
- @Override
- public RestHighLevelClient get() {
- RestClientBuilder restClient = RestClient.builder(hostsToHttpHosts())
- .setMaxRetryTimeoutMillis(Math.toIntExact(configuration.getRequestTimeout().toMillis()));
- return new RestHighLevelClient(restClient);
- }
-}
diff --git a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ClientProviderImplConnectionTest.java b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ClientProviderImplConnectionTest.java
index 72ef84a..516221a 100644
--- a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ClientProviderImplConnectionTest.java
+++ b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ClientProviderImplConnectionTest.java
@@ -60,7 +60,7 @@ public class ClientProviderImplConnectionTest {
Awaitility.await()
.atMost(1, TimeUnit.MINUTES)
.pollInterval(5, TimeUnit.SECONDS)
- .until(() -> isConnected(ClientProviderImpl.fromConfiguration(configuration)));
+ .until(() -> isConnected(new ClientProvider(configuration)));
}
@Test
@@ -73,7 +73,7 @@ public class ClientProviderImplConnectionTest {
Awaitility.await()
.atMost(1, TimeUnit.MINUTES)
.pollInterval(5, TimeUnit.SECONDS)
- .until(() -> isConnected(ClientProviderImpl.fromConfiguration(configuration)));
+ .until(() -> isConnected(new ClientProvider(configuration)));
}
@Test
@@ -90,7 +90,7 @@ public class ClientProviderImplConnectionTest {
Awaitility.await()
.atMost(1, TimeUnit.MINUTES)
.pollInterval(5, TimeUnit.SECONDS)
- .until(() -> isConnected(ClientProviderImpl.fromConfiguration(configuration)));
+ .until(() -> isConnected(new ClientProvider(configuration)));
}
private boolean isConnected(ClientProvider clientProvider) {
diff --git a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ClientProviderImplTest.java b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ClientProviderTest.java
similarity index 81%
rename from backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ClientProviderImplTest.java
rename to backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ClientProviderTest.java
index 949ad74..aef7c35 100644
--- a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ClientProviderImplTest.java
+++ b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ClientProviderTest.java
@@ -19,19 +19,15 @@
package org.apache.james.backends.es;
-import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
-import java.util.Optional;
-
-import org.elasticsearch.common.settings.Settings;
import org.junit.Test;
-public class ClientProviderImplTest {
+public class ClientProviderTest {
@Test
- public void fromConfigurationShouldThrowOnNull() {
- assertThatThrownBy(() -> ClientProviderImpl.fromConfiguration(null))
+ public void constructorShouldThrowOnNull() {
+ assertThatThrownBy(() -> new ClientProvider(null))
.isInstanceOf(NullPointerException.class);
}
}
diff --git a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/DockerElasticSearch.java b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/DockerElasticSearch.java
index 82eb83b..8441a99 100644
--- a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/DockerElasticSearch.java
+++ b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/DockerElasticSearch.java
@@ -112,9 +112,12 @@ public class DockerElasticSearch {
}
}
+ public ElasticSearchConfiguration configuration() {
+ return ElasticSearchConfiguration.builder().addHost(getHttpHost()).build();
+ }
+
public ClientProvider clientProvider() {
- ElasticSearchConfiguration configuration = ElasticSearchConfiguration.builder().addHost(getHttpHost()).build();
- return ClientProviderImpl.fromConfiguration(configuration);
+ return new ClientProvider(configuration());
}
private ElasticSearchAPI esAPI() {
diff --git a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/DockerElasticSearchRule.java b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/DockerElasticSearchRule.java
index 89233a1..a7de917 100644
--- a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/DockerElasticSearchRule.java
+++ b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/DockerElasticSearchRule.java
@@ -38,7 +38,7 @@ public class DockerElasticSearchRule extends ExternalResource {
public ClientProvider clientProvider() {
return dockerElasticSearch.clientProvider();
}
-
+
public void awaitForElasticSearch() {
dockerElasticSearch.flushIndices();
}
diff --git a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ElasticSearchIndexerTest.java b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ElasticSearchIndexerTest.java
index 34abc24..08e4e4f 100644
--- a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ElasticSearchIndexerTest.java
+++ b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ElasticSearchIndexerTest.java
@@ -24,6 +24,8 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.awaitility.Duration.ONE_HUNDRED_MILLISECONDS;
import static org.elasticsearch.index.query.QueryBuilders.termQuery;
+import java.io.IOException;
+
import org.awaitility.Awaitility;
import org.awaitility.Duration;
import org.awaitility.core.ConditionFactory;
@@ -32,6 +34,7 @@ import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.builder.SearchSourceBuilder;
+import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
@@ -52,19 +55,21 @@ public class ElasticSearchIndexerTest {
@Rule
public DockerElasticSearchRule elasticSearch = new DockerElasticSearchRule();
private ElasticSearchIndexer testee;
+ private RestHighLevelClient client;
@Before
public void setup() {
+ client = elasticSearch.clientProvider().get();
new IndexCreationFactory(ElasticSearchConfiguration.DEFAULT_CONFIGURATION)
.useIndex(INDEX_NAME)
.addAlias(ALIAS_NAME)
- .createIndexAndAliases(getESClient());
- testee = new ElasticSearchIndexer(getESClient(),
- ALIAS_NAME, MINIMUM_BATCH_SIZE);
+ .createIndexAndAliases(client);
+ testee = new ElasticSearchIndexer(client, ALIAS_NAME, MINIMUM_BATCH_SIZE);
}
- private RestHighLevelClient getESClient() {
- return elasticSearch.clientProvider().get();
+ @After
+ public void tearDown() throws IOException {
+ client.close();
}
@Test
@@ -75,12 +80,10 @@ public class ElasticSearchIndexerTest {
testee.index(messageId, content);
elasticSearch.awaitForElasticSearch();
- try (RestHighLevelClient client = getESClient()) {
- SearchResponse searchResponse = client.search(
- new SearchRequest(INDEX_NAME.getValue())
- .source(new SearchSourceBuilder().query(QueryBuilders.matchQuery("message", "trying"))));
- assertThat(searchResponse.getHits().getTotalHits()).isEqualTo(1);
- }
+ SearchResponse searchResponse = client.search(
+ new SearchRequest(INDEX_NAME.getValue())
+ .source(new SearchSourceBuilder().query(QueryBuilders.matchQuery("message", "trying"))));
+ assertThat(searchResponse.getHits().getTotalHits()).isEqualTo(1);
}
@Test
@@ -100,19 +103,16 @@ public class ElasticSearchIndexerTest {
testee.update(ImmutableList.of(new UpdatedRepresentation(messageId, "{\"message\": \"mastering out Elasticsearch\"}")));
elasticSearch.awaitForElasticSearch();
- try (RestHighLevelClient client = getESClient()) {
- SearchResponse searchResponse = client.search(
- new SearchRequest(INDEX_NAME.getValue())
- .source(new SearchSourceBuilder().query(QueryBuilders.matchQuery("message", "mastering"))));
- assertThat(searchResponse.getHits().getTotalHits()).isEqualTo(1);
- }
- try (RestHighLevelClient client = getESClient()) {
- SearchResponse searchResponse = client.search(
- new SearchRequest(INDEX_NAME.getValue())
- .source(new SearchSourceBuilder().query(QueryBuilders.matchQuery("field", "unchanged"))));
- assertThat(searchResponse.getHits().getTotalHits()).isEqualTo(1);
- }
+ SearchResponse searchResponse = client.search(
+ new SearchRequest(INDEX_NAME.getValue())
+ .source(new SearchSourceBuilder().query(QueryBuilders.matchQuery("message", "mastering"))));
+ assertThat(searchResponse.getHits().getTotalHits()).isEqualTo(1);
+
+ SearchResponse searchResponse2 = client.search(
+ new SearchRequest(INDEX_NAME.getValue())
+ .source(new SearchSourceBuilder().query(QueryBuilders.matchQuery("field", "unchanged"))));
+ assertThat(searchResponse2.getHits().getTotalHits()).isEqualTo(1);
}
@Test
@@ -150,13 +150,11 @@ public class ElasticSearchIndexerTest {
testee.deleteAllMatchingQuery(termQuery("property", "1"));
elasticSearch.awaitForElasticSearch();
- try (RestHighLevelClient client = getESClient()) {
- CALMLY_AWAIT.atMost(Duration.TEN_SECONDS)
- .until(() -> client.search(
- new SearchRequest(INDEX_NAME.getValue())
- .source(new SearchSourceBuilder().query(QueryBuilders.matchAllQuery())))
- .getHits().getTotalHits() == 0);
- }
+ CALMLY_AWAIT.atMost(Duration.TEN_SECONDS)
+ .until(() -> client.search(
+ new SearchRequest(INDEX_NAME.getValue())
+ .source(new SearchSourceBuilder().query(QueryBuilders.matchAllQuery())))
+ .getHits().getTotalHits() == 0);
}
@Test
@@ -180,13 +178,11 @@ public class ElasticSearchIndexerTest {
testee.deleteAllMatchingQuery(termQuery("property", "1"));
elasticSearch.awaitForElasticSearch();
- try (RestHighLevelClient client = getESClient()) {
- CALMLY_AWAIT.atMost(Duration.TEN_SECONDS)
- .until(() -> client.search(
- new SearchRequest(INDEX_NAME.getValue())
- .source(new SearchSourceBuilder().query(QueryBuilders.matchAllQuery())))
- .getHits().getTotalHits() == 1);
- }
+ CALMLY_AWAIT.atMost(Duration.TEN_SECONDS)
+ .until(() -> client.search(
+ new SearchRequest(INDEX_NAME.getValue())
+ .source(new SearchSourceBuilder().query(QueryBuilders.matchAllQuery())))
+ .getHits().getTotalHits() == 1);
}
@Test
@@ -200,12 +196,10 @@ public class ElasticSearchIndexerTest {
testee.delete(ImmutableList.of(messageId));
elasticSearch.awaitForElasticSearch();
- try (RestHighLevelClient client = getESClient()) {
- SearchResponse searchResponse = client.search(
- new SearchRequest(INDEX_NAME.getValue())
- .source(new SearchSourceBuilder().query(QueryBuilders.matchAllQuery())));
- assertThat(searchResponse.getHits().getTotalHits()).isEqualTo(0);
- }
+ SearchResponse searchResponse = client.search(
+ new SearchRequest(INDEX_NAME.getValue())
+ .source(new SearchSourceBuilder().query(QueryBuilders.matchAllQuery())));
+ assertThat(searchResponse.getHits().getTotalHits()).isEqualTo(0);
}
@Test
@@ -229,12 +223,10 @@ public class ElasticSearchIndexerTest {
testee.delete(ImmutableList.of(messageId, messageId3));
elasticSearch.awaitForElasticSearch();
- try (RestHighLevelClient client = getESClient()) {
- SearchResponse searchResponse = client.search(
- new SearchRequest(INDEX_NAME.getValue())
- .source(new SearchSourceBuilder().query(QueryBuilders.matchAllQuery())));
- assertThat(searchResponse.getHits().getTotalHits()).isEqualTo(1);
- }
+ SearchResponse searchResponse = client.search(
+ new SearchRequest(INDEX_NAME.getValue())
+ .source(new SearchSourceBuilder().query(QueryBuilders.matchAllQuery())));
+ assertThat(searchResponse.getHits().getTotalHits()).isEqualTo(1);
}
@Test
diff --git a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/IndexCreationFactoryTest.java b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/IndexCreationFactoryTest.java
index 50babf5..f204f2c 100644
--- a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/IndexCreationFactoryTest.java
+++ b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/IndexCreationFactoryTest.java
@@ -21,6 +21,10 @@ package org.apache.james.backends.es;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import java.io.IOException;
+
+import org.elasticsearch.client.RestHighLevelClient;
+import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
@@ -31,15 +35,20 @@ public class IndexCreationFactoryTest {
@Rule
public DockerElasticSearchRule elasticSearch = new DockerElasticSearchRule();
- private ClientProvider clientProvider;
+ private RestHighLevelClient client;
@Before
public void setUp() {
- clientProvider = elasticSearch.clientProvider();
+ client = elasticSearch.clientProvider().get();
new IndexCreationFactory(ElasticSearchConfiguration.DEFAULT_CONFIGURATION)
.useIndex(INDEX_NAME)
.addAlias(ALIAS_NAME)
- .createIndexAndAliases(clientProvider.get());
+ .createIndexAndAliases(client);
+ }
+
+ @After
+ public void tearDown() throws IOException {
+ client.close();
}
@Test
@@ -47,7 +56,7 @@ public class IndexCreationFactoryTest {
new IndexCreationFactory(ElasticSearchConfiguration.DEFAULT_CONFIGURATION)
.useIndex(INDEX_NAME)
.addAlias(ALIAS_NAME)
- .createIndexAndAliases(clientProvider.get());
+ .createIndexAndAliases(client);
}
@Test
diff --git a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/NodeMappingFactoryTest.java b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/NodeMappingFactoryTest.java
index dd24ddd..e78c970 100644
--- a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/NodeMappingFactoryTest.java
+++ b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/NodeMappingFactoryTest.java
@@ -22,7 +22,11 @@ package org.apache.james.backends.es;
import static org.assertj.core.api.Assertions.assertThatCode;
import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
+import java.io.IOException;
+
+import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
@@ -34,36 +38,41 @@ public class NodeMappingFactoryTest {
@Rule
public DockerElasticSearchRule elasticSearch = new DockerElasticSearchRule();
- private ClientProvider clientProvider;
+ private RestHighLevelClient client;
@Before
public void setUp() throws Exception {
- clientProvider = elasticSearch.clientProvider();
+ client = elasticSearch.getDockerElasticSearch().clientProvider().get();
new IndexCreationFactory(ElasticSearchConfiguration.DEFAULT_CONFIGURATION)
.useIndex(INDEX_NAME)
.addAlias(ALIAS_NAME)
- .createIndexAndAliases(clientProvider.get());
- NodeMappingFactory.applyMapping(clientProvider.get(),
+ .createIndexAndAliases(client);
+ NodeMappingFactory.applyMapping(client,
INDEX_NAME,
getMappingsSources());
}
+ @After
+ public void tearDown() throws IOException {
+ client.close();
+ }
+
@Test
public void applyMappingShouldNotThrowWhenCalledSeveralTime() throws Exception {
- NodeMappingFactory.applyMapping(clientProvider.get(),
+ NodeMappingFactory.applyMapping(client,
INDEX_NAME,
getMappingsSources());
}
@Test
public void applyMappingShouldNotThrowWhenIncrementalChanges() throws Exception {
- NodeMappingFactory.applyMapping(clientProvider.get(),
+ NodeMappingFactory.applyMapping(client,
INDEX_NAME,
getMappingsSources());
elasticSearch.awaitForElasticSearch();
- assertThatCode(() -> NodeMappingFactory.applyMapping(clientProvider.get(),
+ assertThatCode(() -> NodeMappingFactory.applyMapping(client,
INDEX_NAME,
getOtherMappingsSources()))
.doesNotThrowAnyException();
diff --git a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/search/ScrolledSearchTest.java b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/search/ScrolledSearchTest.java
index f622a63..49549af 100644
--- a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/search/ScrolledSearchTest.java
+++ b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/search/ScrolledSearchTest.java
@@ -24,7 +24,6 @@ import static org.awaitility.Awaitility.await;
import java.io.IOException;
-import org.apache.james.backends.es.ClientProvider;
import org.apache.james.backends.es.DockerElasticSearchRule;
import org.apache.james.backends.es.ElasticSearchConfiguration;
import org.apache.james.backends.es.IndexCreationFactory;
@@ -40,6 +39,7 @@ import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
+import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
@@ -55,120 +55,117 @@ public class ScrolledSearchTest {
@Rule
public DockerElasticSearchRule elasticSearch = new DockerElasticSearchRule();
- private ClientProvider clientProvider;
+ private RestHighLevelClient client;
@Before
public void setUp() {
- clientProvider = elasticSearch.clientProvider();
+ client = elasticSearch.clientProvider().get();
new IndexCreationFactory(ElasticSearchConfiguration.DEFAULT_CONFIGURATION)
.useIndex(INDEX_NAME)
.addAlias(ALIAS_NAME)
- .createIndexAndAliases(clientProvider.get());
+ .createIndexAndAliases(client);
elasticSearch.awaitForElasticSearch();
}
+ @After
+ public void tearDown() throws IOException {
+ client.close();
+ }
+
@Test
public void scrollIterableShouldWorkWhenEmpty() throws Exception {
- try (RestHighLevelClient client = clientProvider.get()) {
- SearchRequest searchRequest = new SearchRequest(INDEX_NAME.getValue())
- .scroll(TIMEOUT)
- .source(new SearchSourceBuilder()
- .query(QueryBuilders.matchAllQuery())
- .size(SIZE));
-
- assertThat(new ScrolledSearch(client, searchRequest).searchHits())
- .isEmpty();
- }
+ SearchRequest searchRequest = new SearchRequest(INDEX_NAME.getValue())
+ .scroll(TIMEOUT)
+ .source(new SearchSourceBuilder()
+ .query(QueryBuilders.matchAllQuery())
+ .size(SIZE));
+
+ assertThat(new ScrolledSearch(client, searchRequest).searchHits())
+ .isEmpty();
}
@Test
public void scrollIterableShouldWorkWhenOneElement() throws Exception {
- try (RestHighLevelClient client = clientProvider.get()) {
- String id = "1";
- client.index(new IndexRequest(INDEX_NAME.getValue())
- .type(NodeMappingFactory.DEFAULT_MAPPING_NAME)
- .id(id)
- .source(MESSAGE, "Sample message"));
-
- elasticSearch.awaitForElasticSearch();
- WAIT_CONDITION.untilAsserted(() -> hasIdsInIndex(client, id));
-
- SearchRequest searchRequest = new SearchRequest(INDEX_NAME.getValue())
- .scroll(TIMEOUT)
- .source(new SearchSourceBuilder()
- .query(QueryBuilders.matchAllQuery())
- .size(SIZE));
-
- assertThat(new ScrolledSearch(client, searchRequest).searchHits())
- .extracting(SearchHit::getId)
- .containsOnly(id);
- }
+ String id = "1";
+ client.index(new IndexRequest(INDEX_NAME.getValue())
+ .type(NodeMappingFactory.DEFAULT_MAPPING_NAME)
+ .id(id)
+ .source(MESSAGE, "Sample message"));
+
+ elasticSearch.awaitForElasticSearch();
+ WAIT_CONDITION.untilAsserted(() -> hasIdsInIndex(client, id));
+
+ SearchRequest searchRequest = new SearchRequest(INDEX_NAME.getValue())
+ .scroll(TIMEOUT)
+ .source(new SearchSourceBuilder()
+ .query(QueryBuilders.matchAllQuery())
+ .size(SIZE));
+
+ assertThat(new ScrolledSearch(client, searchRequest).searchHits())
+ .extracting(SearchHit::getId)
+ .containsOnly(id);
}
@Test
public void scrollIterableShouldWorkWhenSizeElement() throws Exception {
- try (RestHighLevelClient client = clientProvider.get()) {
- String id1 = "1";
- client.index(new IndexRequest(INDEX_NAME.getValue())
- .type(NodeMappingFactory.DEFAULT_MAPPING_NAME)
- .id(id1)
- .source(MESSAGE, "Sample message"));
-
- String id2 = "2";
- client.index(new IndexRequest(INDEX_NAME.getValue())
- .type(NodeMappingFactory.DEFAULT_MAPPING_NAME)
- .id(id2)
- .source(MESSAGE, "Sample message"));
-
- elasticSearch.awaitForElasticSearch();
- WAIT_CONDITION.untilAsserted(() -> hasIdsInIndex(client, id1, id2));
-
- SearchRequest searchRequest = new SearchRequest(INDEX_NAME.getValue())
- .scroll(TIMEOUT)
- .source(new SearchSourceBuilder()
- .query(QueryBuilders.matchAllQuery())
- .size(SIZE));
-
- assertThat(new ScrolledSearch(client, searchRequest).searchHits())
- .extracting(SearchHit::getId)
- .containsOnly(id1, id2);
- }
+ String id1 = "1";
+ client.index(new IndexRequest(INDEX_NAME.getValue())
+ .type(NodeMappingFactory.DEFAULT_MAPPING_NAME)
+ .id(id1)
+ .source(MESSAGE, "Sample message"));
+
+ String id2 = "2";
+ client.index(new IndexRequest(INDEX_NAME.getValue())
+ .type(NodeMappingFactory.DEFAULT_MAPPING_NAME)
+ .id(id2)
+ .source(MESSAGE, "Sample message"));
+
+ elasticSearch.awaitForElasticSearch();
+ WAIT_CONDITION.untilAsserted(() -> hasIdsInIndex(client, id1, id2));
+
+ SearchRequest searchRequest = new SearchRequest(INDEX_NAME.getValue())
+ .scroll(TIMEOUT)
+ .source(new SearchSourceBuilder()
+ .query(QueryBuilders.matchAllQuery())
+ .size(SIZE));
+
+ assertThat(new ScrolledSearch(client, searchRequest).searchHits())
+ .extracting(SearchHit::getId)
+ .containsOnly(id1, id2);
}
@Test
public void scrollIterableShouldWorkWhenMoreThanSizeElement() throws Exception {
- try (RestHighLevelClient client = clientProvider.get()) {
- String id1 = "1";
- client.index(new IndexRequest(INDEX_NAME.getValue())
- .type(NodeMappingFactory.DEFAULT_MAPPING_NAME)
- .id(id1)
- .source(MESSAGE, "Sample message"));
-
- String id2 = "2";
- client.index(new IndexRequest(INDEX_NAME.getValue())
- .type(NodeMappingFactory.DEFAULT_MAPPING_NAME)
- .id(id2)
- .source(MESSAGE, "Sample message"));
-
- String id3 = "3";
- client.index(new IndexRequest(INDEX_NAME.getValue())
- .type(NodeMappingFactory.DEFAULT_MAPPING_NAME)
- .id(id3)
- .source(MESSAGE, "Sample message"));
-
- elasticSearch.awaitForElasticSearch();
- WAIT_CONDITION.untilAsserted(() -> hasIdsInIndex(client, id1, id2, id3));
-
- SearchRequest searchRequest = new SearchRequest(INDEX_NAME.getValue())
- .scroll(TIMEOUT)
- .source(new SearchSourceBuilder()
- .query(QueryBuilders.matchAllQuery())
- .size(SIZE));
-
- assertThat(new ScrolledSearch(client, searchRequest).searchHits())
- .extracting(SearchHit::getId)
- .containsOnly(id1, id2, id3);
- }
+ String id1 = "1";
+ client.index(new IndexRequest(INDEX_NAME.getValue())
+ .type(NodeMappingFactory.DEFAULT_MAPPING_NAME)
+ .id(id1)
+ .source(MESSAGE, "Sample message"));
+
+ String id2 = "2";
+ client.index(new IndexRequest(INDEX_NAME.getValue())
+ .type(NodeMappingFactory.DEFAULT_MAPPING_NAME)
+ .id(id2)
+ .source(MESSAGE, "Sample message"));
+
+ String id3 = "3";
+ client.index(new IndexRequest(INDEX_NAME.getValue())
+ .type(NodeMappingFactory.DEFAULT_MAPPING_NAME)
+ .id(id3)
+ .source(MESSAGE, "Sample message"));
+
+ elasticSearch.awaitForElasticSearch();
+ WAIT_CONDITION.untilAsserted(() -> hasIdsInIndex(client, id1, id2, id3));
+
+ SearchRequest searchRequest = new SearchRequest(INDEX_NAME.getValue())
+ .scroll(TIMEOUT)
+ .source(new SearchSourceBuilder()
+ .query(QueryBuilders.matchAllQuery())
+ .size(SIZE));
+
+ assertThat(new ScrolledSearch(client, searchRequest).searchHits())
+ .extracting(SearchHit::getId)
+ .containsOnly(id1, id2, id3);
}
private void hasIdsInIndex(RestHighLevelClient client, String... ids) throws IOException {
diff --git a/mailbox/elasticsearch/src/test/java/org/apache/james/mailbox/elasticsearch/ElasticSearchIntegrationTest.java b/mailbox/elasticsearch/src/test/java/org/apache/james/mailbox/elasticsearch/ElasticSearchIntegrationTest.java
index d32d51b..0e02175 100644
--- a/mailbox/elasticsearch/src/test/java/org/apache/james/mailbox/elasticsearch/ElasticSearchIntegrationTest.java
+++ b/mailbox/elasticsearch/src/test/java/org/apache/james/mailbox/elasticsearch/ElasticSearchIntegrationTest.java
@@ -50,6 +50,7 @@ import org.apache.james.mailbox.tika.TikaTextExtractor;
import org.apache.james.metrics.api.NoopMetricFactory;
import org.apache.james.mime4j.dom.Message;
import org.elasticsearch.client.RestHighLevelClient;
+import org.junit.After;
import org.junit.ClassRule;
import org.junit.Rule;
import org.junit.Test;
@@ -67,6 +68,7 @@ public class ElasticSearchIntegrationTest extends AbstractMessageSearchIndexTest
@Rule
public DockerElasticSearchRule elasticSearch = new DockerElasticSearchRule();
private TikaTextExtractor textExtractor;
+ private RestHighLevelClient client;
@Override
public void setUp() throws Exception {
@@ -79,6 +81,11 @@ public class ElasticSearchIntegrationTest extends AbstractMessageSearchIndexTest
super.setUp();
}
+ @After
+ public void tearDown() throws IOException {
+ client.close();
+ }
+
@Override
protected void await() {
elasticSearch.awaitForElasticSearch();
@@ -86,11 +93,9 @@ public class ElasticSearchIntegrationTest extends AbstractMessageSearchIndexTest
@Override
protected void initializeMailboxManager() throws IOException {
- RestHighLevelClient client = MailboxIndexCreationUtil.prepareDefaultClient(
+ client = MailboxIndexCreationUtil.prepareDefaultClient(
elasticSearch.clientProvider().get(),
- ElasticSearchConfiguration.builder()
- .addHost(elasticSearch.getDockerElasticSearch().getHttpHost())
- .build());
+ elasticSearch.getDockerElasticSearch().configuration());
InMemoryMessageId.Factory messageIdFactory = new InMemoryMessageId.Factory();
diff --git a/mailbox/plugin/quota-search-elasticsearch/src/test/java/org/apache/james/quota/search/elasticsearch/ElasticSearchQuotaSearchTestSystemExtension.java b/mailbox/plugin/quota-search-elasticsearch/src/test/java/org/apache/james/quota/search/elasticsearch/ElasticSearchQuotaSearchTestSystemExtension.java
index 55dbf0b..9bfe5ac 100644
--- a/mailbox/plugin/quota-search-elasticsearch/src/test/java/org/apache/james/quota/search/elasticsearch/ElasticSearchQuotaSearchTestSystemExtension.java
+++ b/mailbox/plugin/quota-search-elasticsearch/src/test/java/org/apache/james/quota/search/elasticsearch/ElasticSearchQuotaSearchTestSystemExtension.java
@@ -21,6 +21,9 @@ package org.apache.james.quota.search.elasticsearch;
import static org.mockito.Mockito.mock;
+import java.io.IOException;
+
+import org.apache.james.backends.es.ClientProvider;
import org.apache.james.backends.es.DockerElasticSearch;
import org.apache.james.backends.es.DockerElasticSearchSingleton;
import org.apache.james.backends.es.ElasticSearchConfiguration;
@@ -44,6 +47,7 @@ import org.junit.jupiter.api.extension.ParameterResolver;
public class ElasticSearchQuotaSearchTestSystemExtension implements ParameterResolver, BeforeEachCallback, AfterEachCallback {
private final DockerElasticSearch elasticSearch = DockerElasticSearchSingleton.INSTANCE;
+ private RestHighLevelClient client;
@Override
public boolean supportsParameter(ParameterContext parameterContext, ExtensionContext extensionContext) throws ParameterResolutionException {
@@ -53,11 +57,9 @@ public class ElasticSearchQuotaSearchTestSystemExtension implements ParameterRes
@Override
public Object resolveParameter(ParameterContext parameterContext, ExtensionContext extensionContext) throws ParameterResolutionException {
try {
- RestHighLevelClient client = QuotaSearchIndexCreationUtil.prepareDefaultClient(
+ client = QuotaSearchIndexCreationUtil.prepareDefaultClient(
elasticSearch.clientProvider().get(),
- ElasticSearchConfiguration.builder()
- .addHost(elasticSearch.getHttpHost())
- .build());
+ elasticSearch.configuration());
InMemoryIntegrationResources resources = InMemoryIntegrationResources.defaultResources();
@@ -98,7 +100,8 @@ public class ElasticSearchQuotaSearchTestSystemExtension implements ParameterRes
}
@Override
- public void afterEach(ExtensionContext context) {
+ public void afterEach(ExtensionContext context) throws IOException {
+ client.close();
elasticSearch.cleanUpData();
}
}
diff --git a/mailbox/plugin/quota-search-elasticsearch/src/test/java/org/apache/james/quota/search/elasticsearch/events/ElasticSearchQuotaMailboxListenerTest.java b/mailbox/plugin/quota-search-elasticsearch/src/test/java/org/apache/james/quota/search/elasticsearch/events/ElasticSearchQuotaMailboxListenerTest.java
index ef5ef59..69e55e8 100644
--- a/mailbox/plugin/quota-search-elasticsearch/src/test/java/org/apache/james/quota/search/elasticsearch/events/ElasticSearchQuotaMailboxListenerTest.java
+++ b/mailbox/plugin/quota-search-elasticsearch/src/test/java/org/apache/james/quota/search/elasticsearch/events/ElasticSearchQuotaMailboxListenerTest.java
@@ -43,6 +43,7 @@ import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.builder.SearchSourceBuilder;
+import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
@@ -72,6 +73,11 @@ public class ElasticSearchQuotaMailboxListenerTest {
new QuotaRatioToElasticSearchJson());
}
+ @After
+ public void tearDown() throws IOException {
+ client.close();
+ }
+
@Test
public void deserializeElasticSearchQuotaMailboxListenerGroup() throws Exception {
assertThat(Group.deserialize("org.apache.james.quota.search.elasticsearch.events.ElasticSearchQuotaMailboxListener$ElasticSearchQuotaMailboxListenerGroup"))
diff --git a/mpt/impl/imap-mailbox/elasticsearch/src/test/java/org/apache/james/mpt/imapmailbox/elasticsearch/host/ElasticSearchHostSystem.java b/mpt/impl/imap-mailbox/elasticsearch/src/test/java/org/apache/james/mpt/imapmailbox/elasticsearch/host/ElasticSearchHostSystem.java
index e0f3b92..f191c30 100644
--- a/mpt/impl/imap-mailbox/elasticsearch/src/test/java/org/apache/james/mpt/imapmailbox/elasticsearch/host/ElasticSearchHostSystem.java
+++ b/mpt/impl/imap-mailbox/elasticsearch/src/test/java/org/apache/james/mpt/imapmailbox/elasticsearch/host/ElasticSearchHostSystem.java
@@ -19,6 +19,7 @@
package org.apache.james.mpt.imapmailbox.elasticsearch.host;
+import java.io.IOException;
import java.time.ZoneId;
import org.apache.commons.lang.NotImplementedException;
@@ -61,6 +62,7 @@ public class ElasticSearchHostSystem extends JamesImapHostSystem {
private DockerElasticSearch dockerElasticSearch;
private StoreMailboxManager mailboxManager;
+ private RestHighLevelClient client;
@Override
public void beforeTest() throws Exception {
@@ -71,12 +73,13 @@ public class ElasticSearchHostSystem extends JamesImapHostSystem {
}
@Override
- public void afterTest() {
+ public void afterTest() throws IOException {
+ client.close();
dockerElasticSearch.cleanUpData();
}
private void initFields() throws Exception {
- RestHighLevelClient client = MailboxIndexCreationUtil.prepareDefaultClient(
+ client = MailboxIndexCreationUtil.prepareDefaultClient(
dockerElasticSearch.clientProvider().get(),
ElasticSearchConfiguration.builder()
.addHost(dockerElasticSearch.getHttpHost())
diff --git a/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchClientModule.java b/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchClientModule.java
index 020a365..d6a6c11 100644
--- a/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchClientModule.java
+++ b/server/container/guice/cassandra-guice/src/main/java/org/apache/james/modules/mailbox/ElasticSearchClientModule.java
@@ -19,50 +19,18 @@
package org.apache.james.modules.mailbox;
-import java.io.IOException;
-import java.time.Duration;
-import java.time.LocalDateTime;
-
-import javax.inject.Singleton;
-
-import org.apache.commons.lang3.time.DurationFormatUtils;
-import org.apache.james.backends.es.ClientProviderImpl;
-import org.apache.james.backends.es.ElasticSearchConfiguration;
+import org.apache.james.backends.es.ClientProvider;
import org.elasticsearch.client.RestHighLevelClient;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import com.google.inject.AbstractModule;
-import com.google.inject.Provides;
-
-import reactor.core.publisher.Mono;
-import reactor.core.scheduler.Schedulers;
+import com.google.inject.Scopes;
public class ElasticSearchClientModule extends AbstractModule {
- private static final Logger LOGGER = LoggerFactory.getLogger(ElasticSearchClientModule.class);
-
@Override
protected void configure() {
+ bind(ClientProvider.class).in(Scopes.SINGLETON);
+ bind(RestHighLevelClient.class).toProvider(ClientProvider.class);
}
- @Provides
- @Singleton
- protected RestHighLevelClient provideClient(ElasticSearchConfiguration configuration) {
- Duration waitDelay = Duration.ofMillis(configuration.getMinDelay());
- Duration forever = Duration.ofMillis(Long.MAX_VALUE);
- return Mono.fromCallable(() -> connectToCluster(configuration))
- .doOnError(e -> LOGGER.warn("Error establishing ElasticSearch connection. Next retry scheduled in {}",
- DurationFormatUtils.formatDurationWords(waitDelay.toMillis(), true, true), e))
- .retryBackoff(configuration.getMaxRetries(), waitDelay, forever, Schedulers.elastic())
- .publishOn(Schedulers.elastic())
- .block();
- }
-
- private RestHighLevelClient connectToCluster(ElasticSearchConfiguration configuration) throws IOException {
- LOGGER.info("Trying to connect to ElasticSearch service at {}", LocalDateTime.now());
-
- return ClientProviderImpl.fromConfiguration(configuration)
- .get();
- }
}
diff --git a/server/container/guice/cassandra-guice/src/test/java/org/apache/james/modules/TestDockerElasticSearchModule.java b/server/container/guice/cassandra-guice/src/test/java/org/apache/james/modules/TestDockerElasticSearchModule.java
index 70e9362..fe87e53 100644
--- a/server/container/guice/cassandra-guice/src/test/java/org/apache/james/modules/TestDockerElasticSearchModule.java
+++ b/server/container/guice/cassandra-guice/src/test/java/org/apache/james/modules/TestDockerElasticSearchModule.java
@@ -19,18 +19,11 @@
package org.apache.james.modules;
-import java.io.IOException;
-
-import javax.inject.Singleton;
-
import org.apache.james.CleanupTasksPerformer;
import org.apache.james.backends.es.DockerElasticSearch;
import org.apache.james.backends.es.ElasticSearchConfiguration;
-import org.apache.james.mailbox.elasticsearch.MailboxIndexCreationUtil;
-import org.elasticsearch.client.RestHighLevelClient;
import com.google.inject.AbstractModule;
-import com.google.inject.Provides;
import com.google.inject.multibindings.Multibinder;
public class TestDockerElasticSearchModule extends AbstractModule {
@@ -59,17 +52,10 @@ public class TestDockerElasticSearchModule extends AbstractModule {
@Override
protected void configure() {
+ bind(ElasticSearchConfiguration.class).toInstance(elasticSearch.configuration());
Multibinder.newSetBinder(binder(), CleanupTasksPerformer.CleanupTask.class)
.addBinding()
.toInstance(new ESContainerCleanUp(elasticSearch));
}
- @Provides
- @Singleton
- protected RestHighLevelClient provideClientProvider() throws IOException {
- RestHighLevelClient client = elasticSearch.clientProvider().get();
- return MailboxIndexCreationUtil.prepareDefaultClient(client, ElasticSearchConfiguration.builder()
- .addHost(elasticSearch.getHttpHost())
- .build());
- }
}
diff --git a/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/ElasticSearchQuotaSearchExtension.java b/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/ElasticSearchQuotaSearchExtension.java
index f062b60..a469985 100644
--- a/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/ElasticSearchQuotaSearchExtension.java
+++ b/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/ElasticSearchQuotaSearchExtension.java
@@ -21,6 +21,8 @@ package org.apache.james.webadmin.routes;
import static org.mockito.Mockito.mock;
+import java.io.IOException;
+
import org.apache.james.backends.es.DockerElasticSearch;
import org.apache.james.backends.es.DockerElasticSearchSingleton;
import org.apache.james.backends.es.ElasticSearchConfiguration;
@@ -50,6 +52,7 @@ public class ElasticSearchQuotaSearchExtension implements ParameterResolver, Bef
private final DockerElasticSearch elasticSearch = DockerElasticSearchSingleton.INSTANCE;
private WebAdminQuotaSearchTestSystem restQuotaSearchTestSystem;
private TemporaryFolder temporaryFolder = new TemporaryFolder();
+ private RestHighLevelClient client;
@Override
public void beforeEach(ExtensionContext context) {
@@ -57,7 +60,7 @@ public class ElasticSearchQuotaSearchExtension implements ParameterResolver, Bef
temporaryFolder.create();
elasticSearch.start();
- RestHighLevelClient client = QuotaSearchIndexCreationUtil.prepareDefaultClient(
+ client = QuotaSearchIndexCreationUtil.prepareDefaultClient(
elasticSearch.clientProvider().get(),
ElasticSearchConfiguration.builder()
.addHost(elasticSearch.getHttpHost())
@@ -98,9 +101,9 @@ public class ElasticSearchQuotaSearchExtension implements ParameterResolver, Bef
}
@Override
- public void afterEach(ExtensionContext context) {
+ public void afterEach(ExtensionContext context) throws IOException {
restQuotaSearchTestSystem.getWebAdminServer().destroy();
-
+ client.close();
elasticSearch.cleanUpData();
temporaryFolder.delete();
}
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 16/28: JAMES-2794 Avoid relying on a DTO in
RabbitMQ MailQueue logic
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 1785ab5fda5e74c4eaf4f1b41aa3f9a28bdad277
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Thu Jun 20 10:15:15 2019 +0700
JAMES-2794 Avoid relying on a DTO in RabbitMQ MailQueue logic
---
.../main/java/org/apache/james/queue/rabbitmq/MailReferenceDTO.java | 4 ----
1 file changed, 4 deletions(-)
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailReferenceDTO.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailReferenceDTO.java
index 4d88292..84c79dc 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailReferenceDTO.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailReferenceDTO.java
@@ -199,10 +199,6 @@ class MailReferenceDTO {
return bodyBlobId;
}
- EnqueueId retrieveEnqueueId() {
- return EnqueueId.ofSerialized(enQueueId);
- }
-
MailReference toMailReference(BlobId.Factory blobIdFactory) {
MimeMessagePartsId messagePartsId = MimeMessagePartsId.builder()
.headerBlobId(blobIdFactory.from(headerBlobId))
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 04/28: JAMES-2794 Create an EnqueueId POJO in
RabbitMQMailQueue
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 3e6f9f3dbf3443faeb8873972a1af966ec57fd0d
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Tue Jun 18 11:28:45 2019 +0700
JAMES-2794 Create an EnqueueId POJO in RabbitMQMailQueue
The aim is not to identify a given email but rather identify a given
enqueue hence I do not reuse the MailKey class, that marks the intent to
identify a given mail, regardless of specific enqueues.
---
.../model/MailKey.java => EnQueueId.java} | 45 +++++++++++++---------
.../model/{MailKeyTest.java => EnQueueIdTest.java} | 32 ++++++++++++++-
2 files changed, 56 insertions(+), 21 deletions(-)
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/model/MailKey.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/EnQueueId.java
similarity index 62%
rename from server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/model/MailKey.java
rename to server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/EnQueueId.java
index 4f17a71..133172d 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/model/MailKey.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/EnQueueId.java
@@ -17,49 +17,56 @@
* under the License. *
****************************************************************/
-package org.apache.james.queue.rabbitmq.view.cassandra.model;
+package org.apache.james.queue.rabbitmq;
import java.util.Objects;
+import java.util.UUID;
-import org.apache.mailet.Mail;
-
+import com.datastax.driver.core.utils.UUIDs;
import com.google.common.base.Preconditions;
-public class MailKey {
+public class EnQueueId {
+
+ public static EnQueueId generate() {
+ return of(UUIDs.timeBased());
+ }
- public static MailKey fromMail(Mail mail) {
- return of(mail.getName());
+ public static EnQueueId of(UUID uuid) {
+ Preconditions.checkNotNull(uuid);
+ return new EnQueueId(uuid);
}
- public static MailKey of(String mailKey) {
- return new MailKey(mailKey);
+ public static EnQueueId ofSerialized(String serialized) {
+ Preconditions.checkNotNull(serialized);
+ return of(UUID.fromString(serialized));
}
- private final String mailKey;
+ private final UUID id;
- private MailKey(String mailKey) {
- Preconditions.checkNotNull(mailKey);
- Preconditions.checkArgument(!mailKey.isEmpty());
+ private EnQueueId(UUID id) {
+ this.id = id;
+ }
- this.mailKey = mailKey;
+ public UUID asUUID() {
+ return id;
}
- public String getMailKey() {
- return mailKey;
+ public String serialize() {
+ return id.toString();
}
@Override
public final boolean equals(Object o) {
- if (o instanceof MailKey) {
- MailKey mailKey1 = (MailKey) o;
+ if (o instanceof EnQueueId) {
+ EnQueueId enQueueId = (EnQueueId) o;
- return Objects.equals(this.mailKey, mailKey1.mailKey);
+ return Objects.equals(this.id, enQueueId.id);
}
return false;
}
@Override
public final int hashCode() {
- return Objects.hash(mailKey);
+ return Objects.hash(id);
}
}
diff --git a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/model/MailKeyTest.java b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/model/EnQueueIdTest.java
similarity index 60%
rename from server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/model/MailKeyTest.java
rename to server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/model/EnQueueIdTest.java
index c1a854e..2f21c34 100644
--- a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/model/MailKeyTest.java
+++ b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/model/EnQueueIdTest.java
@@ -19,15 +19,43 @@
package org.apache.james.queue.rabbitmq.view.cassandra.model;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.UUID;
+
+import org.apache.james.queue.rabbitmq.EnQueueId;
import org.junit.jupiter.api.Test;
import nl.jqno.equalsverifier.EqualsVerifier;
-class MailKeyTest {
+class EnQueueIdTest {
+ private static final UUID UUID_1 = UUID.fromString("110e8400-e29b-11d4-a716-446655440000");
@Test
void shouldMatchBeanContract() {
- EqualsVerifier.forClass(MailKey.class)
+ EqualsVerifier.forClass(EnQueueId.class)
.verify();
}
+
+ @Test
+ void ofSerializedShouldDeserializeTheGivenEnqueueId() {
+ assertThat(EnQueueId.ofSerialized(UUID_1.toString()))
+ .isEqualTo(EnQueueId.of(UUID_1));
+ }
+
+ @Test
+ void serializeShouldReturnAStringRepresentation() {
+ EnQueueId enQueueId = EnQueueId.of(UUID_1);
+
+ assertThat(enQueueId.serialize())
+ .isEqualTo(UUID_1.toString());
+ }
+
+ @Test
+ void ofSerializedShouldRevertSerialize() {
+ EnQueueId enQueueId = EnQueueId.of(UUID_1);
+
+ assertThat(EnQueueId.ofSerialized(enQueueId.serialize()))
+ .isEqualTo(enQueueId);
+ }
}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 21/28: JAMES-2803 ElasticSearch cluster name is no
longer useful when using REST API
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit a7b2cf7b74ca19c4a3391b03514a4e0b18eff1f3
Author: Matthieu Baechler <ma...@apache.org>
AuthorDate: Wed Jun 19 15:49:22 2019 +0200
JAMES-2803 ElasticSearch cluster name is no longer useful when using REST API
---
.../james/backends/es/ClientProviderImpl.java | 11 ------
.../backends/es/ElasticSearchConfiguration.java | 21 +---------
.../es/ElasticSearchConfigurationTest.java | 46 ----------------------
3 files changed, 2 insertions(+), 76 deletions(-)
diff --git a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ClientProviderImpl.java b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ClientProviderImpl.java
index 7816e8d..98611df 100644
--- a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ClientProviderImpl.java
+++ b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ClientProviderImpl.java
@@ -22,9 +22,7 @@ import org.apache.http.HttpHost;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
-import org.elasticsearch.common.settings.Settings;
-import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
public class ClientProviderImpl implements ClientProvider {
@@ -34,7 +32,6 @@ public class ClientProviderImpl implements ClientProvider {
return new ClientProviderImpl(configuration);
}
- private static final String CLUSTER_NAME_SETTING = "cluster.name";
private static final String HTTP_HOST_SCHEME = "http";
private final ElasticSearchConfiguration configuration;
@@ -54,12 +51,4 @@ public class ClientProviderImpl implements ClientProvider {
.setMaxRetryTimeoutMillis(Math.toIntExact(configuration.getRequestTimeout().toMillis()));
return new RestHighLevelClient(restClient);
}
-
- @VisibleForTesting Settings settings() {
- return configuration.getClusterName()
- .map(clusterName -> Settings.builder()
- .put(CLUSTER_NAME_SETTING, clusterName)
- .build())
- .orElse(Settings.EMPTY);
- }
}
diff --git a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ElasticSearchConfiguration.java b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ElasticSearchConfiguration.java
index ed3199d..24635f5 100644
--- a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ElasticSearchConfiguration.java
+++ b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ElasticSearchConfiguration.java
@@ -42,7 +42,6 @@ public class ElasticSearchConfiguration {
public static class Builder {
private final ImmutableList.Builder<Host> hosts;
- private Optional<String> clusterName;
private Optional<Integer> nbShards;
private Optional<Integer> nbReplica;
private Optional<Integer> minDelay;
@@ -51,7 +50,6 @@ public class ElasticSearchConfiguration {
public Builder() {
hosts = ImmutableList.builder();
- clusterName = Optional.empty();
nbShards = Optional.empty();
nbReplica = Optional.empty();
minDelay = Optional.empty();
@@ -64,11 +62,6 @@ public class ElasticSearchConfiguration {
return this;
}
- public Builder clusterName(String clusterName) {
- this.clusterName = Optional.ofNullable(clusterName);
- return this;
- }
-
public Builder addHosts(Collection<Host> hosts) {
this.hosts.addAll(hosts);
return this;
@@ -106,7 +99,6 @@ public class ElasticSearchConfiguration {
Preconditions.checkState(!hosts.isEmpty(), "You need to specify ElasticSearch host");
return new ElasticSearchConfiguration(
hosts,
- clusterName,
nbShards.orElse(DEFAULT_NB_SHARDS),
nbReplica.orElse(DEFAULT_NB_REPLICA),
minDelay.orElse(DEFAULT_CONNECTION_MIN_DELAY),
@@ -120,7 +112,6 @@ public class ElasticSearchConfiguration {
}
public static final String ELASTICSEARCH_HOSTS = "elasticsearch.hosts";
- public static final String ELASTICSEARCH_CLUSTER_NAME = "elasticsearch.clusterName";
public static final String ELASTICSEARCH_MASTER_HOST = "elasticsearch.masterHost";
public static final String ELASTICSEARCH_PORT = "elasticsearch.port";
public static final String ELASTICSEARCH_NB_REPLICA = "elasticsearch.nb.replica";
@@ -144,7 +135,6 @@ public class ElasticSearchConfiguration {
public static ElasticSearchConfiguration fromProperties(Configuration configuration) throws ConfigurationException {
return builder()
.addHosts(getHosts(configuration))
- .clusterName(configuration.getString(ELASTICSEARCH_CLUSTER_NAME))
.nbShards(configuration.getInteger(ELASTICSEARCH_NB_SHARDS, DEFAULT_NB_SHARDS))
.nbReplica(configuration.getInteger(ELASTICSEARCH_NB_REPLICA, DEFAULT_NB_REPLICA))
.minDelay(Optional.ofNullable(configuration.getInteger(ELASTICSEARCH_RETRY_CONNECTION_MIN_DELAY, null)))
@@ -189,16 +179,14 @@ public class ElasticSearchConfiguration {
}
private final ImmutableList<Host> hosts;
- private final Optional<String> clusterName;
private final int nbShards;
private final int nbReplica;
private final int minDelay;
private final int maxRetries;
private final Duration requestTimeout;
- private ElasticSearchConfiguration(ImmutableList<Host> hosts, Optional<String> clusterName, int nbShards, int nbReplica, int minDelay, int maxRetries, Duration requestTimeout) {
+ private ElasticSearchConfiguration(ImmutableList<Host> hosts, int nbShards, int nbReplica, int minDelay, int maxRetries, Duration requestTimeout) {
this.hosts = hosts;
- this.clusterName = clusterName;
this.nbShards = nbShards;
this.nbReplica = nbReplica;
this.minDelay = minDelay;
@@ -210,10 +198,6 @@ public class ElasticSearchConfiguration {
return hosts;
}
- public Optional<String> getClusterName() {
- return clusterName;
- }
-
public int getNbShards() {
return nbShards;
}
@@ -240,7 +224,6 @@ public class ElasticSearchConfiguration {
ElasticSearchConfiguration that = (ElasticSearchConfiguration) o;
return Objects.equals(this.nbShards, that.nbShards)
- && Objects.equals(this.clusterName, that.clusterName)
&& Objects.equals(this.nbReplica, that.nbReplica)
&& Objects.equals(this.minDelay, that.minDelay)
&& Objects.equals(this.maxRetries, that.maxRetries)
@@ -252,6 +235,6 @@ public class ElasticSearchConfiguration {
@Override
public final int hashCode() {
- return Objects.hash(hosts, clusterName, nbShards, nbReplica, minDelay, maxRetries, requestTimeout);
+ return Objects.hash(hosts, nbShards, nbReplica, minDelay, maxRetries, requestTimeout);
}
}
diff --git a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ElasticSearchConfigurationTest.java b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ElasticSearchConfigurationTest.java
index c970bd3..18167a1 100644
--- a/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ElasticSearchConfigurationTest.java
+++ b/backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/ElasticSearchConfigurationTest.java
@@ -30,7 +30,6 @@ import org.apache.james.util.Host;
import org.junit.Test;
import com.google.common.collect.ImmutableList;
-
import nl.jqno.equalsverifier.EqualsVerifier;
public class ElasticSearchConfigurationTest {
@@ -192,51 +191,6 @@ public class ElasticSearchConfigurationTest {
}
@Test
- public void clusterNameShouldBeEmptyWhenNotGiven() throws ConfigurationException {
- PropertiesConfiguration configuration = new PropertiesConfiguration();
- String hostname = "myHost";
- configuration.addProperty("elasticsearch.masterHost", hostname);
- int port = 9200;
- configuration.addProperty("elasticsearch.port", port);
-
- ElasticSearchConfiguration elasticSearchConfiguration = ElasticSearchConfiguration.fromProperties(configuration);
-
- assertThat(elasticSearchConfiguration.getClusterName())
- .isEmpty();
- }
-
- @Test
- public void clusterNameShouldBeEmptyWhenNull() throws ConfigurationException {
- PropertiesConfiguration configuration = new PropertiesConfiguration();
- String hostname = "myHost";
- configuration.addProperty("elasticsearch.masterHost", hostname);
- int port = 9200;
- configuration.addProperty("elasticsearch.port", port);
- configuration.addProperty("elasticsearch.clusterName", null);
-
- ElasticSearchConfiguration elasticSearchConfiguration = ElasticSearchConfiguration.fromProperties(configuration);
-
- assertThat(elasticSearchConfiguration.getClusterName())
- .isEmpty();
- }
-
- @Test
- public void clusterNameShouldKeepTheValueWhenGiven() throws ConfigurationException {
- PropertiesConfiguration configuration = new PropertiesConfiguration();
- String hostname = "myHost";
- configuration.addProperty("elasticsearch.masterHost", hostname);
- int port = 9200;
- configuration.addProperty("elasticsearch.port", port);
- String clusterName = "myClusterName";
- configuration.addProperty("elasticsearch.clusterName", clusterName);
-
- ElasticSearchConfiguration elasticSearchConfiguration = ElasticSearchConfiguration.fromProperties(configuration);
-
- assertThat(elasticSearchConfiguration.getClusterName())
- .contains(clusterName);
- }
-
- @Test
public void validateHostsConfigurationOptionsShouldThrowWhenNoHostSpecify() {
assertThatThrownBy(() ->
ElasticSearchConfiguration.validateHostsConfigurationOptions(
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 18/28: JAMES-2794 Replace a pair by
MailWithEnqueueId
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 16617ab9ecdff41181c18c71e14f37d58d1b8b9b
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Tue Jun 25 08:28:36 2019 +0700
JAMES-2794 Replace a pair by MailWithEnqueueId
---
.../org/apache/james/queue/rabbitmq/Dequeuer.java | 24 ++++++-------
.../apache/james/queue/rabbitmq/MailLoader.java | 5 ++-
.../james/queue/rabbitmq/MailWithEnqueueId.java | 40 ++++++++++++++++++++++
.../queue/rabbitmq/RabbitMQMailQueueFactory.java | 4 +--
4 files changed, 54 insertions(+), 19 deletions(-)
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 01db228..94b803c 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
@@ -25,7 +25,6 @@ import java.io.IOException;
import java.util.function.Consumer;
import java.util.function.Function;
-import org.apache.commons.lang3.tuple.Pair;
import org.apache.james.metrics.api.Metric;
import org.apache.james.metrics.api.MetricFactory;
import org.apache.james.queue.api.MailQueue;
@@ -50,10 +49,10 @@ class Dequeuer {
private final EnqueueId enqueueId;
private final Mail mail;
- private RabbitMQMailQueueItem(Consumer<Boolean> ack, EnqueueId enqueueId, Mail mail) {
+ private RabbitMQMailQueueItem(Consumer<Boolean> ack, MailWithEnqueueId mailWithEnqueueId) {
this.ack = ack;
- this.enqueueId = enqueueId;
- this.mail = mail;
+ this.enqueueId = mailWithEnqueueId.getEnqueueId();
+ this.mail = mailWithEnqueueId.getMail();
}
@Override
@@ -71,12 +70,12 @@ class Dequeuer {
}
}
- private final Function<MailReferenceDTO, Pair<EnqueueId, Mail>> mailLoader;
+ private final Function<MailReferenceDTO, MailWithEnqueueId> mailLoader;
private final Metric dequeueMetric;
private final MailReferenceSerializer mailReferenceSerializer;
private final MailQueueView mailQueueView;
- Dequeuer(MailQueueName name, RabbitClient rabbitClient, Function<MailReferenceDTO, Pair<EnqueueId, Mail>> mailLoader,
+ Dequeuer(MailQueueName name, RabbitClient rabbitClient, Function<MailReferenceDTO, MailWithEnqueueId> mailLoader,
MailReferenceSerializer serializer, MetricFactory metricFactory,
MailQueueView mailQueueView) {
this.mailLoader = mailLoader;
@@ -108,28 +107,27 @@ class Dequeuer {
private Mono<RabbitMQMailQueueItem> loadItem(AcknowledgableDelivery response) {
try {
- Pair<EnqueueId, Mail> idAndMail = loadMail(response);
- Mail mail = idAndMail.getRight();
- ThrowingConsumer<Boolean> ack = ack(response, mail);
- return Mono.just(new RabbitMQMailQueueItem(ack, idAndMail.getLeft(), mail));
+ MailWithEnqueueId mailWithEnqueueId = loadMail(response);
+ ThrowingConsumer<Boolean> ack = ack(response, mailWithEnqueueId);
+ return Mono.just(new RabbitMQMailQueueItem(ack, mailWithEnqueueId));
} catch (MailQueue.MailQueueException e) {
return Mono.error(e);
}
}
- private ThrowingConsumer<Boolean> ack(AcknowledgableDelivery response, Mail mail) {
+ private ThrowingConsumer<Boolean> ack(AcknowledgableDelivery response, MailWithEnqueueId mailWithEnqueueId) {
return success -> {
if (success) {
dequeueMetric.increment();
response.ack();
- mailQueueView.delete(DeleteCondition.withName(mail.getName()));
+ mailQueueView.delete(DeleteCondition.withName(mailWithEnqueueId.getMail().getName()));
} else {
response.nack(REQUEUE);
}
};
}
- private Pair<EnqueueId, Mail> loadMail(Delivery response) throws MailQueue.MailQueueException {
+ private MailWithEnqueueId loadMail(Delivery response) throws MailQueue.MailQueueException {
MailReferenceDTO mailDTO = toMailReference(response);
return mailLoader.apply(mailDTO);
}
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailLoader.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailLoader.java
index 98fdc5c..a616909 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailLoader.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailLoader.java
@@ -23,7 +23,6 @@ import javax.mail.MessagingException;
import javax.mail.internet.AddressException;
import javax.mail.internet.MimeMessage;
-import org.apache.commons.lang3.tuple.Pair;
import org.apache.james.blob.api.BlobId;
import org.apache.james.blob.api.Store;
import org.apache.james.blob.mail.MimeMessagePartsId;
@@ -39,14 +38,14 @@ class MailLoader {
this.blobIdFactory = blobIdFactory;
}
- Pair<EnqueueId, Mail> load(MailReferenceDTO dto) throws MailQueue.MailQueueException {
+ MailWithEnqueueId load(MailReferenceDTO dto) throws MailQueue.MailQueueException {
try {
MailReference mailReference = dto.toMailReference(blobIdFactory);
Mail mail = mailReference.getMail();
MimeMessage mimeMessage = mimeMessageStore.read(mailReference.getPartsId()).block();
mail.setMessage(mimeMessage);
- return Pair.of(mailReference.getEnqueueId(), mail);
+ return new MailWithEnqueueId(mailReference.getEnqueueId(), mail);
} catch (AddressException e) {
throw new MailQueue.MailQueueException("Failed to parse mail address", e);
} catch (MessagingException e) {
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailWithEnqueueId.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailWithEnqueueId.java
new file mode 100644
index 0000000..f256fbf
--- /dev/null
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailWithEnqueueId.java
@@ -0,0 +1,40 @@
+/****************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one *
+ * or more contributor license agreements. See the NOTICE file *
+ * distributed with this work for additional information *
+ * regarding copyright ownership. The ASF licenses this file *
+ * to you under the Apache License, Version 2.0 (the *
+ * "License"); you may not use this file except in compliance *
+ * with the License. You may obtain a copy of the License at *
+ * *
+ * http://www.apache.org/licenses/LICENSE-2.0 *
+ * *
+ * Unless required by applicable law or agreed to in writing, *
+ * software distributed under the License is distributed on an *
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY *
+ * KIND, either express or implied. See the License for the *
+ * specific language governing permissions and limitations *
+ * under the License. *
+ ****************************************************************/
+
+package org.apache.james.queue.rabbitmq;
+
+import org.apache.mailet.Mail;
+
+public class MailWithEnqueueId {
+ private final EnqueueId enqueueId;
+ private final Mail mail;
+
+ MailWithEnqueueId(EnqueueId enqueueId, Mail mail) {
+ this.enqueueId = enqueueId;
+ this.mail = mail;
+ }
+
+ public EnqueueId getEnqueueId() {
+ return enqueueId;
+ }
+
+ public Mail getMail() {
+ return mail;
+ }
+}
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 fdf24a1..adaf3de 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
@@ -30,7 +30,6 @@ import java.util.function.Function;
import javax.inject.Inject;
import javax.mail.internet.MimeMessage;
-import org.apache.commons.lang3.tuple.Pair;
import org.apache.james.blob.api.BlobId;
import org.apache.james.blob.api.Store;
import org.apache.james.blob.mail.MimeMessagePartsId;
@@ -40,7 +39,6 @@ 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.rabbitmq.view.api.MailQueueView;
-import org.apache.mailet.Mail;
import com.github.fge.lambdas.Throwing;
import com.google.common.annotations.VisibleForTesting;
@@ -54,7 +52,7 @@ public class RabbitMQMailQueueFactory implements MailQueueFactory<RabbitMQMailQu
private final RabbitClient rabbitClient;
private final Store<MimeMessage, MimeMessagePartsId> mimeMessageStore;
private final MailReferenceSerializer mailReferenceSerializer;
- private final Function<MailReferenceDTO, Pair<EnqueueId, Mail>> mailLoader;
+ private final Function<MailReferenceDTO, MailWithEnqueueId> mailLoader;
private final MailQueueView.Factory mailQueueViewFactory;
private final Clock clock;
private final MailQueueItemDecoratorFactory decoratorFactory;
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 28/28: JAMES-2774 log when fail to dispatch in
EventDispatcher
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 581b64bff46f52e61a7f91ef893396eaa336fe9e
Author: Tran Tien Duc <dt...@linagora.com>
AuthorDate: Fri Jun 21 17:28:46 2019 +0700
JAMES-2774 log when fail to dispatch in EventDispatcher
---
.../src/main/java/org/apache/james/mailbox/events/EventDispatcher.java | 1 +
1 file changed, 1 insertion(+)
diff --git a/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/EventDispatcher.java b/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/EventDispatcher.java
index 311d6ed..b9d7884 100644
--- a/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/EventDispatcher.java
+++ b/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/EventDispatcher.java
@@ -81,6 +81,7 @@ class EventDispatcher {
dispatchToLocalListeners(event, keys),
dispatchToRemoteListeners(serializeEvent(event), keys))
.subscribeOn(Schedulers.elastic())
+ .doOnError(throwable -> LOGGER.error("error while dispatching event", throwable))
.then()
.subscribeWith(MonoProcessor.create());
}
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org