You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@james.apache.org by bt...@apache.org on 2022/04/19 04:04:20 UTC
[james-project] 29/49: JAMES-3737 Reactive implementation for IMAP CLOSE
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit b2c12f4a51c9509c82cd945cea416830970d1225
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Wed Mar 30 22:41:35 2022 +0700
JAMES-3737 Reactive implementation for IMAP CLOSE
---
.../org/apache/james/mailbox/MessageManager.java | 7 +++
.../cassandra/mail/CassandraMessageMapper.java | 17 +++++--
.../james/mailbox/store/StoreMessageManager.java | 52 +++++++++++++---------
.../james/mailbox/store/mail/MessageMapper.java | 12 +++++
.../james/imap/processor/CloseProcessor.java | 43 +++++++++---------
5 files changed, 86 insertions(+), 45 deletions(-)
diff --git a/mailbox/api/src/main/java/org/apache/james/mailbox/MessageManager.java b/mailbox/api/src/main/java/org/apache/james/mailbox/MessageManager.java
index 25ec494f4f..a7d1f1924c 100644
--- a/mailbox/api/src/main/java/org/apache/james/mailbox/MessageManager.java
+++ b/mailbox/api/src/main/java/org/apache/james/mailbox/MessageManager.java
@@ -63,6 +63,7 @@ import org.apache.james.mime4j.dom.Message;
import org.apache.james.mime4j.message.DefaultMessageWriter;
import org.reactivestreams.Publisher;
+import com.github.fge.lambdas.Throwing;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
@@ -131,6 +132,12 @@ public interface MessageManager {
*/
Iterator<MessageUid> expunge(MessageRange set, MailboxSession mailboxSession) throws MailboxException;
+ default Flux<MessageUid> expungeReactive(MessageRange set, MailboxSession mailboxSession) {
+ return Flux.fromStream(Throwing.supplier(() -> StreamSupport.stream(
+ Spliterators.spliteratorUnknownSize(expunge(set, mailboxSession), Spliterator.ORDERED),
+ false)));
+ }
+
/**
* Deletes a list of messages given their uids in the mailbox.
*/
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 117aa4050f..a511db6f6b 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
@@ -311,6 +311,19 @@ public class CassandraMessageMapper implements MessageMapper {
@Override
public Map<MessageUid, MessageMetaData> deleteMessages(Mailbox mailbox, List<MessageUid> uids) {
+ return deleteMessagesReactive(mailbox, uids)
+ .block();
+ }
+
+ @Override
+ public Flux<MessageUid> retrieveMessagesMarkedForDeletionReactive(Mailbox mailbox, MessageRange messageRange) {
+ CassandraId mailboxId = (CassandraId) mailbox.getMailboxId();
+
+ return deletedMessageDAO.retrieveDeletedMessage(mailboxId, messageRange);
+ }
+
+ @Override
+ public Mono<Map<MessageUid, MessageMetaData>> deleteMessagesReactive(Mailbox mailbox, List<MessageUid> uids) {
CassandraId mailboxId = (CassandraId) mailbox.getMailboxId();
return Flux.fromIterable(MessageRange.toRanges(uids))
@@ -319,9 +332,7 @@ public class CassandraMessageMapper implements MessageMapper {
.flatMap(this::expungeOne, cassandraConfiguration.getExpungeChunkSize())
.collect(ImmutableMap.toImmutableMap(MailboxMessage::getUid, MailboxMessage::metaData))
.flatMap(messageMap -> indexTableHandler.updateIndexOnDelete(mailboxId, messageMap.values())
- .thenReturn(messageMap))
- .subscribeOn(Schedulers.elastic())
- .block();
+ .thenReturn(messageMap));
}
private Mono<SimpleMailboxMessage> expungeOne(ComposedMessageIdWithMetaData metaData) {
diff --git a/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreMessageManager.java b/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreMessageManager.java
index 47ee981fc0..15d54828cd 100644
--- a/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreMessageManager.java
+++ b/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreMessageManager.java
@@ -256,50 +256,59 @@ public class StoreMessageManager implements MessageManager {
throw new ReadOnlyException(getMailboxPath());
}
- List<MessageUid> uids = retrieveMessagesMarkedForDeletion(set, mailboxSession);
- Map<MessageUid, MessageMetaData> deletedMessages = deleteMessages(uids, mailboxSession);
+ List<MessageUid> uids = MailboxReactorUtils.block(retrieveMessagesMarkedForDeletion(set, mailboxSession));
+ Map<MessageUid, MessageMetaData> deletedMessages = MailboxReactorUtils.block(deleteMessages(uids, mailboxSession));
- dispatchExpungeEvent(mailboxSession, deletedMessages);
+ dispatchExpungeEvent(mailboxSession, deletedMessages).block();
return deletedMessages.keySet().iterator();
}
- private List<MessageUid> retrieveMessagesMarkedForDeletion(MessageRange messageRange, MailboxSession session) throws MailboxException {
+ @Override
+ public Flux<MessageUid> expungeReactive(MessageRange set, MailboxSession mailboxSession) {
+ if (!isWriteable(mailboxSession)) {
+ return Flux.error(new ReadOnlyException(getMailboxEntity().generateAssociatedPath()));
+ }
+
+ return retrieveMessagesMarkedForDeletion(set, mailboxSession)
+ .flatMap(uids -> deleteMessages(uids, mailboxSession))
+ .flatMap(deletedMessages -> dispatchExpungeEvent(mailboxSession, deletedMessages).thenReturn(deletedMessages))
+ .flatMapIterable(Map::keySet);
+ }
+
+ private Mono<List<MessageUid>> retrieveMessagesMarkedForDeletion(MessageRange messageRange, MailboxSession session) {
MessageMapper messageMapper = mapperFactory.getMessageMapper(session);
- return messageMapper.execute(
- () -> messageMapper.retrieveMessagesMarkedForDeletion(getMailboxEntity(), messageRange));
+ return messageMapper.executeReactive(
+ messageMapper.retrieveMessagesMarkedForDeletionReactive(getMailboxEntity(), messageRange).collectList());
}
@Override
public void delete(List<MessageUid> messageUids, MailboxSession mailboxSession) throws MailboxException {
- Map<MessageUid, MessageMetaData> deletedMessages = deleteMessages(messageUids, mailboxSession);
+ Map<MessageUid, MessageMetaData> deletedMessages = MailboxReactorUtils.block(deleteMessages(messageUids, mailboxSession));
- dispatchExpungeEvent(mailboxSession, deletedMessages);
+ dispatchExpungeEvent(mailboxSession, deletedMessages).block();
}
- private Map<MessageUid, MessageMetaData> deleteMessages(List<MessageUid> messageUids, MailboxSession session) throws MailboxException {
+ private Mono<Map<MessageUid, MessageMetaData>> deleteMessages(List<MessageUid> messageUids, MailboxSession session) {
if (messageUids.isEmpty()) {
- return ImmutableMap.of();
+ return Mono.just(ImmutableMap.of());
}
MessageMapper messageMapper = mapperFactory.getMessageMapper(session);
- runPredeletionHooks(messageUids, session);
+ return runPredeletionHooks(messageUids, session)
+ .then(messageMapper.executeReactive(messageMapper.deleteMessagesReactive(getMailboxEntity(), messageUids)));
- return messageMapper.execute(
- () -> messageMapper.deleteMessages(getMailboxEntity(), messageUids));
}
- private void dispatchExpungeEvent(MailboxSession mailboxSession, Map<MessageUid, MessageMetaData> deletedMessages) throws MailboxException {
- eventBus.dispatch(EventFactory.expunged()
+ private Mono<Void> dispatchExpungeEvent(MailboxSession mailboxSession, Map<MessageUid, MessageMetaData> deletedMessages) {
+ return eventBus.dispatch(EventFactory.expunged()
.randomEventId()
.mailboxSession(mailboxSession)
.mailbox(getMailboxEntity())
.metaData(ImmutableSortedMap.copyOf(deletedMessages))
.build(),
- new MailboxIdRegistrationKey(mailbox.getMailboxId()))
- .subscribeOn(Schedulers.elastic())
- .block();
+ new MailboxIdRegistrationKey(mailbox.getMailboxId()));
}
@Override
@@ -547,7 +556,7 @@ public class StoreMessageManager implements MessageManager {
}
@Override
- public boolean isWriteable(MailboxSession session) throws MailboxException {
+ public boolean isWriteable(MailboxSession session) {
return storeRightManager.isReadWrite(session, mailbox, getSharedPermanentFlags(session));
}
@@ -820,17 +829,16 @@ public class StoreMessageManager implements MessageManager {
return Mono.empty();
}
- private void runPredeletionHooks(List<MessageUid> uids, MailboxSession session) {
+ private Mono<Void> runPredeletionHooks(List<MessageUid> uids, MailboxSession session) {
MessageMapper messageMapper = mapperFactory.getMessageMapper(session);
Mono<DeleteOperation> deleteOperation = Flux.fromIterable(MessageRange.toRanges(uids))
- .publishOn(Schedulers.elastic())
.flatMap(range -> messageMapper.findInMailboxReactive(mailbox, range, FetchType.METADATA, UNLIMITED), DEFAULT_CONCURRENCY)
.map(mailboxMessage -> MetadataWithMailboxId.from(mailboxMessage.metaData(), mailboxMessage.getMailboxId()))
.collect(ImmutableList.toImmutableList())
.map(DeleteOperation::from);
- deleteOperation.flatMap(preDeletionHooks::runHooks).block();
+ return deleteOperation.flatMap(preDeletionHooks::runHooks).then();
}
@Override
diff --git a/mailbox/store/src/main/java/org/apache/james/mailbox/store/mail/MessageMapper.java b/mailbox/store/src/main/java/org/apache/james/mailbox/store/mail/MessageMapper.java
index 63f471a0f9..0fa6e16629 100644
--- a/mailbox/store/src/main/java/org/apache/james/mailbox/store/mail/MessageMapper.java
+++ b/mailbox/store/src/main/java/org/apache/james/mailbox/store/mail/MessageMapper.java
@@ -96,6 +96,14 @@ public interface MessageMapper extends Mapper {
*/
List<MessageUid> retrieveMessagesMarkedForDeletion(Mailbox mailbox, MessageRange messageRange) throws MailboxException;
+ default Flux<MessageUid> retrieveMessagesMarkedForDeletionReactive(Mailbox mailbox, MessageRange messageRange) {
+ try {
+ return Flux.fromIterable(retrieveMessagesMarkedForDeletion(mailbox, messageRange));
+ } catch (MailboxException e) {
+ return Flux.error(e);
+ }
+ }
+
/**
* Return the count of messages in the mailbox
*/
@@ -119,6 +127,10 @@ public interface MessageMapper extends Mapper {
*/
Map<MessageUid, MessageMetaData> deleteMessages(Mailbox mailbox, List<MessageUid> uids) throws MailboxException;
+ default Mono<Map<MessageUid, MessageMetaData>> deleteMessagesReactive(Mailbox mailbox, List<MessageUid> uids) {
+ return Mono.fromCallable(() -> deleteMessages(mailbox, uids));
+ }
+
/**
* Return the uid of the first unseen message. If non can be found null will get returned
*/
diff --git a/protocols/imap/src/main/java/org/apache/james/imap/processor/CloseProcessor.java b/protocols/imap/src/main/java/org/apache/james/imap/processor/CloseProcessor.java
index 0c37adc751..eeea6b5fd4 100644
--- a/protocols/imap/src/main/java/org/apache/james/imap/processor/CloseProcessor.java
+++ b/protocols/imap/src/main/java/org/apache/james/imap/processor/CloseProcessor.java
@@ -27,7 +27,6 @@ import org.apache.james.imap.api.process.ImapSession;
import org.apache.james.imap.message.request.CloseRequest;
import org.apache.james.mailbox.MailboxManager;
import org.apache.james.mailbox.MailboxSession;
-import org.apache.james.mailbox.MessageManager;
import org.apache.james.mailbox.exception.MailboxException;
import org.apache.james.mailbox.model.MailboxACL;
import org.apache.james.mailbox.model.MessageRange;
@@ -36,6 +35,10 @@ import org.apache.james.util.MDCBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.github.fge.lambdas.Throwing;
+
+import reactor.core.publisher.Mono;
+
public class CloseProcessor extends AbstractMailboxProcessor<CloseRequest> {
private static final Logger LOGGER = LoggerFactory.getLogger(CloseProcessor.class);
@@ -45,25 +48,25 @@ public class CloseProcessor extends AbstractMailboxProcessor<CloseRequest> {
}
@Override
- protected void processRequest(CloseRequest request, ImapSession session, Responder responder) {
- try {
- MessageManager mailbox = getSelectedMailbox(session)
- .orElseThrow(() -> new MailboxException("Session not in SELECTED state"));
- final MailboxSession mailboxSession = session.getMailboxSession();
- if (getMailboxManager().hasRight(mailbox.getMailboxEntity(), MailboxACL.Right.PerformExpunge, mailboxSession)) {
- mailbox.expunge(MessageRange.all(), mailboxSession);
- session.deselect().block();
-
- // Don't send HIGHESTMODSEQ when close. Like correct in the ERRATA of RFC5162
- //
- // See http://www.rfc-editor.org/errata_search.php?rfc=5162
- okComplete(request, responder);
- }
-
- } catch (MailboxException e) {
- LOGGER.error("Close failed for mailbox {}", session.getSelected().getMailboxId(), e);
- no(request, responder, HumanReadableText.GENERIC_FAILURE_DURING_PROCESSING);
- }
+ protected Mono<Void> processRequestReactive(CloseRequest request, ImapSession session, Responder responder) {
+ MailboxSession mailboxSession = session.getMailboxSession();
+ return getSelectedMailboxReactive(session)
+ .switchIfEmpty(Mono.error(() -> new MailboxException("Session not in SELECTED state")))
+ .flatMap(Throwing.function(mailbox -> {
+ if (getMailboxManager().hasRight(mailbox.getMailboxEntity(), MailboxACL.Right.PerformExpunge, mailboxSession)) {
+ return mailbox.expungeReactive(MessageRange.all(), mailboxSession)
+ .then(session.deselect())
+ // Don't send HIGHESTMODSEQ when close. Like correct in the ERRATA of RFC5162
+ // See http://www.rfc-editor.org/errata_search.php?rfc=5162
+ .then(Mono.fromRunnable(() -> okComplete(request, responder)));
+ }
+ return Mono.empty();
+ }))
+ .onErrorResume(MailboxException.class, e -> {
+ LOGGER.error("Close failed for mailbox {}", session.getSelected().getMailboxId(), e);
+ no(request, responder, HumanReadableText.GENERIC_FAILURE_DURING_PROCESSING);
+ return Mono.empty();
+ }).then();
}
@Override
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org