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 9de81f848d8b8d4e48adb4c0e9582bfb0bfd0041 Author: Benoit Tellier <[email protected]> AuthorDate: Fri May 7 18:31:12 2021 +0700 JAMES-2979 MailDispatcher: Use immediate schedule to avoid allocating threads upon nested block calls This test mimics the previous behaviour of LocalDelivery ``` @Test void testNestedBlocksWithElasticScheduler() { // mono1 corresponds to the append to the mailbox (blocking) Mono<Void> mono1 = Mono.fromRunnable(() -> { System.out.println("mono1 running on " + Thread.currentThread().getName()); }); // mono2 corresponds to the retries perforned in MailDispatcher Mono<Void> mono2 = Mono.fromRunnable(() -> { System.out.println("mono2 running on " + Thread.currentThread().getName()); mono1.subscribeOn(Schedulers.elastic()).block(); }); // This is a spooler thread running LocalDelivery System.out.println("Current thread " + Thread.currentThread().getName()); mono2.subscribeOn(Schedulers.elastic()).block(); } ``` Output: ``` Current thread main mono2 running on elastic-2 mono1 running on elastic-3 ``` One thread doing the work and two waiting... With the new paradigm (waiting for a fully reactive version) ``` @Test void testNestedBlockWithImmediateScheduler() { // mono1 corresponds to the append to the mailbox (blocking) Mono<Void> mono1 = Mono.fromRunnable(() -> { System.out.println("mono1 running on " + Thread.currentThread().getName()); }); // mono2 corresponds to the retries perforned in MailDispatcher Mono<Void> mono2 = Mono.fromRunnable(() -> { System.out.println("mono2 running on " + Thread.currentThread().getName()); mono1.subscribeOn(Schedulers.immediate()).block(); }); // This is a spooler thread running LocalDelivery System.out.println("Current thread " + Thread.currentThread().getName()); mono2.subscribeOn(Schedulers.immediate()).block(); } ``` We get... ``` Current thread main mono2 running on main mono1 running on main ``` Way better! We do mobilize only one thread that would anyway be blocked. --- .../org/apache/james/transport/mailets/delivery/MailDispatcher.java | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/server/mailet/mailets/src/main/java/org/apache/james/transport/mailets/delivery/MailDispatcher.java b/server/mailet/mailets/src/main/java/org/apache/james/transport/mailets/delivery/MailDispatcher.java index ec66410..2a7bf6a 100644 --- a/server/mailet/mailets/src/main/java/org/apache/james/transport/mailets/delivery/MailDispatcher.java +++ b/server/mailet/mailets/src/main/java/org/apache/james/transport/mailets/delivery/MailDispatcher.java @@ -43,7 +43,6 @@ import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; import reactor.core.publisher.Mono; -import reactor.core.scheduler.Scheduler; import reactor.core.scheduler.Schedulers; import reactor.util.retry.Retry; @@ -90,13 +89,11 @@ public class MailDispatcher { private final MailStore mailStore; private final boolean consume; private final MailetContext mailetContext; - private final Scheduler scheduler; private MailDispatcher(MailStore mailStore, boolean consume, MailetContext mailetContext) { this.mailStore = mailStore; this.consume = consume; this.mailetContext = mailetContext; - this.scheduler = Schedulers.elastic(); } public void dispatch(Mail mail) throws MessagingException { @@ -142,7 +139,7 @@ public class MailDispatcher { Map<String, List<String>> savedHeaders = saveHeaders(mail, recipient); addSpecificHeadersForRecipient(mail, message, recipient); - storeMailWithRetry(mail, recipient).block(); + storeMailWithRetry(mail, recipient).subscribeOn(Schedulers.immediate()).block(); restoreHeaders(mail.getMessage(), savedHeaders); } catch (Exception ex) { @@ -156,7 +153,6 @@ public class MailDispatcher { private Mono<Void> storeMailWithRetry(Mail mail, MailAddress recipient) { return Mono.fromRunnable((ThrowingRunnable)() -> mailStore.storeMail(recipient, mail)) .doOnError(error -> LOGGER.warn("Error While storing mail. This error will be retried.", error)) - .subscribeOn(scheduler) .retryWhen(Retry.backoff(RETRIES, FIRST_BACKOFF).maxBackoff(MAX_BACKOFF).scheduler(Schedulers.elastic())) .then(); } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
