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/05/26 08:41:23 UTC
[james-project] branch master updated: JAMES-3737 Reactive IMAP CREATE (#1011)
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
The following commit(s) were added to refs/heads/master by this push:
new e031f46444 JAMES-3737 Reactive IMAP CREATE (#1011)
e031f46444 is described below
commit e031f46444c8b1affc79be7e745ad9738b3d5d3e
Author: Benoit TELLIER <bt...@linagora.com>
AuthorDate: Thu May 26 15:41:18 2022 +0700
JAMES-3737 Reactive IMAP CREATE (#1011)
---
.../org/apache/james/mailbox/MailboxManager.java | 5 +
.../james/mailbox/store/StoreMailboxManager.java | 116 +++++++++++----------
.../james/imap/processor/CreateProcessor.java | 45 ++++----
.../james/healthcheck/MailReceptionCheck.java | 2 +-
.../mailets/delivery/MailboxAppenderImpl.java | 5 +-
.../jmap/http/DefaultMailboxesProvisioner.java | 26 ++---
.../jmap/http/DefaultMailboxesProvisionerTest.java | 2 +-
.../DefaultMailboxesProvisionerThreadTest.java | 1 +
.../james/jmap/http/MailboxesProvisioner.scala | 28 ++---
.../webadmin/service/CreateMissingParentsTask.java | 17 ++-
.../CreateMissingParentsRequestToTaskTest.java | 7 +-
11 files changed, 124 insertions(+), 130 deletions(-)
diff --git a/mailbox/api/src/main/java/org/apache/james/mailbox/MailboxManager.java b/mailbox/api/src/main/java/org/apache/james/mailbox/MailboxManager.java
index 1dfa2efec8..3b6acbe0d2 100644
--- a/mailbox/api/src/main/java/org/apache/james/mailbox/MailboxManager.java
+++ b/mailbox/api/src/main/java/org/apache/james/mailbox/MailboxManager.java
@@ -161,6 +161,11 @@ public interface MailboxManager extends RequestAware, RightManager, MailboxAnnot
*/
Optional<MailboxId> createMailbox(MailboxPath mailboxPath, MailboxSession mailboxSession) throws MailboxException;
+ default Publisher<MailboxId> createMailboxReactive(MailboxPath mailboxPath, MailboxSession mailboxSession) {
+ return Mono.fromCallable(() -> createMailbox(mailboxPath, mailboxSession))
+ .handle((optional, sink) -> optional.ifPresent(sink::next));
+ }
+
/**
* Delete the mailbox with the name
*/
diff --git a/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreMailboxManager.java b/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreMailboxManager.java
index c3385d53af..3ef21f12e3 100644
--- a/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreMailboxManager.java
+++ b/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreMailboxManager.java
@@ -24,14 +24,12 @@ import static org.apache.james.mailbox.store.MailboxReactorUtils.blockOptional;
import static org.apache.james.mailbox.store.mail.AbstractMessageMapper.UNLIMITED;
import java.time.Duration;
-import java.util.ArrayList;
import java.util.EnumSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.function.Function;
-import java.util.stream.Stream;
import javax.inject.Inject;
@@ -52,6 +50,7 @@ import org.apache.james.mailbox.exception.InboxAlreadyCreated;
import org.apache.james.mailbox.exception.InsufficientRightsException;
import org.apache.james.mailbox.exception.MailboxException;
import org.apache.james.mailbox.exception.MailboxExistsException;
+import org.apache.james.mailbox.exception.MailboxNameException;
import org.apache.james.mailbox.exception.MailboxNotFoundException;
import org.apache.james.mailbox.exception.SubscriptionException;
import org.apache.james.mailbox.exception.UnsupportedRightException;
@@ -98,7 +97,6 @@ import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
-import com.google.common.collect.Iterables;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
@@ -334,94 +332,98 @@ public class StoreMailboxManager implements MailboxManager {
@Override
public Optional<MailboxId> createMailbox(MailboxPath mailboxPath, MailboxSession mailboxSession) throws MailboxException {
+ return MailboxReactorUtils.blockOptional(createMailboxReactive(mailboxPath, mailboxSession));
+ }
+
+ public Mono<MailboxId> createMailboxReactive(MailboxPath mailboxPath, MailboxSession mailboxSession) {
LOGGER.debug("createMailbox {}", mailboxPath);
- assertMailboxPathBelongToUser(mailboxSession, mailboxPath);
+ return assertMailboxPathBelongToUserReactive(mailboxSession, mailboxPath)
+ .then(doCreateMailboxReactive(mailboxPath, mailboxSession));
+ }
+ private Mono<MailboxId> doCreateMailboxReactive(MailboxPath mailboxPath, MailboxSession mailboxSession) {
if (mailboxPath.getName().isEmpty()) {
LOGGER.warn("Ignoring mailbox with empty name");
+ return Mono.empty();
} else {
- MailboxPath sanitizedMailboxPath = mailboxPath.sanitize(mailboxSession.getPathDelimiter());
- sanitizedMailboxPath.assertAcceptable(mailboxSession.getPathDelimiter());
-
- if (block(mailboxExists(sanitizedMailboxPath, mailboxSession))) {
- throw new MailboxExistsException(sanitizedMailboxPath.asString());
- }
-
- List<MailboxId> mailboxIds = createMailboxesForPath(mailboxSession, sanitizedMailboxPath);
-
- if (!mailboxIds.isEmpty()) {
- return Optional.ofNullable(Iterables.getLast(mailboxIds));
+ try {
+ MailboxPath sanitizedMailboxPath = mailboxPath.sanitize(mailboxSession.getPathDelimiter());
+ sanitizedMailboxPath.assertAcceptable(mailboxSession.getPathDelimiter());
+
+ return mailboxExists(sanitizedMailboxPath, mailboxSession)
+ .flatMap(exists -> {
+ if (exists) {
+ return Mono.error(new MailboxExistsException(sanitizedMailboxPath.asString()));
+ } else {
+ return createMailboxesForPath(mailboxSession, sanitizedMailboxPath).takeLast(1).next();
+ }
+ });
+ } catch (MailboxNameException e) {
+ return Mono.error(e);
}
}
- return Optional.empty();
}
- private List<MailboxId> createMailboxesForPath(MailboxSession mailboxSession, MailboxPath sanitizedMailboxPath) {
+ private Flux<MailboxId> createMailboxesForPath(MailboxSession mailboxSession, MailboxPath sanitizedMailboxPath) {
// Create parents first
// If any creation fails then the mailbox will not be created
// TODO: transaction
List<MailboxPath> intermediatePaths = sanitizedMailboxPath.getHierarchyLevels(getDelimiter());
boolean isRootPath = intermediatePaths.size() == 1;
- return intermediatePaths
- .stream()
- .flatMap(Throwing.<MailboxPath, Stream<MailboxId>>function(mailboxPath -> manageMailboxCreation(mailboxSession, isRootPath, mailboxPath)).sneakyThrow())
- .collect(ImmutableList.toImmutableList());
+ return Flux.fromIterable(intermediatePaths)
+ .concatMap(path -> manageMailboxCreation(mailboxSession, isRootPath, path));
}
- private Stream<MailboxId> manageMailboxCreation(MailboxSession mailboxSession, boolean isRootPath, MailboxPath mailboxPath) throws MailboxException {
+ private Mono<MailboxId> manageMailboxCreation(MailboxSession mailboxSession, boolean isRootPath, MailboxPath mailboxPath) {
if (mailboxPath.isInbox()) {
- if (block(hasInbox(mailboxSession))) {
- return duplicatedINBOXCreation(isRootPath, mailboxPath);
- }
-
- return performConcurrentMailboxCreation(mailboxSession, MailboxPath.inbox(mailboxSession)).stream();
+ return Mono.from(hasInbox(mailboxSession))
+ .flatMap(hasInbox -> {
+ if (hasInbox) {
+ return duplicatedINBOXCreation(isRootPath, mailboxPath);
+ }
+ return performConcurrentMailboxCreation(mailboxSession, MailboxPath.inbox(mailboxSession));
+ });
}
- return performConcurrentMailboxCreation(mailboxSession, mailboxPath).stream();
+ return performConcurrentMailboxCreation(mailboxSession, mailboxPath);
}
- private Stream<MailboxId> duplicatedINBOXCreation(boolean isRootPath, MailboxPath mailbox) throws InboxAlreadyCreated {
+ private Mono<MailboxId> duplicatedINBOXCreation(boolean isRootPath, MailboxPath mailbox) {
if (isRootPath) {
- throw new InboxAlreadyCreated(mailbox.getName());
+ return Mono.error(new InboxAlreadyCreated(mailbox.getName()));
}
- return Stream.empty();
+ return Mono.empty();
}
- private List<MailboxId> performConcurrentMailboxCreation(MailboxSession mailboxSession, MailboxPath mailboxPath) throws MailboxException {
- List<MailboxId> mailboxIds = new ArrayList<>();
+ private Mono<MailboxId> performConcurrentMailboxCreation(MailboxSession mailboxSession, MailboxPath mailboxPath) {
MailboxMapper mapper = mailboxSessionMapperFactory.getMailboxMapper(mailboxSession);
- locker.executeWithLock(mailboxPath, () ->
- block(mapper.executeReactive(mapper.create(mailboxPath, UidValidity.generate())
- .doOnNext(mailbox -> mailboxIds.add(mailbox.getMailboxId()))
- .flatMap(mailbox ->
- // notify listeners
- eventBus.dispatch(EventFactory.mailboxAdded()
- .randomEventId()
- .mailboxSession(mailboxSession)
- .mailbox(mailbox)
- .build(),
- new MailboxIdRegistrationKey(mailbox.getMailboxId()))))
- .onErrorResume(e -> {
- if (e instanceof MailboxExistsException) {
- LOGGER.info("{} mailbox was created concurrently", mailboxPath.asString());
- } else if (e instanceof MailboxException) {
- return Mono.error(e);
- }
- return Mono.empty();
- })), MailboxPathLocker.LockType.Write);
-
- return mailboxIds;
+ return Mono.from(locker.executeReactiveWithLockReactive(mailboxPath,
+ mapper.executeReactive(mapper.create(mailboxPath, UidValidity.generate())
+ .flatMap(mailbox ->
+ // notify listeners
+ eventBus.dispatch(EventFactory.mailboxAdded()
+ .randomEventId()
+ .mailboxSession(mailboxSession)
+ .mailbox(mailbox)
+ .build(),
+ new MailboxIdRegistrationKey(mailbox.getMailboxId()))
+ .thenReturn(mailbox.getMailboxId()))
+ .onErrorResume(MailboxExistsException.class, e -> {
+ LOGGER.info("{} mailbox was created concurrently", mailboxPath.asString());
+ return Mono.empty();
+ })), MailboxPathLocker.LockType.Write));
}
- private void assertMailboxPathBelongToUser(MailboxSession mailboxSession, MailboxPath mailboxPath) throws MailboxException {
+ private Mono<Void> assertMailboxPathBelongToUserReactive(MailboxSession mailboxSession, MailboxPath mailboxPath) {
if (!mailboxPath.belongsTo(mailboxSession)) {
- throw new InsufficientRightsException("mailboxPath '" + mailboxPath.asString() + "'"
- + " does not belong to user '" + mailboxSession.getUser().asString() + "'");
+ return Mono.error(new InsufficientRightsException("mailboxPath '" + mailboxPath.asString() + "'"
+ + " does not belong to user '" + mailboxSession.getUser().asString() + "'"));
}
+ return Mono.empty();
}
@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 e3345edee6..8c85c83117 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
@@ -19,6 +19,8 @@
package org.apache.james.imap.processor;
+import static org.apache.james.util.ReactorUtils.logOnError;
+
import org.apache.james.imap.api.display.HumanReadableText;
import org.apache.james.imap.api.message.response.StatusResponseFactory;
import org.apache.james.imap.api.process.ImapSession;
@@ -28,12 +30,13 @@ import org.apache.james.mailbox.MailboxManager;
import org.apache.james.mailbox.exception.MailboxException;
import org.apache.james.mailbox.exception.MailboxExistsException;
import org.apache.james.mailbox.exception.TooLongMailboxNameException;
-import org.apache.james.mailbox.model.MailboxPath;
import org.apache.james.metrics.api.MetricFactory;
import org.apache.james.util.MDCBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import reactor.core.publisher.Mono;
+
public class CreateProcessor extends AbstractMailboxProcessor<CreateRequest> {
private static final Logger LOGGER = LoggerFactory.getLogger(CreateProcessor.class);
@@ -43,23 +46,29 @@ public class CreateProcessor extends AbstractMailboxProcessor<CreateRequest> {
}
@Override
- protected void processRequest(CreateRequest request, ImapSession session, Responder responder) {
- final MailboxPath mailboxPath = PathConverter.forSession(session).buildFullPath(request.getMailboxName());
- try {
- final MailboxManager mailboxManager = getMailboxManager();
- mailboxManager.createMailbox(mailboxPath, session.getMailboxSession());
- unsolicitedResponses(session, responder, false).block();
- okComplete(request, responder);
- } catch (MailboxExistsException e) {
- LOGGER.debug("Create failed for mailbox {} as it already exists", mailboxPath, e);
- no(request, responder, HumanReadableText.MAILBOX_EXISTS);
- } catch (TooLongMailboxNameException e) {
- LOGGER.debug("The mailbox name length is over limit: {}", mailboxPath.getName(), e);
- taggedBad(request, responder, HumanReadableText.FAILURE_MAILBOX_NAME);
- } catch (MailboxException e) {
- LOGGER.error("Create failed for mailbox {}", mailboxPath, e);
- no(request, responder, HumanReadableText.GENERIC_FAILURE_DURING_PROCESSING);
- }
+ protected Mono<Void> processRequestReactive(CreateRequest request, ImapSession session, Responder responder) {
+ MailboxManager mailboxManager = getMailboxManager();
+
+ return Mono.fromCallable(() -> PathConverter.forSession(session).buildFullPath(request.getMailboxName()))
+ .flatMap(mailboxPath -> Mono.from(mailboxManager.createMailboxReactive(mailboxPath, session.getMailboxSession()))
+ .then(unsolicitedResponses(session, responder, false))
+ .then(Mono.fromRunnable(() -> okComplete(request, responder)))
+ .doOnEach(logOnError(MailboxExistsException.class, e -> LOGGER.debug("Create failed for mailbox {} as it already exists", mailboxPath, e)))
+ .onErrorResume(MailboxExistsException.class, e -> {
+ no(request, responder, HumanReadableText.MAILBOX_EXISTS);
+ return Mono.empty();
+ })
+ .doOnEach(logOnError(TooLongMailboxNameException.class, e -> LOGGER.debug("The mailbox name length is over limit: {}", mailboxPath.getName(), e)))
+ .onErrorResume(TooLongMailboxNameException.class, e -> {
+ taggedBad(request, responder, HumanReadableText.FAILURE_MAILBOX_NAME);
+ return Mono.empty();
+ })
+ .doOnEach(logOnError(MailboxException.class, e -> LOGGER.error("Create failed for mailbox {}", mailboxPath, e)))
+ .onErrorResume(TooLongMailboxNameException.class, e -> {
+ no(request, responder, HumanReadableText.GENERIC_FAILURE_DURING_PROCESSING);
+ return Mono.empty();
+ })
+ .then());
}
@Override
diff --git a/server/container/feature-checks/src/main/java/org/apache/james/healthcheck/MailReceptionCheck.java b/server/container/feature-checks/src/main/java/org/apache/james/healthcheck/MailReceptionCheck.java
index 054b16a559..518f4c6690 100644
--- a/server/container/feature-checks/src/main/java/org/apache/james/healthcheck/MailReceptionCheck.java
+++ b/server/container/feature-checks/src/main/java/org/apache/james/healthcheck/MailReceptionCheck.java
@@ -230,7 +230,7 @@ public class MailReceptionCheck implements HealthCheck {
private Mono<MessageManager> retrieveInbox(Username username, MailboxSession session) {
MailboxPath mailboxPath = MailboxPath.inbox(username);
return Mono.from(mailboxManager.getMailboxReactive(mailboxPath, session))
- .onErrorResume(MailboxNotFoundException.class, e -> Mono.fromCallable(() -> mailboxManager.createMailbox(mailboxPath, session))
+ .onErrorResume(MailboxNotFoundException.class, e -> Mono.from(mailboxManager.createMailboxReactive(mailboxPath, session))
.then(Mono.from(mailboxManager.getMailboxReactive(mailboxPath, session))));
}
diff --git a/server/mailet/mailets/src/main/java/org/apache/james/transport/mailets/delivery/MailboxAppenderImpl.java b/server/mailet/mailets/src/main/java/org/apache/james/transport/mailets/delivery/MailboxAppenderImpl.java
index 326ba0a212..0f38b8ae4a 100644
--- a/server/mailet/mailets/src/main/java/org/apache/james/transport/mailets/delivery/MailboxAppenderImpl.java
+++ b/server/mailet/mailets/src/main/java/org/apache/james/transport/mailets/delivery/MailboxAppenderImpl.java
@@ -41,11 +41,9 @@ import org.apache.james.server.core.MimeMessageUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.github.fge.lambdas.Throwing;
import com.google.common.base.Strings;
import reactor.core.publisher.Mono;
-import reactor.core.scheduler.Schedulers;
public class MailboxAppenderImpl implements MailboxAppender {
private static final Logger LOGGER = LoggerFactory.getLogger(MailboxAppenderImpl.class);
@@ -119,8 +117,7 @@ public class MailboxAppenderImpl implements MailboxAppender {
private Mono<MessageManager> createMailboxIfNotExist(MailboxSession session, MailboxPath path) {
return Mono.from(mailboxManager.getMailboxReactive(path, session))
.onErrorResume(MailboxNotFoundException.class, e ->
- Mono.fromRunnable(Throwing.runnable(() -> mailboxManager.createMailbox(path, session)).sneakyThrow())
- .subscribeOn(Schedulers.elastic())
+ Mono.from(mailboxManager.createMailboxReactive(path, session))
.then(Mono.from(mailboxManager.getMailboxReactive(path, session)))
.onErrorResume(MailboxExistsException.class, e2 -> {
LOGGER.info("Mailbox {} have been created concurrently", path);
diff --git a/server/protocols/jmap-draft/src/main/java/org/apache/james/jmap/http/DefaultMailboxesProvisioner.java b/server/protocols/jmap-draft/src/main/java/org/apache/james/jmap/http/DefaultMailboxesProvisioner.java
index ebf4a2f9ed..b42091b7d5 100644
--- a/server/protocols/jmap-draft/src/main/java/org/apache/james/jmap/http/DefaultMailboxesProvisioner.java
+++ b/server/protocols/jmap-draft/src/main/java/org/apache/james/jmap/http/DefaultMailboxesProvisioner.java
@@ -21,7 +21,6 @@ package org.apache.james.jmap.http;
import static org.apache.james.util.FunctionalUtils.negate;
import static org.apache.james.util.ReactorUtils.DEFAULT_CONCURRENCY;
-import java.util.Optional;
import java.util.function.Function;
import javax.inject.Inject;
@@ -30,9 +29,7 @@ import org.apache.james.mailbox.DefaultMailboxes;
import org.apache.james.mailbox.MailboxManager;
import org.apache.james.mailbox.MailboxSession;
import org.apache.james.mailbox.SubscriptionManager;
-import org.apache.james.mailbox.exception.MailboxException;
import org.apache.james.mailbox.exception.MailboxExistsException;
-import org.apache.james.mailbox.model.MailboxId;
import org.apache.james.mailbox.model.MailboxPath;
import org.apache.james.metrics.api.MetricFactory;
import org.slf4j.Logger;
@@ -42,7 +39,6 @@ import com.google.common.annotations.VisibleForTesting;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
-import reactor.core.scheduler.Schedulers;
public class DefaultMailboxesProvisioner {
private static final Logger LOGGER = LoggerFactory.getLogger(DefaultMailboxesProvisioner.class);
@@ -69,8 +65,7 @@ public class DefaultMailboxesProvisioner {
return Flux.fromIterable(DefaultMailboxes.DEFAULT_MAILBOXES)
.map(toMailboxPath(session))
.filterWhen(mailboxPath -> mailboxDoesntExist(mailboxPath, session), DEFAULT_CONCURRENCY)
- .concatMap(mailboxPath -> Mono.fromRunnable(() -> createMailbox(mailboxPath, session))
- .subscribeOn(Schedulers.elastic()))
+ .concatMap(mailboxPath -> createMailbox(mailboxPath, session))
.then();
}
@@ -83,17 +78,12 @@ public class DefaultMailboxesProvisioner {
return mailbox -> MailboxPath.forUser(session.getUser(), mailbox);
}
- private void createMailbox(MailboxPath mailboxPath, MailboxSession session) {
- try {
- Optional<MailboxId> mailboxId = mailboxManager.createMailbox(mailboxPath, session);
- if (mailboxId.isPresent()) {
- subscriptionManager.subscribe(session, mailboxPath.getName());
- }
- LOGGER.info("Provisioning {}. {} created.", mailboxPath, mailboxId);
- } catch (MailboxExistsException e) {
- LOGGER.info("Mailbox {} have been created concurrently", mailboxPath);
- } catch (MailboxException e) {
- throw new RuntimeException(e);
- }
+ private Mono<Void> createMailbox(MailboxPath mailboxPath, MailboxSession session) {
+ return Mono.from(mailboxManager.createMailboxReactive(mailboxPath, session))
+ .flatMap(id -> Mono.from(subscriptionManager.subscribeReactive(mailboxPath.getName(), session)))
+ .onErrorResume(MailboxExistsException.class, e -> {
+ LOGGER.info("Mailbox {} have been created concurrently", mailboxPath);
+ return Mono.empty();
+ });
}
}
diff --git a/server/protocols/jmap-draft/src/test/java/org/apache/james/jmap/http/DefaultMailboxesProvisionerTest.java b/server/protocols/jmap-draft/src/test/java/org/apache/james/jmap/http/DefaultMailboxesProvisionerTest.java
index 159cbe8c20..6ea328e49a 100644
--- a/server/protocols/jmap-draft/src/test/java/org/apache/james/jmap/http/DefaultMailboxesProvisionerTest.java
+++ b/server/protocols/jmap-draft/src/test/java/org/apache/james/jmap/http/DefaultMailboxesProvisionerTest.java
@@ -90,7 +90,7 @@ public class DefaultMailboxesProvisionerTest {
@Test
public void createMailboxesIfNeededShouldNotGenerateExceptionsInConcurrentEnvironment() throws Exception {
ConcurrentTestRunner.builder()
- .operation((threadNumber, step) -> testee.createMailboxesIfNeeded(session).block())
+ .reactorOperation((threadNumber, step) -> testee.createMailboxesIfNeeded(session))
.threadCount(10)
.runSuccessfullyWithin(Duration.ofSeconds(10));
diff --git a/server/protocols/jmap-draft/src/test/java/org/apache/james/jmap/http/DefaultMailboxesProvisionerThreadTest.java b/server/protocols/jmap-draft/src/test/java/org/apache/james/jmap/http/DefaultMailboxesProvisionerThreadTest.java
index 5b04f7bc36..85f93449fe 100644
--- a/server/protocols/jmap-draft/src/test/java/org/apache/james/jmap/http/DefaultMailboxesProvisionerThreadTest.java
+++ b/server/protocols/jmap-draft/src/test/java/org/apache/james/jmap/http/DefaultMailboxesProvisionerThreadTest.java
@@ -55,6 +55,7 @@ public class DefaultMailboxesProvisionerThreadTest {
public void before() {
session = MailboxSessionUtil.create(USERNAME);
mailboxManager = mock(MailboxManager.class);
+ when(mailboxManager.createMailboxReactive(any(), any())).thenReturn(Mono.empty());
subscriptionManager = mock(SubscriptionManager.class);
testee = new DefaultMailboxesProvisioner(mailboxManager, subscriptionManager, new RecordingMetricFactory());
}
diff --git a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/http/MailboxesProvisioner.scala b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/http/MailboxesProvisioner.scala
index 391d47e6b5..3fddd83217 100644
--- a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/http/MailboxesProvisioner.scala
+++ b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/http/MailboxesProvisioner.scala
@@ -19,17 +19,14 @@
package org.apache.james.jmap.http
-import java.util.Optional
-
import javax.inject.Inject
import org.apache.james.core.Username
import org.apache.james.mailbox.exception.{MailboxException, MailboxExistsException}
-import org.apache.james.mailbox.model.{MailboxId, MailboxPath}
+import org.apache.james.mailbox.model.MailboxPath
import org.apache.james.mailbox.{DefaultMailboxes, MailboxManager, MailboxSession, SubscriptionManager}
import org.apache.james.metrics.api.MetricFactory
import org.slf4j.{Logger, LoggerFactory}
import reactor.core.scala.publisher.{SFlux, SMono}
-import reactor.core.scheduler.Schedulers
import scala.jdk.CollectionConverters._
@@ -48,9 +45,8 @@ class MailboxesProvisioner @Inject() (mailboxManager: MailboxManager,
SFlux.fromIterable(DefaultMailboxes.DEFAULT_MAILBOXES.asScala)
.map(toMailboxPath(session))
- .filterWhen((mailboxPath: MailboxPath) => mailboxDoesntExist(mailboxPath, session))
- .concatMap((mailboxPath: MailboxPath) => SMono.fromCallable(() => createMailbox(mailboxPath, session))
- .subscribeOn(Schedulers.elastic))
+ .filterWhen(mailboxPath => mailboxDoesntExist(mailboxPath, session))
+ .concatMap(mailboxPath => createMailbox(mailboxPath, session))
.`then`
}
@@ -66,16 +62,14 @@ class MailboxesProvisioner @Inject() (mailboxManager: MailboxManager,
private def toMailboxPath(session: MailboxSession): String => MailboxPath =
(mailbox: String) => MailboxPath.forUser(session.getUser, mailbox)
- private def createMailbox(mailboxPath: MailboxPath, session: MailboxSession): Unit = {
- try {
- val mailboxId: Optional[MailboxId] = mailboxManager.createMailbox(mailboxPath, session)
- if (mailboxId.isPresent) {
- subscriptionManager.subscribe(session, mailboxPath.getName)
+ private def createMailbox(mailboxPath: MailboxPath, session: MailboxSession): SMono[Unit] = {
+ SMono(mailboxManager.createMailboxReactive(mailboxPath, session))
+ .flatMap(id => SMono(subscriptionManager.subscribeReactive(mailboxPath.getName, session)))
+ .onErrorResume {
+ case _: MailboxExistsException => LOGGER.info("Mailbox {} have been created concurrently", mailboxPath)
+ SMono.empty
+ case e => SMono.error(e)
}
- LOGGER.info("Provisioning {}. {} created.", mailboxPath, mailboxId)
- } catch {
- case e: MailboxExistsException => LOGGER.info("Mailbox {} have been created concurrently", mailboxPath)
- case e: MailboxException => throw new RuntimeException(e)
- }
+ .`then`()
}
}
diff --git a/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/CreateMissingParentsTask.java b/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/CreateMissingParentsTask.java
index 5d8981d728..d8da1c6eba 100644
--- a/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/CreateMissingParentsTask.java
+++ b/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/CreateMissingParentsTask.java
@@ -44,7 +44,6 @@ import org.apache.james.task.TaskExecutionDetails;
import org.apache.james.task.TaskType;
import com.fasterxml.jackson.annotation.JsonProperty;
-import com.github.fge.lambdas.Throwing;
import com.google.common.collect.ImmutableSet;
import reactor.core.publisher.Flux;
@@ -158,11 +157,9 @@ public class CreateMissingParentsTask implements Task {
}
private Mono<Result> createMailbox(MailboxPath path) {
- return Mono.fromRunnable(() -> {
- MailboxSession ownerSession = mailboxManager.createSystemSession(path.getUser());
- Optional<MailboxId> mailboxId = Throwing.supplier(() -> mailboxManager.createMailbox(path, ownerSession)).sneakyThrow().get();
- recordSuccess(mailboxId);
- })
+ MailboxSession ownerSession = mailboxManager.createSystemSession(path.getUser());
+ return Mono.from(mailboxManager.createMailboxReactive(path, ownerSession))
+ .doOnNext(this::recordSuccess)
.then(Mono.just(Result.COMPLETED))
.onErrorResume(e -> {
LOGGER.error("Error creating missing parent mailbox: {}", path.getName(), e);
@@ -185,11 +182,9 @@ public class CreateMissingParentsTask implements Task {
totalFailure.get()));
}
- private void recordSuccess(Optional<MailboxId> mailboxId) {
- if (mailboxId.isPresent()) {
- created.add(mailboxId.get());
- totalCreated.incrementAndGet();
- }
+ private void recordSuccess(MailboxId mailboxId) {
+ created.add(mailboxId);
+ totalCreated.incrementAndGet();
}
private void recordFailure(MailboxPath mailboxPath) {
diff --git a/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/service/CreateMissingParentsRequestToTaskTest.java b/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/service/CreateMissingParentsRequestToTaskTest.java
index 1657b2cd3e..d4ef57235f 100644
--- a/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/service/CreateMissingParentsRequestToTaskTest.java
+++ b/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/service/CreateMissingParentsRequestToTaskTest.java
@@ -27,7 +27,7 @@ import static org.hamcrest.Matchers.hasSize;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.notNullValue;
import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.doReturn;
import org.apache.james.core.Username;
import org.apache.james.json.DTOConverter;
@@ -52,6 +52,7 @@ import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import io.restassured.RestAssured;
+import reactor.core.publisher.Mono;
import spark.Service;
class CreateMissingParentsRequestToTaskTest {
@@ -248,9 +249,9 @@ class CreateMissingParentsRequestToTaskTest {
mailboxManager.createMailbox(path, session);
mailboxManager.deleteMailbox(MailboxPath.forUser(USERNAME, "INBOX"), session);
- doThrow(new RuntimeException())
+ doReturn(Mono.error(new RuntimeException()))
.when(mailboxManager)
- .createMailbox(any(), any());
+ .createMailboxReactive(any(), any());
String taskId = with()
.queryParam("action", "createMissingParents")
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org