james-server-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From btell...@apache.org
Subject [3/5] james-project git commit: JAMES-2544 RabbitMQ browse Cassandra DAOs and tests
Date Wed, 26 Sep 2018 02:23:39 GMT
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 <dtran@linagora.com>
Authored: Wed Sep 12 20:15:49 2018 +0700
Committer: Benoit Tellier <btellier@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


Mime
View raw message