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:07 UTC
[james-project] branch master updated (fdd2fd4 -> 3118607)
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git.
from fdd2fd4 [maven-release-plugin] Spring dockerfile should rely on 3.4.0-SNAPSHOT
new 9691d5d JAMES-2659 Stop RabbitMQ on each test class
new 7b154f0 JAMES-2659 don't use Reactor when Stream is enough
new ee538d9 JAMES-2659 checkstyle fix
new f0e2597 JAMES-2659 remove needless complexity: subscribeWith(MonoProcessor) triggers an eager computation and replay the computed value like .cache
new 5edbc26 JAMES-2659 simplify call to local listeners
new a29f84a JAMES-2659 Implement MailboxListenerRegistry on top of a ConcurrentHashMap to limit lock-contention and optimize for iteration rather than addition/removal
new c038bc2 JAMES-2659 implement a simpler API for MailboxListenerRegistry
new 5d64652 JAMES-2659 Rename ambiguous Registry
new 77839d7 JAMES-2632 Don't search for max quota twice when you can compute it
new 8dce1ec JAMES-2632 Use the same method for all max quota computations
new 5dacd82 JAMES-2632 JMAP should not load mailbox meta-data to access resolved ACL
new 3118607 JAMES-2632 Reading metadata is enough when moving messages by messageIds
The 12 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../backend/rabbitmq/DockerRabbitMQTestRule.java | 20 ++--
.../org/apache/james/mailbox/MessageManager.java | 8 ++
.../james/mailbox/quota/MaxQuotaManager.java | 25 +++-
.../quota/CassandraPerUserMaxQuotaManager.java | 30 -----
.../james/mailbox/events/EventDispatcher.java | 76 ++++++------
.../mailbox/events/KeyRegistrationHandler.java | 24 ++--
.../mailbox/events/LocalListenerRegistry.java | 105 +++++++++++++++++
.../mailbox/events/MailboxListenerRegistry.java | 59 ----------
.../james/mailbox/events/RabbitMQEventBus.java | 7 +-
...tryTest.java => LocalListenerRegistryTest.java} | 128 ++++++++++-----------
.../james/mailbox/events/RabbitMQEventBusTest.java | 1 -
.../jpa/quota/JPAPerUserMaxQuotaManager.java | 22 ----
.../quota/InMemoryPerUserMaxQuotaManager.java | 17 ---
.../james/mailbox/store/StoreMessageIdManager.java | 4 +-
.../james/mailbox/store/StoreMessageManager.java | 8 +-
.../mailbox/store/quota/StoreQuotaManager.java | 13 ++-
.../mailbox/store/quota/StoreQuotaManagerTest.java | 14 ++-
.../apache/james/jmap/model/MailboxFactory.java | 4 +-
18 files changed, 290 insertions(+), 275 deletions(-)
create mode 100644 mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/LocalListenerRegistry.java
delete mode 100644 mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/MailboxListenerRegistry.java
rename mailbox/event/event-rabbitmq/src/test/java/org/apache/james/mailbox/events/{MailboxListenerRegistryTest.java => LocalListenerRegistryTest.java} (63%)
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 01/12: JAMES-2659 Stop RabbitMQ on each test class
Posted by 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 9691d5d7751a3e607117d7f78f3cbca3a336ae0d
Author: Antoine Duprat <ad...@linagora.com>
AuthorDate: Tue Feb 12 17:26:34 2019 +0100
JAMES-2659 Stop RabbitMQ on each test class
---
.../backend/rabbitmq/DockerRabbitMQTestRule.java | 20 ++++++++++----------
1 file changed, 10 insertions(+), 10 deletions(-)
diff --git a/backends-common/rabbitmq/src/test/java/org/apache/james/backend/rabbitmq/DockerRabbitMQTestRule.java b/backends-common/rabbitmq/src/test/java/org/apache/james/backend/rabbitmq/DockerRabbitMQTestRule.java
index 0814049..9216fe1 100644
--- a/backends-common/rabbitmq/src/test/java/org/apache/james/backend/rabbitmq/DockerRabbitMQTestRule.java
+++ b/backends-common/rabbitmq/src/test/java/org/apache/james/backend/rabbitmq/DockerRabbitMQTestRule.java
@@ -18,24 +18,24 @@
****************************************************************/
package org.apache.james.backend.rabbitmq;
+import org.junit.rules.ExternalResource;
import org.junit.rules.TestRule;
import org.junit.runner.Description;
import org.junit.runners.model.Statement;
-public class DockerRabbitMQTestRule implements TestRule {
+public class DockerRabbitMQTestRule extends ExternalResource {
private DockerRabbitMQ dockerRabbitMQ;
@Override
- public Statement apply(Statement base, Description description) {
- return new Statement() {
- @Override
- public void evaluate() throws Throwable {
- dockerRabbitMQ = DockerRabbitMQ.withoutCookie();
- dockerRabbitMQ.start();
- base.evaluate();
- }
- };
+ protected void before() throws Throwable {
+ dockerRabbitMQ = DockerRabbitMQ.withoutCookie();
+ dockerRabbitMQ.start();
+ }
+
+ @Override
+ protected void after() {
+ dockerRabbitMQ.stop();
}
public DockerRabbitMQ getDockerRabbitMQ() {
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 10/12: JAMES-2632 Use the same method for all max
quota computations
Posted by 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 8dce1ec1481a8c9af84d13211b7b47163321b49c
Author: Raphael Ouazana <ra...@linagora.com>
AuthorDate: Tue Dec 18 12:09:24 2018 +0100
JAMES-2632 Use the same method for all max quota computations
---
.../james/mailbox/quota/MaxQuotaManager.java | 11 ++++++--
.../quota/CassandraPerUserMaxQuotaManager.java | 30 ----------------------
.../jpa/quota/JPAPerUserMaxQuotaManager.java | 22 ----------------
.../quota/InMemoryPerUserMaxQuotaManager.java | 17 ------------
4 files changed, 9 insertions(+), 71 deletions(-)
diff --git a/mailbox/api/src/main/java/org/apache/james/mailbox/quota/MaxQuotaManager.java b/mailbox/api/src/main/java/org/apache/james/mailbox/quota/MaxQuotaManager.java
index d027a32..6cb89b7 100644
--- a/mailbox/api/src/main/java/org/apache/james/mailbox/quota/MaxQuotaManager.java
+++ b/mailbox/api/src/main/java/org/apache/james/mailbox/quota/MaxQuotaManager.java
@@ -27,6 +27,7 @@ import org.apache.james.core.quota.QuotaCount;
import org.apache.james.core.quota.QuotaSize;
import org.apache.james.mailbox.exception.MailboxException;
import org.apache.james.mailbox.model.Quota;
+import org.apache.james.mailbox.model.Quota.Scope;
import org.apache.james.mailbox.model.QuotaRoot;
import org.apache.james.util.OptionalUtils;
@@ -114,7 +115,10 @@ public interface MaxQuotaManager {
* @param quotaRoot Quota root argument from RFC 2087 ( correspond to the user owning this mailbox )
* @return The maximum storage in bytes if any
*/
- Optional<QuotaSize> getMaxStorage(QuotaRoot quotaRoot) throws MailboxException;
+ default Optional<QuotaSize> getMaxStorage(QuotaRoot quotaRoot) throws MailboxException {
+ Map<Scope, QuotaSize> maxStorageDetails = listMaxStorageDetails(quotaRoot);
+ return getMaxStorage(maxStorageDetails);
+ }
default Optional<QuotaSize> getMaxStorage(Map<Quota.Scope, QuotaSize> maxStorageDetails) {
return OptionalUtils.or(
@@ -129,7 +133,10 @@ public interface MaxQuotaManager {
* @param quotaRoot Quota root argument from RFC 2087 ( correspond to the user owning this mailbox )
* @return maximum of allowed message count
*/
- Optional<QuotaCount> getMaxMessage(QuotaRoot quotaRoot) throws MailboxException;
+ default Optional<QuotaCount> getMaxMessage(QuotaRoot quotaRoot) throws MailboxException {
+ Map<Scope, QuotaCount> maxMessagesDetails = listMaxMessagesDetails(quotaRoot);
+ return getMaxMessage(maxMessagesDetails);
+ }
default Optional<QuotaCount> getMaxMessage(Map<Quota.Scope, QuotaCount> maxMessagesDetails) {
return OptionalUtils.or(
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraPerUserMaxQuotaManager.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraPerUserMaxQuotaManager.java
index e4f7327..c278ae6 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraPerUserMaxQuotaManager.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraPerUserMaxQuotaManager.java
@@ -22,7 +22,6 @@ package org.apache.james.mailbox.cassandra.quota;
import java.util.Map;
import java.util.Optional;
import java.util.function.Function;
-import java.util.function.Supplier;
import java.util.stream.Stream;
import javax.inject.Inject;
@@ -34,7 +33,6 @@ import org.apache.james.core.quota.QuotaSize;
import org.apache.james.mailbox.model.Quota;
import org.apache.james.mailbox.model.QuotaRoot;
import org.apache.james.mailbox.quota.MaxQuotaManager;
-import org.apache.james.util.OptionalUtils;
import com.github.fge.lambdas.Throwing;
import com.github.steveash.guavate.Guavate;
@@ -135,34 +133,6 @@ public class CassandraPerUserMaxQuotaManager implements MaxQuotaManager {
}
@Override
- public Optional<QuotaSize> getMaxStorage(QuotaRoot quotaRoot) {
- Supplier<Optional<QuotaSize>> domainQuotaSupplier = Throwing.supplier(() -> quotaRoot.getDomain()
- .flatMap(this::getDomainMaxStorage)).sneakyThrow();
- Supplier<Optional<QuotaSize>> globalDomainSupplier = Throwing.supplier(this::getGlobalMaxStorage).sneakyThrow();
-
- return Stream
- .of(() -> perUserQuota.getMaxStorage(quotaRoot),
- domainQuotaSupplier,
- globalDomainSupplier)
- .flatMap(supplier -> OptionalUtils.toStream(supplier.get()))
- .findFirst();
- }
-
- @Override
- public Optional<QuotaCount> getMaxMessage(QuotaRoot quotaRoot) {
- Supplier<Optional<QuotaCount>> domainQuotaSupplier = Throwing.supplier(() -> quotaRoot.getDomain()
- .flatMap(this::getDomainMaxMessage)).sneakyThrow();
- Supplier<Optional<QuotaCount>> globalDomainSupplier = Throwing.supplier(this::getGlobalMaxMessage).sneakyThrow();
-
- return Stream
- .of(() -> perUserQuota.getMaxMessage(quotaRoot),
- domainQuotaSupplier,
- globalDomainSupplier)
- .flatMap(supplier -> OptionalUtils.toStream(supplier.get()))
- .findFirst();
- }
-
- @Override
public Map<Quota.Scope, QuotaCount> listMaxMessagesDetails(QuotaRoot quotaRoot) {
Function<Domain, Optional<QuotaCount>> domainQuotaSupplier = Throwing.function(this::getDomainMaxMessage).sneakyThrow();
return Stream.of(
diff --git a/mailbox/jpa/src/main/java/org/apache/james/mailbox/jpa/quota/JPAPerUserMaxQuotaManager.java b/mailbox/jpa/src/main/java/org/apache/james/mailbox/jpa/quota/JPAPerUserMaxQuotaManager.java
index dd24bd2..53b6ccd 100644
--- a/mailbox/jpa/src/main/java/org/apache/james/mailbox/jpa/quota/JPAPerUserMaxQuotaManager.java
+++ b/mailbox/jpa/src/main/java/org/apache/james/mailbox/jpa/quota/JPAPerUserMaxQuotaManager.java
@@ -22,7 +22,6 @@ package org.apache.james.mailbox.jpa.quota;
import java.util.Map;
import java.util.Optional;
import java.util.function.Function;
-import java.util.function.Supplier;
import java.util.stream.Stream;
import javax.inject.Inject;
@@ -34,7 +33,6 @@ import org.apache.james.core.quota.QuotaSize;
import org.apache.james.mailbox.model.Quota;
import org.apache.james.mailbox.model.QuotaRoot;
import org.apache.james.mailbox.quota.MaxQuotaManager;
-import org.apache.james.util.OptionalUtils;
import com.github.fge.lambdas.Throwing;
import com.github.steveash.guavate.Guavate;
@@ -119,26 +117,6 @@ public class JPAPerUserMaxQuotaManager implements MaxQuotaManager {
}
@Override
- public Optional<QuotaSize> getMaxStorage(QuotaRoot quotaRoot) {
- return Stream
- .of((Supplier<Optional<QuotaSize>>) () -> dao.getMaxStorage(quotaRoot),
- () -> quotaRoot.getDomain().flatMap(this::getDomainMaxStorage),
- this::getGlobalMaxStorage)
- .flatMap(supplier -> OptionalUtils.toStream(supplier.get()))
- .findFirst();
- }
-
- @Override
- public Optional<QuotaCount> getMaxMessage(QuotaRoot quotaRoot) {
- return Stream
- .of((Supplier<Optional<QuotaCount>>) () -> dao.getMaxMessage(quotaRoot),
- () -> quotaRoot.getDomain().flatMap(this::getDomainMaxMessage),
- this::getGlobalMaxMessage)
- .flatMap(supplier -> OptionalUtils.toStream(supplier.get()))
- .findFirst();
- }
-
- @Override
public Map<Quota.Scope, QuotaCount> listMaxMessagesDetails(QuotaRoot quotaRoot) {
Function<Domain, Optional<QuotaCount>> domainQuotaFunction = Throwing.function(this::getDomainMaxMessage).sneakyThrow();
return Stream.of(
diff --git a/mailbox/memory/src/main/java/org/apache/james/mailbox/inmemory/quota/InMemoryPerUserMaxQuotaManager.java b/mailbox/memory/src/main/java/org/apache/james/mailbox/inmemory/quota/InMemoryPerUserMaxQuotaManager.java
index b2b56a7..848118a 100644
--- a/mailbox/memory/src/main/java/org/apache/james/mailbox/inmemory/quota/InMemoryPerUserMaxQuotaManager.java
+++ b/mailbox/memory/src/main/java/org/apache/james/mailbox/inmemory/quota/InMemoryPerUserMaxQuotaManager.java
@@ -31,7 +31,6 @@ import org.apache.james.core.quota.QuotaSize;
import org.apache.james.mailbox.model.Quota;
import org.apache.james.mailbox.model.QuotaRoot;
import org.apache.james.mailbox.quota.MaxQuotaManager;
-import org.apache.james.util.OptionalUtils;
import com.github.fge.lambdas.Throwing;
import com.github.steveash.guavate.Guavate;
@@ -73,22 +72,6 @@ public class InMemoryPerUserMaxQuotaManager implements MaxQuotaManager {
}
@Override
- public Optional<QuotaSize> getMaxStorage(QuotaRoot quotaRoot) {
- return OptionalUtils.or(
- Optional.ofNullable(userMaxStorage.get(quotaRoot.getValue())),
- quotaRoot.getDomain().flatMap(this::getDomainMaxStorage),
- maxStorage);
- }
-
- @Override
- public Optional<QuotaCount> getMaxMessage(QuotaRoot quotaRoot) {
- return OptionalUtils.or(
- Optional.ofNullable(userMaxMessage.get(quotaRoot.getValue())),
- quotaRoot.getDomain().flatMap(this::getDomainMaxMessage),
- maxMessage);
- }
-
- @Override
public Map<Quota.Scope, QuotaCount> listMaxMessagesDetails(QuotaRoot quotaRoot) {
Function<Domain, Optional<QuotaCount>> domainQuotaFunction = Throwing.function(this::getDomainMaxMessage).sneakyThrow();
return Stream.of(
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 11/12: JAMES-2632 JMAP should not load mailbox
meta-data to access resolved ACL
Posted by 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 5dacd82838d03d8c9eef08256029d002975c5ece
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Wed Jan 30 16:09:04 2019 +0700
JAMES-2632 JMAP should not load mailbox meta-data to access resolved ACL
This saves:
- A highest UID read
- A highest MODSEQ read
So 2 backends reads per mailbox on each JMAP GetMailbox call
---
.../src/main/java/org/apache/james/mailbox/MessageManager.java | 8 ++++++++
.../java/org/apache/james/mailbox/store/StoreMessageManager.java | 8 +++++++-
.../src/main/java/org/apache/james/jmap/model/MailboxFactory.java | 4 +---
3 files changed, 16 insertions(+), 4 deletions(-)
diff --git a/mailbox/api/src/main/java/org/apache/james/mailbox/MessageManager.java b/mailbox/api/src/main/java/org/apache/james/mailbox/MessageManager.java
index 41e79c8..ee4596b 100644
--- a/mailbox/api/src/main/java/org/apache/james/mailbox/MessageManager.java
+++ b/mailbox/api/src/main/java/org/apache/james/mailbox/MessageManager.java
@@ -35,6 +35,7 @@ import javax.mail.Flags;
import org.apache.james.mailbox.MailboxManager.MessageCapabilities;
import org.apache.james.mailbox.exception.MailboxException;
import org.apache.james.mailbox.exception.UnsupportedCriteriaException;
+import org.apache.james.mailbox.exception.UnsupportedRightException;
import org.apache.james.mailbox.model.ComposedMessageId;
import org.apache.james.mailbox.model.MailboxACL;
import org.apache.james.mailbox.model.MailboxCounters;
@@ -440,4 +441,11 @@ public interface MessageManager {
MailboxACL getACL();
}
+
+ /**
+ * Get resolved ACL on this Mailbox for the given Session
+ *
+ * The result will be the same as calling {MessageManager#getMetaDtata().getAcl()} but will load fewer data
+ */
+ MailboxACL getResolvedAcl(MailboxSession mailboxSession) throws UnsupportedRightException;
}
diff --git a/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreMessageManager.java b/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreMessageManager.java
index ca098fb..020eb0e 100644
--- a/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreMessageManager.java
+++ b/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreMessageManager.java
@@ -50,6 +50,7 @@ import org.apache.james.mailbox.events.MailboxIdRegistrationKey;
import org.apache.james.mailbox.events.MailboxListener;
import org.apache.james.mailbox.exception.MailboxException;
import org.apache.james.mailbox.exception.ReadOnlyException;
+import org.apache.james.mailbox.exception.UnsupportedRightException;
import org.apache.james.mailbox.model.ComposedMessageId;
import org.apache.james.mailbox.model.MailboxACL;
import org.apache.james.mailbox.model.MailboxCounters;
@@ -473,7 +474,7 @@ public class StoreMessageManager implements org.apache.james.mailbox.MessageMana
@Override
public MetaData getMetaData(boolean resetRecent, MailboxSession mailboxSession, MetaData.FetchGroup fetchGroup) throws MailboxException {
- MailboxACL resolvedAcl = storeRightManager.getResolvedMailboxACL(mailbox, mailboxSession);
+ MailboxACL resolvedAcl = getResolvedAcl(mailboxSession);
boolean hasReadRight = storeRightManager.hasRight(mailbox, MailboxACL.Right.Read, mailboxSession);
if (!hasReadRight) {
return MailboxMetaData.sensibleInformationFree(resolvedAcl, getMailboxEntity().getUidValidity(), isWriteable(mailboxSession), isModSeqPermanent(mailboxSession));
@@ -524,6 +525,11 @@ public class StoreMessageManager implements org.apache.james.mailbox.MessageMana
return new MailboxMetaData(recent, permanentFlags, uidValidity, uidNext, highestModSeq, messageCount, unseenCount, firstUnseen, isWriteable(mailboxSession), isModSeqPermanent(mailboxSession), resolvedAcl);
}
+ @Override
+ public MailboxACL getResolvedAcl(MailboxSession mailboxSession) throws UnsupportedRightException {
+ return storeRightManager.getResolvedMailboxACL(mailbox, mailboxSession);
+ }
+
/**
* Check if the given {@link Flags} contains {@link Flags} which are not
* included in the returned {@link Flags} of
diff --git a/server/protocols/jmap/src/main/java/org/apache/james/jmap/model/MailboxFactory.java b/server/protocols/jmap/src/main/java/org/apache/james/jmap/model/MailboxFactory.java
index 94310a2..0080339 100644
--- a/server/protocols/jmap/src/main/java/org/apache/james/jmap/model/MailboxFactory.java
+++ b/server/protocols/jmap/src/main/java/org/apache/james/jmap/model/MailboxFactory.java
@@ -51,7 +51,6 @@ import com.google.common.base.Preconditions;
import com.google.common.base.Splitter;
public class MailboxFactory {
- public static final boolean NO_RESET_RECENT = false;
private final MailboxManager mailboxManager;
private final QuotaManager quotaManager;
private final QuotaRootResolver quotaRootResolver;
@@ -113,9 +112,8 @@ public class MailboxFactory {
boolean isOwner = mailboxPath.belongsTo(mailboxSession);
Optional<Role> role = Role.from(mailboxPath.getName());
MailboxCounters mailboxCounters = messageManager.getMailboxCounters(mailboxSession);
- MessageManager.MetaData metaData = messageManager.getMetaData(NO_RESET_RECENT, mailboxSession, MessageManager.MetaData.FetchGroup.NO_COUNT);
- Rights rights = Rights.fromACL(metaData.getACL())
+ Rights rights = Rights.fromACL(messageManager.getResolvedAcl(mailboxSession))
.removeEntriesFor(Username.forMailboxPath(mailboxPath));
Username username = Username.fromSession(mailboxSession);
Quotas quotas = getQuotas(mailboxPath);
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 02/12: JAMES-2659 don't use Reactor when Stream is
enough
Posted by 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 7b154f0b5fc272025e748f95942af1cbb94bd674
Author: Matthieu Baechler <ma...@apache.org>
AuthorDate: Thu Feb 7 14:26:51 2019 +0100
JAMES-2659 don't use Reactor when Stream is enough
---
.../james/mailbox/events/EventDispatcher.java | 22 ++++++++--------------
1 file changed, 8 insertions(+), 14 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 e6f6e1d..981fb95 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
@@ -26,9 +26,11 @@ import static org.apache.james.mailbox.events.RabbitMQEventBus.MAILBOX_EVENT_EXC
import java.nio.charset.StandardCharsets;
import java.util.Set;
+import java.util.stream.Stream;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.james.event.json.EventSerializer;
+import org.apache.james.mailbox.events.RoutingKeyConverter.RoutingKey;
import org.apache.james.util.MDCBuilder;
import org.apache.james.util.MDCStructuredLogger;
import org.apache.james.util.StructuredLogger;
@@ -38,7 +40,6 @@ import org.slf4j.LoggerFactory;
import com.github.fge.lambdas.Throwing;
import com.google.common.collect.ImmutableMap;
import com.rabbitmq.client.AMQP;
-
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoProcessor;
@@ -74,10 +75,7 @@ class EventDispatcher {
}
Mono<Void> dispatch(Event event, Set<RegistrationKey> keys) {
- Mono<byte[]> serializedEvent = Mono.just(event)
- .publishOn(Schedulers.parallel())
- .map(this::serializeEvent)
- .cache();
+ byte[] serializedEvent = serializeEvent(event);
Mono<Void> distantDispatchMono = doDispatch(serializedEvent, keys).cache()
.subscribeWith(MonoProcessor.create());
@@ -114,17 +112,13 @@ class EventDispatcher {
.addField(EventBus.StructuredLoggingFields.REGISTRATION_KEYS, keys);
}
- private Mono<Void> doDispatch(Mono<byte[]> serializedEvent, Set<RegistrationKey> keys) {
- Flux<RoutingKeyConverter.RoutingKey> routingKeys = Flux.concat(
- Mono.just(RoutingKeyConverter.RoutingKey.empty()),
- Flux.fromIterable(keys)
- .map(RoutingKeyConverter.RoutingKey::of));
+ private Mono<Void> doDispatch(byte[] serializedEvent, Set<RegistrationKey> keys) {
+ Stream<RoutingKey> routingKeys = Stream.concat(Stream.of(RoutingKey.empty()), keys.stream().map(RoutingKey::of));
- Flux<OutboundMessage> outboundMessages = routingKeys
- .flatMap(routingKey -> serializedEvent
- .map(payload -> new OutboundMessage(MAILBOX_EVENT_EXCHANGE_NAME, routingKey.asString(), basicProperties, payload)));
+ Stream<OutboundMessage> outboundMessages = routingKeys
+ .map(routingKey -> new OutboundMessage(MAILBOX_EVENT_EXCHANGE_NAME, routingKey.asString(), basicProperties, serializedEvent));
- return sender.send(outboundMessages);
+ return sender.send(Flux.fromStream(outboundMessages));
}
private byte[] serializeEvent(Event event) {
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 08/12: JAMES-2659 Rename ambiguous Registry
Posted by 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 5d64652ac1263d8a03a3efe96475835bb82ec56f
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Tue Feb 12 11:06:56 2019 +0700
JAMES-2659 Rename ambiguous Registry
Make it explicit that we keep track of locally registered listeners
---
.../james/mailbox/events/EventDispatcher.java | 9 ++++----
.../mailbox/events/KeyRegistrationHandler.java | 10 ++++-----
...nerRegistry.java => LocalListenerRegistry.java} | 13 ++++++------
.../james/mailbox/events/RabbitMQEventBus.java | 7 ++++---
...tryTest.java => LocalListenerRegistryTest.java} | 24 +++++++++++-----------
5 files changed, 33 insertions(+), 30 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 2f90983..09c777e 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
@@ -39,6 +39,7 @@ import org.slf4j.LoggerFactory;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.rabbitmq.client.AMQP;
+
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoProcessor;
@@ -53,14 +54,14 @@ class EventDispatcher {
private final EventSerializer eventSerializer;
private final Sender sender;
- private final MailboxListenerRegistry mailboxListenerRegistry;
+ private final LocalListenerRegistry localListenerRegistry;
private final AMQP.BasicProperties basicProperties;
private final MailboxListenerExecutor mailboxListenerExecutor;
- EventDispatcher(EventBusId eventBusId, EventSerializer eventSerializer, Sender sender, MailboxListenerRegistry mailboxListenerRegistry, MailboxListenerExecutor mailboxListenerExecutor) {
+ EventDispatcher(EventBusId eventBusId, EventSerializer eventSerializer, Sender sender, LocalListenerRegistry localListenerRegistry, MailboxListenerExecutor mailboxListenerExecutor) {
this.eventSerializer = eventSerializer;
this.sender = sender;
- this.mailboxListenerRegistry = mailboxListenerRegistry;
+ this.localListenerRegistry = localListenerRegistry;
this.basicProperties = new AMQP.BasicProperties.Builder()
.headers(ImmutableMap.of(EVENT_BUS_ID, eventBusId.asString()))
.build();
@@ -86,7 +87,7 @@ class EventDispatcher {
private Mono<Void> dispatchToLocalListeners(Event event, Set<RegistrationKey> keys) {
return Flux.fromIterable(keys)
.subscribeOn(Schedulers.elastic())
- .flatMap(key -> mailboxListenerRegistry.getLocalMailboxListeners(key)
+ .flatMap(key -> localListenerRegistry.getLocalMailboxListeners(key)
.map(listener -> Tuples.of(key, listener)))
.filter(pair -> pair.getT2().getExecutionMode().equals(MailboxListener.ExecutionMode.SYNCHRONOUS))
.flatMap(pair -> executeListener(event, pair.getT2(), pair.getT1()))
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 dac9d93..59ed7c1 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
@@ -53,7 +53,7 @@ class KeyRegistrationHandler {
private static final Logger LOGGER = LoggerFactory.getLogger(KeyRegistrationHandler.class);
private final EventBusId eventBusId;
- private final MailboxListenerRegistry mailboxListenerRegistry;
+ private final LocalListenerRegistry localListenerRegistry;
private final EventSerializer eventSerializer;
private final Sender sender;
private final RoutingKeyConverter routingKeyConverter;
@@ -63,12 +63,12 @@ class KeyRegistrationHandler {
private final MailboxListenerExecutor mailboxListenerExecutor;
private Optional<Disposable> receiverSubscriber;
- KeyRegistrationHandler(EventBusId eventBusId, EventSerializer eventSerializer, Sender sender, Mono<Connection> connectionMono, RoutingKeyConverter routingKeyConverter, MailboxListenerRegistry mailboxListenerRegistry, MailboxListenerExecutor mailboxListenerExecutor) {
+ KeyRegistrationHandler(EventBusId eventBusId, EventSerializer eventSerializer, Sender sender, Mono<Connection> connectionMono, RoutingKeyConverter routingKeyConverter, LocalListenerRegistry localListenerRegistry, MailboxListenerExecutor mailboxListenerExecutor) {
this.eventBusId = eventBusId;
this.eventSerializer = eventSerializer;
this.sender = sender;
this.routingKeyConverter = routingKeyConverter;
- this.mailboxListenerRegistry = mailboxListenerRegistry;
+ this.localListenerRegistry = localListenerRegistry;
this.receiver = RabbitFlux.createReceiver(new ReceiverOptions().connectionMono(connectionMono));
this.mailboxListenerExecutor = mailboxListenerExecutor;
this.registrationQueue = new RegistrationQueueName();
@@ -99,7 +99,7 @@ class KeyRegistrationHandler {
}
Registration register(MailboxListener listener, RegistrationKey key) {
- MailboxListenerRegistry.Registration registration = mailboxListenerRegistry.addListener(key, listener);
+ LocalListenerRegistry.LocalRegistration registration = localListenerRegistry.addListener(key, listener);
if (registration.isFirstListener()) {
registrationBinder.bind(key).block();
}
@@ -122,7 +122,7 @@ class KeyRegistrationHandler {
RegistrationKey registrationKey = routingKeyConverter.toRegistrationKey(routingKey);
Event event = toEvent(delivery);
- return mailboxListenerRegistry.getLocalMailboxListeners(registrationKey)
+ return localListenerRegistry.getLocalMailboxListeners(registrationKey)
.filter(listener -> !isLocalSynchronousListeners(eventBusId, listener))
.flatMap(listener -> Mono.fromRunnable(Throwing.runnable(() -> executeListener(listener, event, registrationKey)))
.doOnError(e -> structuredLogger(event, registrationKey)
diff --git a/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/MailboxListenerRegistry.java b/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/LocalListenerRegistry.java
similarity index 91%
rename from mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/MailboxListenerRegistry.java
rename to mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/LocalListenerRegistry.java
index 35ebf3d..faa0d23 100644
--- a/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/MailboxListenerRegistry.java
+++ b/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/LocalListenerRegistry.java
@@ -28,19 +28,20 @@ import java.util.function.Supplier;
import com.github.steveash.guavate.Guavate;
import com.google.common.collect.ImmutableSet;
+
import reactor.core.publisher.Flux;
-class MailboxListenerRegistry {
+class LocalListenerRegistry {
interface RemovalStatus {
boolean lastListenerRemoved();
}
- public static class Registration {
+ public static class LocalRegistration {
private final boolean firstListener;
private final Supplier<RemovalStatus> unregister;
- public Registration(boolean firstListener, Supplier<RemovalStatus> unregister) {
+ public LocalRegistration(boolean firstListener, Supplier<RemovalStatus> unregister) {
this.firstListener = firstListener;
this.unregister = unregister;
}
@@ -56,11 +57,11 @@ class MailboxListenerRegistry {
private final ConcurrentHashMap<RegistrationKey, ImmutableSet<MailboxListener>> listenersByKey;
- MailboxListenerRegistry() {
+ LocalListenerRegistry() {
this.listenersByKey = new ConcurrentHashMap<>();
}
- Registration addListener(RegistrationKey registrationKey, MailboxListener listener) {
+ LocalRegistration addListener(RegistrationKey registrationKey, MailboxListener listener) {
AtomicBoolean firstListener = new AtomicBoolean(false);
listenersByKey.compute(registrationKey, (key, listeners) ->
Optional.ofNullable(listeners)
@@ -70,7 +71,7 @@ class MailboxListenerRegistry {
return ImmutableSet.of(listener);
})
);
- return new Registration(firstListener.get(), () -> removeListener(registrationKey, listener));
+ return new LocalRegistration(firstListener.get(), () -> removeListener(registrationKey, listener));
}
private RemovalStatus removeListener(RegistrationKey registrationKey, MailboxListener listener) {
diff --git a/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/RabbitMQEventBus.java b/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/RabbitMQEventBus.java
index c800850..c742107 100644
--- a/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/RabbitMQEventBus.java
+++ b/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/RabbitMQEventBus.java
@@ -30,6 +30,7 @@ import org.apache.james.metrics.api.MetricFactory;
import com.github.fge.lambdas.Throwing;
import com.rabbitmq.client.Connection;
+
import reactor.core.publisher.Mono;
import reactor.rabbitmq.RabbitFlux;
import reactor.rabbitmq.Sender;
@@ -75,10 +76,10 @@ public class RabbitMQEventBus implements EventBus {
if (!isRunning && !isStopping) {
sender = RabbitFlux.createSender(new SenderOptions().connectionMono(connectionMono)
.resourceManagementChannelMono(connectionMono.map(Throwing.function(Connection::createChannel))));
- MailboxListenerRegistry mailboxListenerRegistry = new MailboxListenerRegistry();
- keyRegistrationHandler = new KeyRegistrationHandler(eventBusId, eventSerializer, sender, connectionMono, routingKeyConverter, mailboxListenerRegistry, mailboxListenerExecutor);
+ LocalListenerRegistry localListenerRegistry = new LocalListenerRegistry();
+ keyRegistrationHandler = new KeyRegistrationHandler(eventBusId, eventSerializer, sender, connectionMono, routingKeyConverter, localListenerRegistry, mailboxListenerExecutor);
groupRegistrationHandler = new GroupRegistrationHandler(eventSerializer, sender, connectionMono, retryBackoff, eventDeadLetters, mailboxListenerExecutor);
- eventDispatcher = new EventDispatcher(eventBusId, eventSerializer, sender, mailboxListenerRegistry, mailboxListenerExecutor);
+ eventDispatcher = new EventDispatcher(eventBusId, eventSerializer, sender, localListenerRegistry, mailboxListenerExecutor);
eventDispatcher.start();
keyRegistrationHandler.start();
diff --git a/mailbox/event/event-rabbitmq/src/test/java/org/apache/james/mailbox/events/MailboxListenerRegistryTest.java b/mailbox/event/event-rabbitmq/src/test/java/org/apache/james/mailbox/events/LocalListenerRegistryTest.java
similarity index 89%
rename from mailbox/event/event-rabbitmq/src/test/java/org/apache/james/mailbox/events/MailboxListenerRegistryTest.java
rename to mailbox/event/event-rabbitmq/src/test/java/org/apache/james/mailbox/events/LocalListenerRegistryTest.java
index 2d732dc..05a6a13 100644
--- a/mailbox/event/event-rabbitmq/src/test/java/org/apache/james/mailbox/events/MailboxListenerRegistryTest.java
+++ b/mailbox/event/event-rabbitmq/src/test/java/org/apache/james/mailbox/events/LocalListenerRegistryTest.java
@@ -34,14 +34,14 @@ import org.junit.jupiter.api.Test;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
-class MailboxListenerRegistryTest {
+class LocalListenerRegistryTest {
private static final MailboxIdRegistrationKey KEY_1 = new MailboxIdRegistrationKey(TestId.of(42));
- private MailboxListenerRegistry testee;
+ private LocalListenerRegistry testee;
@BeforeEach
void setUp() {
- testee = new MailboxListenerRegistry();
+ testee = new LocalListenerRegistry();
}
@Test
@@ -75,7 +75,7 @@ class MailboxListenerRegistryTest {
MailboxListener listener1 = event -> {};
MailboxListener listener2 = event -> {};
testee.addListener(KEY_1, listener1);
- MailboxListenerRegistry.Registration registration = testee.addListener(KEY_1, listener2);
+ LocalListenerRegistry.LocalRegistration registration = testee.addListener(KEY_1, listener2);
registration.unregister();
@@ -105,7 +105,7 @@ class MailboxListenerRegistryTest {
MailboxListener listener = event -> {};
MailboxListener listener2 = event -> {};
- MailboxListenerRegistry.Registration registration = testee.addListener(KEY_1, listener);
+ LocalListenerRegistry.LocalRegistration registration = testee.addListener(KEY_1, listener);
testee.addListener(KEY_1, listener2);
assertThat(registration.unregister().lastListenerRemoved()).isFalse();
@@ -116,7 +116,7 @@ class MailboxListenerRegistryTest {
MailboxListener listener = event -> {};
- MailboxListenerRegistry.Registration registration = testee.addListener(KEY_1, listener);
+ LocalListenerRegistry.LocalRegistration registration = testee.addListener(KEY_1, listener);
assertThat(registration.unregister().lastListenerRemoved()).isTrue();
}
@@ -167,7 +167,7 @@ class MailboxListenerRegistryTest {
void getLocalMailboxListenersShouldReturnEmptyWhenRemoveAddedListener() throws Exception {
MailboxListener listener1 = event -> {};
- MailboxListenerRegistry.Registration registration = testee.addListener(KEY_1, listener1);
+ LocalListenerRegistry.LocalRegistration registration = testee.addListener(KEY_1, listener1);
ConcurrentTestRunner.builder()
.operation(((threadNumber, operationNumber) -> registration.unregister()))
@@ -190,17 +190,17 @@ class MailboxListenerRegistryTest {
ConcurrentTestRunner.builder()
.operation((threadNumber, operationNumber) -> {
if (threadNumber % 3 == 0) {
- MailboxListenerRegistry.Registration registration = testee.addListener(KEY_1, listener1);
+ LocalListenerRegistry.LocalRegistration registration = testee.addListener(KEY_1, listener1);
if (registration.isFirstListener()) {
firstListenerCount.incrementAndGet();
}
} else if (threadNumber % 3 == 1) {
- MailboxListenerRegistry.Registration registration = testee.addListener(KEY_1, listener2);
+ LocalListenerRegistry.LocalRegistration registration = testee.addListener(KEY_1, listener2);
if (registration.isFirstListener()) {
firstListenerCount.incrementAndGet();
}
} else if (threadNumber % 3 == 2) {
- MailboxListenerRegistry.Registration registration = testee.addListener(KEY_1, listener3);
+ LocalListenerRegistry.LocalRegistration registration = testee.addListener(KEY_1, listener3);
if (registration.isFirstListener()) {
firstListenerCount.incrementAndGet();
}
@@ -218,7 +218,7 @@ class MailboxListenerRegistryTest {
MailboxListener listener1 = event -> {};
AtomicInteger lastListenerRemoved = new AtomicInteger(0);
- MailboxListenerRegistry.Registration registration = testee.addListener(KEY_1, listener1);
+ LocalListenerRegistry.LocalRegistration registration = testee.addListener(KEY_1, listener1);
ConcurrentTestRunner.builder()
.operation(((threadNumber, operationNumber) -> {
if (registration.unregister().lastListenerRemoved()) {
@@ -244,7 +244,7 @@ class MailboxListenerRegistryTest {
testee.addListener(KEY_1, listener2);
testee.addListener(KEY_1, listener3);
testee.addListener(KEY_1, listener4);
- MailboxListenerRegistry.Registration registration5 = testee.addListener(KEY_1, listener5);
+ LocalListenerRegistry.LocalRegistration registration5 = testee.addListener(KEY_1, listener5);
Mono<List<MailboxListener>> listeners = testee.getLocalMailboxListeners(KEY_1)
.publishOn(Schedulers.elastic())
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 09/12: JAMES-2632 Don't search for max quota twice
when you can compute it
Posted by 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 77839d70a308abc949a2bce36a209f0e7b057537
Author: Raphael Ouazana <ra...@linagora.com>
AuthorDate: Mon Dec 17 17:32:25 2018 +0100
JAMES-2632 Don't search for max quota twice when you can compute it
---
.../org/apache/james/mailbox/quota/MaxQuotaManager.java | 14 ++++++++++++++
.../james/mailbox/store/quota/StoreQuotaManager.java | 13 +++++++++----
.../james/mailbox/store/quota/StoreQuotaManagerTest.java | 14 ++++++++++----
3 files changed, 33 insertions(+), 8 deletions(-)
diff --git a/mailbox/api/src/main/java/org/apache/james/mailbox/quota/MaxQuotaManager.java b/mailbox/api/src/main/java/org/apache/james/mailbox/quota/MaxQuotaManager.java
index c5b6f68..d027a32 100644
--- a/mailbox/api/src/main/java/org/apache/james/mailbox/quota/MaxQuotaManager.java
+++ b/mailbox/api/src/main/java/org/apache/james/mailbox/quota/MaxQuotaManager.java
@@ -116,6 +116,13 @@ public interface MaxQuotaManager {
*/
Optional<QuotaSize> getMaxStorage(QuotaRoot quotaRoot) throws MailboxException;
+ default Optional<QuotaSize> getMaxStorage(Map<Quota.Scope, QuotaSize> maxStorageDetails) {
+ return OptionalUtils.or(
+ Optional.ofNullable(maxStorageDetails.get(Quota.Scope.User)),
+ Optional.ofNullable(maxStorageDetails.get(Quota.Scope.Domain)),
+ Optional.ofNullable(maxStorageDetails.get(Quota.Scope.Global)));
+ }
+
/**
* Return the maximum message count which is allowed for the given {@link QuotaRoot} (in fact the user which the session is bound to)
*
@@ -124,6 +131,13 @@ public interface MaxQuotaManager {
*/
Optional<QuotaCount> getMaxMessage(QuotaRoot quotaRoot) throws MailboxException;
+ default Optional<QuotaCount> getMaxMessage(Map<Quota.Scope, QuotaCount> maxMessagesDetails) {
+ return OptionalUtils.or(
+ Optional.ofNullable(maxMessagesDetails.get(Quota.Scope.User)),
+ Optional.ofNullable(maxMessagesDetails.get(Quota.Scope.Domain)),
+ Optional.ofNullable(maxMessagesDetails.get(Quota.Scope.Global)));
+ }
+
Map<Quota.Scope, QuotaCount> listMaxMessagesDetails(QuotaRoot quotaRoot);
Map<Quota.Scope, QuotaSize> listMaxStorageDetails(QuotaRoot quotaRoot);
diff --git a/mailbox/store/src/main/java/org/apache/james/mailbox/store/quota/StoreQuotaManager.java b/mailbox/store/src/main/java/org/apache/james/mailbox/store/quota/StoreQuotaManager.java
index d2825cb..2889270 100644
--- a/mailbox/store/src/main/java/org/apache/james/mailbox/store/quota/StoreQuotaManager.java
+++ b/mailbox/store/src/main/java/org/apache/james/mailbox/store/quota/StoreQuotaManager.java
@@ -19,12 +19,15 @@
package org.apache.james.mailbox.store.quota;
+import java.util.Map;
+
import javax.inject.Inject;
import org.apache.james.core.quota.QuotaCount;
import org.apache.james.core.quota.QuotaSize;
import org.apache.james.mailbox.exception.MailboxException;
import org.apache.james.mailbox.model.Quota;
+import org.apache.james.mailbox.model.Quota.Scope;
import org.apache.james.mailbox.model.QuotaRoot;
import org.apache.james.mailbox.quota.CurrentQuotaManager;
import org.apache.james.mailbox.quota.MaxQuotaManager;
@@ -47,20 +50,22 @@ public class StoreQuotaManager implements QuotaManager {
@Override
public Quota<QuotaCount> getMessageQuota(QuotaRoot quotaRoot) throws MailboxException {
+ Map<Scope, QuotaCount> maxMessageDetails = maxQuotaManager.listMaxMessagesDetails(quotaRoot);
return Quota.<QuotaCount>builder()
.used(currentQuotaManager.getCurrentMessageCount(quotaRoot))
- .computedLimit(maxQuotaManager.getMaxMessage(quotaRoot).orElse(QuotaCount.unlimited()))
- .limitsByScope(maxQuotaManager.listMaxMessagesDetails(quotaRoot))
+ .computedLimit(maxQuotaManager.getMaxMessage(maxMessageDetails).orElse(QuotaCount.unlimited()))
+ .limitsByScope(maxMessageDetails)
.build();
}
@Override
public Quota<QuotaSize> getStorageQuota(QuotaRoot quotaRoot) throws MailboxException {
+ Map<Scope, QuotaSize> maxStorageDetails = maxQuotaManager.listMaxStorageDetails(quotaRoot);
return Quota.<QuotaSize>builder()
.used(currentQuotaManager.getCurrentStorage(quotaRoot))
- .computedLimit(maxQuotaManager.getMaxStorage(quotaRoot).orElse(QuotaSize.unlimited()))
- .limitsByScope(maxQuotaManager.listMaxStorageDetails(quotaRoot))
+ .computedLimit(maxQuotaManager.getMaxStorage(maxStorageDetails).orElse(QuotaSize.unlimited()))
+ .limitsByScope(maxStorageDetails)
.build();
}
diff --git a/mailbox/store/src/test/java/org/apache/james/mailbox/store/quota/StoreQuotaManagerTest.java b/mailbox/store/src/test/java/org/apache/james/mailbox/store/quota/StoreQuotaManagerTest.java
index ee8dee4..2fde29d 100644
--- a/mailbox/store/src/test/java/org/apache/james/mailbox/store/quota/StoreQuotaManagerTest.java
+++ b/mailbox/store/src/test/java/org/apache/james/mailbox/store/quota/StoreQuotaManagerTest.java
@@ -20,9 +20,11 @@
package org.apache.james.mailbox.store.quota;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
+import java.util.Map;
import java.util.Optional;
import org.apache.james.core.quota.QuotaCount;
@@ -49,34 +51,38 @@ public class StoreQuotaManagerTest {
quotaRoot = QuotaRoot.quotaRoot("benwa", Optional.empty());
}
+ @SuppressWarnings("unchecked")
@Test
public void getMessageQuotaShouldWorkWithNumericValues() throws Exception {
- when(mockedMaxQuotaManager.getMaxMessage(quotaRoot)).thenReturn(Optional.of(QuotaCount.count(360L)));
+ when(mockedMaxQuotaManager.getMaxMessage(any(Map.class))).thenReturn(Optional.of(QuotaCount.count(360L)));
when(mockedCurrentQuotaManager.getCurrentMessageCount(quotaRoot)).thenReturn(QuotaCount.count(36L));
assertThat(testee.getMessageQuota(quotaRoot)).isEqualTo(
Quota.<QuotaCount>builder().used(QuotaCount.count(36)).computedLimit(QuotaCount.count(360)).build());
}
+ @SuppressWarnings("unchecked")
@Test
public void getStorageQuotaShouldWorkWithNumericValues() throws Exception {
- when(mockedMaxQuotaManager.getMaxStorage(quotaRoot)).thenReturn(Optional.of(QuotaSize.size(360L)));
+ when(mockedMaxQuotaManager.getMaxStorage(any(Map.class))).thenReturn(Optional.of(QuotaSize.size(360L)));
when(mockedCurrentQuotaManager.getCurrentStorage(quotaRoot)).thenReturn(QuotaSize.size(36L));
assertThat(testee.getStorageQuota(quotaRoot)).isEqualTo(
Quota.<QuotaSize>builder().used(QuotaSize.size(36)).computedLimit(QuotaSize.size(360)).build());
}
+ @SuppressWarnings("unchecked")
@Test
public void getStorageQuotaShouldCalculateCurrentQuotaWhenUnlimited() throws Exception {
- when(mockedMaxQuotaManager.getMaxStorage(quotaRoot)).thenReturn(Optional.of(QuotaSize.unlimited()));
+ when(mockedMaxQuotaManager.getMaxStorage(any(Map.class))).thenReturn(Optional.of(QuotaSize.unlimited()));
when(mockedCurrentQuotaManager.getCurrentStorage(quotaRoot)).thenReturn(QuotaSize.size(36L));
assertThat(testee.getStorageQuota(quotaRoot)).isEqualTo(
Quota.<QuotaSize>builder().used(QuotaSize.size(36)).computedLimit(QuotaSize.unlimited()).build());
}
+ @SuppressWarnings("unchecked")
@Test
public void getMessageQuotaShouldCalculateCurrentQuotaWhenUnlimited() throws Exception {
- when(mockedMaxQuotaManager.getMaxMessage(quotaRoot)).thenReturn(Optional.of(QuotaCount.unlimited()));
+ when(mockedMaxQuotaManager.getMaxMessage(any(Map.class))).thenReturn(Optional.of(QuotaCount.unlimited()));
when(mockedCurrentQuotaManager.getCurrentMessageCount(quotaRoot)).thenReturn(QuotaCount.count(36L));
assertThat(testee.getMessageQuota(quotaRoot)).isEqualTo(
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 05/12: JAMES-2659 simplify call to local listeners
Posted by 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 5edbc264ff63f056a0b3fcf6d1073919377ddcc1
Author: Matthieu Baechler <ma...@apache.org>
AuthorDate: Thu Feb 7 15:21:02 2019 +0100
JAMES-2659 simplify call to local listeners
---
.../james/mailbox/events/EventDispatcher.java | 32 +++++++++++++---------
1 file changed, 19 insertions(+), 13 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 d7733b5..2f90983 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
@@ -28,7 +28,6 @@ import java.nio.charset.StandardCharsets;
import java.util.Set;
import java.util.stream.Stream;
-import org.apache.commons.lang3.tuple.Pair;
import org.apache.james.event.json.EventSerializer;
import org.apache.james.mailbox.events.RoutingKeyConverter.RoutingKey;
import org.apache.james.util.MDCBuilder;
@@ -37,8 +36,8 @@ import org.apache.james.util.StructuredLogger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.github.fge.lambdas.Throwing;
import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
import com.rabbitmq.client.AMQP;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
@@ -47,6 +46,7 @@ import reactor.core.scheduler.Schedulers;
import reactor.rabbitmq.ExchangeSpecification;
import reactor.rabbitmq.OutboundMessage;
import reactor.rabbitmq.Sender;
+import reactor.util.function.Tuples;
class EventDispatcher {
private static final Logger LOGGER = LoggerFactory.getLogger(EventDispatcher.class);
@@ -87,20 +87,26 @@ class EventDispatcher {
return Flux.fromIterable(keys)
.subscribeOn(Schedulers.elastic())
.flatMap(key -> mailboxListenerRegistry.getLocalMailboxListeners(key)
- .map(listener -> Pair.of(key, listener)))
- .filter(pair -> pair.getRight().getExecutionMode().equals(MailboxListener.ExecutionMode.SYNCHRONOUS))
- .flatMap(pair -> Mono.fromRunnable(Throwing.runnable(() -> executeListener(event, pair.getRight(), pair.getLeft())))
- .doOnError(e -> structuredLogger(event, keys)
- .log(logger -> logger.error("Exception happens when dispatching event", e)))
- .onErrorResume(e -> Mono.empty()))
+ .map(listener -> Tuples.of(key, listener)))
+ .filter(pair -> pair.getT2().getExecutionMode().equals(MailboxListener.ExecutionMode.SYNCHRONOUS))
+ .flatMap(pair -> executeListener(event, pair.getT2(), pair.getT1()))
.then();
}
- private void executeListener(Event event, MailboxListener mailboxListener, RegistrationKey registrationKey) throws Exception {
- mailboxListenerExecutor.execute(mailboxListener,
- MDCBuilder.create()
- .addContext(EventBus.StructuredLoggingFields.REGISTRATION_KEY, registrationKey),
- event);
+ private Mono<Void> executeListener(Event event, MailboxListener mailboxListener, RegistrationKey registrationKey) {
+ return Mono.from((sink) -> {
+ try {
+ mailboxListenerExecutor.execute(mailboxListener,
+ MDCBuilder.create()
+ .addContext(EventBus.StructuredLoggingFields.REGISTRATION_KEY, registrationKey),
+ event);
+ } catch (Exception e) {
+ structuredLogger(event, ImmutableSet.of(registrationKey))
+ .log(logger -> logger.error("Exception happens when dispatching event", e));
+ }
+ sink.onComplete();
+ });
+
}
private StructuredLogger structuredLogger(Event event, Set<RegistrationKey> keys) {
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 07/12: JAMES-2659 implement a simpler API for
MailboxListenerRegistry
Posted by 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 c038bc28ab2fe73cfbaa16a119fb681beebe895d
Author: Matthieu Baechler <ma...@apache.org>
AuthorDate: Fri Feb 8 13:40:43 2019 +0100
JAMES-2659 implement a simpler API for MailboxListenerRegistry
---
.../mailbox/events/KeyRegistrationHandler.java | 16 +--
.../mailbox/events/MailboxListenerRegistry.java | 47 ++++++--
.../events/MailboxListenerRegistryTest.java | 122 ++++++++++-----------
3 files changed, 106 insertions(+), 79 deletions(-)
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 f0e2acf..dac9d93 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
@@ -99,13 +99,15 @@ class KeyRegistrationHandler {
}
Registration register(MailboxListener listener, RegistrationKey key) {
- Runnable bindIfEmpty = () -> registrationBinder.bind(key).block();
- Runnable unbindIfEmpty = () -> registrationBinder.unbind(key).block();
- Runnable unregister = () -> mailboxListenerRegistry.removeListener(key, listener, unbindIfEmpty);
-
- KeyRegistration keyRegistration = new KeyRegistration(unregister);
- mailboxListenerRegistry.addListener(key, listener, bindIfEmpty);
- return keyRegistration;
+ MailboxListenerRegistry.Registration registration = mailboxListenerRegistry.addListener(key, listener);
+ if (registration.isFirstListener()) {
+ registrationBinder.bind(key).block();
+ }
+ return new KeyRegistration(() -> {
+ if (registration.unregister().lastListenerRemoved()) {
+ registrationBinder.unbind(key).block();
+ }
+ });
}
private Mono<Void> handleDelivery(Delivery delivery) {
diff --git a/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/MailboxListenerRegistry.java b/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/MailboxListenerRegistry.java
index cd5cc52..35ebf3d 100644
--- a/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/MailboxListenerRegistry.java
+++ b/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/MailboxListenerRegistry.java
@@ -23,44 +23,77 @@ import static com.google.common.base.Predicates.not;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
import com.github.steveash.guavate.Guavate;
import com.google.common.collect.ImmutableSet;
import reactor.core.publisher.Flux;
class MailboxListenerRegistry {
+
+ interface RemovalStatus {
+ boolean lastListenerRemoved();
+ }
+
+ public static class Registration {
+ private final boolean firstListener;
+ private final Supplier<RemovalStatus> unregister;
+
+ public Registration(boolean firstListener, Supplier<RemovalStatus> unregister) {
+ this.firstListener = firstListener;
+ this.unregister = unregister;
+ }
+
+ public boolean isFirstListener() {
+ return firstListener;
+ }
+
+ public RemovalStatus unregister() {
+ return unregister.get();
+ }
+ }
+
private final ConcurrentHashMap<RegistrationKey, ImmutableSet<MailboxListener>> listenersByKey;
MailboxListenerRegistry() {
this.listenersByKey = new ConcurrentHashMap<>();
}
- void addListener(RegistrationKey registrationKey, MailboxListener listener, Runnable runIfEmpty) {
+ Registration addListener(RegistrationKey registrationKey, MailboxListener listener) {
+ AtomicBoolean firstListener = new AtomicBoolean(false);
listenersByKey.compute(registrationKey, (key, listeners) ->
Optional.ofNullable(listeners)
.map(set -> ImmutableSet.<MailboxListener>builder().addAll(set).add(listener).build())
.orElseGet(() -> {
- runIfEmpty.run();
+ firstListener.set(true);
return ImmutableSet.of(listener);
})
);
+ return new Registration(firstListener.get(), () -> removeListener(registrationKey, listener));
}
- void removeListener(RegistrationKey registrationKey, MailboxListener listener, Runnable runIfEmpty) {
+ private RemovalStatus removeListener(RegistrationKey registrationKey, MailboxListener listener) {
+ AtomicBoolean lastListenerRemoved = new AtomicBoolean(false);
listenersByKey.compute(registrationKey, (key, listeners) -> {
boolean listenersContainRequested = Optional.ofNullable(listeners).orElse(ImmutableSet.of()).contains(listener);
if (listenersContainRequested) {
- return removeListenerFromSet(listener, runIfEmpty, listeners);
+ ImmutableSet<MailboxListener> remainingListeners = removeListenerFromSet(listener, listeners);
+ if (remainingListeners.isEmpty()) {
+ lastListenerRemoved.set(true);
+ return null;
+ }
+ return remainingListeners;
}
return listeners;
});
+ return lastListenerRemoved::get;
}
- private ImmutableSet<MailboxListener> removeListenerFromSet(MailboxListener listener, Runnable runIfEmpty, ImmutableSet<MailboxListener> listeners) {
+ private ImmutableSet<MailboxListener> removeListenerFromSet(MailboxListener listener, ImmutableSet<MailboxListener> listeners) {
ImmutableSet<MailboxListener> remainingListeners = listeners.stream().filter(not(listener::equals)).collect(Guavate.toImmutableSet());
if (remainingListeners.isEmpty()) {
- runIfEmpty.run();
- return null;
+ return ImmutableSet.of();
}
return remainingListeners;
}
diff --git a/mailbox/event/event-rabbitmq/src/test/java/org/apache/james/mailbox/events/MailboxListenerRegistryTest.java b/mailbox/event/event-rabbitmq/src/test/java/org/apache/james/mailbox/events/MailboxListenerRegistryTest.java
index 6320274..2d732dc 100644
--- a/mailbox/event/event-rabbitmq/src/test/java/org/apache/james/mailbox/events/MailboxListenerRegistryTest.java
+++ b/mailbox/event/event-rabbitmq/src/test/java/org/apache/james/mailbox/events/MailboxListenerRegistryTest.java
@@ -23,7 +23,6 @@ import static org.assertj.core.api.Assertions.assertThat;
import java.time.Duration;
import java.util.List;
-import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.james.mailbox.model.TestId;
@@ -37,8 +36,6 @@ import reactor.core.scheduler.Schedulers;
class MailboxListenerRegistryTest {
private static final MailboxIdRegistrationKey KEY_1 = new MailboxIdRegistrationKey(TestId.of(42));
- private static final Runnable NOOP = () -> {
- };
private MailboxListenerRegistry testee;
@@ -56,7 +53,7 @@ class MailboxListenerRegistryTest {
@Test
void getLocalMailboxListenersShouldReturnPreviouslyAddedListener() {
MailboxListener listener = event -> {};
- testee.addListener(KEY_1, listener, NOOP);
+ testee.addListener(KEY_1, listener);
assertThat(testee.getLocalMailboxListeners(KEY_1).collectList().block())
.containsOnly(listener);
@@ -66,8 +63,8 @@ class MailboxListenerRegistryTest {
void getLocalMailboxListenersShouldReturnPreviouslyAddedListeners() {
MailboxListener listener1 = event -> {};
MailboxListener listener2 = event -> {};
- testee.addListener(KEY_1, listener1, NOOP);
- testee.addListener(KEY_1, listener2, NOOP);
+ testee.addListener(KEY_1, listener1);
+ testee.addListener(KEY_1, listener2);
assertThat(testee.getLocalMailboxListeners(KEY_1).collectList().block())
.containsOnly(listener1, listener2);
@@ -77,68 +74,51 @@ class MailboxListenerRegistryTest {
void getLocalMailboxListenersShouldNotReturnRemovedListeners() {
MailboxListener listener1 = event -> {};
MailboxListener listener2 = event -> {};
- testee.addListener(KEY_1, listener1, NOOP);
- testee.addListener(KEY_1, listener2, NOOP);
+ testee.addListener(KEY_1, listener1);
+ MailboxListenerRegistry.Registration registration = testee.addListener(KEY_1, listener2);
- testee.removeListener(KEY_1, listener2, NOOP);
+ registration.unregister();
assertThat(testee.getLocalMailboxListeners(KEY_1).collectList().block())
.containsOnly(listener1);
}
@Test
- void addListenerShouldRunTaskWhenNoPreviouslyRegisteredListeners() {
+ void addListenerShouldReturnFirstListenerWhenNoPreviouslyRegisteredListeners() {
MailboxListener listener = event -> {};
- AtomicBoolean atomicBoolean = new AtomicBoolean(false);
- testee.addListener(KEY_1, listener, () -> atomicBoolean.set(true));
-
- assertThat(atomicBoolean).isTrue();
- }
-
- @Test
- void addListenerShouldNotRunTaskWhenPreviouslyRegisteredListeners() {
- MailboxListener listener = event -> {};
-
- AtomicBoolean atomicBoolean = new AtomicBoolean(false);
- testee.addListener(KEY_1, listener, NOOP);
- testee.addListener(KEY_1, listener, () -> atomicBoolean.set(true));
-
- assertThat(atomicBoolean).isFalse();
+ assertThat(testee.addListener(KEY_1, listener).isFirstListener()).isTrue();
}
@Test
- void removeListenerShouldNotRunTaskWhenNoListener() {
+ void addListenerShouldNotReturnFirstListenerWhenPreviouslyRegisteredListeners() {
MailboxListener listener = event -> {};
+ MailboxListener listener2 = event -> {};
- AtomicBoolean atomicBoolean = new AtomicBoolean(false);
- testee.removeListener(KEY_1, listener, () -> atomicBoolean.set(true));
+ testee.addListener(KEY_1, listener);
- assertThat(atomicBoolean).isFalse();
+ assertThat(testee.addListener(KEY_1, listener2).isFirstListener()).isFalse();
}
@Test
- void removeListenerShouldNotRunTaskWhenSeveralListener() {
+ void removeListenerShouldNotReturnLastListenerRemovedWhenSeveralListener() {
MailboxListener listener = event -> {};
MailboxListener listener2 = event -> {};
- AtomicBoolean atomicBoolean = new AtomicBoolean(false);
- testee.addListener(KEY_1, listener, NOOP);
- testee.addListener(KEY_1, listener2, NOOP);
- testee.removeListener(KEY_1, listener, () -> atomicBoolean.set(true));
+ MailboxListenerRegistry.Registration registration = testee.addListener(KEY_1, listener);
+ testee.addListener(KEY_1, listener2);
- assertThat(atomicBoolean).isFalse();
+ assertThat(registration.unregister().lastListenerRemoved()).isFalse();
}
@Test
- void removeListenerShouldRunTaskWhenOneListener() {
+ void removeListenerShouldReturnLastListenerRemovedWhenOneListener() {
MailboxListener listener = event -> {};
- AtomicBoolean atomicBoolean = new AtomicBoolean(false);
- testee.addListener(KEY_1, listener, NOOP);
- testee.removeListener(KEY_1, listener, () -> atomicBoolean.set(true));
- assertThat(atomicBoolean).isTrue();
+ MailboxListenerRegistry.Registration registration = testee.addListener(KEY_1, listener);
+
+ assertThat(registration.unregister().lastListenerRemoved()).isTrue();
}
@Nested
@@ -150,7 +130,7 @@ class MailboxListenerRegistryTest {
MailboxListener listener = event -> {};
ConcurrentTestRunner.builder()
- .operation((threadNumber, operationNumber) -> testee.addListener(KEY_1, listener, NOOP))
+ .operation((threadNumber, operationNumber) -> testee.addListener(KEY_1, listener))
.threadCount(10)
.operationCount(10)
.runSuccessfullyWithin(ONE_SECOND);
@@ -168,11 +148,11 @@ class MailboxListenerRegistryTest {
ConcurrentTestRunner.builder()
.operation((threadNumber, operationNumber) -> {
if (threadNumber % 3 == 0) {
- testee.addListener(KEY_1, listener1, NOOP);
+ testee.addListener(KEY_1, listener1);
} else if (threadNumber % 3 == 1) {
- testee.addListener(KEY_1, listener2, NOOP);
+ testee.addListener(KEY_1, listener2);
} else if (threadNumber % 3 == 2) {
- testee.addListener(KEY_1, listener3, NOOP);
+ testee.addListener(KEY_1, listener3);
}
})
.threadCount(6)
@@ -187,11 +167,10 @@ class MailboxListenerRegistryTest {
void getLocalMailboxListenersShouldReturnEmptyWhenRemoveAddedListener() throws Exception {
MailboxListener listener1 = event -> {};
- testee.addListener(KEY_1, listener1, NOOP);
+ MailboxListenerRegistry.Registration registration = testee.addListener(KEY_1, listener1);
ConcurrentTestRunner.builder()
- .operation(((threadNumber, operationNumber) ->
- testee.removeListener(KEY_1, listener1, NOOP)))
+ .operation(((threadNumber, operationNumber) -> registration.unregister()))
.threadCount(10)
.operationCount(10)
.runSuccessfullyWithin(ONE_SECOND);
@@ -201,65 +180,78 @@ class MailboxListenerRegistryTest {
}
@Test
- void addListenerOnlyRunTaskOnceForEmptyRegistry() throws Exception {
+ void addListenerOnlyReturnIsFirstListenerForEmptyRegistry() throws Exception {
MailboxListener listener1 = event -> {};
MailboxListener listener2 = event -> {};
MailboxListener listener3 = event -> {};
- AtomicInteger runIfEmptyCount = new AtomicInteger(0);
+ AtomicInteger firstListenerCount = new AtomicInteger(0);
ConcurrentTestRunner.builder()
.operation((threadNumber, operationNumber) -> {
if (threadNumber % 3 == 0) {
- testee.addListener(KEY_1, listener1, runIfEmptyCount::incrementAndGet);
+ MailboxListenerRegistry.Registration registration = testee.addListener(KEY_1, listener1);
+ if (registration.isFirstListener()) {
+ firstListenerCount.incrementAndGet();
+ }
} else if (threadNumber % 3 == 1) {
- testee.addListener(KEY_1, listener2, runIfEmptyCount::incrementAndGet);
+ MailboxListenerRegistry.Registration registration = testee.addListener(KEY_1, listener2);
+ if (registration.isFirstListener()) {
+ firstListenerCount.incrementAndGet();
+ }
} else if (threadNumber % 3 == 2) {
- testee.addListener(KEY_1, listener3, runIfEmptyCount::incrementAndGet);
+ MailboxListenerRegistry.Registration registration = testee.addListener(KEY_1, listener3);
+ if (registration.isFirstListener()) {
+ firstListenerCount.incrementAndGet();
+ }
}
})
.threadCount(6)
.operationCount(10)
.runSuccessfullyWithin(ONE_SECOND);
- assertThat(runIfEmptyCount.get()).isEqualTo(1);
+ assertThat(firstListenerCount.get()).isEqualTo(1);
}
@Test
- void removeListenerOnlyRunTaskOnceForEmptyRegistry() throws Exception {
+ void removeListenerOnlyReturnLastListenerRemovedForEmptyRegistry() throws Exception {
MailboxListener listener1 = event -> {};
- AtomicInteger runIfEmptyCount = new AtomicInteger(0);
+ AtomicInteger lastListenerRemoved = new AtomicInteger(0);
- testee.addListener(KEY_1, listener1, NOOP);
+ MailboxListenerRegistry.Registration registration = testee.addListener(KEY_1, listener1);
ConcurrentTestRunner.builder()
- .operation(((threadNumber, operationNumber) -> testee.removeListener(KEY_1, listener1, runIfEmptyCount::incrementAndGet)))
+ .operation(((threadNumber, operationNumber) -> {
+ if (registration.unregister().lastListenerRemoved()) {
+ lastListenerRemoved.incrementAndGet();
+ }
+ }))
.threadCount(10)
.operationCount(10)
.runSuccessfullyWithin(ONE_SECOND);
- assertThat(runIfEmptyCount.get()).isEqualTo(1);
+ assertThat(lastListenerRemoved.get()).isEqualTo(1);
}
@Test
- void iterationShouldPerformOnASnapshotOfListenersSet() throws Exception {
+ void iterationShouldPerformOnASnapshotOfListenersSet() {
MailboxListener listener1 = event -> {};
MailboxListener listener2 = event -> {};
MailboxListener listener3 = event -> {};
MailboxListener listener4 = event -> {};
MailboxListener listener5 = event -> {};
- testee.addListener(KEY_1, listener1, NOOP);
- testee.addListener(KEY_1, listener2, NOOP);
- testee.addListener(KEY_1, listener3, NOOP);
- testee.addListener(KEY_1, listener4, NOOP);
- testee.addListener(KEY_1, listener5, NOOP);
+ testee.addListener(KEY_1, listener1);
+ testee.addListener(KEY_1, listener2);
+ testee.addListener(KEY_1, listener3);
+ testee.addListener(KEY_1, listener4);
+ MailboxListenerRegistry.Registration registration5 = testee.addListener(KEY_1, listener5);
Mono<List<MailboxListener>> listeners = testee.getLocalMailboxListeners(KEY_1)
.publishOn(Schedulers.elastic())
.delayElements(Duration.ofMillis(100))
.collectList();
- testee.removeListener(KEY_1, listener5, NOOP);
+ registration5.unregister();
assertThat(listeners.block(Duration.ofSeconds(10))).hasSize(5);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 04/12: JAMES-2659 remove needless complexity:
subscribeWith(MonoProcessor) triggers an eager computation and replay the
computed value like .cache
Posted by 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 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
[james-project] 12/12: JAMES-2632 Reading metadata is enough when
moving messages by messageIds
Posted by 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 311860722d68a13c882a2a6cefc890f32dfba224
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Mon Feb 11 14:50:32 2019 +0700
JAMES-2632 Reading metadata is enough when moving messages by messageIds
The full message is no more shipped in the emitted events so we can lower
the level of reads.
---
.../java/org/apache/james/mailbox/store/StoreMessageIdManager.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreMessageIdManager.java b/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreMessageIdManager.java
index 55a82fe..3b0b9f2 100644
--- a/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreMessageIdManager.java
+++ b/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreMessageIdManager.java
@@ -258,7 +258,7 @@ public class StoreMessageIdManager implements MessageIdManager {
public void setInMailboxesNoCheck(MessageId messageId, MailboxId targetMailboxId, MailboxSession mailboxSession) throws MailboxException {
MessageIdMapper messageIdMapper = mailboxSessionMapperFactory.getMessageIdMapper(mailboxSession);
- List<MailboxMessage> currentMailboxMessages = messageIdMapper.find(ImmutableList.of(messageId), MessageMapper.FetchType.Full);
+ List<MailboxMessage> currentMailboxMessages = messageIdMapper.find(ImmutableList.of(messageId), MessageMapper.FetchType.Metadata);
MessageMoves messageMoves = MessageMoves.builder()
.previousMailboxIds(toMailboxIds(currentMailboxMessages))
@@ -273,7 +273,7 @@ public class StoreMessageIdManager implements MessageIdManager {
private List<MailboxMessage> findRelatedMailboxMessages(MessageId messageId, MailboxSession mailboxSession) throws MailboxException {
MessageIdMapper messageIdMapper = mailboxSessionMapperFactory.getMessageIdMapper(mailboxSession);
- return messageIdMapper.find(ImmutableList.of(messageId), MessageMapper.FetchType.Full)
+ return messageIdMapper.find(ImmutableList.of(messageId), MessageMapper.FetchType.Metadata)
.stream()
.filter(hasRightsOn(mailboxSession, Right.Read))
.collect(Guavate.toImmutableList());
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 03/12: JAMES-2659 checkstyle fix
Posted by 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 ee538d9f904b2854585e80622c4ad6d6f4cac65b
Author: Matthieu Baechler <ma...@apache.org>
AuthorDate: Fri Feb 8 13:47:10 2019 +0100
JAMES-2659 checkstyle fix
---
.../test/java/org/apache/james/mailbox/events/RabbitMQEventBusTest.java | 1 -
1 file changed, 1 deletion(-)
diff --git a/mailbox/event/event-rabbitmq/src/test/java/org/apache/james/mailbox/events/RabbitMQEventBusTest.java b/mailbox/event/event-rabbitmq/src/test/java/org/apache/james/mailbox/events/RabbitMQEventBusTest.java
index 1dacb5f..c1150c1 100644
--- a/mailbox/event/event-rabbitmq/src/test/java/org/apache/james/mailbox/events/RabbitMQEventBusTest.java
+++ b/mailbox/event/event-rabbitmq/src/test/java/org/apache/james/mailbox/events/RabbitMQEventBusTest.java
@@ -66,7 +66,6 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Nested;
-import org.junit.jupiter.api.RepeatedTest;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
import org.mockito.stubbing.Answer;
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 06/12: JAMES-2659 Implement MailboxListenerRegistry
on top of a ConcurrentHashMap to limit lock-contention and optimize for
iteration rather than addition/removal
Posted by 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 a29f84abfd0ad19fa523456567cbae268df36646
Author: Matthieu Baechler <ma...@apache.org>
AuthorDate: Fri Feb 8 11:45:30 2019 +0100
JAMES-2659 Implement MailboxListenerRegistry on top of a ConcurrentHashMap to limit lock-contention and optimize for iteration rather than addition/removal
---
.../mailbox/events/MailboxListenerRegistry.java | 50 ++++++++++++++--------
1 file changed, 31 insertions(+), 19 deletions(-)
diff --git a/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/MailboxListenerRegistry.java b/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/MailboxListenerRegistry.java
index 4b440db..cd5cc52 100644
--- a/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/MailboxListenerRegistry.java
+++ b/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/MailboxListenerRegistry.java
@@ -19,41 +19,53 @@
package org.apache.james.mailbox.events;
-import com.google.common.collect.HashMultimap;
-import com.google.common.collect.ImmutableSet;
-import com.google.common.collect.Multimap;
-import com.google.common.collect.Multimaps;
+import static com.google.common.base.Predicates.not;
+
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+import com.github.steveash.guavate.Guavate;
+import com.google.common.collect.ImmutableSet;
import reactor.core.publisher.Flux;
class MailboxListenerRegistry {
- private final Multimap<RegistrationKey, MailboxListener> listeners;
+ private final ConcurrentHashMap<RegistrationKey, ImmutableSet<MailboxListener>> listenersByKey;
MailboxListenerRegistry() {
- this.listeners = Multimaps.synchronizedMultimap(HashMultimap.create());
+ this.listenersByKey = new ConcurrentHashMap<>();
}
void addListener(RegistrationKey registrationKey, MailboxListener listener, Runnable runIfEmpty) {
- synchronized (listeners) {
- if (listeners.get(registrationKey).isEmpty()) {
- runIfEmpty.run();
- }
- listeners.put(registrationKey, listener);
- }
+ listenersByKey.compute(registrationKey, (key, listeners) ->
+ Optional.ofNullable(listeners)
+ .map(set -> ImmutableSet.<MailboxListener>builder().addAll(set).add(listener).build())
+ .orElseGet(() -> {
+ runIfEmpty.run();
+ return ImmutableSet.of(listener);
+ })
+ );
}
void removeListener(RegistrationKey registrationKey, MailboxListener listener, Runnable runIfEmpty) {
- synchronized (listeners) {
- boolean wasRemoved = listeners.remove(registrationKey, listener);
- if (wasRemoved && listeners.get(registrationKey).isEmpty()) {
- runIfEmpty.run();
+ listenersByKey.compute(registrationKey, (key, listeners) -> {
+ boolean listenersContainRequested = Optional.ofNullable(listeners).orElse(ImmutableSet.of()).contains(listener);
+ if (listenersContainRequested) {
+ return removeListenerFromSet(listener, runIfEmpty, listeners);
}
+ return listeners;
+ });
+ }
+
+ private ImmutableSet<MailboxListener> removeListenerFromSet(MailboxListener listener, Runnable runIfEmpty, ImmutableSet<MailboxListener> listeners) {
+ ImmutableSet<MailboxListener> remainingListeners = listeners.stream().filter(not(listener::equals)).collect(Guavate.toImmutableSet());
+ if (remainingListeners.isEmpty()) {
+ runIfEmpty.run();
+ return null;
}
+ return remainingListeners;
}
Flux<MailboxListener> getLocalMailboxListeners(RegistrationKey registrationKey) {
- synchronized (listeners) {
- return Flux.fromIterable(ImmutableSet.copyOf(listeners.get(registrationKey)));
- }
+ return Flux.fromIterable(listenersByKey.getOrDefault(registrationKey, ImmutableSet.of()));
}
}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org