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 bt...@apache.org on 2019/02/13 08:18:11 UTC
[james-project] 04/12: JAMES-2659 remove needless complexity:
subscribeWith(MonoProcessor) triggers an eager computation and replay the
computed value like .cache
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 f0e259726650a8ab18fa970bff2087ee85d3d74b
Author: Matthieu Baechler <ma...@apache.org>
AuthorDate: Thu Feb 7 15:01:27 2019 +0100
JAMES-2659 remove needless complexity: subscribeWith(MonoProcessor) triggers an eager computation and replay the computed value like .cache
---
.../apache/james/mailbox/events/EventDispatcher.java | 19 +++++++++----------
1 file changed, 9 insertions(+), 10 deletions(-)
diff --git a/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/EventDispatcher.java b/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/EventDispatcher.java
index 981fb95..d7733b5 100644
--- a/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/EventDispatcher.java
+++ b/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/EventDispatcher.java
@@ -75,12 +75,16 @@ class EventDispatcher {
}
Mono<Void> dispatch(Event event, Set<RegistrationKey> keys) {
- byte[] serializedEvent = serializeEvent(event);
-
- Mono<Void> distantDispatchMono = doDispatch(serializedEvent, keys).cache()
+ return Flux
+ .concat(
+ dispatchToLocalListeners(event, keys),
+ dispatchToRemoteListeners(serializeEvent(event), keys))
+ .then()
.subscribeWith(MonoProcessor.create());
+ }
- Mono<Void> localListenerDelivery = Flux.fromIterable(keys)
+ private Mono<Void> dispatchToLocalListeners(Event event, Set<RegistrationKey> keys) {
+ return Flux.fromIterable(keys)
.subscribeOn(Schedulers.elastic())
.flatMap(key -> mailboxListenerRegistry.getLocalMailboxListeners(key)
.map(listener -> Pair.of(key, listener)))
@@ -89,11 +93,6 @@ class EventDispatcher {
.doOnError(e -> structuredLogger(event, keys)
.log(logger -> logger.error("Exception happens when dispatching event", e)))
.onErrorResume(e -> Mono.empty()))
- .cache()
- .then()
- .subscribeWith(MonoProcessor.create());
-
- return Flux.concat(localListenerDelivery, distantDispatchMono)
.then();
}
@@ -112,7 +111,7 @@ class EventDispatcher {
.addField(EventBus.StructuredLoggingFields.REGISTRATION_KEYS, keys);
}
- private Mono<Void> doDispatch(byte[] serializedEvent, Set<RegistrationKey> keys) {
+ private Mono<Void> dispatchToRemoteListeners(byte[] serializedEvent, Set<RegistrationKey> keys) {
Stream<RoutingKey> routingKeys = Stream.concat(Stream.of(RoutingKey.empty()), keys.stream().map(RoutingKey::of));
Stream<OutboundMessage> outboundMessages = routingKeys
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org