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 47de17f568a7bbfb37be87674a6cbe0a6f7bc171 Author: Benoit Tellier <[email protected]> AuthorDate: Sat May 8 14:50:59 2021 +0700 [REFACTORING] RabbitMQ receivers can be blocking Detected thanks to https://github.com/reactor/BlockHound They are blocking in case of connection recovery. Doing that on the parallel pool is bad as it "steals" threads dedicated to non blocking tasks and slows the entire application down. --- .../src/main/java/org/apache/james/events/GroupRegistration.java | 1 + .../src/main/java/org/apache/james/events/KeyRegistrationHandler.java | 1 + 2 files changed, 2 insertions(+) diff --git a/event-bus/distributed/src/main/java/org/apache/james/events/GroupRegistration.java b/event-bus/distributed/src/main/java/org/apache/james/events/GroupRegistration.java index d0013ff..a33638b 100644 --- a/event-bus/distributed/src/main/java/org/apache/james/events/GroupRegistration.java +++ b/event-bus/distributed/src/main/java/org/apache/james/events/GroupRegistration.java @@ -133,6 +133,7 @@ class GroupRegistration implements Registration { .publishOn(Schedulers.parallel()) .filter(delivery -> Objects.nonNull(delivery.getBody())) .flatMap(this::deliver, EventBus.EXECUTION_RATE) + .subscribeOn(Schedulers.elastic()) .subscribe(); } diff --git a/event-bus/distributed/src/main/java/org/apache/james/events/KeyRegistrationHandler.java b/event-bus/distributed/src/main/java/org/apache/james/events/KeyRegistrationHandler.java index a6088be..d62f59f 100644 --- a/event-bus/distributed/src/main/java/org/apache/james/events/KeyRegistrationHandler.java +++ b/event-bus/distributed/src/main/java/org/apache/james/events/KeyRegistrationHandler.java @@ -96,6 +96,7 @@ class KeyRegistrationHandler { 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()); } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
