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/06/07 06:49:40 UTC
[james-project] 08/09: [REACTOR] Reactify MessageManager::delete
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 e6fa38a9e15cdc6efd224007655d3f722d0feda4
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Fri Jun 3 11:30:51 2022 +0700
[REACTOR] Reactify MessageManager::delete
THis enable dropping a little bunch of blocking schedulers...
---
.../org/apache/james/mailbox/MessageManager.java | 4 ++++
.../james/mailbox/store/StoreMessageManager.java | 6 ++++++
.../james/healthcheck/MailReceptionCheck.java | 24 ++++++++++------------
.../webadmin/service/ExpireMailboxService.java | 9 ++------
.../webadmin/service/UserMailboxesService.java | 19 +++++++----------
5 files changed, 30 insertions(+), 32 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 d9c3d722ff..ece9c03b75 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
@@ -144,6 +144,10 @@ public interface MessageManager {
*/
void delete(List<MessageUid> uids, MailboxSession mailboxSession) throws MailboxException;
+ default Mono<Void> deleteReactive(List<MessageUid> uids, MailboxSession mailboxSession) {
+ return Mono.fromRunnable(Throwing.runnable(() -> delete(uids, mailboxSession)));
+ }
+
/**
* Sets flags on messages within the given range. The new flags are returned
* for each message altered.
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 8032d279da..0218ae6460 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
@@ -289,6 +289,12 @@ public class StoreMessageManager implements MessageManager {
dispatchExpungeEvent(mailboxSession, deletedMessages).block();
}
+ @Override
+ public Mono<Void> deleteReactive(List<MessageUid> uids, MailboxSession mailboxSession) {
+ return deleteMessages(uids, mailboxSession)
+ .flatMap(deleteMessages -> dispatchExpungeEvent(mailboxSession, deleteMessages));
+ }
+
private Mono<Map<MessageUid, MessageMetaData>> deleteMessages(List<MessageUid> messageUids, MailboxSession session) {
if (messageUids.isEmpty()) {
return Mono.just(ImmutableMap.of());
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 646b7a1e88..c8f31d0844 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
@@ -21,6 +21,7 @@ package org.apache.james.healthcheck;
import static reactor.core.publisher.Sinks.EmitFailureHandler.FAIL_FAST;
+import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.util.Objects;
@@ -46,7 +47,6 @@ import org.apache.james.mailbox.MailboxSession;
import org.apache.james.mailbox.MessageManager;
import org.apache.james.mailbox.events.MailboxEvents.Added;
import org.apache.james.mailbox.events.MailboxIdRegistrationKey;
-import org.apache.james.mailbox.exception.MailboxException;
import org.apache.james.mailbox.exception.MailboxNotFoundException;
import org.apache.james.mailbox.model.FetchGroup;
import org.apache.james.mailbox.model.MailboxPath;
@@ -54,7 +54,6 @@ import org.apache.james.mailbox.model.MessageRange;
import org.apache.james.server.core.MailImpl;
import org.apache.james.user.api.UsersRepository;
import org.apache.james.util.DurationParser;
-import org.apache.james.util.ReactorUtils;
import org.apache.mailet.MailetContext;
import org.reactivestreams.Publisher;
import org.slf4j.Logger;
@@ -220,7 +219,6 @@ public class MailReceptionCheck implements HealthCheck {
registration -> sendMail(username)
.flatMap(content -> checkReceived(session, listener, mailbox, content)),
Registration::unregister))
- .subscribeOn(ReactorUtils.BLOCKING_CALL_WRAPPER)
.timeout(configuration.getTimeout(), Mono.error(() -> new RuntimeException("HealthCheck email was not received after " + configuration.getTimeout().toMillis() + "ms")))
.onErrorResume(e -> {
LOGGER.error("Mail reception check failed", e);
@@ -238,17 +236,16 @@ public class MailReceptionCheck implements HealthCheck {
private Mono<Result> checkReceived(MailboxSession session, AwaitReceptionListener listener, MessageManager mailbox, Content content) {
return listener.addedEvents()
.flatMapIterable(Added::getUids)
- .flatMap(uid -> Mono.fromCallable(() -> mailbox.getMessages(MessageRange.one(uid), FetchGroup.FULL_CONTENT, session)))
- .flatMapIterable(ImmutableList::copyOf)
- .filter(Throwing.predicate(messageResult -> IOUtils.toString(messageResult.getBody().getInputStream()).contains(content.asString())))
+ .flatMap(uid -> Flux.from(mailbox.getMessagesReactive(MessageRange.one(uid), FetchGroup.FULL_CONTENT, session)))
+ .filter(Throwing.predicate(messageResult -> IOUtils.toString(messageResult.getBody().getInputStream(), StandardCharsets.US_ASCII)
+ .contains(content.asString())))
// Cleanup our testing mail
- .doOnNext(messageResult -> {
- try {
- mailbox.delete(ImmutableList.of(messageResult.getUid()), session);
- } catch (MailboxException e) {
+ .concatMap(messageResult -> Mono.from(mailbox.deleteReactive(ImmutableList.of(messageResult.getUid()), session))
+ .onErrorResume(e -> {
LOGGER.warn("Failed to delete Health check testing email", e);
- }
- })
+ return Mono.empty();
+ })
+ .thenReturn(messageResult))
.map(any -> Result.healthy(componentName()))
.next();
}
@@ -270,6 +267,7 @@ public class MailReceptionCheck implements HealthCheck {
.build(),
mail -> Mono.fromRunnable(Throwing.runnable(() -> mailetContext.sendMail(mail))),
LifecycleUtil::dispose))
- .thenReturn(content);
+ .thenReturn(content)
+ .subscribeOn(Schedulers.boundedElastic());
}
}
diff --git a/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/ExpireMailboxService.java b/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/ExpireMailboxService.java
index 7692f1a295..dd07367540 100644
--- a/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/ExpireMailboxService.java
+++ b/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/ExpireMailboxService.java
@@ -56,7 +56,6 @@ import com.google.common.base.Preconditions;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
-import reactor.core.scheduler.Schedulers;
public class ExpireMailboxService {
@@ -237,12 +236,8 @@ public class ExpireMailboxService {
if (uids.isEmpty()) {
return Mono.just(0);
} else {
- return Mono.fromCallable(
- () -> {
- mgr.delete(uids, session);
- return uids.size();
- })
- .subscribeOn(Schedulers.elastic());
+ return Mono.from(mgr.deleteReactive(uids, session))
+ .thenReturn(uids.size());
}
}
}
diff --git a/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/UserMailboxesService.java b/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/UserMailboxesService.java
index 6de9223a09..a7bf0d7291 100644
--- a/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/UserMailboxesService.java
+++ b/server/protocols/webadmin/webadmin-mailbox/src/main/java/org/apache/james/webadmin/service/UserMailboxesService.java
@@ -42,7 +42,6 @@ import org.apache.james.task.Task;
import org.apache.james.task.Task.Result;
import org.apache.james.user.api.UsersRepository;
import org.apache.james.user.api.UsersRepositoryException;
-import org.apache.james.util.ReactorUtils;
import org.apache.james.webadmin.dto.MailboxResponse;
import org.apache.james.webadmin.utils.MailboxHaveChildrenException;
import org.apache.james.webadmin.validation.MailboxName;
@@ -123,17 +122,13 @@ public class UserMailboxesService {
}
private Mono<Result> deleteMessage(MessageManager messageManager, MessageUid messageUid, MailboxSession mailboxSession, ClearMailboxContentTask.Context context) {
- return Mono.fromCallable(() -> {
- try {
- messageManager.delete(List.of(messageUid), mailboxSession);
- context.incrementSuccesses();
- return Result.COMPLETED;
- } catch (MailboxException e) {
- context.incrementMessageFails();
- return Result.PARTIAL;
- }
- })
- .subscribeOn(ReactorUtils.BLOCKING_CALL_WRAPPER);
+ return Mono.from(messageManager.deleteReactive(List.of(messageUid), mailboxSession))
+ .thenReturn(Result.COMPLETED)
+ .doOnNext(next -> context.incrementSuccesses())
+ .onErrorResume(e -> {
+ context.incrementMessageFails();
+ return Mono.just(Result.PARTIAL);
+ });
}
public void deleteMailbox(Username username, MailboxName mailboxName) throws MailboxException, UsersRepositoryException, MailboxHaveChildrenException {
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org