[james-project] 12/15: JAMES-2979 MailDispatcher: Use immediate schedule to avoid allocating threads upon nested block calls

[bt...@apache.org]

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 <bt...@linagora.com>
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: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org