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 ro...@apache.org on 2016/12/16 15:43:25 UTC
[21/24] james-project git commit: JAMES-1886 Improve future handling
JAMES-1886 Improve future handling
Project: http://git-wip-us.apache.org/repos/asf/james-project/repo
Commit: http://git-wip-us.apache.org/repos/asf/james-project/commit/775624be
Tree: http://git-wip-us.apache.org/repos/asf/james-project/tree/775624be
Diff: http://git-wip-us.apache.org/repos/asf/james-project/diff/775624be
Branch: refs/heads/master
Commit: 775624be1f102bd25f561ad2ac2ada15af865bc6
Parents: e3e61bf
Author: Benoit Tellier <bt...@linagora.com>
Authored: Wed Dec 14 11:27:57 2016 +0700
Committer: Benoit Tellier <bt...@linagora.com>
Committed: Fri Dec 16 20:36:56 2016 +0700
----------------------------------------------------------------------
.../cassandra/utils/CassandraAsyncExecutor.java | 3 +-
.../cassandra/mail/AttachmentLoader.java | 6 +-
.../mail/CassandraMessageIdMapper.java | 49 ++++++++------
.../cassandra/mail/CassandraMessageMapper.java | 71 +++++++++++++-------
4 files changed, 79 insertions(+), 50 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/james-project/blob/775624be/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 4559b67..a7515f5 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
@@ -53,7 +53,7 @@ public class CassandraAsyncExecutor {
public CompletableFuture<Void> executeVoid(Statement statement) {
return FutureConverter.toCompletableFuture(session.executeAsync(statement))
- .thenAccept(this::consume);
+ .thenAccept(result -> {});
}
public CompletableFuture<Optional<Row>> executeSingleRow(Statement statement) {
@@ -62,5 +62,4 @@ public class CassandraAsyncExecutor {
.thenApply(Optional::ofNullable);
}
- private <U> void consume(U value) {}
}
http://git-wip-us.apache.org/repos/asf/james-project/blob/775624be/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/AttachmentLoader.java
----------------------------------------------------------------------
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/AttachmentLoader.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/AttachmentLoader.java
index bcf35f0..c9a81d0 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/AttachmentLoader.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/AttachmentLoader.java
@@ -43,12 +43,12 @@ public class AttachmentLoader {
public Collection<MessageAttachment> getAttachments(Set<CassandraMessageDAO.MessageAttachmentRepresentation> attachmentRepresentations) {
- Map<AttachmentId, CassandraMessageDAO.MessageAttachmentRepresentation> attachmentRepresentationByIds = attachmentRepresentations.stream()
+ Map<AttachmentId, CassandraMessageDAO.MessageAttachmentRepresentation> attachmentRepresentationsById = attachmentRepresentations.stream()
.collect(Guavate.toImmutableMap(CassandraMessageDAO.MessageAttachmentRepresentation::getAttachmentId, Function.identity()));
- Map<AttachmentId, Attachment> attachmentById = attachmentsById(attachmentRepresentationByIds.keySet());
+ Map<AttachmentId, Attachment> attachmentsById = attachmentsById(attachmentRepresentationsById.keySet());
- return MapMerger.merge(attachmentById, attachmentRepresentationByIds, this::constructMessageAttachment).values();
+ return MapMerger.merge(attachmentsById, attachmentRepresentationsById, this::constructMessageAttachment).values();
}
private MessageAttachment constructMessageAttachment(Attachment attachment, CassandraMessageDAO.MessageAttachmentRepresentation messageAttachmentRepresentation) {
http://git-wip-us.apache.org/repos/asf/james-project/blob/775624be/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 6632944..77e150c 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
@@ -123,18 +123,20 @@ public class CassandraMessageIdMapper implements MessageIdMapper {
@Override
public void save(MailboxMessage mailboxMessage) throws MailboxException {
CassandraId mailboxId = (CassandraId) mailboxMessage.getMailboxId();
- messageDAO.save(mailboxMapper.findMailboxById(mailboxId), mailboxMessage).join();
CassandraMessageId messageId = (CassandraMessageId) mailboxMessage.getMessageId();
ComposedMessageIdWithMetaData composedMessageIdWithMetaData = ComposedMessageIdWithMetaData.builder()
.composedMessageId(new ComposedMessageId(mailboxId, messageId, mailboxMessage.getUid()))
.flags(mailboxMessage.createFlags())
.modSeq(mailboxMessage.getModSeq())
.build();
- CompletableFuture.allOf(imapUidDAO.insert(composedMessageIdWithMetaData),
- messageIdDAO.insert(composedMessageIdWithMetaData),
- mailboxCounterDAO.incrementCount(mailboxId),
- incrementUnseenOnSave(mailboxId, mailboxMessage.createFlags()))
- .join();
+ messageDAO.save(mailboxMapper.findMailboxById(mailboxId), mailboxMessage)
+ .thenCompose(voidValue -> CompletableFuture.allOf(
+ imapUidDAO.insert(composedMessageIdWithMetaData),
+ messageIdDAO.insert(composedMessageIdWithMetaData)))
+ .thenCompose(voidValue -> CompletableFuture.allOf(
+ mailboxCounterDAO.incrementCount(mailboxId),
+ incrementUnseenOnSave(mailboxId, mailboxMessage.createFlags())))
+ .join();
}
private CompletableFuture<Void> incrementUnseenOnSave(CassandraId mailboxId, Flags flags) {
@@ -147,31 +149,39 @@ public class CassandraMessageIdMapper implements MessageIdMapper {
@Override
public void delete(MessageId messageId, List<MailboxId> mailboxIds) {
CassandraMessageId cassandraMessageId = (CassandraMessageId) messageId;
- mailboxIds.forEach(mailboxId -> retrieveAndDeleteIndices(cassandraMessageId, Optional.of((CassandraId) mailboxId)).join());
+ mailboxIds.stream()
+ .map(mailboxId -> retrieveAndDeleteIndices(cassandraMessageId, Optional.of((CassandraId) mailboxId)))
+ .reduce((f1, f2) -> CompletableFuture.allOf(f1, f2))
+ .orElse(CompletableFuture.completedFuture(null))
+ .join();
}
private CompletableFuture<Void> retrieveAndDeleteIndices(CassandraMessageId messageId, Optional<CassandraId> mailboxId) {
return imapUidDAO.retrieve(messageId, mailboxId)
- .thenAccept(composedMessageIds -> composedMessageIds
+ .thenCompose(composedMessageIds -> composedMessageIds
.map(this::deleteIds)
- .reduce((f1, f2) -> CompletableFuture.allOf(f1, f2)));
+ .reduce((f1, f2) -> CompletableFuture.allOf(f1, f2))
+ .orElse(CompletableFuture.completedFuture(null)));
}
@Override
public void delete(MessageId messageId) {
CassandraMessageId cassandraMessageId = (CassandraMessageId) messageId;
- messageDAO.delete(cassandraMessageId).join();
- retrieveAndDeleteIndices(cassandraMessageId, Optional.empty()).join();
+ retrieveAndDeleteIndices(cassandraMessageId, Optional.empty())
+ .thenCompose(voidValue -> messageDAO.delete(cassandraMessageId))
+ .join();
}
private CompletableFuture<Void> deleteIds(ComposedMessageIdWithMetaData metaData) {
CassandraMessageId messageId = (CassandraMessageId) metaData.getComposedMessageId().getMessageId();
CassandraId mailboxId = (CassandraId) metaData.getComposedMessageId().getMailboxId();
- return CompletableFuture.allOf(imapUidDAO.delete(messageId, mailboxId),
- messageIdDAO.delete(mailboxId, metaData.getComposedMessageId().getUid()),
- mailboxCounterDAO.decrementCount(mailboxId),
- decrementUnseenOnDelete(mailboxId, metaData.getFlags()));
+ return CompletableFuture.allOf(
+ imapUidDAO.delete(messageId, mailboxId),
+ messageIdDAO.delete(mailboxId, metaData.getComposedMessageId().getUid()))
+ .thenCompose(voidValue -> CompletableFuture.allOf(
+ mailboxCounterDAO.decrementCount(mailboxId),
+ decrementUnseenOnDelete(mailboxId, metaData.getFlags())));
}
private CompletableFuture<Void> decrementUnseenOnDelete(CassandraId mailboxId, Flags flags) {
@@ -218,14 +228,14 @@ public class CassandraMessageIdMapper implements MessageIdMapper {
private CompletableFuture<Void> incrementCountIfNeeded(Flags oldFlags, Flags newFlags, CassandraId cassandraId) {
if (oldFlags.contains(Flags.Flag.SEEN) && !newFlags.contains(Flags.Flag.SEEN)) {
- mailboxCounterDAO.incrementUnseen(cassandraId).join();
+ return mailboxCounterDAO.incrementUnseen(cassandraId);
}
return CompletableFuture.completedFuture(null);
}
private CompletableFuture<Void> decrementCountIfNeeded(Flags oldFlags, Flags newFlags, CassandraId cassandraId) {
if (!oldFlags.contains(Flags.Flag.SEEN) && newFlags.contains(Flags.Flag.SEEN)) {
- mailboxCounterDAO.decrementUnseen(cassandraId).join();
+ return mailboxCounterDAO.decrementUnseen(cassandraId);
}
return CompletableFuture.completedFuture(null);
}
@@ -258,7 +268,8 @@ public class CassandraMessageIdMapper implements MessageIdMapper {
}
private boolean updateFlags(ComposedMessageIdWithMetaData composedMessageIdWithMetaData, long oldModSeq) {
- return imapUidDAO.updateMetadata(composedMessageIdWithMetaData, oldModSeq).join()
- && messageIdDAO.updateMetadata(composedMessageIdWithMetaData, oldModSeq).join();
+ CompletableFuture<Boolean> imapUidFuture = imapUidDAO.updateMetadata(composedMessageIdWithMetaData, oldModSeq);
+ CompletableFuture<Boolean> messageIdFuture = messageIdDAO.updateMetadata(composedMessageIdWithMetaData, oldModSeq);
+ return imapUidFuture.join() && messageIdFuture.join();
}
}
http://git-wip-us.apache.org/repos/asf/james-project/blob/775624be/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 67c0c62..5295430 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
@@ -103,23 +103,31 @@ public class CassandraMessageMapper implements MessageMapper {
@Override
public void delete(Mailbox mailbox, MailboxMessage message) {
CassandraId mailboxId = (CassandraId) mailbox.getMailboxId();
- messageIdDAO.retrieve(mailboxId, message.getUid()).join()
- .ifPresent(this::deleteUsingMailboxId);
+
+ messageIdDAO.retrieve(mailboxId, message.getUid())
+ .thenCompose(optional -> optional.map(this::deleteUsingMailboxId)
+ .orElse(CompletableFuture.completedFuture(null)))
+ .join();
}
- private void deleteUsingMailboxId(ComposedMessageIdWithMetaData composedMessageIdWithMetaData) {
+ private CompletableFuture<Void> deleteUsingMailboxId(ComposedMessageIdWithMetaData composedMessageIdWithMetaData) {
ComposedMessageId composedMessageId = composedMessageIdWithMetaData.getComposedMessageId();
CassandraMessageId messageId = (CassandraMessageId) composedMessageId.getMessageId();
CassandraId mailboxId = (CassandraId) composedMessageId.getMailboxId();
MessageUid uid = composedMessageId.getUid();
- CompletableFuture.allOf(imapUidDAO.delete(messageId, mailboxId),
- messageIdDAO.delete(mailboxId, uid))
- .join();
+ return CompletableFuture.allOf(
+ imapUidDAO.delete(messageId, mailboxId),
+ messageIdDAO.delete(mailboxId, uid)
+ ).thenCompose(voidValue -> CompletableFuture.allOf(
+ mailboxCounterDAO.decrementCount(mailboxId),
+ decrementUnseenOnDelete(mailboxId, composedMessageIdWithMetaData.getFlags())));
+ }
- mailboxCounterDAO.decrementCount(mailboxId).join();
- if (!composedMessageIdWithMetaData.getFlags().contains(Flag.SEEN)) {
- mailboxCounterDAO.decrementUnseen(mailboxId).join();
+ private CompletableFuture<Void> decrementUnseenOnDelete(CassandraId mailboxId, Flags flags) {
+ if (flags.contains(Flags.Flag.SEEN)) {
+ return CompletableFuture.completedFuture(null);
}
+ return mailboxCounterDAO.decrementUnseen(mailboxId);
}
private CompletableFuture<Optional<ComposedMessageIdWithMetaData>> retrieveMessageId(CassandraId mailboxId, MailboxMessage message) {
@@ -186,8 +194,9 @@ public class CassandraMessageMapper implements MessageMapper {
public MessageMetaData move(Mailbox destinationMailbox, MailboxMessage original) throws MailboxException {
CassandraId originalMailboxId = (CassandraId) original.getMailboxId();
MessageMetaData messageMetaData = copy(destinationMailbox, original);
- retrieveMessageId(originalMailboxId, original).join()
- .ifPresent(this::deleteUsingMailboxId);
+ retrieveMessageId(originalMailboxId, original)
+ .thenCompose(optional -> optional.map(this::deleteUsingMailboxId).orElse(CompletableFuture.completedFuture(null)))
+ .join();
return messageMetaData;
}
@@ -205,13 +214,20 @@ public class CassandraMessageMapper implements MessageMapper {
public MessageMetaData add(Mailbox mailbox, MailboxMessage message) throws MailboxException {
message.setUid(uidProvider.nextUid(mailboxSession, mailbox));
message.setModSeq(modSeqProvider.nextModSeq(mailboxSession, mailbox));
- MessageMetaData messageMetaData = save(mailbox, message);
CassandraId mailboxId = (CassandraId) mailbox.getMailboxId();
- if (!message.isSeen()) {
- mailboxCounterDAO.incrementUnseen(mailboxId).join();
+ save(mailbox, message)
+ .thenCompose(voidValue -> CompletableFuture.allOf(
+ incrementUnseenOnSave(mailboxId, message.createFlags()),
+ mailboxCounterDAO.incrementCount(mailboxId)))
+ .join();
+ return new SimpleMessageMetaData(message);
+ }
+
+ private CompletableFuture<Void> incrementUnseenOnSave(CassandraId mailboxId, Flags flags) {
+ if (flags.contains(Flags.Flag.SEEN)) {
+ return CompletableFuture.completedFuture(null);
}
- mailboxCounterDAO.incrementCount(mailboxId).join();
- return messageMetaData;
+ return mailboxCounterDAO.incrementUnseen(mailboxId);
}
@Override
@@ -221,7 +237,9 @@ public class CassandraMessageMapper implements MessageMapper {
.map(message -> updateFlagsOnMessage(mailbox, flagUpdateCalculator, message))
.filter(Optional::isPresent)
.map(Optional::get)
- .peek((updatedFlags) -> manageUnseenMessageCounts(mailbox, updatedFlags.getOldFlags(), updatedFlags.getNewFlags()))
+ .map((updatedFlags) -> manageUnseenMessageCounts(mailbox, updatedFlags.getOldFlags(), updatedFlags.getNewFlags())
+ .thenApply(voidValue -> updatedFlags))
+ .map(CompletableFuture::join)
.collect(Collectors.toList()) // This collect is here as we need to consume all the stream before returning result
.iterator();
}
@@ -242,11 +260,10 @@ public class CassandraMessageMapper implements MessageMapper {
return uidProvider.lastUid(mailboxSession, mailbox);
}
- private MessageMetaData save(Mailbox mailbox, MailboxMessage message) throws MailboxException {
+ private CompletableFuture<Void> save(Mailbox mailbox, MailboxMessage message) throws MailboxException {
CassandraId mailboxId = (CassandraId) mailbox.getMailboxId();
- insertIds(message, mailboxId).join();
- messageDAO.save(mailbox, message).join();
- return new SimpleMessageMetaData(message);
+ return messageDAO.save(mailbox, message)
+ .thenCompose(aVoid -> insertIds(message, mailboxId));
}
private CompletableFuture<Void> insertIds(MailboxMessage message, CassandraId mailboxId) {
@@ -259,14 +276,15 @@ public class CassandraMessageMapper implements MessageMapper {
imapUidDAO.insert(composedMessageIdWithMetaData));
}
- private void manageUnseenMessageCounts(Mailbox mailbox, Flags oldFlags, Flags newFlags) {
+ private CompletableFuture<Void> manageUnseenMessageCounts(Mailbox mailbox, Flags oldFlags, Flags newFlags) {
CassandraId mailboxId = (CassandraId) mailbox.getMailboxId();
if (oldFlags.contains(Flag.SEEN) && !newFlags.contains(Flag.SEEN)) {
- mailboxCounterDAO.incrementUnseen(mailboxId).join();
+ return mailboxCounterDAO.incrementUnseen(mailboxId);
}
if (!oldFlags.contains(Flag.SEEN) && newFlags.contains(Flag.SEEN)) {
- mailboxCounterDAO.decrementUnseen(mailboxId).join();
+ return mailboxCounterDAO.decrementUnseen(mailboxId);
}
+ return CompletableFuture.completedFuture(null);
}
private Optional<UpdatedFlags> updateFlagsOnMessage(Mailbox mailbox, FlagsUpdateCalculator flagUpdateCalculator, MailboxMessage message) {
@@ -298,8 +316,9 @@ public class CassandraMessageMapper implements MessageMapper {
.modSeq(message.getModSeq())
.flags(message.createFlags())
.build();
- return messageIdDAO.updateMetadata(composedMessageIdWithMetaData, oldModSeq).join() &&
- imapUidDAO.updateMetadata(composedMessageIdWithMetaData, oldModSeq).join();
+ CompletableFuture<Boolean> messageIdFuture = messageIdDAO.updateMetadata(composedMessageIdWithMetaData, oldModSeq);
+ CompletableFuture<Boolean> imapUidFuture = imapUidDAO.updateMetadata(composedMessageIdWithMetaData, oldModSeq);
+ return messageIdFuture.join() && imapUidFuture.join();
}
private Optional<UpdatedFlags> handleRetries(Mailbox mailbox, FlagsUpdateCalculator flagUpdateCalculator, MailboxMessage message) {
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org