Various JMAP performance enhancements

event-bus/distributed/src/main/java/org/apache/james/events/GroupRegistration.java

@@ -133,6 +133,7 @@ private Disposable consumeWorkQueue() {
             .publishOn(Schedulers.parallel())
             .filter(delivery -> Objects.nonNull(delivery.getBody()))
             .flatMap(this::deliver, EventBus.EXECUTION_RATE)
+            .subscribeOn(Schedulers.elastic())
             .subscribe();
     }
 
@@ -144,7 +145,7 @@ private Mono<Void> deliver(AcknowledgableDelivery acknowledgableDelivery) {
             .flatMap(event -> delayGenerator.delayIfHaveTo(currentRetryCount)
                 .flatMap(any -> runListener(event))
                 .onErrorResume(throwable -> retryHandler.handleRetry(event, currentRetryCount, throwable))
-                .then(Mono.<Void>fromRunnable(acknowledgableDelivery::ack)))
+                .then(Mono.<Void>fromRunnable(acknowledgableDelivery::ack).subscribeOn(Schedulers.elastic())))
             .onErrorResume(e -> {
                 LOGGER.error("Unable to process delivery for group {}", group, e);
                 return Mono.fromRunnable(() -> acknowledgableDelivery.nack(!REQUEUE));

event-bus/distributed/src/main/java/org/apache/james/events/KeyRegistrationHandler.java

@@ -96,6 +96,7 @@ void start() {
         receiverSubscriber = Optional.of(receiver.consumeAutoAck(registrationQueue.asString(), new ConsumeOptions().qos(EventBus.EXECUTION_RATE))
             .subscribeOn(Schedulers.parallel())
             .flatMap(this::handleDelivery, EventBus.EXECUTION_RATE)
+            .subscribeOn(Schedulers.elastic())
             .subscribe());
     }
 

event-sourcing/event-sourcing-core/src/main/scala/org/apache/james/eventsourcing/EventBus.scala

@@ -19,11 +19,9 @@
 package org.apache.james.eventsourcing
 
 import javax.inject.Inject
-
 import org.apache.james.eventsourcing.eventstore.{EventStore, EventStoreFailedException}
 import org.reactivestreams.Publisher
 import org.slf4j.LoggerFactory
-
 import reactor.core.scala.publisher.{SFlux, SMono}
 
 object EventBus {
@@ -32,27 +30,20 @@ object EventBus {
 
 class EventBus @Inject() (eventStore: EventStore, subscribers: Set[Subscriber]) {
   @throws[EventStoreFailedException]
-  def publish(events: Iterable[Event]): SMono[Void] = {
+  def publish(events: Iterable[Event]): SMono[Void] =
     SMono(eventStore.appendAll(events))
         .`then`(runHandlers(events, subscribers))
 
-  }
-
-  def runHandlers(events: Iterable[Event], subscribers: Set[Subscriber]): SMono[Void] = {
+  def runHandlers(events: Iterable[Event], subscribers: Set[Subscriber]): SMono[Void] =
     SFlux.fromIterable(events.flatMap((event: Event) => subscribers.map(subscriber => (event, subscriber))))
-      .flatMap(infos => runHandler(infos._1, infos._2))
+      .flatMapSequential(infos => runHandler(infos._1, infos._2))
       .`then`()
       .`then`(SMono.empty)
-  }
-
-  def runHandler(event: Event, subscriber: Subscriber): Publisher[Void] = SMono.fromCallable(() => handle(event, subscriber)).`then`(SMono.empty)
 
-  private def handle(event : Event, subscriber: Subscriber) : Unit = {
-    try {
-      subscriber.handle(event)
-    } catch {
-      case e: Exception =>
+  def runHandler(event: Event, subscriber: Subscriber): Publisher[Void] =
+    SMono(ReactiveSubscriber.asReactiveSubscriber(subscriber).handleReactive(event))
+      .onErrorResume(e => {
         EventBus.LOGGER.error("Error while calling {} for {}", subscriber, event, e)
-    }
-  }
+        SMono.empty
+      })
 }

event-sourcing/event-sourcing-core/src/main/scala/org/apache/james/eventsourcing/Subscriber.scala

@@ -18,6 +18,31 @@
  * ***************************************************************/
 package org.apache.james.eventsourcing
 
+import org.reactivestreams.Publisher
+import reactor.core.scala.publisher.SMono
+import reactor.core.scheduler.Schedulers
+
 trait Subscriber {
   def handle(event: Event) : Unit
+}
+
+trait ReactiveSubscriber extends Subscriber {
+  def handleReactive(event: Event): Publisher[Void]
+
+  override def handle(event: Event) : Unit = SMono(handleReactive(event)).block()
+}
+
+object ReactiveSubscriber {
+  def asReactiveSubscriber(subscriber: Subscriber): ReactiveSubscriber = subscriber match {
+    case reactive: ReactiveSubscriber => reactive
+    case nonReactive => new ReactiveSubscriberWrapper(nonReactive)
+  }
+}
+
+class ReactiveSubscriberWrapper(delegate: Subscriber) extends ReactiveSubscriber {
+  override def handle(event: Event) : Unit = delegate.handle(event)
+
+  def handleReactive(event: Event): Publisher[Void] = SMono.fromCallable(() => handle(event))
+    .subscribeOn(Schedulers.elastic())
+    .`then`()
 }

mailbox/api/src/main/java/org/apache/james/mailbox/MailboxManager.java

@@ -139,6 +139,10 @@ enum SearchCapabilities {
      */
     MessageManager getMailbox(MailboxId mailboxId, MailboxSession session) throws MailboxException;
 
+    Publisher<MessageManager> getMailboxReactive(MailboxId mailboxId, MailboxSession session);
+
+    Publisher<MessageManager> getMailboxReactive(MailboxPath mailboxPath, MailboxSession session);
+
     /**
      * Creates a new mailbox. Any intermediary mailboxes missing from the
      * hierarchy should be created.

mailbox/api/src/main/java/org/apache/james/mailbox/quota/MaxQuotaManager.java

@@ -32,9 +32,13 @@
 import org.apache.james.mailbox.model.Quota;
 import org.apache.james.mailbox.model.Quota.Scope;
 import org.apache.james.mailbox.model.QuotaRoot;
+import org.reactivestreams.Publisher;
 
 import com.github.fge.lambdas.Throwing;
 
+import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
+
 /**
  * This interface describe how to set the max quotas for users
  * Part of RFC 2087 implementation
@@ -150,12 +154,30 @@ default Optional<QuotaCountLimit> getMaxMessage(Map<Quota.Scope, QuotaCountLimit
 
     Map<Quota.Scope, QuotaCountLimit> listMaxMessagesDetails(QuotaRoot quotaRoot);
 
+    default Publisher<Map<Scope, QuotaCountLimit>> listMaxMessagesDetailsReactive(QuotaRoot quotaRoot) {
+        return Mono.fromCallable(() -> listMaxMessagesDetails(quotaRoot))
+            .subscribeOn(Schedulers.elastic());
+    }
+
     Map<Quota.Scope, QuotaSizeLimit> listMaxStorageDetails(QuotaRoot quotaRoot);
 
+    default Publisher<Map<Quota.Scope, QuotaSizeLimit>> listMaxStorageDetailsReactive(QuotaRoot quotaRoot) {
+        return Mono.fromCallable(() -> listMaxStorageDetails(quotaRoot))
+            .subscribeOn(Schedulers.elastic());
+    }
+
+
     default QuotaDetails quotaDetails(QuotaRoot quotaRoot) {
         return new QuotaDetails(listMaxMessagesDetails(quotaRoot), listMaxStorageDetails(quotaRoot));
     }
 
+    default Publisher<QuotaDetails> quotaDetailsReactive(QuotaRoot quotaRoot) {
+        return Mono.zip(
+            Mono.from(listMaxMessagesDetailsReactive(quotaRoot)),
+            Mono.from(listMaxStorageDetailsReactive(quotaRoot)))
+            .map(tuple -> new QuotaDetails(tuple.getT1(), tuple.getT2()));
+    }
+
     Optional<QuotaCountLimit> getDomainMaxMessage(Domain domain);
 
     void setDomainMaxMessage(Domain domain, QuotaCountLimit count) throws MailboxException;

mailbox/api/src/main/java/org/apache/james/mailbox/quota/QuotaManager.java

@@ -25,6 +25,7 @@
 import org.apache.james.mailbox.exception.MailboxException;
 import org.apache.james.mailbox.model.Quota;
 import org.apache.james.mailbox.model.QuotaRoot;
+import org.reactivestreams.Publisher;
 
 
 /**
@@ -68,4 +69,6 @@ public Quota<QuotaSizeLimit, QuotaSizeUsage> getStorageQuota() {
     Quota<QuotaSizeLimit, QuotaSizeUsage> getStorageQuota(QuotaRoot quotaRoot) throws MailboxException;
 
     Quotas getQuotas(QuotaRoot quotaRoot) throws MailboxException;
+
+    Publisher<Quotas> getQuotasReactive(QuotaRoot quotaRoot);
 }

mailbox/api/src/main/java/org/apache/james/mailbox/quota/QuotaRootResolver.java

@@ -37,10 +37,14 @@ public interface QuotaRootResolver extends QuotaRootDeserializer {
      */
     QuotaRoot getQuotaRoot(MailboxPath mailboxPath) throws MailboxException;
 
+    Publisher<QuotaRoot> getQuotaRootReactive(MailboxPath mailboxPath);
+
     QuotaRoot getQuotaRoot(MailboxId mailboxId) throws MailboxException;
 
     QuotaRoot getQuotaRoot(Mailbox mailbox) throws MailboxException;
 
+    Publisher<QuotaRoot> getQuotaRootReactive(Mailbox mailbox);
+
     Publisher<QuotaRoot> getQuotaRootReactive(MailboxId mailboxId);
 
     Publisher<Mailbox> retrieveAssociatedMailboxes(QuotaRoot quotaRoot, MailboxSession mailboxSession);

mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/DeleteMessageListener.java

@@ -62,6 +62,7 @@
 import org.apache.james.mailbox.model.MessageRange;
 import org.apache.james.mailbox.store.mail.MessageMapper;
 import org.apache.james.util.streams.Limit;
+import org.reactivestreams.Publisher;
 
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -77,7 +78,7 @@
  * Mailbox listener failures lead to eventBus retrying their execution, it ensures the result of the deletion to be
  * idempotent.
  */
-public class DeleteMessageListener implements EventListener.GroupEventListener {
+public class DeleteMessageListener implements EventListener.ReactiveGroupEventListener {
     private static final Optional<CassandraId> ALL_MAILBOXES = Optional.empty();
 
     public static class DeleteMessageListenerGroup extends Group {
@@ -138,21 +139,20 @@ public boolean isHandling(Event event) {
     }
 
     @Override
-    public void event(Event event) {
+    public Publisher<Void> reactiveEvent(Event event) {
         if (event instanceof Expunged) {
             Expunged expunged = (Expunged) event;
 
-            handleMessageDeletion(expunged)
-                .block();
+            return handleMessageDeletion(expunged);
         }
         if (event instanceof MailboxDeletion) {
             MailboxDeletion mailboxDeletion = (MailboxDeletion) event;
 
             CassandraId mailboxId = (CassandraId) mailboxDeletion.getMailboxId();
 
-            handleMailboxDeletion(mailboxId)
-                .block();
+            return handleMailboxDeletion(mailboxId);
         }
+        return Mono.empty();
     }
 
     private Mono<Void> handleMailboxDeletion(CassandraId mailboxId) {

mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMailboxMapper.java

@@ -147,7 +147,8 @@ private Mono<Mailbox> performPathReadRepair(Mailbox mailboxPathEntry) {
     }
 
     private boolean shouldReadRepair() {
-        return secureRandom.nextFloat() < cassandraConfiguration.getMailboxReadRepair();
+        return cassandraConfiguration.getMailboxReadRepair() > 0
+            && secureRandom.nextFloat() < cassandraConfiguration.getMailboxReadRepair();
     }
 
 

mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdMapper.java

@@ -49,7 +49,6 @@
 import org.apache.james.mailbox.store.mail.MailboxMapper;
 import org.apache.james.mailbox.store.mail.MessageIdMapper;
 import org.apache.james.mailbox.store.mail.MessageMapper.FetchType;
-import org.apache.james.mailbox.store.mail.ModSeqProvider;
 import org.apache.james.mailbox.store.mail.model.MailboxMessage;
 import org.apache.james.util.FunctionalUtils;
 import org.apache.james.util.ReactorUtils;
@@ -80,14 +79,14 @@ public class CassandraMessageIdMapper implements MessageIdMapper {
     private final CassandraMessageDAO messageDAO;
     private final CassandraMessageDAOV3 messageDAOV3;
     private final CassandraIndexTableHandler indexTableHandler;
-    private final ModSeqProvider modSeqProvider;
+    private final CassandraModSeqProvider modSeqProvider;
     private final AttachmentLoader attachmentLoader;
     private final CassandraConfiguration cassandraConfiguration;
 
     public CassandraMessageIdMapper(MailboxMapper mailboxMapper, CassandraMailboxDAO mailboxDAO, CassandraAttachmentMapper attachmentMapper,
                                     CassandraMessageIdToImapUidDAO imapUidDAO, CassandraMessageIdDAO messageIdDAO,
                                     CassandraMessageDAO messageDAO, CassandraMessageDAOV3 messageDAOV3, CassandraIndexTableHandler indexTableHandler,
-                                    ModSeqProvider modSeqProvider, CassandraConfiguration cassandraConfiguration) {
+                                    CassandraModSeqProvider modSeqProvider, CassandraConfiguration cassandraConfiguration) {
 
         this.mailboxMapper = mailboxMapper;
         this.mailboxDAO = mailboxDAO;
@@ -305,11 +304,11 @@ private Mono<Pair<Flags, ComposedMessageIdWithMetaData>> updateFlags(Flags newSt
         if (identicalFlags(oldComposedId, newFlags)) {
             return Mono.just(Pair.of(oldComposedId.getFlags(), oldComposedId));
         } else {
-            return Mono
-                .fromCallable(() -> new ComposedMessageIdWithMetaData(
+            return modSeqProvider.nextModSeq(cassandraId)
+                .map(modSeq -> new ComposedMessageIdWithMetaData(
                     oldComposedId.getComposedMessageId(),
                     newFlags,
-                    modSeqProvider.nextModSeq(cassandraId)))
+                    modSeq))
             .flatMap(newComposedId -> updateFlags(oldComposedId, newComposedId));
         }
     }

mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapper.java

@@ -180,9 +180,12 @@ private Mono<Task.Result> fixCounters(Mailbox mailbox) {
     }
 
     private boolean shouldReadRepair(MailboxCounters counters) {
+        boolean activated = cassandraConfiguration.getMailboxCountersReadRepairChanceMax() != 0 || cassandraConfiguration.getMailboxCountersReadRepairChanceOneHundred() != 0;
         double ponderedReadRepairChance = cassandraConfiguration.getMailboxCountersReadRepairChanceOneHundred() * (100.0 / counters.getUnseen());
-        return secureRandom.nextFloat() < Math.min(
-            cassandraConfiguration.getMailboxCountersReadRepairChanceMax(), ponderedReadRepairChance);
+        return activated &&
+            secureRandom.nextFloat() < Math.min(
+                cassandraConfiguration.getMailboxCountersReadRepairChanceMax(),
+                ponderedReadRepairChance);
     }
 
     @Override

mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/eventsourcing/acl/AclV2DAOSubscriber.java

@@ -20,28 +20,30 @@
 package org.apache.james.mailbox.cassandra.mail.eventsourcing.acl;
 
 import org.apache.james.eventsourcing.Event;
-import org.apache.james.eventsourcing.Subscriber;
+import org.apache.james.eventsourcing.ReactiveSubscriber;
 import org.apache.james.mailbox.cassandra.mail.CassandraACLDAOV2;
 
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 
-public class AclV2DAOSubscriber implements Subscriber {
+public class AclV2DAOSubscriber implements ReactiveSubscriber {
     private final CassandraACLDAOV2 acldaov2;
 
     public AclV2DAOSubscriber(CassandraACLDAOV2 acldaov2) {
         this.acldaov2 = acldaov2;
     }
 
     @Override
-    public void handle(Event event) {
+    public Mono<Void> handleReactive(Event event) {
         if (event instanceof ACLUpdated) {
             ACLUpdated aclUpdated = (ACLUpdated) event;
 
-            Flux.fromStream(
+            return Flux.fromStream(
                 aclUpdated.getAclDiff()
                     .commands())
                 .flatMap(command -> acldaov2.updateACL(aclUpdated.mailboxId(), command))
-                .blockLast();
+                .then();
         }
+        return Mono.empty();
     }
 }

mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/eventsourcing/acl/UserRightsDAOSubscriber.java

@@ -20,22 +20,25 @@
 package org.apache.james.mailbox.cassandra.mail.eventsourcing.acl;
 
 import org.apache.james.eventsourcing.Event;
-import org.apache.james.eventsourcing.Subscriber;
+import org.apache.james.eventsourcing.ReactiveSubscriber;
 import org.apache.james.mailbox.cassandra.mail.CassandraUserMailboxRightsDAO;
+import org.reactivestreams.Publisher;
 
-public class UserRightsDAOSubscriber implements Subscriber {
+import reactor.core.publisher.Mono;
+
+public class UserRightsDAOSubscriber implements ReactiveSubscriber {
     private final CassandraUserMailboxRightsDAO userRightsDAO;
 
     public UserRightsDAOSubscriber(CassandraUserMailboxRightsDAO userRightsDAO) {
         this.userRightsDAO = userRightsDAO;
     }
 
     @Override
-    public void handle(Event event) {
+    public Publisher<Void> handleReactive(Event event) {
         if (event instanceof ACLUpdated) {
             ACLUpdated aclUpdated = (ACLUpdated) event;
-            userRightsDAO.update(aclUpdated.mailboxId(), aclUpdated.getAclDiff())
-                .block();
+            return userRightsDAO.update(aclUpdated.mailboxId(), aclUpdated.getAclDiff());
         }
+        return Mono.empty();
     }
 }