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:39 UTC
[3/5] james-project git commit: JAMES-2544 RabbitMQ browse Cassandra
DAOs and tests
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