You are viewing a plain text version of this content. The canonical link for it is here.
Posted to server-dev@james.apache.org by ro...@apache.org on 2019/02/15 14:54:30 UTC
[james-project] 01/03: MAILBOX-374 Enforce bounded parallelism for
the EventBus
This is an automated email from the ASF dual-hosted git repository.
rouazana pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit c6d1945974b86880c5940b4a452b6a19cffc92b0
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Fri Feb 15 15:46:29 2019 +0700
MAILBOX-374 Enforce bounded parallelism for the EventBus
---
.../org/apache/james/mailbox/events/EventBus.java | 2 ++
.../apache/james/mailbox/events/GroupContract.java | 36 ++++++++++++++++++++++
.../apache/james/mailbox/events/KeyContract.java | 27 ++++++++++++++++
.../james/mailbox/events/GroupRegistration.java | 3 +-
.../mailbox/events/KeyRegistrationHandler.java | 3 +-
5 files changed, 69 insertions(+), 2 deletions(-)
diff --git a/mailbox/api/src/main/java/org/apache/james/mailbox/events/EventBus.java b/mailbox/api/src/main/java/org/apache/james/mailbox/events/EventBus.java
index 0b319f7..896c6a3 100644
--- a/mailbox/api/src/main/java/org/apache/james/mailbox/events/EventBus.java
+++ b/mailbox/api/src/main/java/org/apache/james/mailbox/events/EventBus.java
@@ -26,6 +26,8 @@ import com.google.common.collect.ImmutableSet;
import reactor.core.publisher.Mono;
public interface EventBus {
+ int EXECUTION_RATE = 10;
+
interface StructuredLoggingFields {
String EVENT_ID = "eventId";
String EVENT_CLASS = "eventClass";
diff --git a/mailbox/api/src/test/java/org/apache/james/mailbox/events/GroupContract.java b/mailbox/api/src/test/java/org/apache/james/mailbox/events/GroupContract.java
index 491fa88..1ff08f2 100644
--- a/mailbox/api/src/test/java/org/apache/james/mailbox/events/GroupContract.java
+++ b/mailbox/api/src/test/java/org/apache/james/mailbox/events/GroupContract.java
@@ -40,7 +40,10 @@ import static org.mockito.Mockito.timeout;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import java.time.Duration;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.IntStream;
import org.apache.james.core.User;
import org.apache.james.mailbox.MailboxSession;
@@ -54,6 +57,39 @@ import com.google.common.collect.ImmutableSortedMap;
public interface GroupContract {
interface SingleEventBusGroupContract extends EventBusContract {
+
+ @Test
+ default void groupDeliveryShouldNotExceedRate() {
+ int eventCount = 50;
+ AtomicInteger nbCalls = new AtomicInteger(0);
+ AtomicInteger finishedExecutions = new AtomicInteger(0);
+ AtomicBoolean rateExceeded = new AtomicBoolean(false);
+
+ eventBus().register(new MailboxListener.GroupMailboxListener() {
+ @Override
+ public Group getDefaultGroup() {
+ return new GenericGroup("group");
+ }
+
+ @Override
+ public void event(Event event) throws Exception {
+ if (nbCalls.get() - finishedExecutions.get() > EventBus.EXECUTION_RATE) {
+ rateExceeded.set(true);
+ }
+ nbCalls.incrementAndGet();
+ Thread.sleep(Duration.ofMillis(200).toMillis());
+ finishedExecutions.incrementAndGet();
+
+ }
+ }, GROUP_A);
+
+ IntStream.range(0, eventCount)
+ .forEach(i -> eventBus().dispatch(EVENT, NO_KEYS).block());
+
+ WAIT_CONDITION.atMost(com.jayway.awaitility.Duration.TEN_MINUTES).until(() -> finishedExecutions.get() == eventCount);
+ assertThat(rateExceeded).isFalse();
+ }
+
@Test
default void listenersShouldBeAbleToDispatch() {
AtomicBoolean successfulRetry = new AtomicBoolean(false);
diff --git a/mailbox/api/src/test/java/org/apache/james/mailbox/events/KeyContract.java b/mailbox/api/src/test/java/org/apache/james/mailbox/events/KeyContract.java
index b457e6d..e007eff 100644
--- a/mailbox/api/src/test/java/org/apache/james/mailbox/events/KeyContract.java
+++ b/mailbox/api/src/test/java/org/apache/james/mailbox/events/KeyContract.java
@@ -43,6 +43,9 @@ import static org.mockito.Mockito.when;
import java.time.Duration;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.IntStream;
import org.apache.james.core.User;
import org.apache.james.mailbox.MailboxSession;
@@ -57,6 +60,30 @@ public interface KeyContract extends EventBusContract {
interface SingleEventBusKeyContract extends EventBusContract {
@Test
+ default void notificationShouldNotExceedRate() {
+ int eventCount = 50;
+ AtomicInteger nbCalls = new AtomicInteger(0);
+ AtomicInteger finishedExecutions = new AtomicInteger(0);
+ AtomicBoolean rateExceeded = new AtomicBoolean(false);
+
+ eventBus().register(event -> {
+ if (nbCalls.get() - finishedExecutions.get() > EventBus.EXECUTION_RATE) {
+ rateExceeded.set(true);
+ }
+ nbCalls.incrementAndGet();
+ Thread.sleep(Duration.ofMillis(200).toMillis());
+ finishedExecutions.incrementAndGet();
+
+ }, KEY_1);
+
+ IntStream.range(0, eventCount)
+ .forEach(i -> eventBus().dispatch(EVENT, KEY_1).block());
+
+ WAIT_CONDITION.atMost(com.jayway.awaitility.Duration.TEN_MINUTES).until(() -> finishedExecutions.get() == eventCount);
+ assertThat(rateExceeded).isFalse();
+ }
+
+ @Test
default void registeredListenersShouldNotReceiveNoopEvents() throws Exception {
MailboxListener listener = newListener();
diff --git a/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/GroupRegistration.java b/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/GroupRegistration.java
index 426ae46..549de7e 100644
--- a/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/GroupRegistration.java
+++ b/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/GroupRegistration.java
@@ -45,6 +45,7 @@ import reactor.core.publisher.MonoProcessor;
import reactor.core.scheduler.Schedulers;
import reactor.rabbitmq.AcknowledgableDelivery;
import reactor.rabbitmq.BindingSpecification;
+import reactor.rabbitmq.ConsumeOptions;
import reactor.rabbitmq.QueueSpecification;
import reactor.rabbitmq.RabbitFlux;
import reactor.rabbitmq.Receiver;
@@ -126,7 +127,7 @@ class GroupRegistration implements Registration {
}
private void subscribeWorkQueue() {
- receiverSubscriber = Optional.of(receiver.consumeManualAck(queueName.asString())
+ receiverSubscriber = Optional.of(receiver.consumeManualAck(queueName.asString(), new ConsumeOptions().qos(EventBus.EXECUTION_RATE))
.subscribeOn(Schedulers.parallel())
.filter(delivery -> Objects.nonNull(delivery.getBody()))
.flatMap(this::deliver)
diff --git a/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/KeyRegistrationHandler.java b/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/KeyRegistrationHandler.java
index 59ed7c1..5f5d2b3 100644
--- a/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/KeyRegistrationHandler.java
+++ b/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/KeyRegistrationHandler.java
@@ -43,6 +43,7 @@ import com.rabbitmq.client.Delivery;
import reactor.core.Disposable;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
+import reactor.rabbitmq.ConsumeOptions;
import reactor.rabbitmq.QueueSpecification;
import reactor.rabbitmq.RabbitFlux;
import reactor.rabbitmq.Receiver;
@@ -85,7 +86,7 @@ class KeyRegistrationHandler {
.doOnSuccess(registrationQueue::initialize)
.block();
- receiverSubscriber = Optional.of(receiver.consumeAutoAck(registrationQueue.asString())
+ receiverSubscriber = Optional.of(receiver.consumeAutoAck(registrationQueue.asString(), new ConsumeOptions().qos(EventBus.EXECUTION_RATE))
.subscribeOn(Schedulers.parallel())
.flatMap(this::handleDelivery)
.subscribe());
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org