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:03:57 UTC
[james-project] 06/49: JAMES-3737 Reactify IMAP unsolicited notifications
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 9698eac1e216be013fdcc4cbaebed962ace141b2
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Fri Mar 25 17:24:04 2022 +0700
JAMES-3737 Reactify IMAP unsolicited notifications
---
.../james/imap/api/process/SelectedMailbox.java | 3 +
.../imap/processor/AbstractMailboxProcessor.java | 208 +++++++++++----------
.../processor/AbstractMessageRangeProcessor.java | 2 +-
.../james/imap/processor/AppendProcessor.java | 19 +-
.../james/imap/processor/CapabilityProcessor.java | 8 +-
.../james/imap/processor/CheckProcessor.java | 8 +-
.../james/imap/processor/CreateProcessor.java | 2 +-
.../james/imap/processor/DeleteProcessor.java | 2 +-
.../james/imap/processor/EnableProcessor.java | 9 +-
.../james/imap/processor/ExpungeProcessor.java | 2 +-
.../apache/james/imap/processor/IdleProcessor.java | 8 +-
.../james/imap/processor/NamespaceProcessor.java | 8 +-
.../apache/james/imap/processor/NoopProcessor.java | 8 +-
.../james/imap/processor/RenameProcessor.java | 2 +-
.../james/imap/processor/StatusProcessor.java | 2 +-
.../james/imap/processor/StoreProcessor.java | 2 +-
.../james/imap/processor/SubscribeProcessor.java | 4 +-
.../james/imap/processor/UnsubscribeProcessor.java | 4 +-
.../imap/processor/base/SelectedMailboxImpl.java | 5 +
.../james/imap/processor/fetch/FetchProcessor.java | 15 +-
.../james/imapserver/netty/IMAPServerTest.java | 2 +
21 files changed, 175 insertions(+), 148 deletions(-)
diff --git a/protocols/imap/src/main/java/org/apache/james/imap/api/process/SelectedMailbox.java b/protocols/imap/src/main/java/org/apache/james/imap/api/process/SelectedMailbox.java
index 222a2a60d6..b97582eb70 100644
--- a/protocols/imap/src/main/java/org/apache/james/imap/api/process/SelectedMailbox.java
+++ b/protocols/imap/src/main/java/org/apache/james/imap/api/process/SelectedMailbox.java
@@ -25,6 +25,7 @@ import java.util.Optional;
import javax.mail.Flags;
import org.apache.james.events.EventListener;
+import org.apache.james.mailbox.MessageManager;
import org.apache.james.mailbox.MessageUid;
import org.apache.james.mailbox.NullableMessageSequenceNumber;
import org.apache.james.mailbox.exception.MailboxException;
@@ -99,6 +100,8 @@ public interface SelectedMailbox {
*/
MailboxId getMailboxId();
+ MessageManager getMessageManager();
+
/**
* Is the given uid recent ?
*
diff --git a/protocols/imap/src/main/java/org/apache/james/imap/processor/AbstractMailboxProcessor.java b/protocols/imap/src/main/java/org/apache/james/imap/processor/AbstractMailboxProcessor.java
index df2d1bb6f2..079d959db0 100644
--- a/protocols/imap/src/main/java/org/apache/james/imap/processor/AbstractMailboxProcessor.java
+++ b/protocols/imap/src/main/java/org/apache/james/imap/processor/AbstractMailboxProcessor.java
@@ -57,7 +57,6 @@ import org.apache.james.mailbox.ModSeq;
import org.apache.james.mailbox.NullableMessageSequenceNumber;
import org.apache.james.mailbox.exception.MailboxException;
import org.apache.james.mailbox.exception.MessageRangeException;
-import org.apache.james.mailbox.model.ComposedMessageIdWithMetaData;
import org.apache.james.mailbox.model.MailboxPath;
import org.apache.james.mailbox.model.MessageRange;
import org.apache.james.mailbox.model.MessageRange.Type;
@@ -123,53 +122,53 @@ public abstract class AbstractMailboxProcessor<R extends ImapRequest> extends Ab
responder.respond(untaggedOk);
}
- protected void unsolicitedResponses(ImapSession session, ImapProcessor.Responder responder, boolean useUids) {
- unsolicitedResponses(session, responder, false, useUids);
+ protected Mono<Void> unsolicitedResponses(ImapSession session, ImapProcessor.Responder responder, boolean useUids) {
+ return unsolicitedResponses(session, responder, false, useUids);
}
/**
* Sends any unsolicited responses to the client, such as EXISTS and FLAGS
* responses when the selected mailbox is modified by another user.
*/
- protected void unsolicitedResponses(ImapSession session, ImapProcessor.Responder responder, boolean omitExpunged, boolean useUid) {
+ protected Mono<Void> unsolicitedResponses(ImapSession session, ImapProcessor.Responder responder, boolean omitExpunged, boolean useUid) {
final SelectedMailbox selected = session.getSelected();
if (selected == null) {
LOGGER.debug("No mailbox selected");
+ return Mono.empty();
} else {
- unsolicitedResponses(session, responder, selected, omitExpunged, useUid);
+ return unsolicitedResponses(session, responder, selected, omitExpunged, useUid);
}
}
- private void unsolicitedResponses(ImapSession session, ImapProcessor.Responder responder, SelectedMailbox selected, boolean omitExpunged, boolean useUid) {
- final boolean sizeChanged = selected.isSizeChanged();
- // New message response
- if (sizeChanged) {
- addExistsResponses(session, selected, responder);
- }
- // Expunged messages
- if (!omitExpunged) {
- final Collection<MessageUid> expungedUids = selected.expungedUids();
- if (!expungedUids.isEmpty()) {
- // Check if QRESYNC was enabled. If so we MUST use VANISHED responses
- if (EnableProcessor.getEnabledCapabilities(session).contains(ImapConstants.SUPPORTS_QRESYNC)) {
- addVanishedResponse(selected, expungedUids, responder);
- } else {
- addExpungedResponses(selected, expungedUids, responder);
- }
- // Only reset the events if we send the EXPUNGE or VANISHED responses. See IMAP-286
- selected.resetExpungedUids();
+ private Mono<Void> unsolicitedResponses(ImapSession session, ImapProcessor.Responder responder, SelectedMailbox selected, boolean omitExpunged, boolean useUid) {
+ return Mono.fromRunnable(() -> {
+ final boolean sizeChanged = selected.isSizeChanged();
+ // New message response
+ if (sizeChanged) {
+ addExistsResponses(selected, responder);
}
+ // Expunged messages
+ if (!omitExpunged) {
+ final Collection<MessageUid> expungedUids = selected.expungedUids();
+ if (!expungedUids.isEmpty()) {
+ // Check if QRESYNC was enabled. If so we MUST use VANISHED responses
+ if (EnableProcessor.getEnabledCapabilities(session).contains(ImapConstants.SUPPORTS_QRESYNC)) {
+ addVanishedResponse(selected, expungedUids, responder);
+ } else {
+ addExpungedResponses(selected, expungedUids, responder);
+ }
+ // Only reset the events if we send the EXPUNGE or VANISHED responses. See IMAP-286
+ selected.resetExpungedUids();
+ }
- }
- if (sizeChanged || (selected.isRecentUidRemoved() && !omitExpunged)) {
- addRecentResponses(selected, responder);
- selected.resetRecentUidRemoved();
- }
-
- // Message updates
- addFlagsResponses(session, selected, responder, useUid);
-
- selected.resetEvents();
+ }
+ if (sizeChanged || (selected.isRecentUidRemoved() && !omitExpunged)) {
+ addRecentResponses(selected, responder);
+ selected.resetRecentUidRemoved();
+ }
+ })
+ .then(Mono.defer(() -> addFlagsResponses(session, selected, responder, useUid)))
+ .doFinally(type -> selected.resetEvents());
}
private void addExpungedResponses(SelectedMailbox selected, Collection<MessageUid> expungedUids, ImapProcessor.Responder responder) {
@@ -192,92 +191,101 @@ public abstract class AbstractMailboxProcessor<R extends ImapRequest> extends Ab
responder.respond(new VanishedResponse(uidRange, false));
}
- private void addFlagsResponses(ImapSession session, SelectedMailbox selected, ImapProcessor.Responder responder, boolean useUid) {
+ private Mono<Void> addFlagsResponses(ImapSession session, SelectedMailbox selected, ImapProcessor.Responder responder, boolean useUid) {
+ MessageManager messageManager = selected.getMessageManager();
+ MailboxSession mailboxSession = session.getMailboxSession();
+
+ final Collection<MessageUid> flagUpdateUids = selected.flagUpdateUids();
+ if (!flagUpdateUids.isEmpty()) {
+ return addApplicableFlagResponse(session, selected, responder, useUid)
+ .then(Flux.fromIterable(MessageRange.toRanges(flagUpdateUids))
+ .concatMap(range ->
+ addFlagsResponses(session, selected, responder, useUid, range, messageManager, mailboxSession))
+ .then())
+ .onErrorResume(MailboxException.class, e -> {
+ handleResponseException(responder, e, HumanReadableText.FAILURE_TO_LOAD_FLAGS, session);
+ return Mono.empty();
+ });
+ } else {
+ return addApplicableFlagResponse(session, selected, responder, useUid)
+ .onErrorResume(MailboxException.class, e -> {
+ handleResponseException(responder, e, HumanReadableText.FAILURE_TO_LOAD_FLAGS, session);
+ return Mono.empty();
+ });
+ }
+ }
+
+ private Mono<Void> addApplicableFlagResponse(ImapSession session, SelectedMailbox selected, ImapProcessor.Responder responder, boolean useUid) {
try {
// To be lazily initialized only if needed, which is in minority of cases.
- MessageManager messageManager = null;
- MailboxMetaData metaData = null;
- final MailboxSession mailboxSession = session.getMailboxSession();
+ MessageManager messageManager = selected.getMessageManager();
+ MailboxSession mailboxSession = session.getMailboxSession();
// Check if we need to send a FLAGS and PERMANENTFLAGS response before the FETCH response
// This is the case if some new flag/keyword was used
// See IMAP-303
if (selected.hasNewApplicableFlags()) {
- messageManager = getMailbox(session, selected);
flags(responder, selected);
- metaData = messageManager.getMetaData(false, mailboxSession, MailboxMetaData.FetchGroup.NO_COUNT);
-
- permanentFlags(responder, metaData, selected);
- selected.resetNewApplicableFlags();
- }
-
- final Collection<MessageUid> flagUpdateUids = selected.flagUpdateUids();
- if (!flagUpdateUids.isEmpty()) {
- Iterator<MessageRange> ranges = MessageRange.toRanges(flagUpdateUids).iterator();
- if (messageManager == null) {
- messageManager = getMailbox(session, selected);
- }
- boolean isModSeqPermanent = true;
- while (ranges.hasNext()) {
- addFlagsResponses(session, selected, responder, useUid, ranges.next(), messageManager, isModSeqPermanent, mailboxSession);
- }
+ return messageManager.getMetaDataReactive(false, mailboxSession, MailboxMetaData.FetchGroup.NO_COUNT)
+ .doOnNext(metaData -> permanentFlags(responder, metaData, selected))
+ .doOnNext(any -> selected.resetNewApplicableFlags())
+ .onErrorResume(MailboxException.class, e -> {
+ handleResponseException(responder, e, HumanReadableText.FAILURE_TO_LOAD_FLAGS, session);
+ return Mono.empty();
+ }).then();
}
-
+ return Mono.empty();
} catch (MailboxException e) {
handleResponseException(responder, e, HumanReadableText.FAILURE_TO_LOAD_FLAGS, session);
+ return Mono.empty();
}
-
}
- private void addFlagsResponses(ImapSession session,
+ private Mono<Void> addFlagsResponses(ImapSession session,
SelectedMailbox selected,
ImapProcessor.Responder responder,
boolean useUid,
MessageRange messageSet, MessageManager mailbox,
- boolean isModSeqPermanent,
- MailboxSession mailboxSession) throws MailboxException {
- Iterator<ComposedMessageIdWithMetaData> it = Flux.from(
- mailbox.listMessagesMetadata(messageSet, mailboxSession))
- .toStream()
- .iterator();
+ MailboxSession mailboxSession) {
final boolean qresyncEnabled = EnableProcessor.getEnabledCapabilities(session).contains(ImapConstants.SUPPORTS_QRESYNC);
final boolean condstoreEnabled = EnableProcessor.getEnabledCapabilities(session).contains(ImapConstants.SUPPORTS_CONDSTORE);
- while (it.hasNext()) {
- ComposedMessageIdWithMetaData mr = it.next();
- final MessageUid uid = mr.getComposedMessageId().getUid();
- selected.msn(uid).fold(() -> {
- LOGGER.debug("No message found with uid {} in the uid<->msn mapping for mailbox {}. This may be because it was deleted by a concurrent session. So skip it..", uid, selected.getMailboxId().serialize());
- // skip this as it was not found in the mapping
- //
- // See IMAP-346
- return null;
- }, msn -> {
-
- final Flags flags = mr.getFlags();
- final MessageUid uidOut;
- if (useUid || qresyncEnabled) {
- uidOut = uid;
- } else {
- uidOut = null;
- }
- if (selected.isRecent(uid)) {
- flags.add(Flags.Flag.RECENT);
- } else {
- flags.remove(Flags.Flag.RECENT);
- }
- final FetchResponse response;
-
- // Check if we also need to return the MODSEQ in the response. This is true if CONDSTORE or
- // if QRESYNC was enabled, and the mailbox supports the permant storage of mod-sequences
- if ((condstoreEnabled || qresyncEnabled) && isModSeqPermanent) {
- response = new FetchResponse(msn, flags, uidOut, mr.getModSeq(), null, null, null, null, null, null);
- } else {
- response = new FetchResponse(msn, flags, uidOut, null, null, null, null, null, null, null);
- }
- responder.respond(response);
- return null;
- });
- }
+ return Flux.from(
+ mailbox.listMessagesMetadata(messageSet, mailboxSession))
+ .doOnNext(Throwing.consumer(mr -> {
+ MessageUid uid = mr.getComposedMessageId().getUid();
+ selected.msn(uid).fold(() -> {
+ LOGGER.debug("No message found with uid {} in the uid<->msn mapping for mailbox {}. This may be because it was deleted by a concurrent session. So skip it..", uid, selected.getMailboxId().serialize());
+ // skip this as it was not found in the mapping
+ //
+ // See IMAP-346
+ return null;
+ }, msn -> {
+
+ final Flags flags = mr.getFlags();
+ final MessageUid uidOut;
+ if (useUid || qresyncEnabled) {
+ uidOut = uid;
+ } else {
+ uidOut = null;
+ }
+ if (selected.isRecent(uid)) {
+ flags.add(Flags.Flag.RECENT);
+ } else {
+ flags.remove(Flags.Flag.RECENT);
+ }
+ final FetchResponse response;
+
+ // Check if we also need to return the MODSEQ in the response. This is true if CONDSTORE or
+ // if QRESYNC was enabled, and the mailbox supports the permant storage of mod-sequences
+ if (condstoreEnabled || qresyncEnabled) {
+ response = new FetchResponse(msn, flags, uidOut, mr.getModSeq(), null, null, null, null, null, null);
+ } else {
+ response = new FetchResponse(msn, flags, uidOut, null, null, null, null, null, null, null);
+ }
+ responder.respond(response);
+ return null;
+ });
+ })).then();
}
protected void condstoreEnablingCommand(ImapSession session, Responder responder, MailboxMetaData metaData, boolean sendHighestModSeq) {
@@ -304,7 +312,7 @@ public abstract class AbstractMailboxProcessor<R extends ImapRequest> extends Ab
responder.respond(response);
}
- private void addExistsResponses(ImapSession session, SelectedMailbox selected, ImapProcessor.Responder responder) {
+ private void addExistsResponses(SelectedMailbox selected, ImapProcessor.Responder responder) {
final long existsCount = selected.existsCount();
final ExistsResponse response = new ExistsResponse(existsCount);
responder.respond(response);
diff --git a/protocols/imap/src/main/java/org/apache/james/imap/processor/AbstractMessageRangeProcessor.java b/protocols/imap/src/main/java/org/apache/james/imap/processor/AbstractMessageRangeProcessor.java
index 11d3cdf7df..f4a1660fc9 100644
--- a/protocols/imap/src/main/java/org/apache/james/imap/processor/AbstractMessageRangeProcessor.java
+++ b/protocols/imap/src/main/java/org/apache/james/imap/processor/AbstractMessageRangeProcessor.java
@@ -75,7 +75,7 @@ public abstract class AbstractMessageRangeProcessor<R extends AbstractMessageRan
no(request, responder, HumanReadableText.FAILURE_NO_SUCH_MAILBOX, StatusResponse.ResponseCode.tryCreate());
} else {
StatusResponse.ResponseCode code = handleRanges(request, session, targetMailbox, mailboxSession);
- unsolicitedResponses(session, responder, request.isUseUids());
+ unsolicitedResponses(session, responder, request.isUseUids()).block();
okComplete(request, code, responder);
}
} catch (MessageRangeException e) {
diff --git a/protocols/imap/src/main/java/org/apache/james/imap/processor/AppendProcessor.java b/protocols/imap/src/main/java/org/apache/james/imap/processor/AppendProcessor.java
index 1ea46030cb..c94c1505e6 100644
--- a/protocols/imap/src/main/java/org/apache/james/imap/processor/AppendProcessor.java
+++ b/protocols/imap/src/main/java/org/apache/james/imap/processor/AppendProcessor.java
@@ -116,21 +116,20 @@ public class AppendProcessor extends AbstractMailboxProcessor<AppendRequest> {
.isRecent(!isSelectedMailbox)
.build(message), mailboxSession))
.map(MessageManager.AppendResult::getId)
- .doOnNext(Throwing.<ComposedMessageId>consumer(messageId -> {
- if (isSelectedMailbox) {
- selectedMailbox.addRecent(messageId.getUid());
- }
-
+ .map(Throwing.<ComposedMessageId, ComposedMessageId>function(messageId -> {
+ if (isSelectedMailbox) {
+ selectedMailbox.addRecent(messageId.getUid());
+ }
+ return messageId;
+ }).sneakyThrow())
+ .flatMap(messageId -> unsolicitedResponses(session, responder, false).thenReturn(messageId))
+ .doOnNext(Throwing.consumer(messageId -> {
// get folder UIDVALIDITY
UidValidity uidValidity = mailbox
.getMailboxEntity()
.getUidValidity();
-
- unsolicitedResponses(session, responder, false);
-
- // in case of MULTIAPPEND support we will push more then one UID here
okComplete(request, ResponseCode.appendUid(uidValidity, new UidRange[] { new UidRange(messageId.getUid()) }), responder);
- }).sneakyThrow())
+ }))
.onErrorResume(MailboxNotFoundException.class, e -> {
// Indicates that the mailbox does not exist
// So TRY CREATE
diff --git a/protocols/imap/src/main/java/org/apache/james/imap/processor/CapabilityProcessor.java b/protocols/imap/src/main/java/org/apache/james/imap/processor/CapabilityProcessor.java
index da8125f76b..ec268a523b 100644
--- a/protocols/imap/src/main/java/org/apache/james/imap/processor/CapabilityProcessor.java
+++ b/protocols/imap/src/main/java/org/apache/james/imap/processor/CapabilityProcessor.java
@@ -43,6 +43,8 @@ import org.apache.james.util.MDCBuilder;
import com.google.common.collect.ImmutableList;
+import reactor.core.publisher.Mono;
+
public class CapabilityProcessor extends AbstractMailboxProcessor<CapabilityRequest> implements CapabilityImplementingProcessor {
private static final List<Capability> CAPS = ImmutableList.of(
@@ -76,11 +78,11 @@ public class CapabilityProcessor extends AbstractMailboxProcessor<CapabilityRequ
}
@Override
- protected void processRequest(CapabilityRequest request, ImapSession session, Responder responder) {
+ protected Mono<Void> processRequestReactive(CapabilityRequest request, ImapSession session, Responder responder) {
final CapabilityResponse result = new CapabilityResponse(getSupportedCapabilities(session));
responder.respond(result);
- unsolicitedResponses(session, responder, false);
- okComplete(request, responder);
+ return unsolicitedResponses(session, responder, false)
+ .then(Mono.fromRunnable(() -> okComplete(request, responder)));
}
/**
diff --git a/protocols/imap/src/main/java/org/apache/james/imap/processor/CheckProcessor.java b/protocols/imap/src/main/java/org/apache/james/imap/processor/CheckProcessor.java
index a691d02746..bfe87646c0 100644
--- a/protocols/imap/src/main/java/org/apache/james/imap/processor/CheckProcessor.java
+++ b/protocols/imap/src/main/java/org/apache/james/imap/processor/CheckProcessor.java
@@ -28,6 +28,8 @@ import org.apache.james.mailbox.MailboxManager;
import org.apache.james.metrics.api.MetricFactory;
import org.apache.james.util.MDCBuilder;
+import reactor.core.publisher.Mono;
+
public class CheckProcessor extends AbstractMailboxProcessor<CheckRequest> {
public CheckProcessor(MailboxManager mailboxManager, StatusResponseFactory factory,
@@ -36,9 +38,9 @@ public class CheckProcessor extends AbstractMailboxProcessor<CheckRequest> {
}
@Override
- protected void processRequest(CheckRequest request, ImapSession session, Responder responder) {
- unsolicitedResponses(session, responder, false);
- okComplete(request, responder);
+ protected Mono<Void> processRequestReactive(CheckRequest request, ImapSession session, Responder responder) {
+ return unsolicitedResponses(session, responder, false)
+ .then(Mono.fromRunnable(() -> okComplete(request, responder)));
}
@Override
diff --git a/protocols/imap/src/main/java/org/apache/james/imap/processor/CreateProcessor.java b/protocols/imap/src/main/java/org/apache/james/imap/processor/CreateProcessor.java
index 9f87356ff8..1e480741a7 100644
--- a/protocols/imap/src/main/java/org/apache/james/imap/processor/CreateProcessor.java
+++ b/protocols/imap/src/main/java/org/apache/james/imap/processor/CreateProcessor.java
@@ -50,7 +50,7 @@ public class CreateProcessor extends AbstractMailboxProcessor<CreateRequest> {
try {
final MailboxManager mailboxManager = getMailboxManager();
mailboxManager.createMailbox(mailboxPath, session.getMailboxSession());
- unsolicitedResponses(session, responder, false);
+ unsolicitedResponses(session, responder, false).block();
okComplete(request, responder);
} catch (MailboxExistsException e) {
LOGGER.debug("Create failed for mailbox {} as it already exists", mailboxPath, e);
diff --git a/protocols/imap/src/main/java/org/apache/james/imap/processor/DeleteProcessor.java b/protocols/imap/src/main/java/org/apache/james/imap/processor/DeleteProcessor.java
index c5bcc701a2..7b777a1eab 100644
--- a/protocols/imap/src/main/java/org/apache/james/imap/processor/DeleteProcessor.java
+++ b/protocols/imap/src/main/java/org/apache/james/imap/processor/DeleteProcessor.java
@@ -54,7 +54,7 @@ public class DeleteProcessor extends AbstractMailboxProcessor<DeleteRequest> {
}
final MailboxManager mailboxManager = getMailboxManager();
mailboxManager.deleteMailbox(mailboxPath, session.getMailboxSession());
- unsolicitedResponses(session, responder, false);
+ unsolicitedResponses(session, responder, false).block();
okComplete(request, responder);
} catch (MailboxNotFoundException e) {
LOGGER.debug("Delete failed for mailbox {} as it doesn't exist", mailboxPath, e);
diff --git a/protocols/imap/src/main/java/org/apache/james/imap/processor/EnableProcessor.java b/protocols/imap/src/main/java/org/apache/james/imap/processor/EnableProcessor.java
index 284ced27b3..dda5308969 100644
--- a/protocols/imap/src/main/java/org/apache/james/imap/processor/EnableProcessor.java
+++ b/protocols/imap/src/main/java/org/apache/james/imap/processor/EnableProcessor.java
@@ -44,6 +44,8 @@ import org.slf4j.LoggerFactory;
import com.google.common.collect.ImmutableList;
+import reactor.core.publisher.Mono;
+
public class EnableProcessor extends AbstractMailboxProcessor<EnableRequest> implements CapabilityImplementingProcessor {
private static final Logger LOGGER = LoggerFactory.getLogger(EnableProcessor.class);
@@ -67,18 +69,19 @@ public class EnableProcessor extends AbstractMailboxProcessor<EnableRequest> imp
@Override
- protected void processRequest(EnableRequest request, ImapSession session, Responder responder) {
+ protected Mono<Void> processRequestReactive(EnableRequest request, ImapSession session, Responder responder) {
try {
List<Capability> caps = request.getCapabilities();
Set<Capability> enabledCaps = enable(request, responder, session, caps.iterator());
responder.respond(new EnableResponse(enabledCaps));
- unsolicitedResponses(session, responder, false);
- okComplete(request, responder);
+ return unsolicitedResponses(session, responder, false)
+ .then(Mono.fromRunnable(() -> okComplete(request, responder)));
} catch (EnableException e) {
LOGGER.info("Unable to enable extension", e);
taggedBad(request, responder, HumanReadableText.FAILED);
+ return Mono.empty();
}
}
diff --git a/protocols/imap/src/main/java/org/apache/james/imap/processor/ExpungeProcessor.java b/protocols/imap/src/main/java/org/apache/james/imap/processor/ExpungeProcessor.java
index b806dfc354..09663d0c75 100644
--- a/protocols/imap/src/main/java/org/apache/james/imap/processor/ExpungeProcessor.java
+++ b/protocols/imap/src/main/java/org/apache/james/imap/processor/ExpungeProcessor.java
@@ -88,7 +88,7 @@ public class ExpungeProcessor extends AbstractMailboxProcessor<ExpungeRequest> i
}
}
- unsolicitedResponses(session, responder, false);
+ unsolicitedResponses(session, responder, false).block();
// Check if QRESYNC was enabled and at least one message was expunged. If so we need to respond with an OK response that contain the HIGHESTMODSEQ
diff --git a/protocols/imap/src/main/java/org/apache/james/imap/processor/IdleProcessor.java b/protocols/imap/src/main/java/org/apache/james/imap/processor/IdleProcessor.java
index dab830218d..acda2e5a24 100644
--- a/protocols/imap/src/main/java/org/apache/james/imap/processor/IdleProcessor.java
+++ b/protocols/imap/src/main/java/org/apache/james/imap/processor/IdleProcessor.java
@@ -50,6 +50,8 @@ import org.slf4j.LoggerFactory;
import com.google.common.collect.ImmutableList;
+import reactor.core.publisher.Mono;
+
public class IdleProcessor extends AbstractMailboxProcessor<IdleRequest> implements CapabilityImplementingProcessor {
private static final Logger LOGGER = LoggerFactory.getLogger(IdleProcessor.class);
@@ -73,7 +75,7 @@ public class IdleProcessor extends AbstractMailboxProcessor<IdleRequest> impleme
}
@Override
- protected void processRequest(IdleRequest request, ImapSession session, Responder responder) {
+ protected Mono<Void> processRequestReactive(IdleRequest request, ImapSession session, Responder responder) {
SelectedMailbox sm = session.getSelected();
if (sm != null) {
sm.registerIdle(new IdleMailboxListener(session, responder));
@@ -137,7 +139,7 @@ public class IdleProcessor extends AbstractMailboxProcessor<IdleRequest> impleme
// Write the response after the listener was add
// IMAP-341
responder.respond(new ContinuationResponse(HumanReadableText.IDLING));
- unsolicitedResponses(session, responder, false);
+ return unsolicitedResponses(session, responder, false);
}
@Override
@@ -162,7 +164,7 @@ public class IdleProcessor extends AbstractMailboxProcessor<IdleRequest> impleme
@Override
public void event(Event event) {
- unsolicitedResponses(session, responder, false);
+ unsolicitedResponses(session, responder, false).block();
}
@Override
diff --git a/protocols/imap/src/main/java/org/apache/james/imap/processor/NamespaceProcessor.java b/protocols/imap/src/main/java/org/apache/james/imap/processor/NamespaceProcessor.java
index 31dd581e6b..9d8d364332 100644
--- a/protocols/imap/src/main/java/org/apache/james/imap/processor/NamespaceProcessor.java
+++ b/protocols/imap/src/main/java/org/apache/james/imap/processor/NamespaceProcessor.java
@@ -37,6 +37,8 @@ import org.apache.james.util.MDCBuilder;
import com.google.common.collect.ImmutableList;
+import reactor.core.publisher.Mono;
+
/**
* Processes a NAMESPACE command into a suitable set of responses.
*/
@@ -48,15 +50,15 @@ public class NamespaceProcessor extends AbstractMailboxProcessor<NamespaceReques
}
@Override
- protected void processRequest(NamespaceRequest request, ImapSession session, Responder responder) {
+ protected Mono<Void> processRequestReactive(NamespaceRequest request, ImapSession session, Responder responder) {
final MailboxSession mailboxSession = session.getMailboxSession();
final List<NamespaceResponse.Namespace> personalNamespaces = buildPersonalNamespaces(mailboxSession, session);
final List<NamespaceResponse.Namespace> otherUsersNamespaces = buildOtherUsersSpaces(mailboxSession, session);
final List<NamespaceResponse.Namespace> sharedNamespaces = buildSharedNamespaces(mailboxSession, session);
final NamespaceResponse response = new NamespaceResponse(personalNamespaces, otherUsersNamespaces, sharedNamespaces);
responder.respond(response);
- unsolicitedResponses(session, responder, false);
- okComplete(request, responder);
+ return unsolicitedResponses(session, responder, false)
+ .then(Mono.fromRunnable(() -> okComplete(request, responder)));
}
/**
diff --git a/protocols/imap/src/main/java/org/apache/james/imap/processor/NoopProcessor.java b/protocols/imap/src/main/java/org/apache/james/imap/processor/NoopProcessor.java
index 7101da9087..6d55a58d34 100644
--- a/protocols/imap/src/main/java/org/apache/james/imap/processor/NoopProcessor.java
+++ b/protocols/imap/src/main/java/org/apache/james/imap/processor/NoopProcessor.java
@@ -28,16 +28,18 @@ import org.apache.james.mailbox.MailboxManager;
import org.apache.james.metrics.api.MetricFactory;
import org.apache.james.util.MDCBuilder;
+import reactor.core.publisher.Mono;
+
public class NoopProcessor extends AbstractMailboxProcessor<NoopRequest> {
public NoopProcessor(MailboxManager mailboxManager, StatusResponseFactory factory, MetricFactory metricFactory) {
super(NoopRequest.class, mailboxManager, factory, metricFactory);
}
@Override
- protected void processRequest(NoopRequest request, ImapSession session, Responder responder) {
+ protected Mono<Void> processRequestReactive(NoopRequest request, ImapSession session, Responder responder) {
// So, unsolicated responses are returned to check for new mail
- unsolicitedResponses(session, responder, false);
- okComplete(request, responder);
+ return unsolicitedResponses(session, responder, false)
+ .then(Mono.fromRunnable(() -> okComplete(request, responder)));
}
@Override
diff --git a/protocols/imap/src/main/java/org/apache/james/imap/processor/RenameProcessor.java b/protocols/imap/src/main/java/org/apache/james/imap/processor/RenameProcessor.java
index a70d0fef93..9306f6925d 100644
--- a/protocols/imap/src/main/java/org/apache/james/imap/processor/RenameProcessor.java
+++ b/protocols/imap/src/main/java/org/apache/james/imap/processor/RenameProcessor.java
@@ -63,7 +63,7 @@ public class RenameProcessor extends AbstractMailboxProcessor<RenameRequest> {
mailboxManager.createMailbox(existingPath, mailboxsession);
}
okComplete(request, responder);
- unsolicitedResponses(session, responder, false);
+ unsolicitedResponses(session, responder, false).block();
} catch (MailboxExistsException e) {
LOGGER.debug("Rename from {} to {} failed because the target mailbox exists", existingPath, newPath, e);
no(request, responder, HumanReadableText.FAILURE_MAILBOX_EXISTS);
diff --git a/protocols/imap/src/main/java/org/apache/james/imap/processor/StatusProcessor.java b/protocols/imap/src/main/java/org/apache/james/imap/processor/StatusProcessor.java
index c34ed31ccd..e3c67ee3a1 100644
--- a/protocols/imap/src/main/java/org/apache/james/imap/processor/StatusProcessor.java
+++ b/protocols/imap/src/main/java/org/apache/james/imap/processor/StatusProcessor.java
@@ -66,7 +66,7 @@ public class StatusProcessor extends AbstractMailboxProcessor<StatusRequest> {
condstoreEnablingCommand(session, responder, metaData, false);
}
responder.respond(response);
- unsolicitedResponses(session, responder, false);
+ unsolicitedResponses(session, responder, false).block();
okComplete(request, responder);
} catch (MailboxException e) {
diff --git a/protocols/imap/src/main/java/org/apache/james/imap/processor/StoreProcessor.java b/protocols/imap/src/main/java/org/apache/james/imap/processor/StoreProcessor.java
index 7484d58326..dcb185cdf4 100644
--- a/protocols/imap/src/main/java/org/apache/james/imap/processor/StoreProcessor.java
+++ b/protocols/imap/src/main/java/org/apache/james/imap/processor/StoreProcessor.java
@@ -166,7 +166,7 @@ public class StoreProcessor extends AbstractMailboxProcessor<StoreRequest> {
}
final boolean omitExpunged = (!useUids);
- unsolicitedResponses(session, responder, omitExpunged, useUids);
+ unsolicitedResponses(session, responder, omitExpunged, useUids).block();
// check if we had some failed uids which didn't pass the UNCHANGEDSINCE filter
if (failed.isEmpty() && failedMsns.isEmpty()) {
diff --git a/protocols/imap/src/main/java/org/apache/james/imap/processor/SubscribeProcessor.java b/protocols/imap/src/main/java/org/apache/james/imap/processor/SubscribeProcessor.java
index a01e273438..c13175018f 100644
--- a/protocols/imap/src/main/java/org/apache/james/imap/processor/SubscribeProcessor.java
+++ b/protocols/imap/src/main/java/org/apache/james/imap/processor/SubscribeProcessor.java
@@ -49,12 +49,12 @@ public class SubscribeProcessor extends AbstractSubscriptionProcessor<SubscribeR
try {
getSubscriptionManager().subscribe(mailboxSession, mailboxName);
- unsolicitedResponses(session, responder, false);
+ unsolicitedResponses(session, responder, false).block();
okComplete(request, responder);
} catch (SubscriptionException e) {
LOGGER.info("Subscribe failed for mailbox {}", mailboxName, e);
- unsolicitedResponses(session, responder, false);
+ unsolicitedResponses(session, responder, false).block();
no(request, responder, HumanReadableText.GENERIC_SUBSCRIPTION_FAILURE);
}
}
diff --git a/protocols/imap/src/main/java/org/apache/james/imap/processor/UnsubscribeProcessor.java b/protocols/imap/src/main/java/org/apache/james/imap/processor/UnsubscribeProcessor.java
index ddcfa144fd..8d375ba899 100644
--- a/protocols/imap/src/main/java/org/apache/james/imap/processor/UnsubscribeProcessor.java
+++ b/protocols/imap/src/main/java/org/apache/james/imap/processor/UnsubscribeProcessor.java
@@ -49,12 +49,12 @@ public class UnsubscribeProcessor extends AbstractSubscriptionProcessor<Unsubscr
try {
getSubscriptionManager().unsubscribe(mailboxSession, mailboxName);
- unsolicitedResponses(session, responder, false);
+ unsolicitedResponses(session, responder, false).block();
okComplete(request, responder);
} catch (SubscriptionException e) {
LOGGER.info("Unsubscribe failed for mailbox {}", mailboxName, e);
- unsolicitedResponses(session, responder, false);
+ unsolicitedResponses(session, responder, false).block();
no(request, responder, HumanReadableText.GENERIC_SUBSCRIPTION_FAILURE);
}
diff --git a/protocols/imap/src/main/java/org/apache/james/imap/processor/base/SelectedMailboxImpl.java b/protocols/imap/src/main/java/org/apache/james/imap/processor/base/SelectedMailboxImpl.java
index 7bb868a067..b1b2876906 100644
--- a/protocols/imap/src/main/java/org/apache/james/imap/processor/base/SelectedMailboxImpl.java
+++ b/protocols/imap/src/main/java/org/apache/james/imap/processor/base/SelectedMailboxImpl.java
@@ -199,6 +199,11 @@ public class SelectedMailboxImpl implements SelectedMailbox, EventListener {
recentUids.clear();
}
+ @Override
+ public MessageManager getMessageManager() {
+ return messageManager;
+ }
+
@Override
public synchronized boolean removeRecent(MessageUid uid) {
final boolean result = recentUids.remove(uid);
diff --git a/protocols/imap/src/main/java/org/apache/james/imap/processor/fetch/FetchProcessor.java b/protocols/imap/src/main/java/org/apache/james/imap/processor/fetch/FetchProcessor.java
index f2bae05d72..38870f95fe 100644
--- a/protocols/imap/src/main/java/org/apache/james/imap/processor/fetch/FetchProcessor.java
+++ b/protocols/imap/src/main/java/org/apache/james/imap/processor/fetch/FetchProcessor.java
@@ -109,15 +109,12 @@ public class FetchProcessor extends AbstractMailboxProcessor<FetchRequest> {
// If we do so we could prolly save one mailbox access which should give use some more speed up
respondVanished(session.getSelected(), ranges, responder);
}
+ boolean omitExpunged = (!useUids);
return processMessageRanges(session, mailbox, ranges, fetch, mailboxSession, responder)
- .then(Mono.fromRunnable(() -> {
- // Don't send expunge responses if FETCH is used to trigger this
- // processor. See IMAP-284
- final boolean omitExpunged = (!useUids);
- unsolicitedResponses(session, responder, omitExpunged, useUids);
- okComplete(request, responder);
- }))
- .then();
+ // Don't send expunge responses if FETCH is used to trigger this
+ // processor. See IMAP-284
+ .then(Mono.defer(() -> unsolicitedResponses(session, responder, omitExpunged, useUids)))
+ .then(Mono.fromRunnable(() -> okComplete(request, responder)));
}).sneakyThrow())
.onErrorResume(MessageRangeException.class, e -> {
LOGGER.debug("Fetch failed for mailbox {} because of invalid sequence-set {}", session.getSelected().getMailboxId(), idSet, e);
@@ -147,7 +144,7 @@ public class FetchProcessor extends AbstractMailboxProcessor<FetchRequest> {
* {@link org.apache.james.imap.api.process.ImapProcessor.Responder}
*/
private Mono<Void> processMessageRanges(ImapSession session, MessageManager mailbox, List<MessageRange> ranges, FetchData fetch, MailboxSession mailboxSession, Responder responder) throws MailboxException {
- final FetchResponseBuilder builder = new FetchResponseBuilder(new EnvelopeBuilder());
+ FetchResponseBuilder builder = new FetchResponseBuilder(new EnvelopeBuilder());
FetchGroup resultToFetch = FetchDataConverter.getFetchGroup(fetch);
return Flux.fromIterable(ranges)
diff --git a/server/protocols/protocols-imap4/src/test/java/org/apache/james/imapserver/netty/IMAPServerTest.java b/server/protocols/protocols-imap4/src/test/java/org/apache/james/imapserver/netty/IMAPServerTest.java
index 168ffd4db5..baa8d8935a 100644
--- a/server/protocols/protocols-imap4/src/test/java/org/apache/james/imapserver/netty/IMAPServerTest.java
+++ b/server/protocols/protocols-imap4/src/test/java/org/apache/james/imapserver/netty/IMAPServerTest.java
@@ -1479,6 +1479,7 @@ class IMAPServerTest {
readStringUntil(clientConnection, s -> s.contains("a2 OK [READ-WRITE] SELECT completed."));
clientConnection.write(ByteBuffer.wrap(("a3 IDLE\r\n").getBytes(StandardCharsets.UTF_8)));
+ System.out.println("toto");
readStringUntil(clientConnection, s -> s.contains("+ Idling"));
inbox.appendMessage(MessageManager.AppendCommand.builder().build("h: value\r\n\r\nbody".getBytes()), mailboxSession);
@@ -1748,6 +1749,7 @@ class IMAPServerTest {
ImmutableList.Builder<String> result = ImmutableList.builder();
while (true) {
String line = new String(readBytes(channel), StandardCharsets.US_ASCII);
+ System.out.println(line);
result.add(line);
if (condition.test(line)) {
return result.build();
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org