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 2018/09/26 02:23:37 UTC
[1/5] james-project git commit: JAMES-2544 RabbitMQ browse
`MailQueueView` API and cassandra table definitions
Repository: james-project
Updated Branches:
refs/heads/master 9e66ddb13 -> b70ca1a7c
JAMES-2544 RabbitMQ browse `MailQueueView` API and cassandra table definitions
Project: http://git-wip-us.apache.org/repos/asf/james-project/repo
Commit: http://git-wip-us.apache.org/repos/asf/james-project/commit/0523acf4
Tree: http://git-wip-us.apache.org/repos/asf/james-project/tree/0523acf4
Diff: http://git-wip-us.apache.org/repos/asf/james-project/diff/0523acf4
Branch: refs/heads/master
Commit: 0523acf435ff5985cc39a1e5b0454d0bd45a8285
Parents: 9e66ddb
Author: duc <dt...@linagora.com>
Authored: Wed Sep 12 20:12:47 2018 +0700
Committer: Benoit Tellier <bt...@linagora.com>
Committed: Wed Sep 26 09:21:47 2018 +0700
----------------------------------------------------------------------
server/queue/queue-rabbitmq/pom.xml | 1 -
.../james/queue/rabbitmq/MailQueueName.java | 6 +-
.../queue/rabbitmq/view/api/MailQueueView.java | 36 +++++
.../CassandraMailQueueViewConfiguration.java | 91 +++++++++++
.../cassandra/CassandraMailQueueViewModule.java | 121 ++++++++++++++
.../view/cassandra/model/BucketedSlices.java | 129 +++++++++++++++
.../view/cassandra/model/EnqueuedMail.java | 156 +++++++++++++++++++
.../rabbitmq/view/cassandra/model/MailKey.java | 48 ++++++
.../cassandra/model/BucketedSlicesTest.java | 69 ++++++++
9 files changed, 653 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/james-project/blob/0523acf4/server/queue/queue-rabbitmq/pom.xml
----------------------------------------------------------------------
diff --git a/server/queue/queue-rabbitmq/pom.xml b/server/queue/queue-rabbitmq/pom.xml
index 2bc2a58..47b958b 100644
--- a/server/queue/queue-rabbitmq/pom.xml
+++ b/server/queue/queue-rabbitmq/pom.xml
@@ -50,7 +50,6 @@
<dependency>
<groupId>${james.groupId}</groupId>
<artifactId>apache-james-backends-cassandra</artifactId>
- <scope>test</scope>
</dependency>
<dependency>
<groupId>${james.groupId}</groupId>
http://git-wip-us.apache.org/repos/asf/james-project/blob/0523acf4/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailQueueName.java
----------------------------------------------------------------------
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailQueueName.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailQueueName.java
index e059e3f..5c060de 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailQueueName.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/MailQueueName.java
@@ -26,7 +26,7 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.MoreObjects;
import com.google.common.base.Preconditions;
-final class MailQueueName {
+public final class MailQueueName {
static class WorkQueueName {
static Optional<WorkQueueName> fromString(String name) {
@@ -114,7 +114,7 @@ final class MailQueueName {
private static final String EXCHANGE_PREFIX = PREFIX + "-exchange-";
@VisibleForTesting static final String WORKQUEUE_PREFIX = PREFIX + "-workqueue-";
- static MailQueueName fromString(String name) {
+ public static MailQueueName fromString(String name) {
Preconditions.checkNotNull(name);
return new MailQueueName(name);
}
@@ -130,7 +130,7 @@ final class MailQueueName {
this.name = name;
}
- String asString() {
+ public String asString() {
return name;
}
http://git-wip-us.apache.org/repos/asf/james-project/blob/0523acf4/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/api/MailQueueView.java
----------------------------------------------------------------------
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
new file mode 100644
index 0000000..0d70239
--- /dev/null
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/api/MailQueueView.java
@@ -0,0 +1,36 @@
+/****************************************************************
+ * 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.view.api;
+
+import java.util.concurrent.CompletableFuture;
+
+import org.apache.james.queue.api.ManageableMailQueue;
+import org.apache.mailet.Mail;
+
+public interface MailQueueView {
+
+ CompletableFuture<Void> storeMail(Mail mail);
+
+ CompletableFuture<Void> deleteMail(Mail mail);
+
+ ManageableMailQueue.MailQueueIterator browse();
+
+ long getSize();
+}
http://git-wip-us.apache.org/repos/asf/james-project/blob/0523acf4/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueViewConfiguration.java
----------------------------------------------------------------------
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueViewConfiguration.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueViewConfiguration.java
new file mode 100644
index 0000000..b138887
--- /dev/null
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueViewConfiguration.java
@@ -0,0 +1,91 @@
+/****************************************************************
+ * 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.view.cassandra;
+
+import java.time.Duration;
+
+import com.google.common.base.Preconditions;
+
+public class CassandraMailQueueViewConfiguration {
+ interface Builder {
+ @FunctionalInterface
+ interface RequireBucketCount {
+ RequireUpdateBrowseStartPace bucketCount(int bucketCount);
+ }
+
+ @FunctionalInterface
+ interface RequireUpdateBrowseStartPace {
+ RequireSliceWindow updateBrowseStartPace(int updateBrowseStartPace);
+ }
+
+ @FunctionalInterface
+ interface RequireSliceWindow {
+ ReadyToBuild sliceWindow(Duration sliceWindow);
+ }
+
+ class ReadyToBuild {
+ private final int bucketCount;
+ private final int updateBrowseStartPace;
+ private final Duration sliceWindow;
+
+ private ReadyToBuild(int bucketCount, int updateBrowseStartPace, Duration sliceWindow) {
+ this.bucketCount = bucketCount;
+ this.updateBrowseStartPace = updateBrowseStartPace;
+ this.sliceWindow = sliceWindow;
+ }
+
+ public CassandraMailQueueViewConfiguration build() {
+ Preconditions.checkNotNull(sliceWindow, "'sliceWindow' is compulsory");
+ Preconditions.checkState(bucketCount > 0, "'bucketCount' needs to be a strictly positive integer");
+ Preconditions.checkState(updateBrowseStartPace > 0, "'updateBrowseStartPace' needs to be a strictly positive integer");
+
+ return new CassandraMailQueueViewConfiguration(bucketCount, updateBrowseStartPace, sliceWindow);
+ }
+ }
+ }
+
+ public static Builder.RequireBucketCount builder() {
+ return bucketCount -> updateBrowseStartPace -> sliceWindow -> new Builder.ReadyToBuild(bucketCount, updateBrowseStartPace, sliceWindow);
+ }
+
+ private final int bucketCount;
+ private final int updateBrowseStartPace;
+ private final Duration sliceWindow;
+
+ private CassandraMailQueueViewConfiguration(int bucketCount,
+ int updateBrowseStartPace,
+ Duration sliceWindow) {
+ this.bucketCount = bucketCount;
+ this.updateBrowseStartPace = updateBrowseStartPace;
+ this.sliceWindow = sliceWindow;
+ }
+
+ public int getUpdateBrowseStartPace() {
+ return updateBrowseStartPace;
+ }
+
+ public int getBucketCount() {
+ return bucketCount;
+ }
+
+ public Duration getSliceWindow() {
+ return sliceWindow;
+ }
+}
http://git-wip-us.apache.org/repos/asf/james-project/blob/0523acf4/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueViewModule.java
----------------------------------------------------------------------
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
new file mode 100644
index 0000000..6460eb4
--- /dev/null
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueViewModule.java
@@ -0,0 +1,121 @@
+/****************************************************************
+ * 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.view.cassandra;
+
+import static com.datastax.driver.core.DataType.blob;
+import static com.datastax.driver.core.DataType.cint;
+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.schemabuilder.SchemaBuilder.frozen;
+
+import org.apache.james.backends.cassandra.components.CassandraModule;
+
+public interface CassandraMailQueueViewModule {
+
+ interface EnqueuedMailsTable {
+ String TABLE_NAME = "enqueuedMails";
+
+ String QUEUE_NAME = "queueName";
+ String TIME_RANGE_START = "timeRangeStart";
+ String BUCKET_ID = "bucketId";
+
+ String ENQUEUED_TIME = "enqueuedTime";
+ String MAIL_KEY = "mailKey";
+ String HEADER_BLOB_ID = "headerBlobId";
+ String BODY_BLOB_ID = "bodyBlobId";
+ String STATE = "state";
+ String SENDER = "sender";
+ String RECIPIENTS = "recipients";
+ String ATTRIBUTES = "attributes";
+ String ERROR_MESSAGE = "errorMessage";
+ String REMOTE_HOST = "remoteHost";
+ String REMOTE_ADDR = "remoteAddr";
+ String LAST_UPDATED = "lastUpdated";
+ String PER_RECIPIENT_SPECIFIC_HEADERS = "perRecipientSpecificHeaders";
+
+ String HEADER_TYPE = "header";
+ String HEADER_NAME = "headerName";
+ String HEADER_VALUE = "headerValue";
+ }
+
+ interface BrowseStartTable {
+ String TABLE_NAME = "browseStart";
+
+ String QUEUE_NAME = "queueName";
+ String BROWSE_START = "browseStart";
+ }
+
+ interface DeletedMailTable {
+ String TABLE_NAME = "deletedMails";
+
+ String QUEUE_NAME = "queueName";
+ String MAIL_KEY = "mailKey";
+ }
+
+ CassandraModule MODULE = CassandraModule
+ .type(EnqueuedMailsTable.HEADER_TYPE)
+ .statement(statement -> statement
+ .addColumn(EnqueuedMailsTable.HEADER_NAME, text())
+ .addColumn(EnqueuedMailsTable.HEADER_VALUE, text()))
+ .table(EnqueuedMailsTable.TABLE_NAME)
+ .comment("store enqueued mails, if a mail is enqueued into a mail queue, it also being stored in this table," +
+ " when a mail is dequeued from a mail queue, the record associated with that mail still available in this" +
+ " table and will not be deleted immediately regarding to the performance impacts," +
+ " but after some scheduled tasks")
+ .options(options -> options)
+ .statement(statement -> statement
+ .addPartitionKey(EnqueuedMailsTable.QUEUE_NAME, text())
+ .addPartitionKey(EnqueuedMailsTable.TIME_RANGE_START, timestamp())
+ .addPartitionKey(EnqueuedMailsTable.BUCKET_ID, cint())
+ .addClusteringColumn(EnqueuedMailsTable.MAIL_KEY, text())
+ .addColumn(EnqueuedMailsTable.ENQUEUED_TIME, timestamp())
+ .addColumn(EnqueuedMailsTable.STATE, text())
+ .addColumn(EnqueuedMailsTable.HEADER_BLOB_ID, text())
+ .addColumn(EnqueuedMailsTable.BODY_BLOB_ID, text())
+ .addColumn(EnqueuedMailsTable.ATTRIBUTES, map(text(), blob()))
+ .addColumn(EnqueuedMailsTable.ERROR_MESSAGE, text())
+ .addColumn(EnqueuedMailsTable.SENDER, text())
+ .addColumn(EnqueuedMailsTable.RECIPIENTS, list(text()))
+ .addColumn(EnqueuedMailsTable.REMOTE_HOST, text())
+ .addColumn(EnqueuedMailsTable.REMOTE_ADDR, text())
+ .addColumn(EnqueuedMailsTable.LAST_UPDATED, timestamp())
+ .addUDTMapColumn(EnqueuedMailsTable.PER_RECIPIENT_SPECIFIC_HEADERS, text(), frozen(EnqueuedMailsTable.HEADER_TYPE)))
+
+ .table(BrowseStartTable.TABLE_NAME)
+ .comment("this table allows to find the starting point of iteration from the table: "
+ + EnqueuedMailsTable.TABLE_NAME + " in order to make a browse operations through mail queues")
+ .options(options -> options)
+ .statement(statement -> statement
+ .addPartitionKey(BrowseStartTable.QUEUE_NAME, text())
+ .addColumn(BrowseStartTable.BROWSE_START, timestamp()))
+
+ .table(DeletedMailTable.TABLE_NAME)
+ .comment("this table stores the dequeued mails, while browsing mail from table: "
+ + EnqueuedMailsTable.TABLE_NAME + " we need to filter out mails have been dequeued by checking their " +
+ "existence in this table")
+ .options(options -> options)
+ .statement(statement -> statement
+ .addPartitionKey(DeletedMailTable.QUEUE_NAME, text())
+ .addPartitionKey(DeletedMailTable.MAIL_KEY, text()))
+
+ .build();
+}
http://git-wip-us.apache.org/repos/asf/james-project/blob/0523acf4/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/model/BucketedSlices.java
----------------------------------------------------------------------
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/model/BucketedSlices.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/model/BucketedSlices.java
new file mode 100644
index 0000000..f7d5ac1
--- /dev/null
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/model/BucketedSlices.java
@@ -0,0 +1,129 @@
+/****************************************************************
+ * 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.view.cassandra.model;
+
+import java.time.Instant;
+import java.util.Objects;
+import java.util.stream.LongStream;
+import java.util.stream.Stream;
+
+import com.google.common.base.Preconditions;
+
+public class BucketedSlices {
+
+ public static class BucketId {
+
+ public static BucketId of(int bucketId) {
+ return new BucketId(bucketId);
+ }
+
+ private final int value;
+
+ private BucketId(int value) {
+ Preconditions.checkArgument(value >= 0, "sliceWindowSizeInSecond should not be negative");
+
+ this.value = value;
+ }
+
+ public int getValue() {
+ return value;
+ }
+
+ @Override
+ public final boolean equals(Object o) {
+ if (o instanceof BucketId) {
+ BucketId bucketId = (BucketId) o;
+
+ return Objects.equals(this.value, bucketId.value);
+ }
+ return false;
+ }
+
+ @Override
+ public final int hashCode() {
+ return Objects.hash(value);
+ }
+ }
+
+ public static class Slice {
+
+ public static Slice of(Instant sliceStartInstant, long sliceWindowSizeInSecond) {
+ return new Slice(sliceStartInstant, sliceWindowSizeInSecond);
+ }
+
+ public static Stream<Slice> allSlicesTill(Slice firstSlice, Instant endAt) {
+ long sliceCount = calculateSliceCount(firstSlice, endAt);
+ long startAtSeconds = firstSlice.getStartSliceInstant().getEpochSecond();
+ long sliceWindowSizeInSecond = firstSlice.getSliceWindowSizeInSecond();
+
+ return LongStream.range(0, sliceCount)
+ .map(slicePosition -> startAtSeconds + sliceWindowSizeInSecond * slicePosition)
+ .mapToObj(Instant::ofEpochSecond)
+ .map(sliceStartInstant -> Slice.of(sliceStartInstant, firstSlice.getSliceWindowSizeInSecond()));
+ }
+
+ private static long calculateSliceCount(Slice firstSlice, Instant endAt) {
+ long startAtSeconds = firstSlice.getStartSliceInstant().getEpochSecond();
+ long endAtSeconds = endAt.getEpochSecond();
+ long timeDiffInSecond = endAtSeconds - startAtSeconds;
+
+ if (timeDiffInSecond < 0) {
+ return 0;
+ } else {
+ return (timeDiffInSecond / firstSlice.sliceWindowSizeInSecond) + 1;
+ }
+ }
+
+ private final Instant startSliceInstant;
+ private final long sliceWindowSizeInSecond;
+
+ private Slice(Instant startSliceInstant, long sliceWindowSizeInSecond) {
+ Preconditions.checkNotNull(startSliceInstant);
+ Preconditions.checkArgument(sliceWindowSizeInSecond > 0, "sliceWindowSizeInSecond should be positive");
+
+ this.startSliceInstant = startSliceInstant;
+ this.sliceWindowSizeInSecond = sliceWindowSizeInSecond;
+ }
+
+ public Instant getStartSliceInstant() {
+ return startSliceInstant;
+ }
+
+ public long getSliceWindowSizeInSecond() {
+ return sliceWindowSizeInSecond;
+ }
+
+ @Override
+ public final boolean equals(Object o) {
+ if (o instanceof Slice) {
+ Slice slice = (Slice) o;
+
+ return Objects.equals(this.sliceWindowSizeInSecond, slice.sliceWindowSizeInSecond)
+ && Objects.equals(this.startSliceInstant, slice.startSliceInstant);
+ }
+ return false;
+ }
+
+ @Override
+ public final int hashCode() {
+ return Objects.hash(startSliceInstant, sliceWindowSizeInSecond);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/james-project/blob/0523acf4/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/model/EnqueuedMail.java
----------------------------------------------------------------------
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/model/EnqueuedMail.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/model/EnqueuedMail.java
new file mode 100644
index 0000000..6e022b3
--- /dev/null
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/model/EnqueuedMail.java
@@ -0,0 +1,156 @@
+/****************************************************************
+ * 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.view.cassandra.model;
+
+import java.time.Instant;
+import java.util.Comparator;
+import java.util.Objects;
+
+import org.apache.james.queue.rabbitmq.MailQueueName;
+import org.apache.mailet.Mail;
+
+public class EnqueuedMail {
+
+ public interface Builder {
+
+ @FunctionalInterface
+ interface RequireMail {
+ RequireBucketId mail(Mail mail);
+ }
+
+ @FunctionalInterface
+ interface RequireBucketId {
+ RequireTimeRangeStart bucketId(BucketedSlices.BucketId bucketId);
+ }
+
+ @FunctionalInterface
+ interface RequireTimeRangeStart {
+ RequireEnqueuedTime timeRangeStart(Instant timeRangeStart);
+ }
+
+ @FunctionalInterface
+ interface RequireEnqueuedTime {
+ RequireMailKey enqueuedTime(Instant enqueuedTime);
+ }
+
+ @FunctionalInterface
+ interface RequireMailKey {
+ RequireMailQueueName mailKey(MailKey mailKey);
+ }
+
+ @FunctionalInterface
+ interface RequireMailQueueName {
+ LastStage mailQueueName(MailQueueName mailQueueName);
+ }
+
+ class LastStage {
+ private Mail mail;
+ private BucketedSlices.BucketId bucketId;
+ private Instant timeRangeStart;
+ private Instant enqueuedTime;
+ private MailKey mailKey;
+ private MailQueueName mailQueueName;
+
+ private LastStage(Mail mail, BucketedSlices.BucketId bucketId,
+ Instant timeRangeStart, Instant enqueuedTime,
+ MailKey mailKey, MailQueueName mailQueueName) {
+ this.mail = mail;
+ this.bucketId = bucketId;
+ this.timeRangeStart = timeRangeStart;
+ this.enqueuedTime = enqueuedTime;
+ this.mailKey = mailKey;
+ this.mailQueueName = mailQueueName;
+ }
+
+ public EnqueuedMail build() {
+ return new EnqueuedMail(mail, bucketId, timeRangeStart, enqueuedTime, mailKey, mailQueueName);
+ }
+ }
+ }
+
+ public static Builder.RequireMail builder() {
+ return mail -> bucketId -> timeRangeStart -> enqueuedTime -> mailKey -> mailQueueName ->
+ new Builder.LastStage(mail, bucketId, timeRangeStart, enqueuedTime, mailKey, mailQueueName);
+ }
+
+ public static Comparator<EnqueuedMail> getEnqueuedTimeComparator() {
+ return Comparator.comparing(EnqueuedMail::getEnqueuedTime);
+ }
+
+ private final Mail mail;
+ private final BucketedSlices.BucketId bucketId;
+ private final Instant timeRangeStart;
+ private final Instant enqueuedTime;
+ private final MailKey mailKey;
+ private final MailQueueName mailQueueName;
+
+ private EnqueuedMail(Mail mail, BucketedSlices.BucketId bucketId, Instant timeRangeStart,
+ Instant enqueuedTime, MailKey mailKey, MailQueueName mailQueueName) {
+ this.mail = mail;
+ this.bucketId = bucketId;
+ this.timeRangeStart = timeRangeStart;
+ this.enqueuedTime = enqueuedTime;
+ this.mailKey = mailKey;
+ this.mailQueueName = mailQueueName;
+ }
+
+ public Mail getMail() {
+ return mail;
+ }
+
+ public BucketedSlices.BucketId getBucketId() {
+ return bucketId;
+ }
+
+ public MailKey getMailKey() {
+ return mailKey;
+ }
+
+ public MailQueueName getMailQueueName() {
+ return mailQueueName;
+ }
+
+ public Instant getTimeRangeStart() {
+ return timeRangeStart;
+ }
+
+ public Instant getEnqueuedTime() {
+ return enqueuedTime;
+ }
+
+ @Override
+ public final boolean equals(Object o) {
+ if (o instanceof EnqueuedMail) {
+ EnqueuedMail that = (EnqueuedMail) o;
+
+ return Objects.equals(this.bucketId, that.bucketId)
+ && Objects.equals(this.mail, that.mail)
+ && Objects.equals(this.timeRangeStart, that.timeRangeStart)
+ && Objects.equals(this.mailKey, that.mailKey)
+ && Objects.equals(this.mailQueueName, that.mailQueueName);
+ }
+ return false;
+ }
+
+ @Override
+ public final int hashCode() {
+ return Objects.hash(mail, bucketId, timeRangeStart, mailKey, mailQueueName);
+ }
+}
http://git-wip-us.apache.org/repos/asf/james-project/blob/0523acf4/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/model/MailKey.java
----------------------------------------------------------------------
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/view/cassandra/model/MailKey.java
new file mode 100644
index 0000000..c39dc16
--- /dev/null
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/model/MailKey.java
@@ -0,0 +1,48 @@
+/****************************************************************
+ * 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.view.cassandra.model;
+
+import org.apache.mailet.Mail;
+
+import com.google.common.base.Preconditions;
+
+public class MailKey {
+
+ public static MailKey fromMail(Mail mail) {
+ return of(mail.getName());
+ }
+
+ public static MailKey of(String mailKey) {
+ return new MailKey(mailKey);
+ }
+
+ private final String mailKey;
+
+ private MailKey(String mailKey) {
+ Preconditions.checkNotNull(mailKey);
+ Preconditions.checkArgument(!mailKey.isEmpty());
+
+ this.mailKey = mailKey;
+ }
+
+ public String getMailKey() {
+ return mailKey;
+ }
+}
http://git-wip-us.apache.org/repos/asf/james-project/blob/0523acf4/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/model/BucketedSlicesTest.java
----------------------------------------------------------------------
diff --git a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/model/BucketedSlicesTest.java b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/model/BucketedSlicesTest.java
new file mode 100644
index 0000000..110d870
--- /dev/null
+++ b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/model/BucketedSlicesTest.java
@@ -0,0 +1,69 @@
+/****************************************************************
+ * 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.view.cassandra.model;
+
+import static org.apache.james.queue.rabbitmq.view.cassandra.model.BucketedSlices.Slice;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.time.Instant;
+import java.util.stream.Stream;
+
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+
+class BucketedSlicesTest {
+
+ private static final long ONE_HOUR_IN_SECONDS = 3600;
+
+ private static final Instant FIRST_SLICE_INSTANT = Instant.parse("2018-05-20T12:00:00.000Z");
+ private static final Instant FIRST_SLICE_INSTANT_NEXT_HOUR = FIRST_SLICE_INSTANT.plusSeconds(ONE_HOUR_IN_SECONDS);
+ private static final Instant FIRST_SLICE_INSTANT_NEXT_TWO_HOUR = FIRST_SLICE_INSTANT.plusSeconds(ONE_HOUR_IN_SECONDS * 2);
+
+ private static final Slice FIRST_SLICE = Slice.of(FIRST_SLICE_INSTANT, ONE_HOUR_IN_SECONDS);
+ private static final Slice FIRST_SLICE_NEXT_TWO_HOUR = Slice.of(FIRST_SLICE_INSTANT_NEXT_TWO_HOUR, ONE_HOUR_IN_SECONDS);
+
+ @Nested
+ class Validation {
+ }
+
+ @Test
+ void allSlicesTillShouldReturnOnlyFirstSliceWhenEndAtInTheSameInterval() {
+ assertThat(Slice.allSlicesTill(FIRST_SLICE, FIRST_SLICE_INSTANT.plusSeconds(3599)))
+ .containsOnly(FIRST_SLICE);
+ }
+
+ @Test
+ void allSlicesTillShouldReturnAllSlicesBetweenStartAndEndAt() {
+ Stream<Slice> allSlices = Slice.allSlicesTill(FIRST_SLICE, FIRST_SLICE_INSTANT_NEXT_TWO_HOUR.plusSeconds(3599));
+
+ assertThat(allSlices)
+ .containsExactly(
+ FIRST_SLICE,
+ Slice.of(FIRST_SLICE_INSTANT_NEXT_HOUR, ONE_HOUR_IN_SECONDS),
+ Slice.of(FIRST_SLICE_INSTANT_NEXT_TWO_HOUR, ONE_HOUR_IN_SECONDS));
+ }
+
+ @Test
+ void allSlicesTillShouldReturnEmptyIfEndAtBeforeStartSlice() {
+ Stream<Slice> allSlices = Slice.allSlicesTill(FIRST_SLICE_NEXT_TWO_HOUR, FIRST_SLICE_INSTANT);
+
+ assertThat(allSlices).isEmpty();
+ }
+}
\ 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
[5/5] james-project git commit: JAMES-2544 Move enqueuedTime to
MailQueueView level for storing mails
Posted by bt...@apache.org.
JAMES-2544 Move enqueuedTime to MailQueueView level for storing mails
Project: http://git-wip-us.apache.org/repos/asf/james-project/repo
Commit: http://git-wip-us.apache.org/repos/asf/james-project/commit/6be6b165
Tree: http://git-wip-us.apache.org/repos/asf/james-project/tree/6be6b165
Diff: http://git-wip-us.apache.org/repos/asf/james-project/diff/6be6b165
Branch: refs/heads/master
Commit: 6be6b16502634d5d6e02438ef9464f719eaa041b
Parents: 3a53806
Author: duc <dt...@linagora.com>
Authored: Mon Sep 17 14:55:26 2018 +0700
Committer: Benoit Tellier <bt...@linagora.com>
Committed: Wed Sep 26 09:22:06 2018 +0700
----------------------------------------------------------------------
.../apache/james/queue/rabbitmq/Dequeuer.java | 12 +++++--
.../apache/james/queue/rabbitmq/Enqueuer.java | 10 +++++-
.../james/queue/rabbitmq/RabbitMQMailQueue.java | 26 ++++++++------
.../queue/rabbitmq/view/api/MailQueueView.java | 3 +-
.../cassandra/CassandraMailQueueMailStore.java | 8 ++---
.../view/cassandra/CassandraMailQueueView.java | 5 +--
.../queue/rabbitmq/RabbitMQMailQueueTest.java | 36 +++++---------------
.../rabbitmq/RabbitMqMailQueueFactoryTest.java | 3 +-
8 files changed, 53 insertions(+), 50 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/james-project/blob/6be6b165/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/Dequeuer.java
----------------------------------------------------------------------
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 7b81bb6..81f4c61 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
@@ -30,6 +30,7 @@ import java.util.function.Function;
import org.apache.james.metrics.api.Metric;
import org.apache.james.metrics.api.MetricFactory;
import org.apache.james.queue.api.MailQueue;
+import org.apache.james.queue.rabbitmq.view.api.MailQueueView;
import org.apache.mailet.Mail;
import com.github.fge.lambdas.Throwing;
@@ -56,7 +57,7 @@ class Dequeuer {
}
@Override
- public void done(boolean success) throws MailQueue.MailQueueException {
+ public void done(boolean success) {
ack.accept(success);
}
}
@@ -68,16 +69,20 @@ class Dequeuer {
private final Function<MailReferenceDTO, Mail> mailLoader;
private final Metric dequeueMetric;
private final MailReferenceSerializer mailReferenceSerializer;
+ private final MailQueueView mailQueueView;
- Dequeuer(MailQueueName name, RabbitClient rabbitClient, Function<MailReferenceDTO, Mail> mailLoader, MailReferenceSerializer serializer, MetricFactory metricFactory) {
+ Dequeuer(MailQueueName name, RabbitClient rabbitClient, Function<MailReferenceDTO, Mail> mailLoader,
+ MailReferenceSerializer serializer, MetricFactory metricFactory,
+ MailQueueView mailQueueView) {
this.name = name;
this.rabbitClient = rabbitClient;
this.mailLoader = mailLoader;
this.mailReferenceSerializer = serializer;
+ this.mailQueueView = mailQueueView;
this.dequeueMetric = metricFactory.generate(DEQUEUED_METRIC_NAME_PREFIX + name.asString());
}
- MailQueue.MailQueueItem deQueue() throws MailQueue.MailQueueException {
+ MailQueue.MailQueueItem deQueue() {
return pollChannel()
.thenApply(Throwing.function(this::loadItem).sneakyThrow())
.join();
@@ -95,6 +100,7 @@ class Dequeuer {
if (success) {
dequeueMetric.increment();
rabbitClient.ack(deliveryTag);
+ mailQueueView.deleteMail(mail).join();
} else {
rabbitClient.nack(deliveryTag);
}
http://git-wip-us.apache.org/repos/asf/james-project/blob/6be6b165/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/Enqueuer.java
----------------------------------------------------------------------
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 84d4921..f1d4ca4 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
@@ -21,6 +21,7 @@ package org.apache.james.queue.rabbitmq;
import static org.apache.james.queue.api.MailQueue.ENQUEUED_METRIC_NAME_PREFIX;
+import java.time.Clock;
import java.util.concurrent.CompletableFuture;
import javax.mail.MessagingException;
@@ -31,6 +32,7 @@ import org.apache.james.blob.mail.MimeMessagePartsId;
import org.apache.james.metrics.api.Metric;
import org.apache.james.metrics.api.MetricFactory;
import org.apache.james.queue.api.MailQueue;
+import org.apache.james.queue.rabbitmq.view.api.MailQueueView;
import org.apache.mailet.Mail;
import com.fasterxml.jackson.core.JsonProcessingException;
@@ -42,13 +44,18 @@ class Enqueuer {
private final Store<MimeMessage, MimeMessagePartsId> mimeMessageStore;
private final MailReferenceSerializer mailReferenceSerializer;
private final Metric enqueueMetric;
+ private final MailQueueView mailQueueView;
+ private final Clock clock;
Enqueuer(MailQueueName name, RabbitClient rabbitClient, Store<MimeMessage, MimeMessagePartsId> mimeMessageStore,
- MailReferenceSerializer serializer, MetricFactory metricFactory) {
+ MailReferenceSerializer serializer, MetricFactory metricFactory,
+ MailQueueView mailQueueView, Clock clock) {
this.name = name;
this.rabbitClient = rabbitClient;
this.mimeMessageStore = mimeMessageStore;
this.mailReferenceSerializer = serializer;
+ this.mailQueueView = mailQueueView;
+ this.clock = clock;
this.enqueueMetric = metricFactory.generate(ENQUEUED_METRIC_NAME_PREFIX + name.asString());
}
@@ -56,6 +63,7 @@ class Enqueuer {
saveMail(mail)
.thenAccept(Throwing.<MimeMessagePartsId>consumer(partsId -> publishReferenceToRabbit(mail, partsId)).sneakyThrow())
.thenRun(enqueueMetric::increment)
+ .thenCompose(any -> mailQueueView.storeMail(clock.instant(), mail))
.join();
}
http://git-wip-us.apache.org/repos/asf/james-project/blob/6be6b165/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueue.java
----------------------------------------------------------------------
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueue.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueue.java
index 818a83b..c653943 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueue.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueue.java
@@ -19,6 +19,7 @@
package org.apache.james.queue.rabbitmq;
+import java.time.Clock;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
@@ -50,25 +51,29 @@ public class RabbitMQMailQueue implements ManageableMailQueue {
private final MailReferenceSerializer mailReferenceSerializer;
private final Function<MailReferenceDTO, Mail> mailLoader;
private final MailQueueView mailQueueView;
+ private final Clock clock;
@Inject
@VisibleForTesting Factory(MetricFactory metricFactory, RabbitClient rabbitClient,
Store<MimeMessage, MimeMessagePartsId> mimeMessageStore,
BlobId.Factory blobIdFactory,
- MailQueueView mailQueueView) {
+ MailQueueView mailQueueView,
+ Clock clock) {
this.metricFactory = metricFactory;
this.rabbitClient = rabbitClient;
this.mimeMessageStore = mimeMessageStore;
+ this.mailQueueView = mailQueueView;
+ this.clock = clock;
this.mailReferenceSerializer = new MailReferenceSerializer();
this.mailLoader = Throwing.function(new MailLoader(mimeMessageStore, blobIdFactory)::load).sneakyThrow();
- this.mailQueueView = mailQueueView;
}
RabbitMQMailQueue create(MailQueueName mailQueueName) {
return new RabbitMQMailQueue(metricFactory, mailQueueName,
- new Enqueuer(mailQueueName, rabbitClient, mimeMessageStore, mailReferenceSerializer, metricFactory),
- new Dequeuer(mailQueueName, rabbitClient, mailLoader, mailReferenceSerializer, metricFactory),
- mailQueueView);
+ new Enqueuer(mailQueueName, rabbitClient, mimeMessageStore, mailReferenceSerializer,
+ metricFactory, mailQueueView, clock),
+ new Dequeuer(mailQueueName, rabbitClient, mailLoader, mailReferenceSerializer,
+ metricFactory, mailQueueView), mailQueueView);
}
}
@@ -79,13 +84,12 @@ public class RabbitMQMailQueue implements ManageableMailQueue {
private final MailQueueView mailQueueView;
RabbitMQMailQueue(MetricFactory metricFactory, MailQueueName name,
- Enqueuer enqueuer, Dequeuer dequeuer, MailQueueView mailQueueView) {
-
+ Enqueuer enqueuer, Dequeuer dequeuer,
+ MailQueueView mailQueueView) {
+ this.metricFactory = metricFactory;
this.name = name;
this.enqueuer = enqueuer;
this.dequeuer = dequeuer;
-
- this.metricFactory = metricFactory;
this.mailQueueView = mailQueueView;
}
@@ -103,13 +107,13 @@ public class RabbitMQMailQueue implements ManageableMailQueue {
}
@Override
- public void enQueue(Mail mail) throws MailQueueException {
+ public void enQueue(Mail mail) {
metricFactory.runPublishingTimerMetric(ENQUEUED_TIMER_METRIC_NAME_PREFIX + name.asString(),
Throwing.runnable(() -> enqueuer.enQueue(mail)).sneakyThrow());
}
@Override
- public MailQueueItem deQueue() throws MailQueueException {
+ public MailQueueItem deQueue() {
return metricFactory.runPublishingTimerMetric(DEQUEUED_TIMER_METRIC_NAME_PREFIX + name.asString(),
Throwing.supplier(dequeuer::deQueue).sneakyThrow());
}
http://git-wip-us.apache.org/repos/asf/james-project/blob/6be6b165/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/api/MailQueueView.java
----------------------------------------------------------------------
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 0d70239..10ad42d 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
@@ -19,6 +19,7 @@
package org.apache.james.queue.rabbitmq.view.api;
+import java.time.Instant;
import java.util.concurrent.CompletableFuture;
import org.apache.james.queue.api.ManageableMailQueue;
@@ -26,7 +27,7 @@ import org.apache.mailet.Mail;
public interface MailQueueView {
- CompletableFuture<Void> storeMail(Mail mail);
+ CompletableFuture<Void> storeMail(Instant enqueuedTime, Mail mail);
CompletableFuture<Void> deleteMail(Mail mail);
http://git-wip-us.apache.org/repos/asf/james-project/blob/6be6b165/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueMailStore.java
----------------------------------------------------------------------
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueMailStore.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueMailStore.java
index ed696f2..a243407 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueMailStore.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueMailStore.java
@@ -53,8 +53,8 @@ class CassandraMailQueueMailStore {
this.initialInserted = ConcurrentHashMap.newKeySet();
}
- CompletableFuture<Void> storeMailInEnqueueTable(Mail mail, MailQueueName mailQueueName) {
- EnqueuedMail enqueuedMail = convertToEnqueuedMail(mail, mailQueueName);
+ CompletableFuture<Void> storeMailInEnqueueTable(Mail mail, MailQueueName mailQueueName, Instant enqueuedTime) {
+ EnqueuedMail enqueuedMail = convertToEnqueuedMail(mail, mailQueueName, enqueuedTime);
return enqueuedMailsDao.insert(enqueuedMail)
.thenCompose(any -> initBrowseStartIfNeeded(mailQueueName, enqueuedMail.getTimeRangeStart()));
@@ -73,12 +73,12 @@ class CassandraMailQueueMailStore {
.thenAccept(any -> initialInserted.add(mailQueueName));
}
- private EnqueuedMail convertToEnqueuedMail(Mail mail, MailQueueName mailQueueName) {
+ private EnqueuedMail convertToEnqueuedMail(Mail mail, MailQueueName mailQueueName, Instant enqueuedTime) {
return EnqueuedMail.builder()
.mail(mail)
.bucketId(computedBucketId(mail))
.timeRangeStart(currentSliceStartInstant())
- .enqueuedTime(Instant.now())
+ .enqueuedTime(enqueuedTime)
.mailKey(MailKey.fromMail(mail))
.mailQueueName(mailQueueName)
.build();
http://git-wip-us.apache.org/repos/asf/james-project/blob/6be6b165/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueView.java
----------------------------------------------------------------------
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 ee309e6..ad37b5b 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
@@ -19,6 +19,7 @@
package org.apache.james.queue.rabbitmq.view.cassandra;
+import java.time.Instant;
import java.util.concurrent.CompletableFuture;
import javax.inject.Inject;
@@ -66,8 +67,8 @@ public class CassandraMailQueueView implements MailQueueView {
}
@Override
- public CompletableFuture<Void> storeMail(Mail mail) {
- return storeHelper.storeMailInEnqueueTable(mail, mailQueueName);
+ public CompletableFuture<Void> storeMail(Instant enqueuedTime, Mail mail) {
+ return storeHelper.storeMailInEnqueueTable(mail, mailQueueName, enqueuedTime);
}
@Override
http://git-wip-us.apache.org/repos/asf/james-project/blob/6be6b165/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java
----------------------------------------------------------------------
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 92b384a..f1f31be 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
@@ -66,8 +66,6 @@ import org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModu
import org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewTestFactory;
import org.apache.james.util.streams.Iterators;
import org.apache.mailet.Mail;
-import org.junit.jupiter.api.AfterAll;
-import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
@@ -90,7 +88,6 @@ public class RabbitMQMailQueueTest implements ManageableMailQueueContract, MailQ
private static final Instant IN_SLICE_5 = IN_SLICE_1.plus(4, HOURS);
private static final Instant IN_SLICE_6 = IN_SLICE_1.plus(6, HOURS);
-
@RegisterExtension
static CassandraClusterExtension cassandraCluster = new CassandraClusterExtension(CassandraModule.aggregateModules(
CassandraBlobModule.MODULE,
@@ -115,7 +112,7 @@ public class RabbitMQMailQueueTest implements ManageableMailQueueContract, MailQ
.setPort(rabbitMQ.getAdminPort())
.build();
clock = mock(Clock.class);
- when(clock.instant()).thenReturn(Instant.parse("2007-12-03T10:15:30.00Z"));
+ when(clock.instant()).thenReturn(IN_SLICE_1);
ThreadLocalRandom random = ThreadLocalRandom.current();
MailQueueView mailQueueView = CassandraMailQueueViewTestFactory.factory(clock, random, cassandra.getConf(), cassandra.getTypesProvider(),
@@ -136,26 +133,16 @@ public class RabbitMQMailQueueTest implements ManageableMailQueueContract, MailQ
RabbitClient rabbitClient = new RabbitClient(new RabbitChannelPool(rabbitMQConnectionFactory));
RabbitMQMailQueue.Factory factory = new RabbitMQMailQueue.Factory(
- metricTestSystem.getSpyMetricFactory(),
- rabbitClient,
- mimeMessageStore,
- BLOB_ID_FACTORY,
- mailQueueView);
-
+ metricTestSystem.getSpyMetricFactory(),
+ rabbitClient,
+ mimeMessageStore,
+ BLOB_ID_FACTORY,
+ mailQueueView,
+ Clock.systemUTC());
RabbitMQManagementApi mqManagementApi = new RabbitMQManagementApi(rabbitManagementUri, new RabbitMQManagementCredentials("guest", "guest".toCharArray()));
mailQueueFactory = new RabbitMQMailQueueFactory(rabbitClient, mqManagementApi, factory);
}
- @AfterEach
- void tearDown(CassandraCluster cassandra) {
- cassandra.clearTables();
- }
-
- @AfterAll
- static void tearDownClass(CassandraCluster cassandra) {
- cassandra.closeCluster();
- }
-
@Override
public MailQueue getMailQueue() {
return mailQueueFactory.createQueue(SPOOL);
@@ -215,20 +202,15 @@ public class RabbitMQMailQueueTest implements ManageableMailQueueContract, MailQ
when(clock.instant()).thenReturn(IN_SLICE_6);
dequeueMails(5);
-
- dequeueMails(3);
- MailQueue.MailQueueItem item2_4 = mailQueue.deQueue();
- item2_4.done(false);
- dequeueMails(1);
-
dequeueMails(5);
+ dequeueMails(3);
Stream<String> names = Iterators.toStream(mailQueue.browse())
.map(ManageableMailQueue.MailQueueItemView::getMail)
.map(Mail::getName);
assertThat(names)
- .containsExactly("2-4", "5-1", "5-2", "5-3", "5-4", "5-5");
+ .containsExactly("3-4", "3-5", "5-1", "5-2", "5-3", "5-4", "5-5");
}
@Disabled
http://git-wip-us.apache.org/repos/asf/james-project/blob/6be6b165/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMqMailQueueFactoryTest.java
----------------------------------------------------------------------
diff --git a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMqMailQueueFactoryTest.java b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMqMailQueueFactoryTest.java
index bf7d37e..99587dd 100644
--- a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMqMailQueueFactoryTest.java
+++ b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMqMailQueueFactoryTest.java
@@ -24,6 +24,7 @@ import static org.mockito.Mockito.mock;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
+import java.time.Clock;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeoutException;
@@ -80,7 +81,7 @@ class RabbitMqMailQueueFactoryTest implements MailQueueFactoryContract<RabbitMQM
new AsyncRetryExecutor(Executors.newSingleThreadScheduledExecutor()));
RabbitClient rabbitClient = new RabbitClient(new RabbitChannelPool(rabbitMQConnectionFactory));
- RabbitMQMailQueue.Factory factory = new RabbitMQMailQueue.Factory(new NoopMetricFactory(), rabbitClient, mimeMessageStore, BLOB_ID_FACTORY, mailQueueView);
+ RabbitMQMailQueue.Factory factory = new RabbitMQMailQueue.Factory(new NoopMetricFactory(), rabbitClient, mimeMessageStore, BLOB_ID_FACTORY, mailQueueView, Clock.systemUTC());
RabbitMQManagementApi mqManagementApi = new RabbitMQManagementApi(rabbitManagementUri, new RabbitMQManagementCredentials("guest", "guest".toCharArray()));
mailQueueFactory = new RabbitMQMailQueueFactory(rabbitClient, mqManagementApi, factory);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[4/5] james-project git commit: JAMES-2544 Initialize RabbitMQ mail
queues when creating
Posted by bt...@apache.org.
JAMES-2544 Initialize RabbitMQ mail queues when creating
Project: http://git-wip-us.apache.org/repos/asf/james-project/repo
Commit: http://git-wip-us.apache.org/repos/asf/james-project/commit/b70ca1a7
Tree: http://git-wip-us.apache.org/repos/asf/james-project/tree/b70ca1a7
Diff: http://git-wip-us.apache.org/repos/asf/james-project/diff/b70ca1a7
Branch: refs/heads/master
Commit: b70ca1a7c8cc8e9af66f24b3e15ef73a8022d09f
Parents: 6be6b16
Author: Antoine Duprat <ad...@linagora.com>
Authored: Fri Sep 14 15:26:44 2018 +0200
Committer: Benoit Tellier <bt...@linagora.com>
Committed: Wed Sep 26 09:22:06 2018 +0700
----------------------------------------------------------------------
.../james/queue/rabbitmq/RabbitMQMailQueue.java | 2 ++
.../queue/rabbitmq/view/api/MailQueueView.java | 3 +++
.../cassandra/CassandraMailQueueMailStore.java | 19 +++----------------
.../view/cassandra/CassandraMailQueueView.java | 7 ++++++-
.../queue/rabbitmq/RabbitMQMailQueueTest.java | 9 +++++++++
.../CassandraMailQueueViewTestFactory.java | 9 +++++++++
6 files changed, 32 insertions(+), 17 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/james-project/blob/b70ca1a7/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueue.java
----------------------------------------------------------------------
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueue.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueue.java
index c653943..50853aa 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueue.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueue.java
@@ -69,6 +69,8 @@ public class RabbitMQMailQueue implements ManageableMailQueue {
}
RabbitMQMailQueue create(MailQueueName mailQueueName) {
+ mailQueueView.initialize(mailQueueName);
+
return new RabbitMQMailQueue(metricFactory, mailQueueName,
new Enqueuer(mailQueueName, rabbitClient, mimeMessageStore, mailReferenceSerializer,
metricFactory, mailQueueView, clock),
http://git-wip-us.apache.org/repos/asf/james-project/blob/b70ca1a7/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/api/MailQueueView.java
----------------------------------------------------------------------
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 10ad42d..0455ff8 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
@@ -23,10 +23,13 @@ import java.time.Instant;
import java.util.concurrent.CompletableFuture;
import org.apache.james.queue.api.ManageableMailQueue;
+import org.apache.james.queue.rabbitmq.MailQueueName;
import org.apache.mailet.Mail;
public interface MailQueueView {
+ void initialize(MailQueueName mailQueueName);
+
CompletableFuture<Void> storeMail(Instant enqueuedTime, Mail mail);
CompletableFuture<Void> deleteMail(Mail mail);
http://git-wip-us.apache.org/repos/asf/james-project/blob/b70ca1a7/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueMailStore.java
----------------------------------------------------------------------
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueMailStore.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueMailStore.java
index a243407..9339d6e 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueMailStore.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueMailStore.java
@@ -21,9 +21,7 @@ package org.apache.james.queue.rabbitmq.view.cassandra;
import java.time.Clock;
import java.time.Instant;
-import java.util.Set;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ConcurrentHashMap;
import javax.inject.Inject;
@@ -39,7 +37,6 @@ class CassandraMailQueueMailStore {
private final BrowseStartDAO browseStartDao;
private final CassandraMailQueueViewConfiguration configuration;
private final Clock clock;
- private final Set<MailQueueName> initialInserted;
@Inject
CassandraMailQueueMailStore(EnqueuedMailsDAO enqueuedMailsDao,
@@ -50,27 +47,17 @@ class CassandraMailQueueMailStore {
this.browseStartDao = browseStartDao;
this.configuration = configuration;
this.clock = clock;
- this.initialInserted = ConcurrentHashMap.newKeySet();
}
CompletableFuture<Void> storeMailInEnqueueTable(Mail mail, MailQueueName mailQueueName, Instant enqueuedTime) {
EnqueuedMail enqueuedMail = convertToEnqueuedMail(mail, mailQueueName, enqueuedTime);
- return enqueuedMailsDao.insert(enqueuedMail)
- .thenCompose(any -> initBrowseStartIfNeeded(mailQueueName, enqueuedMail.getTimeRangeStart()));
+ return enqueuedMailsDao.insert(enqueuedMail);
}
- private CompletableFuture<Void> initBrowseStartIfNeeded(MailQueueName mailQueueName, Instant sliceStartAt) {
- if (!initialInserted.contains(mailQueueName)) {
- return tryInsertBrowseStart(mailQueueName, sliceStartAt);
- }
- return CompletableFuture.completedFuture(null);
- }
-
- private CompletableFuture<Void> tryInsertBrowseStart(MailQueueName mailQueueName, Instant sliceStartAt) {
+ CompletableFuture<Void> initializeBrowseStart(MailQueueName mailQueueName) {
return browseStartDao
- .insertInitialBrowseStart(mailQueueName, sliceStartAt)
- .thenAccept(any -> initialInserted.add(mailQueueName));
+ .insertInitialBrowseStart(mailQueueName, currentSliceStartInstant());
}
private EnqueuedMail convertToEnqueuedMail(Mail mail, MailQueueName mailQueueName, Instant enqueuedTime) {
http://git-wip-us.apache.org/repos/asf/james-project/blob/b70ca1a7/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueView.java
----------------------------------------------------------------------
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 ad37b5b..1876f88 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
@@ -67,6 +67,12 @@ public class CassandraMailQueueView implements MailQueueView {
}
@Override
+ public void initialize(MailQueueName mailQueueName) {
+ storeHelper.initializeBrowseStart(mailQueueName)
+ .join();
+ }
+
+ @Override
public CompletableFuture<Void> storeMail(Instant enqueuedTime, Mail mail) {
return storeHelper.storeMailInEnqueueTable(mail, mailQueueName, enqueuedTime);
}
@@ -88,5 +94,4 @@ public class CassandraMailQueueView implements MailQueueView {
public long getSize() {
return Iterators.size(browse());
}
-
}
http://git-wip-us.apache.org/repos/asf/james-project/blob/b70ca1a7/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java
----------------------------------------------------------------------
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 f1f31be..8e495b4 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
@@ -213,6 +213,15 @@ public class RabbitMQMailQueueTest implements ManageableMailQueueContract, MailQ
.containsExactly("3-4", "3-5", "5-1", "5-2", "5-3", "5-4", "5-5");
}
+ @Test
+ void mailQueueShouldBeInitializedWhenCreating(CassandraCluster cassandra) {
+ String name = "myQueue";
+ mailQueueFactory.createQueue(name);
+
+ boolean initialized = CassandraMailQueueViewTestFactory.isInitialized(cassandra.getConf(), MailQueueName.fromString(name));
+ assertThat(initialized).isTrue();
+ }
+
@Disabled
@Override
public void clearShouldNotFailWhenBrowsingIterating() {
http://git-wip-us.apache.org/repos/asf/james-project/blob/b70ca1a7/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueViewTestFactory.java
----------------------------------------------------------------------
diff --git a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueViewTestFactory.java b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueViewTestFactory.java
index 95733c2..b37fb91 100644
--- a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueViewTestFactory.java
+++ b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueViewTestFactory.java
@@ -20,10 +20,12 @@
package org.apache.james.queue.rabbitmq.view.cassandra;
import java.time.Clock;
+import java.util.Optional;
import java.util.concurrent.ThreadLocalRandom;
import org.apache.james.backends.cassandra.init.CassandraTypesProvider;
import org.apache.james.backends.cassandra.utils.CassandraUtils;
+import org.apache.james.queue.rabbitmq.MailQueueName;
import com.datastax.driver.core.Session;
@@ -45,4 +47,11 @@ public class CassandraMailQueueViewTestFactory {
cassandraMailQueueBrowser,
cassandraMailQueueMailDelete);
}
+
+ public static boolean isInitialized(Session session, MailQueueName mailQueueName) {
+ BrowseStartDAO browseStartDao = new BrowseStartDAO(session);
+ return browseStartDao.findBrowseStart(mailQueueName)
+ .thenApply(Optional::isPresent)
+ .join();
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[3/5] james-project git commit: JAMES-2544 RabbitMQ browse Cassandra
DAOs and tests
Posted by bt...@apache.org.
JAMES-2544 RabbitMQ browse Cassandra DAOs and tests
Project: http://git-wip-us.apache.org/repos/asf/james-project/repo
Commit: http://git-wip-us.apache.org/repos/asf/james-project/commit/3a53806a
Tree: http://git-wip-us.apache.org/repos/asf/james-project/tree/3a53806a
Diff: http://git-wip-us.apache.org/repos/asf/james-project/diff/3a53806a
Branch: refs/heads/master
Commit: 3a53806a0184960ed50bbbc500d6392870a214a9
Parents: 0523acf
Author: duc <dt...@linagora.com>
Authored: Wed Sep 12 20:15:49 2018 +0700
Committer: Benoit Tellier <bt...@linagora.com>
Committed: Wed Sep 26 09:22:06 2018 +0700
----------------------------------------------------------------------
.../cassandra/utils/CassandraAsyncExecutor.java | 4 +
.../mail/CassandraAttachmentMapper.java | 3 +-
.../cassandra/mail/CassandraMailboxMapper.java | 14 +-
.../mail/CassandraMessageIdMapper.java | 9 +-
.../cassandra/mail/CassandraMessageMapper.java | 14 +-
.../james/util/CompletableFutureUtil.java | 30 +--
.../apache/james/util/FluentFutureStream.java | 127 ++++------
.../james/util/CompletableFutureUtilTest.java | 142 ++---------
.../james/util/FluentFutureStreamTest.java | 57 ++++-
.../queue/api/ManageableMailQueueContract.java | 4 +-
.../james/queue/rabbitmq/RabbitMQMailQueue.java | 47 +++-
.../rabbitmq/view/cassandra/BrowseStartDAO.java | 109 ++++++++
.../cassandra/CassandraMailQueueBrowser.java | 128 ++++++++++
.../cassandra/CassandraMailQueueMailDelete.java | 87 +++++++
.../cassandra/CassandraMailQueueMailStore.java | 98 ++++++++
.../view/cassandra/CassandraMailQueueView.java | 91 +++++++
.../view/cassandra/DeletedMailsDAO.java | 83 +++++++
.../view/cassandra/EnqueuedMailsDAO.java | 144 +++++++++++
.../view/cassandra/EnqueuedMailsDaoUtil.java | 196 +++++++++++++++
.../view/cassandra/model/BucketedSlices.java | 29 +--
.../view/cassandra/model/EnqueuedMail.java | 8 +-
.../rabbitmq/view/cassandra/model/MailKey.java | 17 ++
.../queue/rabbitmq/RabbitMQMailQueueTest.java | 248 +++++++++++++++++--
.../rabbitmq/RabbitMqMailQueueFactoryTest.java | 9 +-
.../view/cassandra/BrowseStartDAOTest.java | 86 +++++++
.../CassandraMailQueueViewTestFactory.java | 48 ++++
.../view/cassandra/DeletedMailsDAOTest.java | 108 ++++++++
.../view/cassandra/EnqueuedMailsDaoTest.java | 126 ++++++++++
.../cassandra/model/BucketedSlicesTest.java | 58 ++++-
.../view/cassandra/model/EnqueuedMailTest.java | 33 +++
.../view/cassandra/model/MailKeyTest.java | 33 +++
31 files changed, 1882 insertions(+), 308 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/james-project/blob/3a53806a/backends-common/cassandra/src/main/java/org/apache/james/backends/cassandra/utils/CassandraAsyncExecutor.java
----------------------------------------------------------------------
diff --git a/backends-common/cassandra/src/main/java/org/apache/james/backends/cassandra/utils/CassandraAsyncExecutor.java b/backends-common/cassandra/src/main/java/org/apache/james/backends/cassandra/utils/CassandraAsyncExecutor.java
index a137c66..7815643 100644
--- a/backends-common/cassandra/src/main/java/org/apache/james/backends/cassandra/utils/CassandraAsyncExecutor.java
+++ b/backends-common/cassandra/src/main/java/org/apache/james/backends/cassandra/utils/CassandraAsyncExecutor.java
@@ -62,4 +62,8 @@ public class CassandraAsyncExecutor {
.thenApply(Optional::ofNullable);
}
+ public CompletableFuture<Boolean> executeReturnExists(Statement statement) {
+ return executeSingleRow(statement)
+ .thenApply(Optional::isPresent);
+ }
}
http://git-wip-us.apache.org/repos/asf/james-project/blob/3a53806a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraAttachmentMapper.java
----------------------------------------------------------------------
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraAttachmentMapper.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraAttachmentMapper.java
index 7975b35..ecb43de 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraAttachmentMapper.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraAttachmentMapper.java
@@ -103,8 +103,7 @@ public class CassandraAttachmentMapper implements AttachmentMapper {
.map(id -> getAttachmentInternal(id)
.thenApply(finalValue -> logNotFound(id, finalValue)));
- return FluentFutureStream
- .ofOptionals(attachments)
+ return FluentFutureStream.of(attachments, FluentFutureStream::unboxOptional)
.collect(Guavate.toImmutableList());
}
http://git-wip-us.apache.org/repos/asf/james-project/blob/3a53806a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMailboxMapper.java
----------------------------------------------------------------------
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMailboxMapper.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMailboxMapper.java
index 5dee291..69c092d 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMailboxMapper.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMailboxMapper.java
@@ -79,8 +79,10 @@ public class CassandraMailboxMapper implements MailboxMapper {
@Override
public void delete(Mailbox mailbox) {
CassandraId mailboxId = (CassandraId) mailbox.getMailboxId();
- FluentFutureStream.ofFutures(mailboxPathDAO.delete(mailbox.generateAssociatedPath()), mailboxPathV2DAO.delete(mailbox.generateAssociatedPath()))
- .thenComposeOnAll(any -> mailboxDAO.delete(mailboxId))
+ FluentFutureStream.ofFutures(
+ mailboxPathDAO.delete(mailbox.generateAssociatedPath()),
+ mailboxPathV2DAO.delete(mailbox.generateAssociatedPath()))
+ .map(any -> mailboxDAO.delete(mailboxId), FluentFutureStream::unboxFuture)
.join();
}
@@ -166,10 +168,10 @@ public class CassandraMailboxMapper implements MailboxMapper {
private List<Mailbox> toMailboxes(MailboxPath path, CompletableFuture<Stream<CassandraIdAndPath>> listUserMailboxes) {
Pattern regex = Pattern.compile(constructEscapedRegexForMailboxNameMatching(path));
-
+
return FluentFutureStream.of(listUserMailboxes)
.filter(idAndPath -> regex.matcher(idAndPath.getMailboxPath().getName()).matches())
- .thenFlatComposeOnOptional(this::retrieveMailbox)
+ .map(this::retrieveMailbox, FluentFutureStream::unboxFutureOptional)
.join()
.collect(Guavate.toImmutableList());
}
@@ -227,7 +229,7 @@ public class CassandraMailboxMapper implements MailboxMapper {
@Override
public List<Mailbox> list() {
return mailboxDAO.retrieveAllMailboxes()
- .thenComposeOnAll(this::toMailboxWithAclFuture)
+ .map(this::toMailboxWithAclFuture, FluentFutureStream::unboxFuture)
.join()
.collect(Guavate.toImmutableList());
}
@@ -283,7 +285,7 @@ public class CassandraMailboxMapper implements MailboxMapper {
public List<Mailbox> findNonPersonalMailboxes(String userName, Right right) {
return FluentFutureStream.of(userMailboxRightsDAO.listRightsForUser(userName)
.thenApply(map -> toAuthorizedMailboxIds(map, right)))
- .thenFlatComposeOnOptional(this::retrieveMailbox)
+ .map(this::retrieveMailbox, FluentFutureStream::unboxFutureOptional)
.join()
.collect(Guavate.toImmutableList());
}
http://git-wip-us.apache.org/repos/asf/james-project/blob/3a53806a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdMapper.java
----------------------------------------------------------------------
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdMapper.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdMapper.java
index c59a60b..2d9a154 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdMapper.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdMapper.java
@@ -98,9 +98,10 @@ public class CassandraMessageIdMapper implements MessageIdMapper {
}
private Stream<SimpleMailboxMessage> findAsStream(Collection<MessageId> messageIds, FetchType fetchType) {
- return FluentFutureStream.ofNestedStreams(
- messageIds.stream()
- .map(messageId -> imapUidDAO.retrieve((CassandraMessageId) messageId, Optional.empty())))
+ return FluentFutureStream.of(
+ messageIds.stream()
+ .map(messageId -> imapUidDAO.retrieve((CassandraMessageId) messageId, Optional.empty())),
+ FluentFutureStream::unboxStream)
.collect(Guavate.toImmutableList())
.thenCompose(composedMessageIds -> messageDAO.retrieveMessages(composedMessageIds, fetchType, Limit.unlimited()))
.thenApply(stream -> stream
@@ -113,7 +114,7 @@ public class CassandraMessageIdMapper implements MessageIdMapper {
}
private CompletableFuture<Stream<SimpleMailboxMessage>> filterMessagesWithExistingMailbox(Stream<SimpleMailboxMessage> stream) {
- return FluentFutureStream.ofOptionals(stream.map(this::keepMessageIfMailboxExists))
+ return FluentFutureStream.of(stream.map(this::keepMessageIfMailboxExists), FluentFutureStream::unboxOptional)
.completableFuture();
}
http://git-wip-us.apache.org/repos/asf/james-project/blob/3a53806a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapper.java
----------------------------------------------------------------------
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapper.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapper.java
index 57b1f81..300e7ef 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapper.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapper.java
@@ -214,10 +214,13 @@ public class CassandraMessageMapper implements MessageMapper {
}
private CompletableFuture<Stream<SimpleMailboxMessage>> expungeUidChunk(CassandraId mailboxId, Collection<MessageUid> uidChunk) {
- return FluentFutureStream.ofOptionals(
- uidChunk.stream().map(uid -> retrieveComposedId(mailboxId, uid)))
+ return FluentFutureStream.of(
+ uidChunk.stream().map(uid -> retrieveComposedId(mailboxId, uid)),
+ FluentFutureStream::unboxOptional)
.performOnAll(this::deleteUsingMailboxId)
- .thenFlatCompose(idWithMetadata -> messageDAO.retrieveMessages(ImmutableList.of(idWithMetadata), FetchType.Metadata, Limit.unlimited()))
+ .map(idWithMetadata -> FluentFutureStream.of(
+ messageDAO.retrieveMessages(ImmutableList.of(idWithMetadata), FetchType.Metadata, Limit.unlimited())),
+ FluentFutureStream::unboxFluentFuture)
.filter(CassandraMessageDAO.MessageResult::isFound)
.map(CassandraMessageDAO.MessageResult::message)
.map(pair -> pair.getKey().toMailboxMessage(ImmutableList.of()))
@@ -299,8 +302,9 @@ public class CassandraMessageMapper implements MessageMapper {
}
private FlagsUpdateStageResult retryUpdatesStage(CassandraId mailboxId, FlagsUpdateCalculator flagsUpdateCalculator, List<MessageUid> failed) {
- Stream<ComposedMessageIdWithMetaData> idsFailed = FluentFutureStream.ofOptionals(
- failed.stream().map(uid -> messageIdDAO.retrieve(mailboxId, uid)))
+ Stream<ComposedMessageIdWithMetaData> idsFailed = FluentFutureStream.of(
+ failed.stream().map(uid -> messageIdDAO.retrieve(mailboxId, uid)),
+ FluentFutureStream::unboxOptional)
.join();
return runUpdateStage(mailboxId, idsFailed, flagsUpdateCalculator);
http://git-wip-us.apache.org/repos/asf/james-project/blob/3a53806a/server/container/util/src/main/java/org/apache/james/util/CompletableFutureUtil.java
----------------------------------------------------------------------
diff --git a/server/container/util/src/main/java/org/apache/james/util/CompletableFutureUtil.java b/server/container/util/src/main/java/org/apache/james/util/CompletableFutureUtil.java
index d2e83af..433de52 100644
--- a/server/container/util/src/main/java/org/apache/james/util/CompletableFutureUtil.java
+++ b/server/container/util/src/main/java/org/apache/james/util/CompletableFutureUtil.java
@@ -19,6 +19,7 @@
package org.apache.james.util;
+import java.util.Comparator;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.function.BiFunction;
@@ -35,11 +36,6 @@ public class CompletableFutureUtil {
.orElse(CompletableFuture.completedFuture(Optional.empty())));
}
- @SafeVarargs
- public static <T> CompletableFuture<Stream<T>> allOfArray(CompletableFuture<T>... futures) {
- return allOf(Stream.of(futures));
- }
-
public static <T, U, V> CompletableFuture<V> combine(CompletableFuture<T> t, CompletableFuture<U> u, BiFunction<T,U,V> combiner) {
return t.thenCompose(valueT ->
u.thenApply(valueU -> combiner.apply(valueT, valueU)));
@@ -81,20 +77,6 @@ public class CompletableFutureUtil {
CompletableFuture.completedFuture(Stream.concat(stream1, stream2))));
}
- public static <T> CompletableFuture<Stream<T>> performOnAll(CompletableFuture<Stream<T>> futurStream, Function<T, CompletableFuture<Void>> action) {
- return thenComposeOnAll(futurStream, value ->
- keepValue(() ->
- action.apply(value),
- value));
- }
-
- public static <T, U> CompletableFuture<Stream<U>> thenComposeOnAll(CompletableFuture<Stream<T>> futurStream, Function<T, CompletableFuture<U>> action) {
- return futurStream
- .thenCompose(stream ->
- CompletableFutureUtil.allOf(
- stream.map(action)));
- }
-
public static <T, U> CompletableFuture<Stream<U>> map(CompletableFuture<Stream<T>> futurStream, Function<T, U> action) {
return futurStream
.thenApply(stream ->
@@ -109,10 +91,6 @@ public class CompletableFutureUtil {
return futureStream.thenApply(stream -> stream.reduce(binaryOperator).orElse(emptyAccumulator));
}
- public static <T> CompletableFuture<T> keepValue(Supplier<CompletableFuture<Void>> supplier, T value) {
- return supplier.get().thenApply(any -> value);
- }
-
public static <T> Function<Boolean, CompletableFuture<Boolean>> composeIfTrue(Supplier<CompletableFuture<T>> composeOperation) {
return b -> {
if (b) {
@@ -121,4 +99,10 @@ public class CompletableFutureUtil {
return CompletableFuture.completedFuture(b);
};
}
+
+ public static <T> CompletableFuture<Stream<T>> sorted(CompletableFuture<Stream<T>> futureStream, Comparator<T> comparator) {
+ return futureStream
+ .thenApply(stream ->
+ stream.sorted(comparator));
+ }
}
http://git-wip-us.apache.org/repos/asf/james-project/blob/3a53806a/server/container/util/src/main/java/org/apache/james/util/FluentFutureStream.java
----------------------------------------------------------------------
diff --git a/server/container/util/src/main/java/org/apache/james/util/FluentFutureStream.java b/server/container/util/src/main/java/org/apache/james/util/FluentFutureStream.java
index 9dcae7a..0b0b86e 100644
--- a/server/container/util/src/main/java/org/apache/james/util/FluentFutureStream.java
+++ b/server/container/util/src/main/java/org/apache/james/util/FluentFutureStream.java
@@ -19,6 +19,8 @@
package org.apache.james.util;
+import java.util.Arrays;
+import java.util.Comparator;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.function.BinaryOperator;
@@ -29,6 +31,31 @@ import java.util.stream.Stream;
public class FluentFutureStream<T> {
+ public static <T> FluentFutureStream<T> unboxStream(FluentFutureStream<Stream<T>> streams) {
+ return FluentFutureStream.of(
+ streams.completableFuture()
+ .thenApply(StreamUtils::flatten));
+ }
+
+ public static <T> FluentFutureStream<T> unboxOptional(FluentFutureStream<Optional<T>> optionals) {
+ return unboxStream(optionals.map(OptionalUtils::toStream));
+ }
+
+ public static <T> FluentFutureStream<T> unboxFuture(FluentFutureStream<CompletableFuture<T>> futures) {
+ return FluentFutureStream.of(futures.completableFuture()
+ .thenCompose(CompletableFutureUtil::allOf));
+ }
+
+ public static <T> FluentFutureStream<T> unboxFluentFuture(FluentFutureStream<FluentFutureStream<T>> futures) {
+ return unboxStream(
+ unboxFuture(
+ futures.map(FluentFutureStream::completableFuture)));
+ }
+
+ public static <T> FluentFutureStream<T> unboxFutureOptional(FluentFutureStream<CompletableFuture<Optional<T>>> futures) {
+ return unboxOptional(unboxFuture(futures));
+ }
+
private final CompletableFuture<Stream<T>> completableFuture;
/**
@@ -38,6 +65,11 @@ public class FluentFutureStream<T> {
return new FluentFutureStream<>(completableFuture);
}
+ public static <T, U> FluentFutureStream<U> of(Stream<CompletableFuture<T>> completableFuture,
+ Function<FluentFutureStream<T>, FluentFutureStream<U>> unboxer) {
+ return unboxer.apply(of(completableFuture));
+ }
+
/**
* Constructs a FluentFutureStream from a Stream of Future
*/
@@ -45,32 +77,9 @@ public class FluentFutureStream<T> {
return new FluentFutureStream<>(CompletableFutureUtil.allOf(completableFutureStream));
}
- /**
- * Constructs a FluentFutureStream from a Stream of Future of Stream.
- *
- * Underlying streams are flatMapped.
- */
- public static <T> FluentFutureStream<T> ofNestedStreams(Stream<CompletableFuture<Stream<T>>> completableFuture) {
- return of(completableFuture)
- .flatMap(Function.identity());
- }
-
- /**
- * Constructs a FluentFutureStream from a Stream of Future of Optionals.
- *
- * Underlying optionals are unboxed.
- */
- public static <T> FluentFutureStream<T> ofOptionals(Stream<CompletableFuture<Optional<T>>> completableFuture) {
- return of(completableFuture)
- .flatMapOptional(Function.identity());
- }
-
- /**
- * Constructs a FluentFutureStream from the supplied futures.
- */
@SafeVarargs
public static <T> FluentFutureStream<T> ofFutures(CompletableFuture<T>... completableFutures) {
- return new FluentFutureStream<>(CompletableFutureUtil.allOfArray(completableFutures));
+ return of(Arrays.stream(completableFutures));
}
private FluentFutureStream(CompletableFuture<Stream<T>> completableFuture) {
@@ -81,8 +90,8 @@ public class FluentFutureStream<T> {
* For all values of the underlying stream, an action will be performed.
*/
public FluentFutureStream<T> performOnAll(Function<T, CompletableFuture<Void>> action) {
- return FluentFutureStream.of(
- CompletableFutureUtil.performOnAll(completableFuture(), action));
+ return map(t -> action.apply(t).thenApply(any -> t),
+ FluentFutureStream::unboxFuture);
}
/**
@@ -93,56 +102,8 @@ public class FluentFutureStream<T> {
CompletableFutureUtil.map(completableFuture(), function));
}
- /**
- * Apply a transformation to all value of the underlying stream.
- *
- * As the supplied transformation produces streams, the results will be flatMapped.
- */
- public <U> FluentFutureStream<U> flatMap(Function<T, Stream<U>> function) {
- return FluentFutureStream.of(completableFuture().thenApply(stream ->
- stream.flatMap(function)));
- }
-
- /**
- * Apply a transformation to all value of the underlying stream.
- *
- * As the supplied transformation produces optionals, the results will be unboxed.
- */
- public <U> FluentFutureStream<U> flatMapOptional(Function<T, Optional<U>> function) {
- return map(function)
- .flatMap(OptionalUtils::toStream);
- }
-
- /**
- * Apply a transformation to all value of the underlying stream.
- *
- * As the supplied transformation produces futures, we need to compose the returned values.
- */
- public <U> FluentFutureStream<U> thenComposeOnAll(Function<T, CompletableFuture<U>> function) {
- return FluentFutureStream.of(
- CompletableFutureUtil.thenComposeOnAll(completableFuture(), function));
- }
-
- /**
- * Apply a transformation to all value of the underlying stream.
- *
- * As the supplied transformation produces futures of stream, we need to compose then flatMap the returned values.
- */
- public <U> FluentFutureStream<U> thenFlatCompose(Function<T, CompletableFuture<Stream<U>>> function) {
- return FluentFutureStream.of(
- CompletableFutureUtil.thenComposeOnAll(completableFuture(), function))
- .flatMap(Function.identity());
- }
-
- /**
- * Apply a transformation to all value of the underlying stream.
- *
- * As the supplied transformation produces futures of optionals, we need to compose then unbox the returned values.
- */
- public <U> FluentFutureStream<U> thenFlatComposeOnOptional(Function<T, CompletableFuture<Optional<U>>> function) {
- return FluentFutureStream.of(
- CompletableFutureUtil.thenComposeOnAll(completableFuture(), function))
- .flatMapOptional(Function.identity());
+ public <U, V> FluentFutureStream<V> map(Function<T, U> function, Function<FluentFutureStream<U>, FluentFutureStream<V>> unboxer) {
+ return unboxer.apply(map(function));
}
/**
@@ -153,6 +114,12 @@ public class FluentFutureStream<T> {
.thenApply(stream -> stream.filter(predicate)));
}
+ public FluentFutureStream<T> thenFilter(Function<T, CompletableFuture<Boolean>> futurePredicate) {
+ return map(t -> futurePredicate.apply(t)
+ .thenApply(isKept -> Optional.of(t).filter(any -> isKept)),
+ FluentFutureStream::unboxFutureOptional);
+ }
+
/**
* Reduces the underlying stream. Reduced value is supplied as a Future of optional, as no empty value is supplied.
*/
@@ -168,6 +135,14 @@ public class FluentFutureStream<T> {
}
/**
+ * sort all elements of the stream by the provided {@code Comparator}.
+ */
+ public FluentFutureStream<T> sorted(Comparator<T> comparator) {
+ return FluentFutureStream.of(
+ CompletableFutureUtil.sorted(completableFuture(), comparator));
+ }
+
+ /**
* Returns a future of the underlying stream.
*/
public CompletableFuture<Stream<T>> completableFuture() {
http://git-wip-us.apache.org/repos/asf/james-project/blob/3a53806a/server/container/util/src/test/java/org/apache/james/util/CompletableFutureUtilTest.java
----------------------------------------------------------------------
diff --git a/server/container/util/src/test/java/org/apache/james/util/CompletableFutureUtilTest.java b/server/container/util/src/test/java/org/apache/james/util/CompletableFutureUtilTest.java
index a42fb39..acbe10f 100644
--- a/server/container/util/src/test/java/org/apache/james/util/CompletableFutureUtilTest.java
+++ b/server/container/util/src/test/java/org/apache/james/util/CompletableFutureUtilTest.java
@@ -27,7 +27,6 @@ import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
-import java.util.function.Supplier;
import java.util.stream.IntStream;
import java.util.stream.Stream;
@@ -171,59 +170,6 @@ public class CompletableFutureUtilTest {
}
@Test
- public void allOfArrayShouldPreserveOrder() {
- long value1 = 18L;
- long value2 = 19L;
- long value3 = 20L;
- long value4 = 21L;
- long value5 = 22L;
- long value6 = 23L;
- long value7 = 24L;
- long value8 = 25L;
- long value9 = 26L;
- long value10 = 27L;
- assertThat(
- CompletableFutureUtil.allOfArray(
- CompletableFuture.completedFuture(value1),
- CompletableFuture.completedFuture(value2),
- CompletableFuture.completedFuture(value3),
- CompletableFuture.completedFuture(value4),
- CompletableFuture.completedFuture(value5),
- CompletableFuture.completedFuture(value6),
- CompletableFuture.completedFuture(value7),
- CompletableFuture.completedFuture(value8),
- CompletableFuture.completedFuture(value9),
- CompletableFuture.completedFuture(value10))
- .join()
- .collect(Guavate.toImmutableList()))
- .containsExactly(value1, value2, value3, value4, value5, value6, value7, value8, value9, value10);
- }
-
- @Test
- public void allOfArrayShouldUnboxNoArgs() {
- assertThat(
- CompletableFutureUtil.allOfArray()
- .join()
- .collect(Guavate.toImmutableList()))
- .isEmpty();
- }
-
- @Test
- public void allOfArrayShouldUnboxArray() {
- long value1 = 18L;
- long value2 = 19L;
- long value3 = 20L;
- assertThat(
- CompletableFutureUtil.allOfArray(
- CompletableFuture.completedFuture(value1),
- CompletableFuture.completedFuture(value2),
- CompletableFuture.completedFuture(value3))
- .join()
- .collect(Guavate.toImmutableList()))
- .containsOnly(value1, value2, value3);
- }
-
- @Test
public void allOfShouldWorkOnVeryLargeStream() {
CompletableFutureUtil.allOf(
IntStream.range(0, 100000)
@@ -257,58 +203,6 @@ public class CompletableFutureUtilTest {
}
@Test
- public void thenComposeOnAllShouldMapOnStreamInsideACompletableFuturOfStreamAndTransformTheResultingStreamOfCompletableFutureIntoACompletableOfStreamAndFlatIt() {
- CompletableFuture<Stream<Integer>> futurOfInteger = CompletableFuture.completedFuture(Stream.of(1, 2, 3));
-
- assertThat(
- CompletableFutureUtil.thenComposeOnAll(futurOfInteger, integer ->
- CompletableFuture.completedFuture(integer * 2))
- .join()
- .collect(Guavate.toImmutableList()))
- .containsExactly(2, 4, 6);
- }
-
- @Test
- public void thenComposeOnAllOnEmptyStreamShouldReturnAnEmptyStream() {
- CompletableFuture<Stream<Integer>> futurOfInteger = CompletableFuture.completedFuture(Stream.of());
-
- assertThat(
- CompletableFutureUtil.thenComposeOnAll(futurOfInteger, integer ->
- CompletableFuture.completedFuture(integer * 2))
- .join()
- .collect(Guavate.toImmutableList()))
- .isEmpty();
- }
-
- @Test
- public void keepValueShouldCompleteWhenTheGivenCompletableFutureEnd() {
- final AtomicInteger numOfFutureExecution = new AtomicInteger(0);
-
- Supplier<CompletableFuture<Void>> future = () ->
- CompletableFuture.runAsync(numOfFutureExecution::incrementAndGet);
-
- assertThat(
- CompletableFutureUtil.keepValue(future, 42)
- .join())
- .isEqualTo(42);
-
- assertThat(
- numOfFutureExecution.get())
- .isEqualTo(1);
- }
-
- @Test
- public void keepValueShouldReturnNullWithNullValue() {
- Supplier<CompletableFuture<Void>> future = () ->
- CompletableFuture.completedFuture(null);
-
- assertThat(
- CompletableFutureUtil.keepValue(future, null)
- .join())
- .isNull();
- }
-
- @Test
public void composeIfTrueShouldReturnTrueWhenTrue() {
assertThat(
CompletableFutureUtil.composeIfTrue(() -> CompletableFuture.completedFuture(null))
@@ -357,7 +251,7 @@ public class CompletableFutureUtilTest {
assertThat(
CompletableFutureUtil.reduce(
(i, j) -> i + j,
- CompletableFutureUtil.<Long>allOfArray())
+ CompletableFuture.completedFuture(Stream.<Long>of()))
.join())
.isEmpty();
}
@@ -367,11 +261,8 @@ public class CompletableFutureUtilTest {
assertThat(
CompletableFutureUtil.reduce(
(i, j) -> i + j,
- CompletableFutureUtil.allOfArray(
- CompletableFuture.completedFuture(1L),
- CompletableFuture.completedFuture(2L),
- CompletableFuture.completedFuture(3L)
- ))
+ CompletableFuture.completedFuture(Stream.of(
+ 1L, 2L, 3L)))
.join())
.contains(6L);
}
@@ -382,7 +273,7 @@ public class CompletableFutureUtilTest {
assertThat(
CompletableFutureUtil.reduce(
(i, j) -> i + j,
- CompletableFutureUtil.<Long>allOfArray(),
+ CompletableFuture.completedFuture(Stream.of()),
identityAccumulator)
.join())
.isEqualTo(identityAccumulator);
@@ -393,11 +284,7 @@ public class CompletableFutureUtilTest {
assertThat(
CompletableFutureUtil.reduce(
(i, j) -> i + j,
- CompletableFutureUtil.allOfArray(
- CompletableFuture.completedFuture(1L),
- CompletableFuture.completedFuture(2L),
- CompletableFuture.completedFuture(3L)
- ),
+ CompletableFuture.completedFuture(Stream.of(1L, 2L,3L)),
0L)
.join())
.isEqualTo(6L);
@@ -420,4 +307,23 @@ public class CompletableFutureUtilTest {
.join())
.isEmpty();
}
+
+ @Test
+ public void sortShouldReturnEmptyWhenEmptyStream() {
+ FluentFutureStream<Long> futureStream = FluentFutureStream.ofFutures();
+ assertThat(futureStream.sorted(Long::compareTo).join())
+ .isEmpty();
+ }
+
+ @Test
+ public void sortShouldReturnTheSortedStream() {
+ FluentFutureStream<Long> futureStream = FluentFutureStream.ofFutures(
+ CompletableFuture.completedFuture(4L),
+ CompletableFuture.completedFuture(3L),
+ CompletableFuture.completedFuture(2L),
+ CompletableFuture.completedFuture(1L));
+
+ assertThat(futureStream.sorted(Long::compareTo).join())
+ .containsExactly(1L, 2L, 3L, 4L);
+ }
}
http://git-wip-us.apache.org/repos/asf/james-project/blob/3a53806a/server/container/util/src/test/java/org/apache/james/util/FluentFutureStreamTest.java
----------------------------------------------------------------------
diff --git a/server/container/util/src/test/java/org/apache/james/util/FluentFutureStreamTest.java b/server/container/util/src/test/java/org/apache/james/util/FluentFutureStreamTest.java
index 0877414..56017d6 100644
--- a/server/container/util/src/test/java/org/apache/james/util/FluentFutureStreamTest.java
+++ b/server/container/util/src/test/java/org/apache/james/util/FluentFutureStreamTest.java
@@ -71,11 +71,12 @@ public class FluentFutureStreamTest {
@Test
public void ofNestedStreamsShouldConstructAFluentFutureStreamWhenProvidedAStreamOfFutureOfStream() {
assertThat(
- FluentFutureStream.ofNestedStreams(
+ FluentFutureStream.<Stream<Integer>, Integer>of(
Stream.of(
CompletableFuture.completedFuture(Stream.of(1, 2)),
CompletableFuture.completedFuture(Stream.of()),
- CompletableFuture.completedFuture(Stream.of(3))))
+ CompletableFuture.completedFuture(Stream.of(3))),
+ FluentFutureStream::unboxStream)
.join()
.collect(Guavate.toImmutableList()))
.containsExactly(1, 2, 3);
@@ -85,12 +86,13 @@ public class FluentFutureStreamTest {
@Test
public void ofOptionalsShouldConstructAFluentFutureStreamWhenProvidedAStreamOfFutureOfOptionals() {
assertThat(
- FluentFutureStream.ofOptionals(
+ FluentFutureStream.<Optional<Integer>, Integer>of(
Stream.of(
CompletableFuture.completedFuture(Optional.of(1)),
CompletableFuture.completedFuture(Optional.of(2)),
CompletableFuture.completedFuture(Optional.empty()),
- CompletableFuture.completedFuture(Optional.of(3))))
+ CompletableFuture.completedFuture(Optional.of(3))),
+ FluentFutureStream::unboxOptional)
.join()
.collect(Guavate.toImmutableList()))
.containsExactly(1, 2, 3);
@@ -126,7 +128,7 @@ public class FluentFutureStreamTest {
FluentFutureStream.of(
CompletableFuture.completedFuture(
Stream.of(1, 2, 3)))
- .flatMap(i -> Stream.of(i, i + 1))
+ .map(i -> Stream.of(i, i + 1), FluentFutureStream::unboxStream)
.join()
.collect(Guavate.toImmutableList()))
.containsExactly(1, 2, 2, 3, 3, 4);
@@ -138,8 +140,9 @@ public class FluentFutureStreamTest {
FluentFutureStream.of(
CompletableFuture.completedFuture(
Stream.of(1, 2, 3)))
- .flatMapOptional(i -> Optional.of(i + 1)
- .filter(j -> j % 2 == 0))
+ .map(i -> Optional.of(i + 1)
+ .filter(j -> j % 2 == 0),
+ FluentFutureStream::unboxOptional)
.join()
.collect(Guavate.toImmutableList()))
.containsExactly(2, 4);
@@ -180,12 +183,24 @@ public class FluentFutureStreamTest {
}
@Test
+ public void thenFilterShouldBeAppliedOnTheUnderlyingStream() {
+ assertThat(
+ FluentFutureStream.of(
+ CompletableFuture.completedFuture(
+ Stream.of(1, 2, 3)))
+ .thenFilter(i -> CompletableFuture.completedFuture(i % 2 == 1))
+ .join()
+ .collect(Guavate.toImmutableList()))
+ .containsExactly(1, 3);
+ }
+
+ @Test
public void thenComposeOnAllShouldTransformUnderlyingValuesAndComposeFutures() {
assertThat(
FluentFutureStream.of(
CompletableFuture.completedFuture(
Stream.of(1, 2, 3)))
- .thenComposeOnAll(i -> CompletableFuture.completedFuture(i + 1))
+ .map(i -> CompletableFuture.completedFuture(i + 1), FluentFutureStream::unboxFuture)
.join()
.collect(Guavate.toImmutableList()))
.containsExactly(2, 3, 4);
@@ -197,7 +212,7 @@ public class FluentFutureStreamTest {
FluentFutureStream.of(
CompletableFuture.completedFuture(
Stream.of(1, 2, 3)))
- .thenFlatCompose(i -> CompletableFuture.completedFuture(Stream.of(i, i + 1)))
+ .map(i -> FluentFutureStream.of(CompletableFuture.completedFuture(Stream.of(i, i + 1))), FluentFutureStream::unboxFluentFuture)
.join()
.collect(Guavate.toImmutableList()))
.containsExactly(1, 2, 2, 3, 3, 4);
@@ -209,8 +224,9 @@ public class FluentFutureStreamTest {
FluentFutureStream.of(
CompletableFuture.completedFuture(
Stream.of(1, 2, 3)))
- .thenFlatComposeOnOptional(i -> CompletableFuture.completedFuture(Optional.of(i + 1)
- .filter(j -> j % 2 == 0)))
+ .map(i -> CompletableFuture.completedFuture(
+ Optional.of(i + 1).filter(j -> j % 2 == 0)),
+ FluentFutureStream::unboxFutureOptional)
.join()
.collect(Guavate.toImmutableList()))
.containsExactly(2, 4);
@@ -255,4 +271,23 @@ public class FluentFutureStreamTest {
.isEmpty();
}
+ @Test
+ public void sortedShouldReturnInOrderElements() {
+ assertThat(
+ FluentFutureStream.of(
+ CompletableFuture.completedFuture(Stream.of(4L, 3L, 2L, 1L)))
+ .sorted(Long::compareTo)
+ .join())
+ .containsExactly(1L, 2L, 3L, 4L);
+ }
+
+ @Test
+ public void sortedShouldReturnEmptyWhenEmpty() {
+ CompletableFuture<Stream<Long>> completableFutureStream = CompletableFuture.completedFuture(Stream.of());
+ assertThat(
+ FluentFutureStream.of(completableFutureStream)
+ .sorted(Long::compareTo)
+ .join())
+ .isEmpty();
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/james-project/blob/3a53806a/server/queue/queue-api/src/test/java/org/apache/james/queue/api/ManageableMailQueueContract.java
----------------------------------------------------------------------
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 2877efe..9bc503c 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
@@ -57,8 +57,8 @@ public interface ManageableMailQueueContract extends MailQueueContract {
@Test
default void getSizeShouldReturnMessageCountWhenSeveralMails() throws Exception {
- getManageableMailQueue().enQueue(defaultMail().build());
- getManageableMailQueue().enQueue(defaultMail().build());
+ getManageableMailQueue().enQueue(defaultMail().name("1").build());
+ getManageableMailQueue().enQueue(defaultMail().name("2").build());
long size = getManageableMailQueue().getSize();
http://git-wip-us.apache.org/repos/asf/james-project/blob/3a53806a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueue.java
----------------------------------------------------------------------
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueue.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueue.java
index e4b8881..818a83b 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueue.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueue.java
@@ -25,11 +25,13 @@ import java.util.function.Function;
import javax.inject.Inject;
import javax.mail.internet.MimeMessage;
+import org.apache.commons.lang3.NotImplementedException;
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.metrics.api.MetricFactory;
-import org.apache.james.queue.api.MailQueue;
+import org.apache.james.queue.api.ManageableMailQueue;
+import org.apache.james.queue.rabbitmq.view.api.MailQueueView;
import org.apache.mailet.Mail;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -37,7 +39,7 @@ import org.slf4j.LoggerFactory;
import com.github.fge.lambdas.Throwing;
import com.google.common.annotations.VisibleForTesting;
-public class RabbitMQMailQueue implements MailQueue {
+public class RabbitMQMailQueue implements ManageableMailQueue {
private static final Logger LOGGER = LoggerFactory.getLogger(RabbitMQMailQueue.class);
@@ -47,21 +49,26 @@ public class RabbitMQMailQueue implements MailQueue {
private final Store<MimeMessage, MimeMessagePartsId> mimeMessageStore;
private final MailReferenceSerializer mailReferenceSerializer;
private final Function<MailReferenceDTO, Mail> mailLoader;
+ private final MailQueueView mailQueueView;
@Inject
@VisibleForTesting Factory(MetricFactory metricFactory, RabbitClient rabbitClient,
- Store<MimeMessage, MimeMessagePartsId> mimeMessageStore, BlobId.Factory blobIdFactory) {
+ Store<MimeMessage, MimeMessagePartsId> mimeMessageStore,
+ BlobId.Factory blobIdFactory,
+ MailQueueView mailQueueView) {
this.metricFactory = metricFactory;
this.rabbitClient = rabbitClient;
this.mimeMessageStore = mimeMessageStore;
this.mailReferenceSerializer = new MailReferenceSerializer();
this.mailLoader = Throwing.function(new MailLoader(mimeMessageStore, blobIdFactory)::load).sneakyThrow();
+ this.mailQueueView = mailQueueView;
}
RabbitMQMailQueue create(MailQueueName mailQueueName) {
return new RabbitMQMailQueue(metricFactory, mailQueueName,
new Enqueuer(mailQueueName, rabbitClient, mimeMessageStore, mailReferenceSerializer, metricFactory),
- new Dequeuer(mailQueueName, rabbitClient, mailLoader, mailReferenceSerializer, metricFactory));
+ new Dequeuer(mailQueueName, rabbitClient, mailLoader, mailReferenceSerializer, metricFactory),
+ mailQueueView);
}
}
@@ -69,15 +76,17 @@ public class RabbitMQMailQueue implements MailQueue {
private final MetricFactory metricFactory;
private final Enqueuer enqueuer;
private final Dequeuer dequeuer;
+ private final MailQueueView mailQueueView;
RabbitMQMailQueue(MetricFactory metricFactory, MailQueueName name,
- Enqueuer enqueuer, Dequeuer dequeuer) {
+ Enqueuer enqueuer, Dequeuer dequeuer, MailQueueView mailQueueView) {
this.name = name;
this.enqueuer = enqueuer;
this.dequeuer = dequeuer;
this.metricFactory = metricFactory;
+ this.mailQueueView = mailQueueView;
}
@Override
@@ -104,4 +113,30 @@ public class RabbitMQMailQueue implements MailQueue {
return metricFactory.runPublishingTimerMetric(DEQUEUED_TIMER_METRIC_NAME_PREFIX + name.asString(),
Throwing.supplier(dequeuer::deQueue).sneakyThrow());
}
-}
\ No newline at end of file
+
+ @Override
+ public long getSize() {
+ return mailQueueView.getSize();
+ }
+
+ @Override
+ public long flush() {
+ LOGGER.warn("Delays are not supported by RabbitMQ. Flush is a NOOP.");
+ return 0;
+ }
+
+ @Override
+ public long clear() {
+ throw new NotImplementedException("Not yet implemented");
+ }
+
+ @Override
+ public long remove(Type type, String value) {
+ throw new NotImplementedException("Not yet implemented");
+ }
+
+ @Override
+ public MailQueueIterator browse() {
+ return mailQueueView.browse();
+ }
+}
http://git-wip-us.apache.org/repos/asf/james-project/blob/3a53806a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/BrowseStartDAO.java
----------------------------------------------------------------------
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/BrowseStartDAO.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/BrowseStartDAO.java
new file mode 100644
index 0000000..098c033
--- /dev/null
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/BrowseStartDAO.java
@@ -0,0 +1,109 @@
+/****************************************************************
+ * 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.view.cassandra;
+
+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 com.datastax.driver.core.querybuilder.QueryBuilder.set;
+import static com.datastax.driver.core.querybuilder.QueryBuilder.update;
+import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.BrowseStartTable.BROWSE_START;
+import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.BrowseStartTable.QUEUE_NAME;
+import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.BrowseStartTable.TABLE_NAME;
+
+import java.time.Instant;
+import java.util.Date;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+
+import javax.inject.Inject;
+
+import org.apache.james.backends.cassandra.utils.CassandraAsyncExecutor;
+import org.apache.james.queue.rabbitmq.MailQueueName;
+
+import com.datastax.driver.core.PreparedStatement;
+import com.datastax.driver.core.Row;
+import com.datastax.driver.core.Session;
+import com.google.common.annotations.VisibleForTesting;
+
+class BrowseStartDAO {
+
+ private final CassandraAsyncExecutor executor;
+ private final PreparedStatement selectOne;
+ private final PreparedStatement insertOne;
+ private final PreparedStatement updateOne;
+
+ @Inject
+ BrowseStartDAO(Session session) {
+ this.executor = new CassandraAsyncExecutor(session);
+
+ this.selectOne = prepareSelectOne(session);
+ this.updateOne = prepareUpdate(session);
+ this.insertOne = prepareInsertOne(session);
+ }
+
+ private PreparedStatement prepareSelectOne(Session session) {
+ return session.prepare(select()
+ .from(TABLE_NAME)
+ .where(eq(QUEUE_NAME, bindMarker(QUEUE_NAME))));
+ }
+
+ private PreparedStatement prepareUpdate(Session session) {
+ return session.prepare(update(TABLE_NAME)
+ .with(set(BROWSE_START, bindMarker(BROWSE_START)))
+ .where(eq(QUEUE_NAME, bindMarker(QUEUE_NAME))));
+ }
+
+ private PreparedStatement prepareInsertOne(Session session) {
+ return session.prepare(insertInto(TABLE_NAME)
+ .ifNotExists()
+ .value(BROWSE_START, bindMarker(BROWSE_START))
+ .value(QUEUE_NAME, bindMarker(QUEUE_NAME)));
+ }
+
+ CompletableFuture<Optional<Instant>> findBrowseStart(MailQueueName queueName) {
+ return selectOne(queueName)
+ .thenApply(optional -> optional.map(this::getBrowseStart));
+ }
+
+ CompletableFuture<Void> updateBrowseStart(MailQueueName mailQueueName, Instant sliceStart) {
+ return executor.executeVoid(updateOne.bind()
+ .setTimestamp(BROWSE_START, Date.from(sliceStart))
+ .setString(QUEUE_NAME, mailQueueName.asString()));
+ }
+
+ CompletableFuture<Void> insertInitialBrowseStart(MailQueueName mailQueueName, Instant sliceStart) {
+ return executor.executeVoid(insertOne.bind()
+ .setTimestamp(BROWSE_START, Date.from(sliceStart))
+ .setString(QUEUE_NAME, mailQueueName.asString()));
+ }
+
+ @VisibleForTesting
+ CompletableFuture<Optional<Row>> selectOne(MailQueueName queueName) {
+ return executor.executeSingleRow(
+ selectOne.bind()
+ .setString(QUEUE_NAME, queueName.asString()));
+ }
+
+ private Instant getBrowseStart(Row row) {
+ return row.getTimestamp(BROWSE_START).toInstant();
+ }
+}
http://git-wip-us.apache.org/repos/asf/james-project/blob/3a53806a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueBrowser.java
----------------------------------------------------------------------
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
new file mode 100644
index 0000000..21dedb8
--- /dev/null
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueBrowser.java
@@ -0,0 +1,128 @@
+/****************************************************************
+ * 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.view.cassandra;
+
+import static org.apache.james.queue.rabbitmq.view.cassandra.model.BucketedSlices.BucketId;
+import static org.apache.james.queue.rabbitmq.view.cassandra.model.BucketedSlices.Slice;
+import static org.apache.james.queue.rabbitmq.view.cassandra.model.BucketedSlices.Slice.allSlicesTill;
+
+import java.time.Clock;
+import java.time.Instant;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+
+import javax.inject.Inject;
+
+import org.apache.james.queue.api.ManageableMailQueue;
+import org.apache.james.queue.rabbitmq.MailQueueName;
+import org.apache.james.queue.rabbitmq.view.cassandra.model.EnqueuedMail;
+import org.apache.james.util.FluentFutureStream;
+
+import com.google.common.base.Preconditions;
+
+class CassandraMailQueueBrowser {
+
+ static class CassandraMailQueueIterator implements ManageableMailQueue.MailQueueIterator {
+
+ private final Iterator<ManageableMailQueue.MailQueueItemView> iterator;
+
+ CassandraMailQueueIterator(Iterator<ManageableMailQueue.MailQueueItemView> iterator) {
+ Preconditions.checkNotNull(iterator);
+
+ this.iterator = iterator;
+ }
+
+ @Override
+ public void close() {}
+
+ @Override
+ public boolean hasNext() {
+ return iterator.hasNext();
+ }
+
+ @Override
+ public ManageableMailQueue.MailQueueItemView next() {
+ return iterator.next();
+ }
+ }
+
+ private final BrowseStartDAO browseStartDao;
+ private final DeletedMailsDAO deletedMailsDao;
+ private final EnqueuedMailsDAO enqueuedMailsDao;
+ private final CassandraMailQueueViewConfiguration configuration;
+ private final Clock clock;
+
+ @Inject
+ CassandraMailQueueBrowser(BrowseStartDAO browseStartDao,
+ DeletedMailsDAO deletedMailsDao,
+ EnqueuedMailsDAO enqueuedMailsDao,
+ CassandraMailQueueViewConfiguration configuration, Clock clock) {
+ this.browseStartDao = browseStartDao;
+ this.deletedMailsDao = deletedMailsDao;
+ this.enqueuedMailsDao = enqueuedMailsDao;
+ this.configuration = configuration;
+ this.clock = clock;
+ }
+
+ CompletableFuture<Stream<ManageableMailQueue.MailQueueItemView>> browse(MailQueueName queueName) {
+ return browseReferences(queueName)
+ .map(EnqueuedMail::getMail)
+ .map(ManageableMailQueue.MailQueueItemView::new)
+ .completableFuture();
+ }
+
+ FluentFutureStream<EnqueuedMail> browseReferences(MailQueueName queueName) {
+ return FluentFutureStream.of(browseStartDao.findBrowseStart(queueName)
+ .thenApply(this::allSlicesStartingAt))
+ .map(slice -> browseSlice(queueName, slice), FluentFutureStream::unboxFluentFuture);
+ }
+
+ private FluentFutureStream<EnqueuedMail> browseSlice(MailQueueName queueName, Slice slice) {
+ return FluentFutureStream.of(
+ allBucketIds()
+ .map(bucketId ->
+ browseBucket(queueName, slice, bucketId).completableFuture()),
+ FluentFutureStream::unboxStream)
+ .sorted(Comparator.comparing(EnqueuedMail::getEnqueuedTime));
+ }
+
+ private FluentFutureStream<EnqueuedMail> browseBucket(MailQueueName queueName, Slice slice, BucketId bucketId) {
+ return FluentFutureStream.of(
+ enqueuedMailsDao.selectEnqueuedMails(queueName, slice, bucketId))
+ .thenFilter(mailReference -> deletedMailsDao.isStillEnqueued(queueName, mailReference.getMailKey()));
+ }
+
+ private Stream<Slice> allSlicesStartingAt(Optional<Instant> maybeBrowseStart) {
+ return maybeBrowseStart
+ .map(browseStart -> Slice.of(browseStart, configuration.getSliceWindow()))
+ .map(startSlice -> allSlicesTill(startSlice, clock.instant()))
+ .orElse(Stream.empty());
+ }
+
+ private Stream<BucketId> allBucketIds() {
+ return IntStream
+ .range(0, configuration.getBucketCount())
+ .mapToObj(BucketId::of);
+ }
+}
http://git-wip-us.apache.org/repos/asf/james-project/blob/3a53806a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueMailDelete.java
----------------------------------------------------------------------
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
new file mode 100644
index 0000000..c8c0a0e
--- /dev/null
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueMailDelete.java
@@ -0,0 +1,87 @@
+/****************************************************************
+ * 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.view.cassandra;
+
+import java.time.Instant;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.stream.Stream;
+
+import javax.inject.Inject;
+
+import org.apache.james.queue.rabbitmq.MailQueueName;
+import org.apache.james.queue.rabbitmq.view.cassandra.model.EnqueuedMail;
+import org.apache.james.queue.rabbitmq.view.cassandra.model.MailKey;
+import org.apache.mailet.Mail;
+
+class CassandraMailQueueMailDelete {
+
+ private final DeletedMailsDAO deletedMailsDao;
+ private final BrowseStartDAO browseStartDao;
+ private final CassandraMailQueueBrowser cassandraMailQueueBrowser;
+ private final CassandraMailQueueViewConfiguration configuration;
+ private final ThreadLocalRandom random;
+
+ @Inject
+ CassandraMailQueueMailDelete(DeletedMailsDAO deletedMailsDao,
+ BrowseStartDAO browseStartDao,
+ CassandraMailQueueBrowser cassandraMailQueueBrowser,
+ CassandraMailQueueViewConfiguration configuration,
+ ThreadLocalRandom random) {
+ this.deletedMailsDao = deletedMailsDao;
+ this.browseStartDao = browseStartDao;
+ this.cassandraMailQueueBrowser = cassandraMailQueueBrowser;
+ this.configuration = configuration;
+ this.random = random;
+ }
+
+ CompletableFuture<Void> considerDeleted(Mail mail, MailQueueName mailQueueName) {
+ return deletedMailsDao
+ .markAsDeleted(mailQueueName, MailKey.fromMail(mail))
+ .thenRunAsync(() -> maybeUpdateBrowseStart(mailQueueName));
+ }
+
+ private void maybeUpdateBrowseStart(MailQueueName mailQueueName) {
+ if (shouldUpdateBrowseStart()) {
+ findNewBrowseStart(mailQueueName)
+ .thenCompose(newBrowseStart -> updateNewBrowseStart(mailQueueName, newBrowseStart))
+ .join();
+ }
+ }
+
+ private CompletableFuture<Optional<Instant>> findNewBrowseStart(MailQueueName mailQueueName) {
+ return cassandraMailQueueBrowser.browseReferences(mailQueueName)
+ .map(EnqueuedMail::getTimeRangeStart)
+ .completableFuture()
+ .thenApply(Stream::findFirst);
+ }
+
+ private CompletableFuture<Void> updateNewBrowseStart(MailQueueName mailQueueName, Optional<Instant> maybeNewBrowseStart) {
+ return maybeNewBrowseStart
+ .map(newBrowseStartInstant -> browseStartDao.updateBrowseStart(mailQueueName, newBrowseStartInstant))
+ .orElse(CompletableFuture.completedFuture(null));
+ }
+
+ private boolean shouldUpdateBrowseStart() {
+ int threshold = configuration.getUpdateBrowseStartPace();
+ return Math.abs(random.nextInt()) % threshold == 0;
+ }
+}
http://git-wip-us.apache.org/repos/asf/james-project/blob/3a53806a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueMailStore.java
----------------------------------------------------------------------
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueMailStore.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueMailStore.java
new file mode 100644
index 0000000..ed696f2
--- /dev/null
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueMailStore.java
@@ -0,0 +1,98 @@
+/****************************************************************
+ * 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.view.cassandra;
+
+import java.time.Clock;
+import java.time.Instant;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+
+import javax.inject.Inject;
+
+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.EnqueuedMail;
+import org.apache.james.queue.rabbitmq.view.cassandra.model.MailKey;
+import org.apache.mailet.Mail;
+
+class CassandraMailQueueMailStore {
+
+ private final EnqueuedMailsDAO enqueuedMailsDao;
+ private final BrowseStartDAO browseStartDao;
+ private final CassandraMailQueueViewConfiguration configuration;
+ private final Clock clock;
+ private final Set<MailQueueName> initialInserted;
+
+ @Inject
+ CassandraMailQueueMailStore(EnqueuedMailsDAO enqueuedMailsDao,
+ BrowseStartDAO browseStartDao,
+ CassandraMailQueueViewConfiguration configuration,
+ Clock clock) {
+ this.enqueuedMailsDao = enqueuedMailsDao;
+ this.browseStartDao = browseStartDao;
+ this.configuration = configuration;
+ this.clock = clock;
+ this.initialInserted = ConcurrentHashMap.newKeySet();
+ }
+
+ CompletableFuture<Void> storeMailInEnqueueTable(Mail mail, MailQueueName mailQueueName) {
+ EnqueuedMail enqueuedMail = convertToEnqueuedMail(mail, mailQueueName);
+
+ return enqueuedMailsDao.insert(enqueuedMail)
+ .thenCompose(any -> initBrowseStartIfNeeded(mailQueueName, enqueuedMail.getTimeRangeStart()));
+ }
+
+ private CompletableFuture<Void> initBrowseStartIfNeeded(MailQueueName mailQueueName, Instant sliceStartAt) {
+ if (!initialInserted.contains(mailQueueName)) {
+ return tryInsertBrowseStart(mailQueueName, sliceStartAt);
+ }
+ return CompletableFuture.completedFuture(null);
+ }
+
+ private CompletableFuture<Void> tryInsertBrowseStart(MailQueueName mailQueueName, Instant sliceStartAt) {
+ return browseStartDao
+ .insertInitialBrowseStart(mailQueueName, sliceStartAt)
+ .thenAccept(any -> initialInserted.add(mailQueueName));
+ }
+
+ private EnqueuedMail convertToEnqueuedMail(Mail mail, MailQueueName mailQueueName) {
+ return EnqueuedMail.builder()
+ .mail(mail)
+ .bucketId(computedBucketId(mail))
+ .timeRangeStart(currentSliceStartInstant())
+ .enqueuedTime(Instant.now())
+ .mailKey(MailKey.fromMail(mail))
+ .mailQueueName(mailQueueName)
+ .build();
+ }
+
+ private Instant currentSliceStartInstant() {
+ long sliceSize = configuration.getSliceWindow().getSeconds();
+ long sliceId = clock.instant().getEpochSecond() / sliceSize;
+ return Instant.ofEpochSecond(sliceId * sliceSize);
+ }
+
+ private BucketId computedBucketId(Mail mail) {
+ int mailKeyHashCode = mail.getName().hashCode();
+ int bucketIdValue = mailKeyHashCode % configuration.getBucketCount();
+ return BucketId.of(bucketIdValue);
+ }
+}
http://git-wip-us.apache.org/repos/asf/james-project/blob/3a53806a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueView.java
----------------------------------------------------------------------
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
new file mode 100644
index 0000000..ee309e6
--- /dev/null
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueView.java
@@ -0,0 +1,91 @@
+/****************************************************************
+ * 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.view.cassandra;
+
+import java.util.concurrent.CompletableFuture;
+
+import javax.inject.Inject;
+
+import org.apache.james.queue.api.ManageableMailQueue;
+import org.apache.james.queue.rabbitmq.MailQueueName;
+import org.apache.james.queue.rabbitmq.view.api.MailQueueView;
+import org.apache.mailet.Mail;
+
+import com.google.common.collect.Iterators;
+
+public class CassandraMailQueueView implements MailQueueView {
+
+ public static class Factory {
+ private final CassandraMailQueueMailStore storeHelper;
+ private final CassandraMailQueueBrowser cassandraMailQueueBrowser;
+ private final CassandraMailQueueMailDelete cassandraMailQueueMailDelete;
+
+ @Inject
+ public Factory(CassandraMailQueueMailStore storeHelper, CassandraMailQueueBrowser cassandraMailQueueBrowser, CassandraMailQueueMailDelete cassandraMailQueueMailDelete) {
+ this.storeHelper = storeHelper;
+ this.cassandraMailQueueBrowser = cassandraMailQueueBrowser;
+ this.cassandraMailQueueMailDelete = cassandraMailQueueMailDelete;
+ }
+
+ public MailQueueView create(MailQueueName mailQueueName) {
+ return new CassandraMailQueueView(storeHelper, mailQueueName, cassandraMailQueueBrowser, cassandraMailQueueMailDelete);
+ }
+ }
+
+ private final CassandraMailQueueMailStore storeHelper;
+ private final CassandraMailQueueBrowser cassandraMailQueueBrowser;
+ private final CassandraMailQueueMailDelete cassandraMailQueueMailDelete;
+
+ private final MailQueueName mailQueueName;
+
+ CassandraMailQueueView(CassandraMailQueueMailStore storeHelper,
+ MailQueueName mailQueueName,
+ CassandraMailQueueBrowser cassandraMailQueueBrowser,
+ CassandraMailQueueMailDelete cassandraMailQueueMailDelete) {
+ this.mailQueueName = mailQueueName;
+ this.storeHelper = storeHelper;
+ this.cassandraMailQueueBrowser = cassandraMailQueueBrowser;
+ this.cassandraMailQueueMailDelete = cassandraMailQueueMailDelete;
+ }
+
+ @Override
+ public CompletableFuture<Void> storeMail(Mail mail) {
+ return storeHelper.storeMailInEnqueueTable(mail, mailQueueName);
+ }
+
+ @Override
+ public CompletableFuture<Void> deleteMail(Mail mail) {
+ return cassandraMailQueueMailDelete.considerDeleted(mail, mailQueueName);
+ }
+
+ @Override
+ public ManageableMailQueue.MailQueueIterator browse() {
+ return new CassandraMailQueueBrowser.CassandraMailQueueIterator(
+ cassandraMailQueueBrowser.browse(mailQueueName)
+ .join()
+ .iterator());
+ }
+
+ @Override
+ public long getSize() {
+ return Iterators.size(browse());
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/james-project/blob/3a53806a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/DeletedMailsDAO.java
----------------------------------------------------------------------
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
new file mode 100644
index 0000000..36ee5f4
--- /dev/null
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/DeletedMailsDAO.java
@@ -0,0 +1,83 @@
+/****************************************************************
+ * 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.view.cassandra;
+
+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.QUEUE_NAME;
+import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.DeletedMailTable.TABLE_NAME;
+
+import java.util.concurrent.CompletableFuture;
+
+import javax.inject.Inject;
+
+import org.apache.james.backends.cassandra.utils.CassandraAsyncExecutor;
+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;
+
+class DeletedMailsDAO {
+
+ private final CassandraAsyncExecutor executor;
+ private final PreparedStatement selectOne;
+ private final PreparedStatement insertOne;
+
+ @Inject
+ DeletedMailsDAO(Session session) {
+ this.executor = new CassandraAsyncExecutor(session);
+ this.selectOne = prepareSelectExist(session);
+ this.insertOne = prepareInsert(session);
+ }
+
+ private PreparedStatement prepareInsert(Session session) {
+ return session.prepare(insertInto(TABLE_NAME)
+ .value(QUEUE_NAME, bindMarker(QUEUE_NAME))
+ .value(MAIL_KEY, bindMarker(MAIL_KEY)));
+ }
+
+ 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))));
+ }
+
+ CompletableFuture<Void> markAsDeleted(MailQueueName mailQueueName, MailKey mailKey) {
+ return executor.executeVoid(insertOne.bind()
+ .setString(QUEUE_NAME, mailQueueName.asString())
+ .setString(MAIL_KEY, mailKey.getMailKey()));
+ }
+
+ CompletableFuture<Boolean> isDeleted(MailQueueName mailQueueName, MailKey mailKey) {
+ return executor.executeReturnExists(
+ selectOne.bind()
+ .setString(QUEUE_NAME, mailQueueName.asString())
+ .setString(MAIL_KEY, mailKey.getMailKey()));
+ }
+
+ CompletableFuture<Boolean> isStillEnqueued(MailQueueName mailQueueName, MailKey mailKey) {
+ return isDeleted(mailQueueName, mailKey).thenApply(b -> !b);
+ }
+}
http://git-wip-us.apache.org/repos/asf/james-project/blob/3a53806a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/EnqueuedMailsDAO.java
----------------------------------------------------------------------
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
new file mode 100644
index 0000000..2b8e11b
--- /dev/null
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/EnqueuedMailsDAO.java
@@ -0,0 +1,144 @@
+/****************************************************************
+ * 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.view.cassandra;
+
+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.EnqueuedMailsTable.ATTRIBUTES;
+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.ERROR_MESSAGE;
+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.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;
+import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.REMOTE_ADDR;
+import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.REMOTE_HOST;
+import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.SENDER;
+import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.STATE;
+import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.TABLE_NAME;
+import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.TIME_RANGE_START;
+import static org.apache.james.queue.rabbitmq.view.cassandra.EnqueuedMailsDaoUtil.asStringList;
+import static org.apache.james.queue.rabbitmq.view.cassandra.EnqueuedMailsDaoUtil.toHeaderMap;
+import static org.apache.james.queue.rabbitmq.view.cassandra.EnqueuedMailsDaoUtil.toRawAttributeMap;
+import static org.apache.james.queue.rabbitmq.view.cassandra.model.BucketedSlices.BucketId;
+import static org.apache.james.queue.rabbitmq.view.cassandra.model.BucketedSlices.Slice;
+
+import java.util.Date;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.Stream;
+
+import javax.inject.Inject;
+
+import org.apache.james.backends.cassandra.init.CassandraTypesProvider;
+import org.apache.james.backends.cassandra.utils.CassandraAsyncExecutor;
+import org.apache.james.backends.cassandra.utils.CassandraUtils;
+import org.apache.james.core.MailAddress;
+import org.apache.james.queue.rabbitmq.MailQueueName;
+import org.apache.james.queue.rabbitmq.view.cassandra.model.EnqueuedMail;
+import org.apache.mailet.Mail;
+
+import com.datastax.driver.core.PreparedStatement;
+import com.datastax.driver.core.Session;
+
+class EnqueuedMailsDAO {
+
+ private final CassandraAsyncExecutor executor;
+ private final PreparedStatement selectFrom;
+ private final PreparedStatement insert;
+ private final CassandraUtils cassandraUtils;
+ private final CassandraTypesProvider cassandraTypesProvider;
+
+ @Inject
+ EnqueuedMailsDAO(Session session, CassandraUtils cassandraUtils, CassandraTypesProvider cassandraTypesProvider) {
+ this.executor = new CassandraAsyncExecutor(session);
+ this.cassandraUtils = cassandraUtils;
+ this.cassandraTypesProvider = cassandraTypesProvider;
+
+ this.selectFrom = prepareSelectFrom(session);
+ this.insert = prepareInsert(session);
+ }
+
+ private PreparedStatement prepareSelectFrom(Session session) {
+ return session.prepare(select()
+ .from(TABLE_NAME)
+ .where(eq(QUEUE_NAME, bindMarker(QUEUE_NAME)))
+ .and(eq(TIME_RANGE_START, bindMarker(TIME_RANGE_START)))
+ .and(eq(BUCKET_ID, bindMarker(BUCKET_ID))));
+ }
+
+ private PreparedStatement prepareInsert(Session session) {
+ return session.prepare(insertInto(TABLE_NAME)
+ .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(ENQUEUED_TIME, bindMarker(ENQUEUED_TIME))
+ .value(STATE, bindMarker(STATE))
+ .value(SENDER, bindMarker(SENDER))
+ .value(RECIPIENTS, bindMarker(RECIPIENTS))
+ .value(ATTRIBUTES, bindMarker(ATTRIBUTES))
+ .value(ERROR_MESSAGE, bindMarker(ERROR_MESSAGE))
+ .value(REMOTE_ADDR, bindMarker(REMOTE_ADDR))
+ .value(REMOTE_HOST, bindMarker(REMOTE_HOST))
+ .value(LAST_UPDATED, bindMarker(LAST_UPDATED))
+ .value(PER_RECIPIENT_SPECIFIC_HEADERS, bindMarker(PER_RECIPIENT_SPECIFIC_HEADERS)));
+ }
+
+ CompletableFuture<Void> insert(EnqueuedMail enqueuedMail) {
+ Mail mail = enqueuedMail.getMail();
+
+ return executor.executeVoid(insert.bind()
+ .setString(QUEUE_NAME, enqueuedMail.getMailQueueName().asString())
+ .setTimestamp(TIME_RANGE_START, Date.from(enqueuedMail.getTimeRangeStart()))
+ .setInt(BUCKET_ID, enqueuedMail.getBucketId().getValue())
+ .setTimestamp(ENQUEUED_TIME, Date.from(enqueuedMail.getEnqueuedTime()))
+ .setString(MAIL_KEY, mail.getName())
+ .setString(STATE, mail.getState())
+ .setString(SENDER, Optional.ofNullable(mail.getSender())
+ .map(MailAddress::asString)
+ .orElse(null))
+ .setList(RECIPIENTS, asStringList(mail.getRecipients()))
+ .setString(ERROR_MESSAGE, mail.getErrorMessage())
+ .setString(REMOTE_ADDR, mail.getRemoteAddr())
+ .setString(REMOTE_HOST, mail.getRemoteHost())
+ .setTimestamp(LAST_UPDATED, mail.getLastUpdated())
+ .setMap(ATTRIBUTES, toRawAttributeMap(mail))
+ .setMap(PER_RECIPIENT_SPECIFIC_HEADERS, toHeaderMap(cassandraTypesProvider, mail.getPerRecipientSpecificHeaders()))
+ );
+ }
+
+ CompletableFuture<Stream<EnqueuedMail>> selectEnqueuedMails(
+ MailQueueName queueName, Slice slice, BucketId bucketId) {
+
+ return executor.execute(
+ selectFrom.bind()
+ .setString(QUEUE_NAME, queueName.asString())
+ .setTimestamp(TIME_RANGE_START, Date.from(slice.getStartSliceInstant()))
+ .setInt(BUCKET_ID, bucketId.getValue()))
+ .thenApply(resultSet -> cassandraUtils.convertToStream(resultSet)
+ .map(EnqueuedMailsDaoUtil::toEnqueuedMail));
+ }
+
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[2/5] james-project git commit: JAMES-2544 RabbitMQ browse Cassandra
DAOs and tests
Posted by bt...@apache.org.
http://git-wip-us.apache.org/repos/asf/james-project/blob/3a53806a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/EnqueuedMailsDaoUtil.java
----------------------------------------------------------------------
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
new file mode 100644
index 0000000..6be8597
--- /dev/null
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/EnqueuedMailsDaoUtil.java
@@ -0,0 +1,196 @@
+/****************************************************************
+ * 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.view.cassandra;
+
+import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.ATTRIBUTES;
+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.ERROR_MESSAGE;
+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.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;
+import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.REMOTE_ADDR;
+import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.REMOTE_HOST;
+import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.SENDER;
+import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.STATE;
+import static org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule.EnqueuedMailsTable.TIME_RANGE_START;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.io.Serializable;
+import java.nio.ByteBuffer;
+import java.time.Instant;
+import java.util.Collection;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import javax.mail.internet.AddressException;
+
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.james.backends.cassandra.init.CassandraTypesProvider;
+import org.apache.james.core.MailAddress;
+import org.apache.james.queue.rabbitmq.MailQueueName;
+import org.apache.james.queue.rabbitmq.view.cassandra.model.BucketedSlices;
+import org.apache.james.queue.rabbitmq.view.cassandra.model.EnqueuedMail;
+import org.apache.james.queue.rabbitmq.view.cassandra.model.MailKey;
+import org.apache.james.server.core.MailImpl;
+import org.apache.james.util.streams.Iterators;
+import org.apache.mailet.Mail;
+import org.apache.mailet.PerRecipientHeaders;
+
+import com.datastax.driver.core.Row;
+import com.datastax.driver.core.UDTValue;
+import com.github.fge.lambdas.Throwing;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+
+public class EnqueuedMailsDaoUtil {
+
+ static EnqueuedMail toEnqueuedMail(Row row) {
+ MailQueueName queueName = MailQueueName.fromString(row.getString(QUEUE_NAME));
+ 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();
+
+ MailAddress sender = Optional.ofNullable(row.getString(SENDER))
+ .map(Throwing.function(MailAddress::new))
+ .orElse(null);
+ List<MailAddress> recipients = row.getList(RECIPIENTS, String.class)
+ .stream()
+ .map(Throwing.function(MailAddress::new))
+ .collect(ImmutableList.toImmutableList());
+ String state = row.getString(STATE);
+ String remoteAddr = row.getString(REMOTE_ADDR);
+ String remoteHost = row.getString(REMOTE_HOST);
+ String errorMessage = row.getString(ERROR_MESSAGE);
+ String name = row.getString(MAIL_KEY);
+ 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));
+
+ MailImpl mail = MailImpl.builder()
+ .name(name)
+ .sender(sender)
+ .recipients(recipients)
+ .lastUpdated(lastUpdated)
+ .errorMessage(errorMessage)
+ .remoteHost(remoteHost)
+ .remoteAddr(remoteAddr)
+ .state(state)
+ .addAllHeadersForRecipients(perRecipientHeaders)
+ .attributes(toAttributes(rawAttributes))
+ .build();
+
+ return EnqueuedMail.builder()
+ .mail(mail)
+ .bucketId(bucketId)
+ .timeRangeStart(timeRangeStart)
+ .enqueuedTime(enqueuedTime)
+ .mailKey(MailKey.of(name))
+ .mailQueueName(queueName)
+ .build();
+ }
+
+ private static Map<String, Serializable> toAttributes(Map<String, ByteBuffer> rowAttributes) {
+ return rowAttributes.entrySet()
+ .stream()
+ .collect(ImmutableMap.toImmutableMap(
+ Map.Entry::getKey,
+ entry -> fromByteBuffer(entry.getValue())));
+ }
+
+ private static Serializable fromByteBuffer(ByteBuffer byteBuffer) {
+ try {
+ byte[] data = new byte[byteBuffer.remaining()];
+ byteBuffer.get(data);
+ ObjectInputStream objectInputStream = new ObjectInputStream(new ByteArrayInputStream(data));
+ return (Serializable) objectInputStream.readObject();
+ } catch (IOException | ClassNotFoundException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static PerRecipientHeaders fromHeaderMap(Map<String, UDTValue> rawMap) {
+ PerRecipientHeaders result = new PerRecipientHeaders();
+
+ rawMap.forEach((key, value) -> result.addHeaderForRecipient(PerRecipientHeaders.Header.builder()
+ .name(value.getString(HEADER_NAME))
+ .value(value.getString(HEADER_VALUE))
+ .build(),
+ toMailAddress(key)));
+ return result;
+ }
+
+ private static MailAddress toMailAddress(String rawValue) {
+ try {
+ return new MailAddress(rawValue);
+ } catch (AddressException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ static ImmutableList<String> asStringList(Collection<MailAddress> mailAddresses) {
+ return mailAddresses.stream()
+ .map(MailAddress::asString)
+ .collect(ImmutableList.toImmutableList());
+ }
+
+ static ImmutableMap<String, ByteBuffer> toRawAttributeMap(Mail mail) {
+ return Iterators.toStream(mail.getAttributeNames())
+ .map(name -> Pair.of(name, mail.getAttribute(name)))
+ .map(pair -> Pair.of(pair.getLeft(), toByteBuffer(pair.getRight())))
+ .collect(ImmutableMap.toImmutableMap(Pair::getLeft, Pair::getRight));
+ }
+
+ private static ByteBuffer toByteBuffer(Serializable serializable) {
+ try {
+ ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+ new ObjectOutputStream(outputStream).writeObject(serializable);
+ return ByteBuffer.wrap(outputStream.toByteArray());
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ static ImmutableMap<String, UDTValue> toHeaderMap(CassandraTypesProvider cassandraTypesProvider,
+ PerRecipientHeaders perRecipientHeaders) {
+ return perRecipientHeaders.getHeadersByRecipient()
+ .asMap()
+ .entrySet()
+ .stream()
+ .flatMap(entry -> entry.getValue().stream().map(value -> Pair.of(entry.getKey(), value)))
+ .map(entry -> Pair.of(entry.getKey().asString(),
+ cassandraTypesProvider.getDefinedUserType(HEADER_TYPE)
+ .newValue()
+ .setString(HEADER_NAME, entry.getRight().getName())
+ .setString(HEADER_VALUE, entry.getRight().getValue())))
+ .collect(ImmutableMap.toImmutableMap(Pair::getLeft, Pair::getRight));
+ }
+}
http://git-wip-us.apache.org/repos/asf/james-project/blob/3a53806a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/model/BucketedSlices.java
----------------------------------------------------------------------
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/model/BucketedSlices.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/model/BucketedSlices.java
index f7d5ac1..1f5d3d5 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/model/BucketedSlices.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/model/BucketedSlices.java
@@ -19,6 +19,7 @@
package org.apache.james.queue.rabbitmq.view.cassandra.model;
+import java.time.Duration;
import java.time.Instant;
import java.util.Objects;
import java.util.stream.LongStream;
@@ -37,7 +38,7 @@ public class BucketedSlices {
private final int value;
private BucketId(int value) {
- Preconditions.checkArgument(value >= 0, "sliceWindowSizeInSecond should not be negative");
+ Preconditions.checkArgument(value >= 0, "bucketId should not be negative");
this.value = value;
}
@@ -64,19 +65,19 @@ public class BucketedSlices {
public static class Slice {
- public static Slice of(Instant sliceStartInstant, long sliceWindowSizeInSecond) {
- return new Slice(sliceStartInstant, sliceWindowSizeInSecond);
+ public static Slice of(Instant sliceStartInstant, Duration sliceWindowSize) {
+ return new Slice(sliceStartInstant, sliceWindowSize);
}
public static Stream<Slice> allSlicesTill(Slice firstSlice, Instant endAt) {
long sliceCount = calculateSliceCount(firstSlice, endAt);
long startAtSeconds = firstSlice.getStartSliceInstant().getEpochSecond();
- long sliceWindowSizeInSecond = firstSlice.getSliceWindowSizeInSecond();
+ long sliceWindowSizeInSecond = firstSlice.getSliceWindowSize().getSeconds();
return LongStream.range(0, sliceCount)
.map(slicePosition -> startAtSeconds + sliceWindowSizeInSecond * slicePosition)
.mapToObj(Instant::ofEpochSecond)
- .map(sliceStartInstant -> Slice.of(sliceStartInstant, firstSlice.getSliceWindowSizeInSecond()));
+ .map(sliceStartInstant -> Slice.of(sliceStartInstant, firstSlice.getSliceWindowSize()));
}
private static long calculateSliceCount(Slice firstSlice, Instant endAt) {
@@ -87,27 +88,27 @@ public class BucketedSlices {
if (timeDiffInSecond < 0) {
return 0;
} else {
- return (timeDiffInSecond / firstSlice.sliceWindowSizeInSecond) + 1;
+ return (timeDiffInSecond / firstSlice.getSliceWindowSize().getSeconds()) + 1;
}
}
private final Instant startSliceInstant;
- private final long sliceWindowSizeInSecond;
+ private final Duration sliceWindowSize;
- private Slice(Instant startSliceInstant, long sliceWindowSizeInSecond) {
+ private Slice(Instant startSliceInstant, Duration sliceWindowSize) {
Preconditions.checkNotNull(startSliceInstant);
- Preconditions.checkArgument(sliceWindowSizeInSecond > 0, "sliceWindowSizeInSecond should be positive");
+ Preconditions.checkNotNull(sliceWindowSize);
this.startSliceInstant = startSliceInstant;
- this.sliceWindowSizeInSecond = sliceWindowSizeInSecond;
+ this.sliceWindowSize = sliceWindowSize;
}
public Instant getStartSliceInstant() {
return startSliceInstant;
}
- public long getSliceWindowSizeInSecond() {
- return sliceWindowSizeInSecond;
+ public Duration getSliceWindowSize() {
+ return sliceWindowSize;
}
@Override
@@ -115,7 +116,7 @@ public class BucketedSlices {
if (o instanceof Slice) {
Slice slice = (Slice) o;
- return Objects.equals(this.sliceWindowSizeInSecond, slice.sliceWindowSizeInSecond)
+ return Objects.equals(this.sliceWindowSize, slice.sliceWindowSize)
&& Objects.equals(this.startSliceInstant, slice.startSliceInstant);
}
return false;
@@ -123,7 +124,7 @@ public class BucketedSlices {
@Override
public final int hashCode() {
- return Objects.hash(startSliceInstant, sliceWindowSizeInSecond);
+ return Objects.hash(startSliceInstant, sliceWindowSize);
}
}
}
http://git-wip-us.apache.org/repos/asf/james-project/blob/3a53806a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/model/EnqueuedMail.java
----------------------------------------------------------------------
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/model/EnqueuedMail.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/model/EnqueuedMail.java
index 6e022b3..93d8e8b 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/model/EnqueuedMail.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/model/EnqueuedMail.java
@@ -20,7 +20,6 @@
package org.apache.james.queue.rabbitmq.view.cassandra.model;
import java.time.Instant;
-import java.util.Comparator;
import java.util.Objects;
import org.apache.james.queue.rabbitmq.MailQueueName;
@@ -90,10 +89,6 @@ public class EnqueuedMail {
new Builder.LastStage(mail, bucketId, timeRangeStart, enqueuedTime, mailKey, mailQueueName);
}
- public static Comparator<EnqueuedMail> getEnqueuedTimeComparator() {
- return Comparator.comparing(EnqueuedMail::getEnqueuedTime);
- }
-
private final Mail mail;
private final BucketedSlices.BucketId bucketId;
private final Instant timeRangeStart;
@@ -143,6 +138,7 @@ public class EnqueuedMail {
return Objects.equals(this.bucketId, that.bucketId)
&& Objects.equals(this.mail, that.mail)
&& Objects.equals(this.timeRangeStart, that.timeRangeStart)
+ && Objects.equals(this.enqueuedTime, that.enqueuedTime)
&& Objects.equals(this.mailKey, that.mailKey)
&& Objects.equals(this.mailQueueName, that.mailQueueName);
}
@@ -151,6 +147,6 @@ public class EnqueuedMail {
@Override
public final int hashCode() {
- return Objects.hash(mail, bucketId, timeRangeStart, mailKey, mailQueueName);
+ return Objects.hash(mail, bucketId, timeRangeStart, enqueuedTime, mailKey, mailQueueName);
}
}
http://git-wip-us.apache.org/repos/asf/james-project/blob/3a53806a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/model/MailKey.java
----------------------------------------------------------------------
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/view/cassandra/model/MailKey.java
index c39dc16..4f17a71 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/view/cassandra/model/MailKey.java
@@ -19,6 +19,8 @@
package org.apache.james.queue.rabbitmq.view.cassandra.model;
+import java.util.Objects;
+
import org.apache.mailet.Mail;
import com.google.common.base.Preconditions;
@@ -45,4 +47,19 @@ public class MailKey {
public String getMailKey() {
return mailKey;
}
+
+ @Override
+ public final boolean equals(Object o) {
+ if (o instanceof MailKey) {
+ MailKey mailKey1 = (MailKey) o;
+
+ return Objects.equals(this.mailKey, mailKey1.mailKey);
+ }
+ return false;
+ }
+
+ @Override
+ public final int hashCode() {
+ return Objects.hash(mailKey);
+ }
}
http://git-wip-us.apache.org/repos/asf/james-project/blob/3a53806a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java
----------------------------------------------------------------------
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 f883883..92b384a 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
@@ -19,11 +19,23 @@
package org.apache.james.queue.rabbitmq;
+import static java.time.temporal.ChronoUnit.HOURS;
+import static org.apache.james.queue.api.Mails.defaultMail;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
+import java.time.Clock;
+import java.time.Duration;
+import java.time.Instant;
import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeoutException;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
import javax.mail.internet.MimeMessage;
@@ -34,7 +46,8 @@ import org.apache.james.backend.rabbitmq.RabbitMQConfiguration;
import org.apache.james.backend.rabbitmq.RabbitMQConnectionFactory;
import org.apache.james.backend.rabbitmq.ReusableDockerRabbitMQExtension;
import org.apache.james.backends.cassandra.CassandraCluster;
-import org.apache.james.backends.cassandra.DockerCassandraExtension;
+import org.apache.james.backends.cassandra.CassandraClusterExtension;
+import org.apache.james.backends.cassandra.components.CassandraModule;
import org.apache.james.backends.cassandra.init.configuration.CassandraConfiguration;
import org.apache.james.blob.api.HashBlobId;
import org.apache.james.blob.api.Store;
@@ -43,33 +56,51 @@ import org.apache.james.blob.cassandra.CassandraBlobsDAO;
import org.apache.james.blob.mail.MimeMessagePartsId;
import org.apache.james.blob.mail.MimeMessageStore;
import org.apache.james.queue.api.MailQueue;
-import org.apache.james.queue.api.MailQueueContract;
import org.apache.james.queue.api.MailQueueMetricContract;
import org.apache.james.queue.api.MailQueueMetricExtension;
+import org.apache.james.queue.api.ManageableMailQueue;
+import org.apache.james.queue.api.ManageableMailQueueContract;
+import org.apache.james.queue.rabbitmq.view.api.MailQueueView;
+import org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewConfiguration;
+import org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewModule;
+import org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueViewTestFactory;
+import org.apache.james.util.streams.Iterators;
+import org.apache.mailet.Mail;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import com.github.fge.lambdas.Throwing;
import com.nurkiewicz.asyncretry.AsyncRetryExecutor;
-@ExtendWith({ReusableDockerRabbitMQExtension.class, DockerCassandraExtension.class})
-public class RabbitMQMailQueueTest implements MailQueueContract, MailQueueMetricContract {
+@ExtendWith(ReusableDockerRabbitMQExtension.class)
+public class RabbitMQMailQueueTest implements ManageableMailQueueContract, MailQueueMetricContract {
private static final HashBlobId.Factory BLOB_ID_FACTORY = new HashBlobId.Factory();
+ private static final int THREE_BUCKET_COUNT = 3;
+ private static final int UPDATE_BROWSE_START_PACE = 2;
+ private static final Duration ONE_HOUR_SLICE_WINDOW = Duration.ofHours(1);
+ private static final String SPOOL = "spool";
+ private static final Instant IN_SLICE_1 = Instant.parse("2007-12-03T10:15:30.00Z");
+ private static final Instant IN_SLICE_2 = IN_SLICE_1.plus(1, HOURS);
+ private static final Instant IN_SLICE_3 = IN_SLICE_1.plus(2, HOURS);
+ private static final Instant IN_SLICE_5 = IN_SLICE_1.plus(4, HOURS);
+ private static final Instant IN_SLICE_6 = IN_SLICE_1.plus(6, HOURS);
- private static CassandraCluster cassandra;
- private RabbitMQMailQueueFactory mailQueueFactory;
+ @RegisterExtension
+ static CassandraClusterExtension cassandraCluster = new CassandraClusterExtension(CassandraModule.aggregateModules(
+ CassandraBlobModule.MODULE,
+ CassandraMailQueueViewModule.MODULE));
- @BeforeAll
- static void setUpClass(DockerCassandraExtension.DockerCassandra dockerCassandra) {
- cassandra = CassandraCluster.create(CassandraBlobModule.MODULE, dockerCassandra.getHost());
- }
+ private RabbitMQMailQueueFactory mailQueueFactory;
+ private Clock clock;
@BeforeEach
- void setup(DockerRabbitMQ rabbitMQ, MailQueueMetricExtension.MailQueueMetricTestSystem metricTestSystem) throws IOException, TimeoutException, URISyntaxException {
+ void setup(DockerRabbitMQ rabbitMQ, CassandraCluster cassandra, MailQueueMetricExtension.MailQueueMetricTestSystem metricTestSystem) throws IOException, TimeoutException, URISyntaxException {
CassandraBlobsDAO blobsDAO = new CassandraBlobsDAO(cassandra.getConf(), CassandraConfiguration.DEFAULT_CONFIGURATION, BLOB_ID_FACTORY);
Store<MimeMessage, MimeMessagePartsId> mimeMessageStore = MimeMessageStore.factory(blobsDAO).mimeMessageStore();
@@ -83,44 +114,217 @@ public class RabbitMQMailQueueTest implements MailQueueContract, MailQueueMetric
.setHost(rabbitMQ.getHostIp())
.setPort(rabbitMQ.getAdminPort())
.build();
+ clock = mock(Clock.class);
+ when(clock.instant()).thenReturn(Instant.parse("2007-12-03T10:15:30.00Z"));
+ ThreadLocalRandom random = ThreadLocalRandom.current();
+
+ MailQueueView mailQueueView = CassandraMailQueueViewTestFactory.factory(clock, random, cassandra.getConf(), cassandra.getTypesProvider(),
+ CassandraMailQueueViewConfiguration.builder()
+ .bucketCount(THREE_BUCKET_COUNT)
+ .updateBrowseStartPace(UPDATE_BROWSE_START_PACE)
+ .sliceWindow(ONE_HOUR_SLICE_WINDOW)
+ .build())
+ .create(MailQueueName.fromString(SPOOL));
RabbitMQConfiguration rabbitMQConfiguration = RabbitMQConfiguration.builder()
.amqpUri(amqpUri)
.managementUri(rabbitManagementUri)
.build();
- RabbitMQConnectionFactory rabbitMQConnectionFactory = new RabbitMQConnectionFactory(
- rabbitMQConfiguration,
- new AsyncRetryExecutor(Executors.newSingleThreadScheduledExecutor()));
+ RabbitMQConnectionFactory rabbitMQConnectionFactory = new RabbitMQConnectionFactory(rabbitMQConfiguration,
+ new AsyncRetryExecutor(Executors.newSingleThreadScheduledExecutor()));
RabbitClient rabbitClient = new RabbitClient(new RabbitChannelPool(rabbitMQConnectionFactory));
RabbitMQMailQueue.Factory factory = new RabbitMQMailQueue.Factory(
- metricTestSystem.getSpyMetricFactory(),
- rabbitClient,
- mimeMessageStore,
- BLOB_ID_FACTORY);
+ metricTestSystem.getSpyMetricFactory(),
+ rabbitClient,
+ mimeMessageStore,
+ BLOB_ID_FACTORY,
+ mailQueueView);
+
RabbitMQManagementApi mqManagementApi = new RabbitMQManagementApi(rabbitManagementUri, new RabbitMQManagementCredentials("guest", "guest".toCharArray()));
mailQueueFactory = new RabbitMQMailQueueFactory(rabbitClient, mqManagementApi, factory);
}
@AfterEach
- void tearDown() {
+ void tearDown(CassandraCluster cassandra) {
cassandra.clearTables();
}
@AfterAll
- static void tearDownClass() {
+ static void tearDownClass(CassandraCluster cassandra) {
cassandra.closeCluster();
}
@Override
public MailQueue getMailQueue() {
- return mailQueueFactory.createQueue("spool");
+ return mailQueueFactory.createQueue(SPOOL);
+ }
+
+ @Override
+ public ManageableMailQueue getManageableMailQueue() {
+ return mailQueueFactory.createQueue(SPOOL);
+ }
+
+ @Test
+ void browseShouldReturnCurrentlyEnqueuedMailFromAllSlices() throws Exception {
+ ManageableMailQueue mailQueue = getManageableMailQueue();
+ int emailCount = 5;
+
+ when(clock.instant()).thenReturn(IN_SLICE_1);
+ enqueueMailsInSlice(1, emailCount);
+
+ when(clock.instant()).thenReturn(IN_SLICE_2);
+ enqueueMailsInSlice(2, emailCount);
+
+ when(clock.instant()).thenReturn(IN_SLICE_3);
+ enqueueMailsInSlice(3, emailCount);
+
+ when(clock.instant()).thenReturn(IN_SLICE_5);
+ enqueueMailsInSlice(5, emailCount);
+
+ when(clock.instant()).thenReturn(IN_SLICE_6);
+ Stream<String> names = Iterators.toStream(mailQueue.browse())
+ .map(ManageableMailQueue.MailQueueItemView::getMail)
+ .map(Mail::getName);
+
+ assertThat(names).containsExactly(
+ "1-1", "1-2", "1-3", "1-4", "1-5",
+ "2-1", "2-2", "2-3", "2-4", "2-5",
+ "3-1", "3-2", "3-3", "3-4", "3-5",
+ "5-1", "5-2", "5-3", "5-4", "5-5");
+ }
+
+ @Test
+ void browseAndDequeueShouldCombineWellWhenDifferentSlices() throws Exception {
+ ManageableMailQueue mailQueue = getManageableMailQueue();
+ int emailCount = 5;
+
+ when(clock.instant()).thenReturn(IN_SLICE_1);
+ enqueueMailsInSlice(1, emailCount);
+
+ when(clock.instant()).thenReturn(IN_SLICE_2);
+ enqueueMailsInSlice(2, emailCount);
+
+ when(clock.instant()).thenReturn(IN_SLICE_3);
+ enqueueMailsInSlice(3, emailCount);
+
+ when(clock.instant()).thenReturn(IN_SLICE_5);
+ enqueueMailsInSlice(5, emailCount);
+
+ when(clock.instant()).thenReturn(IN_SLICE_6);
+
+ dequeueMails(5);
+
+ dequeueMails(3);
+ MailQueue.MailQueueItem item2_4 = mailQueue.deQueue();
+ item2_4.done(false);
+ dequeueMails(1);
+
+ dequeueMails(5);
+
+ Stream<String> names = Iterators.toStream(mailQueue.browse())
+ .map(ManageableMailQueue.MailQueueItemView::getMail)
+ .map(Mail::getName);
+
+ assertThat(names)
+ .containsExactly("2-4", "5-1", "5-2", "5-3", "5-4", "5-5");
+ }
+
+ @Disabled
+ @Override
+ public void clearShouldNotFailWhenBrowsingIterating() {
+
+ }
+
+ @Disabled
+ @Override
+ public void browseShouldNotFailWhenConcurrentClearWhenIterating() {
+
+ }
+
+ @Disabled
+ @Override
+ public void removeShouldNotFailWhenBrowsingIterating() {
+
+ }
+
+ @Disabled
+ @Override
+ public void browseShouldNotFailWhenConcurrentRemoveWhenIterating() {
+
+ }
+
+ @Disabled
+ @Override
+ public void removeByNameShouldRemoveSpecificEmail() {
+
+ }
+
+ @Disabled
+ @Override
+ public void removeBySenderShouldRemoveSpecificEmail() {
+
+ }
+
+ @Disabled
+ @Override
+ public void removeByRecipientShouldRemoveSpecificEmail() {
+
+ }
+
+ @Disabled
+ @Override
+ public void removeByRecipientShouldRemoveSpecificEmailWhenMultipleRecipients() {
+
+ }
+
+ @Disabled
+ @Override
+ public void removeByNameShouldNotFailWhenQueueIsEmpty() {
+
+ }
+
+ @Disabled
+ @Override
+ public void removeBySenderShouldNotFailWhenQueueIsEmpty() {
+
+ }
+
+ @Disabled
+ @Override
+ public void removeByRecipientShouldNotFailWhenQueueIsEmpty() {
+
+ }
+
+ @Disabled
+ @Override
+ public void clearShouldNotFailWhenQueueIsEmpty() {
+
+ }
+
+ @Disabled
+ @Override
+ public void clearShouldRemoveAllElements() {
}
@Disabled("RabbitMQ Mail Queue do not yet implement getSize()")
@Override
public void constructorShouldRegisterGetQueueSizeGauge(MailQueueMetricExtension.MailQueueMetricTestSystem testSystem) {
+ }
+
+ private void enqueueMailsInSlice(int slice, int emailCount) {
+ ManageableMailQueue mailQueue = getManageableMailQueue();
+
+ IntStream.rangeClosed(1, emailCount)
+ .forEach(Throwing.intConsumer(bucketId -> mailQueue.enQueue(defaultMail()
+ .name(slice + "-" + bucketId)
+ .build())));
+ }
+ private void dequeueMails(int times) {
+ ManageableMailQueue mailQueue = getManageableMailQueue();
+ IntStream.rangeClosed(1, times)
+ .forEach(Throwing.intConsumer(bucketId -> mailQueue.deQueue().done(true)));
}
}
http://git-wip-us.apache.org/repos/asf/james-project/blob/3a53806a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMqMailQueueFactoryTest.java
----------------------------------------------------------------------
diff --git a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMqMailQueueFactoryTest.java b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMqMailQueueFactoryTest.java
index 18cca28..bf7d37e 100644
--- a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMqMailQueueFactoryTest.java
+++ b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMqMailQueueFactoryTest.java
@@ -36,11 +36,12 @@ import org.apache.james.backend.rabbitmq.RabbitMQConfiguration;
import org.apache.james.backend.rabbitmq.RabbitMQConnectionFactory;
import org.apache.james.backend.rabbitmq.ReusableDockerRabbitMQExtension;
import org.apache.james.blob.api.HashBlobId;
+import org.apache.james.metrics.api.NoopMetricFactory;
import org.apache.james.blob.api.Store;
import org.apache.james.blob.mail.MimeMessagePartsId;
-import org.apache.james.metrics.api.NoopMetricFactory;
import org.apache.james.queue.api.MailQueueFactory;
import org.apache.james.queue.api.MailQueueFactoryContract;
+import org.apache.james.queue.rabbitmq.view.api.MailQueueView;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.extension.ExtendWith;
@@ -55,6 +56,7 @@ class RabbitMqMailQueueFactoryTest implements MailQueueFactoryContract<RabbitMQM
@BeforeEach
void setup(DockerRabbitMQ rabbitMQ) throws IOException, TimeoutException, URISyntaxException {
Store<MimeMessage, MimeMessagePartsId> mimeMessageStore = mock(Store.class);
+ MailQueueView mailQueueView = mock(MailQueueView.class);
URI amqpUri = new URIBuilder()
.setScheme("amqp")
@@ -78,8 +80,7 @@ class RabbitMqMailQueueFactoryTest implements MailQueueFactoryContract<RabbitMQM
new AsyncRetryExecutor(Executors.newSingleThreadScheduledExecutor()));
RabbitClient rabbitClient = new RabbitClient(new RabbitChannelPool(rabbitMQConnectionFactory));
- RabbitMQMailQueue.Factory factory = new RabbitMQMailQueue.Factory(new NoopMetricFactory(), rabbitClient, mimeMessageStore, BLOB_ID_FACTORY);
-
+ RabbitMQMailQueue.Factory factory = new RabbitMQMailQueue.Factory(new NoopMetricFactory(), rabbitClient, mimeMessageStore, BLOB_ID_FACTORY, mailQueueView);
RabbitMQManagementApi mqManagementApi = new RabbitMQManagementApi(rabbitManagementUri, new RabbitMQManagementCredentials("guest", "guest".toCharArray()));
mailQueueFactory = new RabbitMQMailQueueFactory(rabbitClient, mqManagementApi, factory);
}
@@ -88,4 +89,4 @@ class RabbitMqMailQueueFactoryTest implements MailQueueFactoryContract<RabbitMQM
public MailQueueFactory<RabbitMQMailQueue> getMailQueueFactory() {
return mailQueueFactory;
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/james-project/blob/3a53806a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/BrowseStartDAOTest.java
----------------------------------------------------------------------
diff --git a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/BrowseStartDAOTest.java b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/BrowseStartDAOTest.java
new file mode 100644
index 0000000..ef844c6
--- /dev/null
+++ b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/BrowseStartDAOTest.java
@@ -0,0 +1,86 @@
+/****************************************************************
+ * 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.view.cassandra;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.time.Instant;
+import java.util.Optional;
+
+import org.apache.james.backends.cassandra.CassandraCluster;
+import org.apache.james.backends.cassandra.CassandraClusterExtension;
+import org.apache.james.queue.rabbitmq.MailQueueName;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+class BrowseStartDAOTest {
+
+ 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 Instant NOW = Instant.now();
+ private static final Instant NOW_PLUS_TEN_SECONDS = NOW.plusSeconds(10);
+
+ @RegisterExtension
+ static CassandraClusterExtension cassandraCluster = new CassandraClusterExtension(CassandraMailQueueViewModule.MODULE);
+
+ private BrowseStartDAO testee;
+
+ @BeforeEach
+ void setUp(CassandraCluster cassandra) {
+ testee = new BrowseStartDAO(cassandra.getConf());
+ }
+
+ @Test
+ void findBrowseStartShouldReturnEmptyWhenTableDoesntContainQueueName() {
+ testee.updateBrowseStart(OUT_GOING_1, NOW).join();
+
+ Optional<Instant> firstEnqueuedItemFromQueue2 = testee.findBrowseStart(OUT_GOING_2).join();
+ assertThat(firstEnqueuedItemFromQueue2)
+ .isEmpty();
+ }
+
+ @Test
+ void findBrowseStartShouldReturnInstantWhenTableContainsQueueName() {
+ testee.updateBrowseStart(OUT_GOING_1, NOW).join();
+ testee.updateBrowseStart(OUT_GOING_2, NOW).join();
+
+ Optional<Instant> firstEnqueuedItemFromQueue2 = testee.findBrowseStart(OUT_GOING_2).join();
+ assertThat(firstEnqueuedItemFromQueue2)
+ .isNotEmpty();
+ }
+
+ @Test
+ void updateFirstEnqueuedTimeShouldWork() {
+ testee.updateBrowseStart(OUT_GOING_1, NOW).join();
+
+ assertThat(testee.selectOne(OUT_GOING_1).join())
+ .isNotEmpty();
+ }
+
+ @Test
+ void insertInitialBrowseStartShouldInsertFirstInstant() {
+ testee.insertInitialBrowseStart(OUT_GOING_1, NOW).join();
+ testee.insertInitialBrowseStart(OUT_GOING_1, NOW_PLUS_TEN_SECONDS).join();
+
+ assertThat(testee.findBrowseStart(OUT_GOING_1).join())
+ .contains(NOW);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/james-project/blob/3a53806a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueViewTestFactory.java
----------------------------------------------------------------------
diff --git a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueViewTestFactory.java b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueViewTestFactory.java
new file mode 100644
index 0000000..95733c2
--- /dev/null
+++ b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueViewTestFactory.java
@@ -0,0 +1,48 @@
+/****************************************************************
+ * 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.view.cassandra;
+
+import java.time.Clock;
+import java.util.concurrent.ThreadLocalRandom;
+
+import org.apache.james.backends.cassandra.init.CassandraTypesProvider;
+import org.apache.james.backends.cassandra.utils.CassandraUtils;
+
+import com.datastax.driver.core.Session;
+
+public class CassandraMailQueueViewTestFactory {
+
+ public static CassandraMailQueueView.Factory factory(Clock clock, ThreadLocalRandom random, Session session,
+ CassandraTypesProvider typesProvider,
+ CassandraMailQueueViewConfiguration configuration) {
+ EnqueuedMailsDAO enqueuedMailsDao = new EnqueuedMailsDAO(session, CassandraUtils.WITH_DEFAULT_CONFIGURATION, typesProvider);
+ BrowseStartDAO browseStartDao = new BrowseStartDAO(session);
+ DeletedMailsDAO deletedMailsDao = new DeletedMailsDAO(session);
+
+ CassandraMailQueueBrowser cassandraMailQueueBrowser = new CassandraMailQueueBrowser(browseStartDao, deletedMailsDao, enqueuedMailsDao, configuration, clock);
+ CassandraMailQueueMailStore cassandraMailQueueMailStore = new CassandraMailQueueMailStore(enqueuedMailsDao, browseStartDao, configuration, clock);
+ CassandraMailQueueMailDelete cassandraMailQueueMailDelete = new CassandraMailQueueMailDelete(deletedMailsDao, browseStartDao, cassandraMailQueueBrowser, configuration, random);
+
+ return new CassandraMailQueueView.Factory(
+ cassandraMailQueueMailStore,
+ cassandraMailQueueBrowser,
+ cassandraMailQueueMailDelete);
+ }
+}
http://git-wip-us.apache.org/repos/asf/james-project/blob/3a53806a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/DeletedMailsDAOTest.java
----------------------------------------------------------------------
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
new file mode 100644
index 0000000..d9dc69a
--- /dev/null
+++ b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/DeletedMailsDAOTest.java
@@ -0,0 +1,108 @@
+/****************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one *
+ * or more contributor license agreements. See the NOTICE file *
+ * distributed with this work for additional information *
+ * regarding copyright ownership. The ASF licenses this file *
+ * to you under the Apache License, Version 2.0 (the *
+ * "License"); you may not use this file except in compliance *
+ * with the License. You may obtain a copy of the License at *
+ * *
+ * http://www.apache.org/licenses/LICENSE-2.0 *
+ * *
+ * Unless required by applicable law or agreed to in writing, *
+ * software distributed under the License is distributed on an *
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY *
+ * KIND, either express or implied. See the License for the *
+ * specific language governing permissions and limitations *
+ * under the License. *
+ ****************************************************************/
+
+package org.apache.james.queue.rabbitmq.view.cassandra;
+
+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.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;
+
+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");
+
+ @RegisterExtension
+ static CassandraClusterExtension cassandraCluster = new CassandraClusterExtension(CassandraMailQueueViewModule.MODULE);
+
+ private DeletedMailsDAO testee;
+
+ @BeforeEach
+ void setUp(CassandraCluster cassandra) {
+ testee = new DeletedMailsDAO(cassandra.getConf());
+ }
+
+ @Test
+ void markAsDeletedShouldWork() {
+ Boolean isDeletedBeforeMark = testee
+ .isDeleted(OUT_GOING_1, MAIL_KEY_1)
+ .join();
+ assertThat(isDeletedBeforeMark).isFalse();
+
+ testee.markAsDeleted(OUT_GOING_1, MAIL_KEY_1).join();
+
+ Boolean isDeletedAfterMark = testee
+ .isDeleted(OUT_GOING_1, MAIL_KEY_1)
+ .join();
+
+ assertThat(isDeletedAfterMark).isTrue();
+ }
+
+ @Test
+ void checkDeletedShouldReturnFalseWhenTableDoesntContainBothMailQueueAndMailKey() {
+ testee.markAsDeleted(OUT_GOING_2, MAIL_KEY_2).join();
+
+ Boolean isDeleted = testee
+ .isDeleted(OUT_GOING_1, MAIL_KEY_1)
+ .join();
+
+ assertThat(isDeleted).isFalse();
+ }
+
+ @Test
+ void checkDeletedShouldReturnFalseWhenTableContainsMailQueueButNotMailKey() {
+ testee.markAsDeleted(OUT_GOING_1, MAIL_KEY_2).join();
+
+ Boolean isDeleted = testee
+ .isDeleted(OUT_GOING_1, MAIL_KEY_1)
+ .join();
+
+ assertThat(isDeleted).isFalse();
+ }
+
+ @Test
+ void checkDeletedShouldReturnFalseWhenTableContainsMailKeyButNotMailQueue() {
+ testee.markAsDeleted(OUT_GOING_2, MAIL_KEY_1).join();
+
+ Boolean isDeleted = testee
+ .isDeleted(OUT_GOING_1, MAIL_KEY_1)
+ .join();
+
+ assertThat(isDeleted).isFalse();
+ }
+
+ @Test
+ void checkDeletedShouldReturnTrueWhenTableContainsMailItem() {
+ testee.markAsDeleted(OUT_GOING_1, MAIL_KEY_1).join();
+
+ Boolean isDeleted = testee
+ .isDeleted(OUT_GOING_1, MAIL_KEY_1)
+ .join();
+
+ assertThat(isDeleted).isTrue();
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/james-project/blob/3a53806a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/EnqueuedMailsDaoTest.java
----------------------------------------------------------------------
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
new file mode 100644
index 0000000..94a80a3
--- /dev/null
+++ b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/EnqueuedMailsDaoTest.java
@@ -0,0 +1,126 @@
+/****************************************************************
+ * 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.view.cassandra;
+
+import static org.apache.james.queue.rabbitmq.view.cassandra.model.BucketedSlices.BucketId;
+import static org.apache.james.queue.rabbitmq.view.cassandra.model.BucketedSlices.Slice;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.stream.Stream;
+
+import org.apache.james.backends.cassandra.CassandraCluster;
+import org.apache.james.backends.cassandra.CassandraClusterExtension;
+import org.apache.james.backends.cassandra.utils.CassandraUtils;
+import org.apache.james.queue.rabbitmq.MailQueueName;
+import org.apache.james.queue.rabbitmq.view.cassandra.model.EnqueuedMail;
+import org.apache.james.queue.rabbitmq.view.cassandra.model.MailKey;
+import org.apache.mailet.base.test.FakeMail;
+import org.assertj.core.api.SoftAssertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+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 int BUCKET_ID_VALUE = 10;
+ private static final BucketId BUCKET_ID = BucketId.of(BUCKET_ID_VALUE);
+ private static final Instant NOW = Instant.now();
+ private static final Slice SLICE_OF_NOW = Slice.of(NOW, Duration.ofSeconds(100));
+
+ @RegisterExtension
+ static CassandraClusterExtension cassandraCluster = new CassandraClusterExtension(CassandraMailQueueViewModule.MODULE);
+
+ private EnqueuedMailsDAO testee;
+
+ @BeforeEach
+ void setUp(CassandraCluster cassandra) {
+ testee = new EnqueuedMailsDAO(
+ cassandra.getConf(),
+ CassandraUtils.WITH_DEFAULT_CONFIGURATION,
+ cassandra.getTypesProvider());
+ }
+
+ @Test
+ void insertShouldWork() throws Exception {
+ testee.insert(EnqueuedMail.builder()
+ .mail(FakeMail.builder()
+ .name(MAIL_KEY_1.getMailKey())
+ .build())
+ .bucketId(BucketId.of(BUCKET_ID_VALUE))
+ .timeRangeStart(NOW)
+ .enqueuedTime(NOW)
+ .mailKey(MAIL_KEY_1)
+ .mailQueueName(OUT_GOING_1)
+ .build())
+ .join();
+
+ Stream<EnqueuedMail> selectedEnqueuedMails = testee
+ .selectEnqueuedMails(OUT_GOING_1, SLICE_OF_NOW, BUCKET_ID)
+ .join();
+
+ assertThat(selectedEnqueuedMails).hasSize(1);
+ }
+
+ @Test
+ void selectEnqueuedMailsShouldWork() throws Exception {
+ testee.insert(EnqueuedMail.builder()
+ .mail(FakeMail.builder()
+ .name(MAIL_KEY_1.getMailKey())
+ .build())
+ .bucketId(BucketId.of(BUCKET_ID_VALUE))
+ .timeRangeStart(NOW)
+ .enqueuedTime(NOW)
+ .mailKey(MAIL_KEY_1)
+ .mailQueueName(OUT_GOING_1)
+ .build())
+ .join();
+
+ testee.insert(EnqueuedMail.builder()
+ .mail(FakeMail.builder()
+ .name(MAIL_KEY_1.getMailKey())
+ .build())
+ .bucketId(BucketId.of(BUCKET_ID_VALUE + 1))
+ .timeRangeStart(NOW)
+ .enqueuedTime(NOW)
+ .mailKey(MAIL_KEY_1)
+ .mailQueueName(OUT_GOING_1)
+ .build())
+ .join();
+
+ Stream<EnqueuedMail> selectedEnqueuedMails = testee.selectEnqueuedMails(OUT_GOING_1, SLICE_OF_NOW, BUCKET_ID)
+ .join();
+
+ assertThat(selectedEnqueuedMails)
+ .hasSize(1)
+ .hasOnlyOneElementSatisfying(selectedEnqueuedMail -> {
+ SoftAssertions softly = new SoftAssertions();
+ softly.assertThat(selectedEnqueuedMail.getMailQueueName()).isEqualTo(OUT_GOING_1);
+ softly.assertThat(selectedEnqueuedMail.getBucketId()).isEqualTo(BUCKET_ID);
+ softly.assertThat(selectedEnqueuedMail.getTimeRangeStart()).isEqualTo(NOW);
+ softly.assertThat(selectedEnqueuedMail.getEnqueuedTime()).isEqualTo(NOW);
+ softly.assertThat(selectedEnqueuedMail.getMailKey()).isEqualTo(MAIL_KEY_1);
+ softly.assertAll();
+ });
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/james-project/blob/3a53806a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/model/BucketedSlicesTest.java
----------------------------------------------------------------------
diff --git a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/model/BucketedSlicesTest.java b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/model/BucketedSlicesTest.java
index 110d870..5386050 100644
--- a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/model/BucketedSlicesTest.java
+++ b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/model/BucketedSlicesTest.java
@@ -21,13 +21,17 @@ package org.apache.james.queue.rabbitmq.view.cassandra.model;
import static org.apache.james.queue.rabbitmq.view.cassandra.model.BucketedSlices.Slice;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import java.time.Duration;
import java.time.Instant;
import java.util.stream.Stream;
-import org.junit.jupiter.api.Nested;
+import org.apache.james.queue.rabbitmq.view.cassandra.model.BucketedSlices.BucketId;
import org.junit.jupiter.api.Test;
+import nl.jqno.equalsverifier.EqualsVerifier;
+
class BucketedSlicesTest {
private static final long ONE_HOUR_IN_SECONDS = 3600;
@@ -36,28 +40,64 @@ class BucketedSlicesTest {
private static final Instant FIRST_SLICE_INSTANT_NEXT_HOUR = FIRST_SLICE_INSTANT.plusSeconds(ONE_HOUR_IN_SECONDS);
private static final Instant FIRST_SLICE_INSTANT_NEXT_TWO_HOUR = FIRST_SLICE_INSTANT.plusSeconds(ONE_HOUR_IN_SECONDS * 2);
- private static final Slice FIRST_SLICE = Slice.of(FIRST_SLICE_INSTANT, ONE_HOUR_IN_SECONDS);
- private static final Slice FIRST_SLICE_NEXT_TWO_HOUR = Slice.of(FIRST_SLICE_INSTANT_NEXT_TWO_HOUR, ONE_HOUR_IN_SECONDS);
+ private static final Duration ONE_HOUR_SLICE_WINDOW = Duration.ofSeconds(ONE_HOUR_IN_SECONDS);
+ private static final Slice FIRST_SLICE = Slice.of(FIRST_SLICE_INSTANT, ONE_HOUR_SLICE_WINDOW);
+ private static final Slice FIRST_SLICE_NEXT_TWO_HOUR = Slice.of(FIRST_SLICE_INSTANT_NEXT_TWO_HOUR, Duration.ofSeconds(ONE_HOUR_IN_SECONDS));
+
+ @Test
+ void bucketIdShouldMatchBeanContract() {
+ EqualsVerifier.forClass(BucketId.class)
+ .verify();
+ }
+
+ @Test
+ void sliceShouldMatchBeanContract() {
+ EqualsVerifier.forClass(Slice.class)
+ .verify();
+ }
- @Nested
- class Validation {
+ @Test
+ void bucketIdShouldThrowWhenValueIsNegative() {
+ assertThatThrownBy(() -> BucketId.of(-1))
+ .isInstanceOf(IllegalArgumentException.class);
}
@Test
void allSlicesTillShouldReturnOnlyFirstSliceWhenEndAtInTheSameInterval() {
- assertThat(Slice.allSlicesTill(FIRST_SLICE, FIRST_SLICE_INSTANT.plusSeconds(3599)))
+ assertThat(Slice.allSlicesTill(FIRST_SLICE, FIRST_SLICE_INSTANT.plusSeconds(ONE_HOUR_IN_SECONDS - 1)))
.containsOnly(FIRST_SLICE);
}
@Test
void allSlicesTillShouldReturnAllSlicesBetweenStartAndEndAt() {
- Stream<Slice> allSlices = Slice.allSlicesTill(FIRST_SLICE, FIRST_SLICE_INSTANT_NEXT_TWO_HOUR.plusSeconds(3599));
+ Stream<Slice> allSlices = Slice.allSlicesTill(FIRST_SLICE, FIRST_SLICE_INSTANT_NEXT_TWO_HOUR.plusSeconds(ONE_HOUR_IN_SECONDS - 1));
assertThat(allSlices)
.containsExactly(
FIRST_SLICE,
- Slice.of(FIRST_SLICE_INSTANT_NEXT_HOUR, ONE_HOUR_IN_SECONDS),
- Slice.of(FIRST_SLICE_INSTANT_NEXT_TWO_HOUR, ONE_HOUR_IN_SECONDS));
+ Slice.of(FIRST_SLICE_INSTANT_NEXT_HOUR, ONE_HOUR_SLICE_WINDOW),
+ Slice.of(FIRST_SLICE_INSTANT_NEXT_TWO_HOUR, ONE_HOUR_SLICE_WINDOW));
+ }
+
+ @Test
+ void allSlicesTillShouldReturnSameSlicesWhenEndAtsAreInTheSameInterval() {
+ Stream<Slice> allSlicesEndAtTheStartOfWindow = Slice.allSlicesTill(FIRST_SLICE, FIRST_SLICE_INSTANT_NEXT_TWO_HOUR);
+ Stream<Slice> allSlicesEndAtTheMiddleOfWindow = Slice.allSlicesTill(FIRST_SLICE, FIRST_SLICE_INSTANT_NEXT_TWO_HOUR.plusSeconds(1000));
+ Stream<Slice> allSlicesEndAtTheEndWindow = Slice.allSlicesTill(FIRST_SLICE, FIRST_SLICE_INSTANT_NEXT_TWO_HOUR.plusSeconds(ONE_HOUR_IN_SECONDS - 1));
+
+ Slice [] allSlicesInThreeHours = {
+ FIRST_SLICE,
+ Slice.of(FIRST_SLICE_INSTANT_NEXT_HOUR, ONE_HOUR_SLICE_WINDOW),
+ Slice.of(FIRST_SLICE_INSTANT_NEXT_TWO_HOUR, ONE_HOUR_SLICE_WINDOW)};
+
+ assertThat(allSlicesEndAtTheStartOfWindow)
+ .containsExactly(allSlicesInThreeHours);
+
+ assertThat(allSlicesEndAtTheMiddleOfWindow)
+ .containsExactly(allSlicesInThreeHours);
+
+ assertThat(allSlicesEndAtTheEndWindow)
+ .containsExactly(allSlicesInThreeHours);
}
@Test
http://git-wip-us.apache.org/repos/asf/james-project/blob/3a53806a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/model/EnqueuedMailTest.java
----------------------------------------------------------------------
diff --git a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/model/EnqueuedMailTest.java b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/model/EnqueuedMailTest.java
new file mode 100644
index 0000000..fb0f487
--- /dev/null
+++ b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/model/EnqueuedMailTest.java
@@ -0,0 +1,33 @@
+/****************************************************************
+ * 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.view.cassandra.model;
+
+import org.junit.jupiter.api.Test;
+
+import nl.jqno.equalsverifier.EqualsVerifier;
+
+class EnqueuedMailTest {
+
+ @Test
+ void shouldMatchBeanContract() {
+ EqualsVerifier.forClass(EnqueuedMail.class)
+ .verify();
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/james-project/blob/3a53806a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/model/MailKeyTest.java
----------------------------------------------------------------------
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/MailKeyTest.java
new file mode 100644
index 0000000..c1a854e
--- /dev/null
+++ b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/view/cassandra/model/MailKeyTest.java
@@ -0,0 +1,33 @@
+/****************************************************************
+ * 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.view.cassandra.model;
+
+import org.junit.jupiter.api.Test;
+
+import nl.jqno.equalsverifier.EqualsVerifier;
+
+class MailKeyTest {
+
+ @Test
+ void shouldMatchBeanContract() {
+ EqualsVerifier.forClass(MailKey.class)
+ .verify();
+ }
+}
\ 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