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 rc...@apache.org on 2020/05/13 02:03:21 UTC
[james-project] branch master updated (0018e5c -> 88d05d9)
This is an automated email from the ASF dual-hosted git repository.
rcordier pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git.
from 0018e5c Merge pull request #189 from dleangen/documentation
new d3ae4a1 JAMES-3149 ReactorElasticSearchClient should not call subscribeOn
new f2fbe4f JAMES-3149 EventBus caller should choose on which scheduler he dispatch
new 7795d0b JAMES-3149 CassandraMailQueueBrowser caller should choose on which scheduler he dispatch
new e694f39 [Refactoring] Extract a ReactorUtils::unboxOptional utils
new a4f75c2 [Refactoring] Reactor: use handle for nullable transformations
new db157e6 [Refactoring] Reactor: extract .then() call
new 9215920 [Refactoring] Rely on publishIfPresent when mapping null values
new c01bf3a JAMES-3091 JSON serialization for Mailbox POJO
new 88d05d9 JAMES-3091 Mailbox/Get POJO
The 9 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:
.../cassandra/utils/CassandraAsyncExecutor.java | 4 +-
.../backends/es/ReactorElasticSearchClient.java | 2 +-
.../backends/rabbitmq/SimpleConnectionPool.java | 5 +-
.../apache/james/mailbox/events/GroupContract.java | 5 +-
.../cassandra/mail/CassandraMessageMapper.java | 4 +-
.../cassandra/mail/CassandraModSeqProvider.java | 8 +-
.../cassandra/mail/CassandraUidProvider.java | 7 +-
.../task/SolveMessageInconsistenciesService.java | 6 +-
.../quota/CassandraGlobalMaxQuotaDao.java | 15 ++-
.../quota/CassandraPerDomainMaxQuotaDao.java | 15 ++-
.../quota/CassandraPerUserMaxQuotaDao.java | 15 ++-
.../ElasticSearchListeningMessageSearchIndex.java | 3 +-
.../james/mailbox/events/EventDispatcher.java | 7 +-
.../apache/james/vault/metadata/MetadataDAO.java | 4 +-
.../james/mailbox/store/StoreMailboxManager.java | 49 ++++----
.../james/mailbox/store/StoreMessageIdManager.java | 28 +++--
.../james/mailbox/store/StoreMessageManager.java | 28 +++--
.../james/mailbox/store/StoreRightManager.java | 3 +
.../java/org/apache/james/util/ReactorUtils.java | 5 +
.../sieve/cassandra/CassandraSieveRepository.java | 3 +-
.../cassandra/CassandraMailRepository.java | 4 +-
.../jmap/draft/methods/GetMailboxesMethod.java | 8 +-
server/protocols/jmap-rfc-8621/pom.xml | 5 +
.../org/apache/james/jmap/json/Serializer.scala | 73 ++++++++++-
.../scala/org/apache/james/jmap/mail/Mailbox.scala | 2 +-
.../org/apache/james/jmap/mail/MailboxGet.scala | 36 +++---
.../scala/org/apache/james/jmap/mail/Quotas.scala | 19 ++-
.../james/jmap/json/MailboxSerializationTest.scala | 135 +++++++++++++++++++++
.../org/apache/james/jmap/mail/MailboxTest.scala | 12 +-
.../view/cassandra/CassandraMailQueueBrowser.java | 4 +-
.../view/cassandra/CassandraMailQueueView.java | 8 +-
.../EventsourcingConfigurationManagement.java | 7 +-
.../distributed/RabbitMQTerminationSubscriber.java | 5 +-
.../apache/james/task/SerialTaskManagerWorker.java | 4 +-
34 files changed, 415 insertions(+), 123 deletions(-)
copy backends-common/elasticsearch/src/test/java/org/apache/james/backends/es/RoutingKeyTest.java => server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/mail/MailboxGet.scala (62%)
create mode 100644 server/protocols/jmap-rfc-8621/src/test/scala/org/apache/james/jmap/json/MailboxSerializationTest.scala
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 04/09: [Refactoring] Extract a
ReactorUtils::unboxOptional utils
Posted by rc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rcordier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit e694f392989540a416765c224bee028327651eba
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Mon May 4 10:36:06 2020 +0700
[Refactoring] Extract a ReactorUtils::unboxOptional utils
This reduces the boiler plate around the use of handle, which
optimizes Reactor synchronous transformations.
---
.../james/backends/cassandra/utils/CassandraAsyncExecutor.java | 4 +++-
.../james/mailbox/cassandra/mail/CassandraMessageMapper.java | 4 +++-
.../james/mailbox/cassandra/mail/CassandraModSeqProvider.java | 8 +++++---
.../apache/james/mailbox/cassandra/mail/CassandraUidProvider.java | 7 +++++--
.../cassandra/mail/task/SolveMessageInconsistenciesService.java | 6 ++++--
.../james/mailbox/cassandra/quota/CassandraGlobalMaxQuotaDao.java | 7 +++++--
.../mailbox/cassandra/quota/CassandraPerDomainMaxQuotaDao.java | 7 +++++--
.../mailbox/cassandra/quota/CassandraPerUserMaxQuotaDao.java | 7 +++++--
.../events/ElasticSearchListeningMessageSearchIndex.java | 3 ++-
.../main/java/org/apache/james/vault/metadata/MetadataDAO.java | 4 +++-
.../util/src/main/java/org/apache/james/util/ReactorUtils.java | 5 +++++
.../james/mailrepository/cassandra/CassandraMailRepository.java | 4 +++-
.../org/apache/james/jmap/draft/methods/GetMailboxesMethod.java | 8 +++++---
.../configuration/EventsourcingConfigurationManagement.java | 7 +++++--
.../eventsourcing/distributed/RabbitMQTerminationSubscriber.java | 5 ++++-
.../main/java/org/apache/james/task/SerialTaskManagerWorker.java | 4 +++-
16 files changed, 65 insertions(+), 25 deletions(-)
diff --git a/backends-common/cassandra/src/main/java/org/apache/james/backends/cassandra/utils/CassandraAsyncExecutor.java b/backends-common/cassandra/src/main/java/org/apache/james/backends/cassandra/utils/CassandraAsyncExecutor.java
index 86b192b..24bf48e 100644
--- a/backends-common/cassandra/src/main/java/org/apache/james/backends/cassandra/utils/CassandraAsyncExecutor.java
+++ b/backends-common/cassandra/src/main/java/org/apache/james/backends/cassandra/utils/CassandraAsyncExecutor.java
@@ -19,6 +19,8 @@
package org.apache.james.backends.cassandra.utils;
+import static org.apache.james.util.ReactorUtils.publishIfPresent;
+
import java.util.Optional;
import javax.inject.Inject;
@@ -66,7 +68,7 @@ public class CassandraAsyncExecutor {
public Mono<Row> executeSingleRow(Statement statement) {
return executeSingleRowOptional(statement)
- .handle((t, sink) -> t.ifPresent(sink::next));
+ .handle(publishIfPresent());
}
public Flux<Row> executeRows(Statement statement) {
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapper.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapper.java
index 465c9cf..cb31434 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapper.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapper.java
@@ -19,6 +19,8 @@
package org.apache.james.mailbox.cassandra.mail;
+import static org.apache.james.util.ReactorUtils.publishIfPresent;
+
import java.time.Duration;
import java.util.Comparator;
import java.util.Iterator;
@@ -312,7 +314,7 @@ public class CassandraMessageMapper implements MessageMapper {
if (!failed.isEmpty()) {
Flux<ComposedMessageIdWithMetaData> toUpdate = Flux.fromIterable(failed)
.flatMap(uid -> messageIdDAO.retrieve(mailboxId, uid))
- .handle((t, sink) -> t.ifPresent(sink::next));
+ .handle(publishIfPresent());
return runUpdateStage(mailboxId, toUpdate, flagsUpdateCalculator);
} else {
return Mono.empty();
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraModSeqProvider.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraModSeqProvider.java
index 15f1a2d..fc1647c 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraModSeqProvider.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraModSeqProvider.java
@@ -28,6 +28,7 @@ import static com.datastax.driver.core.querybuilder.QueryBuilder.update;
import static org.apache.james.mailbox.cassandra.table.CassandraMessageModseqTable.MAILBOX_ID;
import static org.apache.james.mailbox.cassandra.table.CassandraMessageModseqTable.NEXT_MODSEQ;
import static org.apache.james.mailbox.cassandra.table.CassandraMessageModseqTable.TABLE_NAME;
+import static org.apache.james.util.ReactorUtils.publishIfPresent;
import java.time.Duration;
import java.util.Optional;
@@ -157,7 +158,8 @@ public class CassandraModSeqProvider implements ModSeqProvider {
insert.bind()
.setUUID(MAILBOX_ID, mailboxId.asUuid())
.setLong(NEXT_MODSEQ, nextModSeq.asLong()))
- .handle((success, sink) -> successToModSeq(nextModSeq, success).ifPresent(sink::next));
+ .map(success -> successToModSeq(nextModSeq, success))
+ .handle(publishIfPresent());
}
private Mono<ModSeq> tryUpdateModSeq(CassandraId mailboxId, ModSeq modSeq) {
@@ -167,7 +169,8 @@ public class CassandraModSeqProvider implements ModSeqProvider {
.setUUID(MAILBOX_ID, mailboxId.asUuid())
.setLong(NEXT_MODSEQ, nextModSeq.asLong())
.setLong(MOD_SEQ_CONDITION, modSeq.asLong()))
- .handle((success, sink) -> successToModSeq(nextModSeq, success).ifPresent(sink::next));
+ .map(success -> successToModSeq(nextModSeq, success))
+ .handle(publishIfPresent());
}
private Optional<ModSeq> successToModSeq(ModSeq modSeq, Boolean success) {
@@ -187,5 +190,4 @@ public class CassandraModSeqProvider implements ModSeqProvider {
.single()
.retryWhen(Retry.backoff(maxModSeqRetries, firstBackoff).scheduler(Schedulers.elastic()));
}
-
}
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraUidProvider.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraUidProvider.java
index fea0d5e..86fa061 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraUidProvider.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraUidProvider.java
@@ -28,6 +28,7 @@ import static com.datastax.driver.core.querybuilder.QueryBuilder.update;
import static org.apache.james.mailbox.cassandra.table.CassandraMessageUidTable.MAILBOX_ID;
import static org.apache.james.mailbox.cassandra.table.CassandraMessageUidTable.NEXT_UID;
import static org.apache.james.mailbox.cassandra.table.CassandraMessageUidTable.TABLE_NAME;
+import static org.apache.james.util.ReactorUtils.publishIfPresent;
import java.time.Duration;
import java.util.Optional;
@@ -135,14 +136,16 @@ public class CassandraUidProvider implements UidProvider {
.setUUID(MAILBOX_ID, mailboxId.asUuid())
.setLong(CONDITION, uid.asLong())
.setLong(NEXT_UID, nextUid.asLong()))
- .handle((success, sink) -> successToUid(nextUid, success).ifPresent(sink::next));
+ .map(success -> successToUid(nextUid, success))
+ .handle(publishIfPresent());
}
private Mono<MessageUid> tryInsert(CassandraId mailboxId) {
return executor.executeReturnApplied(
insertStatement.bind()
.setUUID(MAILBOX_ID, mailboxId.asUuid()))
- .handle((success, sink) -> successToUid(MessageUid.MIN_VALUE, success).ifPresent(sink::next));
+ .map(success -> successToUid(MessageUid.MIN_VALUE, success))
+ .handle(publishIfPresent());
}
private Optional<MessageUid> successToUid(MessageUid uid, Boolean success) {
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/task/SolveMessageInconsistenciesService.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/task/SolveMessageInconsistenciesService.java
index 8518f1e..f0ec09b 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/task/SolveMessageInconsistenciesService.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/task/SolveMessageInconsistenciesService.java
@@ -19,6 +19,8 @@
package org.apache.james.mailbox.cassandra.mail.task;
+import static org.apache.james.util.ReactorUtils.publishIfPresent;
+
import java.util.Collection;
import java.util.Objects;
import java.util.Optional;
@@ -414,7 +416,7 @@ public class SolveMessageInconsistenciesService {
private Mono<Inconsistency> compareWithMessageIdRecord(ComposedMessageIdWithMetaData upToDateMessageFromImapUid) {
return messageIdDAO.retrieve((CassandraId) upToDateMessageFromImapUid.getComposedMessageId().getMailboxId(), upToDateMessageFromImapUid.getComposedMessageId().getUid())
- .flatMap(Mono::justOrEmpty)
+ .handle(publishIfPresent())
.map(messageIdRecord -> {
if (messageIdRecord.equals(upToDateMessageFromImapUid)) {
return NO_INCONSISTENCY;
@@ -433,7 +435,7 @@ public class SolveMessageInconsistenciesService {
private Mono<Inconsistency> detectInconsistencyInMessageId(ComposedMessageIdWithMetaData message) {
return messageIdDAO.retrieve((CassandraId) message.getComposedMessageId().getMailboxId(), message.getComposedMessageId().getUid())
- .flatMap(Mono::justOrEmpty)
+ .handle(publishIfPresent())
.flatMap(upToDateMessage -> messageIdToImapUidDAO.retrieve((CassandraMessageId) message.getComposedMessageId().getMessageId(), Optional.of((CassandraId) message.getComposedMessageId().getMailboxId()))
.map(uidRecord -> NO_INCONSISTENCY)
.switchIfEmpty(Mono.just(new OrphanMessageIdEntry(message)))
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraGlobalMaxQuotaDao.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraGlobalMaxQuotaDao.java
index 74fd9bf..9ee9159 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraGlobalMaxQuotaDao.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraGlobalMaxQuotaDao.java
@@ -24,6 +24,7 @@ import static com.datastax.driver.core.querybuilder.QueryBuilder.delete;
import static com.datastax.driver.core.querybuilder.QueryBuilder.eq;
import static com.datastax.driver.core.querybuilder.QueryBuilder.insertInto;
import static com.datastax.driver.core.querybuilder.QueryBuilder.select;
+import static org.apache.james.util.ReactorUtils.publishIfPresent;
import javax.inject.Inject;
@@ -93,14 +94,16 @@ public class CassandraGlobalMaxQuotaDao {
return queryExecutor.executeSingleRow(getGlobalMaxStatement.bind()
.setString(CassandraGlobalMaxQuota.TYPE, CassandraGlobalMaxQuota.STORAGE))
.flatMap(row -> Mono.justOrEmpty(row.get(CassandraGlobalMaxQuota.VALUE, Long.class)))
- .flatMap(maxStorage -> Mono.justOrEmpty(QuotaCodec.longToQuotaSize(maxStorage)));
+ .map(QuotaCodec::longToQuotaSize)
+ .handle(publishIfPresent());
}
Mono<QuotaCountLimit> getGlobalMaxMessage() {
return queryExecutor.executeSingleRow(getGlobalMaxStatement.bind()
.setString(CassandraGlobalMaxQuota.TYPE, CassandraGlobalMaxQuota.MESSAGE))
.flatMap(row -> Mono.justOrEmpty(row.get(CassandraGlobalMaxQuota.VALUE, Long.class)))
- .flatMap(maxMessages -> Mono.justOrEmpty(QuotaCodec.longToQuotaCount(maxMessages)));
+ .map(QuotaCodec::longToQuotaCount)
+ .handle(publishIfPresent());
}
Mono<Void> removeGlobaltMaxStorage() {
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraPerDomainMaxQuotaDao.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraPerDomainMaxQuotaDao.java
index cb49d3a..38161ef 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraPerDomainMaxQuotaDao.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraPerDomainMaxQuotaDao.java
@@ -24,6 +24,7 @@ import static com.datastax.driver.core.querybuilder.QueryBuilder.delete;
import static com.datastax.driver.core.querybuilder.QueryBuilder.eq;
import static com.datastax.driver.core.querybuilder.QueryBuilder.insertInto;
import static com.datastax.driver.core.querybuilder.QueryBuilder.select;
+import static org.apache.james.util.ReactorUtils.publishIfPresent;
import javax.inject.Inject;
@@ -109,13 +110,15 @@ public class CassandraPerDomainMaxQuotaDao {
Mono<QuotaSizeLimit> getMaxStorage(Domain domain) {
return queryExecutor.executeSingleRow(getMaxStorageStatement.bind(domain.asString()))
.flatMap(row -> Mono.justOrEmpty(row.get(CassandraDomainMaxQuota.STORAGE, Long.class)))
- .flatMap(maxStorage -> Mono.justOrEmpty(QuotaCodec.longToQuotaSize(maxStorage)));
+ .map(QuotaCodec::longToQuotaSize)
+ .handle(publishIfPresent());
}
Mono<QuotaCountLimit> getMaxMessage(Domain domain) {
return queryExecutor.executeSingleRow(getMaxMessageStatement.bind(domain.asString()))
.flatMap(row -> Mono.justOrEmpty(row.get(CassandraDomainMaxQuota.MESSAGE_COUNT, Long.class)))
- .flatMap(maxMessages -> Mono.justOrEmpty(QuotaCodec.longToQuotaCount(maxMessages)));
+ .map(QuotaCodec::longToQuotaCount)
+ .handle(publishIfPresent());
}
Mono<Void> removeMaxMessage(Domain domain) {
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraPerUserMaxQuotaDao.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraPerUserMaxQuotaDao.java
index cff560d..2b4e21b 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraPerUserMaxQuotaDao.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraPerUserMaxQuotaDao.java
@@ -24,6 +24,7 @@ import static com.datastax.driver.core.querybuilder.QueryBuilder.delete;
import static com.datastax.driver.core.querybuilder.QueryBuilder.eq;
import static com.datastax.driver.core.querybuilder.QueryBuilder.insertInto;
import static com.datastax.driver.core.querybuilder.QueryBuilder.select;
+import static org.apache.james.util.ReactorUtils.publishIfPresent;
import javax.inject.Inject;
@@ -109,13 +110,15 @@ public class CassandraPerUserMaxQuotaDao {
Mono<QuotaSizeLimit> getMaxStorage(QuotaRoot quotaRoot) {
return queryExecutor.executeSingleRow(getMaxStorageStatement.bind(quotaRoot.getValue()))
.flatMap(row -> Mono.justOrEmpty(row.get(CassandraMaxQuota.STORAGE, Long.class)))
- .flatMap(maxStorage -> Mono.justOrEmpty(QuotaCodec.longToQuotaSize(maxStorage)));
+ .map(QuotaCodec::longToQuotaSize)
+ .handle(publishIfPresent());
}
Mono<QuotaCountLimit> getMaxMessage(QuotaRoot quotaRoot) {
return queryExecutor.executeSingleRow(getMaxMessageStatement.bind(quotaRoot.getValue()))
.flatMap(row -> Mono.justOrEmpty(row.get(CassandraMaxQuota.MESSAGE_COUNT, Long.class)))
- .flatMap(maxMessages -> Mono.justOrEmpty(QuotaCodec.longToQuotaCount(maxMessages)));
+ .map(QuotaCodec::longToQuotaCount)
+ .handle(publishIfPresent());
}
Mono<Void> removeMaxMessage(QuotaRoot quotaRoot) {
diff --git a/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/events/ElasticSearchListeningMessageSearchIndex.java b/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/events/ElasticSearchListeningMessageSearchIndex.java
index 3d2d099..091db6b 100644
--- a/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/events/ElasticSearchListeningMessageSearchIndex.java
+++ b/mailbox/elasticsearch/src/main/java/org/apache/james/mailbox/elasticsearch/events/ElasticSearchListeningMessageSearchIndex.java
@@ -18,6 +18,7 @@
****************************************************************/
package org.apache.james.mailbox.elasticsearch.events;
+import static org.apache.james.util.ReactorUtils.publishIfPresent;
import static org.elasticsearch.index.query.QueryBuilders.termQuery;
import java.util.Collection;
@@ -128,7 +129,7 @@ public class ElasticSearchListeningMessageSearchIndex extends ListeningMessageSe
return searcher.search(mailboxIds, searchQuery, Optional.empty())
.doOnNext(this::logIfNoMessageId)
.map(SearchResult::getMessageId)
- .flatMap(Mono::justOrEmpty)
+ .handle(publishIfPresent())
.distinct()
.take(limit);
}
diff --git a/mailbox/plugin/deleted-messages-vault-cassandra/src/main/java/org/apache/james/vault/metadata/MetadataDAO.java b/mailbox/plugin/deleted-messages-vault-cassandra/src/main/java/org/apache/james/vault/metadata/MetadataDAO.java
index 87553e9..f03fd4e 100644
--- a/mailbox/plugin/deleted-messages-vault-cassandra/src/main/java/org/apache/james/vault/metadata/MetadataDAO.java
+++ b/mailbox/plugin/deleted-messages-vault-cassandra/src/main/java/org/apache/james/vault/metadata/MetadataDAO.java
@@ -24,6 +24,7 @@ import static com.datastax.driver.core.querybuilder.QueryBuilder.delete;
import static com.datastax.driver.core.querybuilder.QueryBuilder.eq;
import static com.datastax.driver.core.querybuilder.QueryBuilder.insertInto;
import static com.datastax.driver.core.querybuilder.QueryBuilder.select;
+import static org.apache.james.util.ReactorUtils.publishIfPresent;
import static org.apache.james.vault.metadata.DeletedMessageMetadataModule.DeletedMessageMetadataTable.BUCKET_NAME;
import static org.apache.james.vault.metadata.DeletedMessageMetadataModule.DeletedMessageMetadataTable.MESSAGE_ID;
import static org.apache.james.vault.metadata.DeletedMessageMetadataModule.DeletedMessageMetadataTable.OWNER;
@@ -109,7 +110,8 @@ public class MetadataDAO {
.setString(BUCKET_NAME, bucketName.asString())
.setString(OWNER, username.asString()))
.map(row -> row.getString(PAYLOAD))
- .handle((json, sink) -> metadataSerializer.deserialize(json).ifPresent(sink::next));
+ .map(metadataSerializer::deserialize)
+ .handle(publishIfPresent());
}
Flux<MessageId> retrieveMessageIds(BucketName bucketName, Username username) {
diff --git a/server/container/util/src/main/java/org/apache/james/util/ReactorUtils.java b/server/container/util/src/main/java/org/apache/james/util/ReactorUtils.java
index df9c937..a477355 100644
--- a/server/container/util/src/main/java/org/apache/james/util/ReactorUtils.java
+++ b/server/container/util/src/main/java/org/apache/james/util/ReactorUtils.java
@@ -24,11 +24,13 @@ import java.io.InputStream;
import java.nio.ByteBuffer;
import java.util.Iterator;
import java.util.Optional;
+import java.util.function.BiConsumer;
import java.util.function.Consumer;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Signal;
+import reactor.core.publisher.SynchronousSink;
import reactor.util.context.Context;
public class ReactorUtils {
@@ -39,6 +41,9 @@ public class ReactorUtils {
return Mono.fromRunnable(runnable).then(Mono.empty());
}
+ public static <T> BiConsumer<Optional<T>, SynchronousSink<T>> publishIfPresent() {
+ return (element, sink) -> element.ifPresent(sink::next);
+ }
public static InputStream toInputStream(Flux<ByteBuffer> byteArrays) {
return new StreamInputStream(byteArrays.toIterable(1).iterator());
diff --git a/server/mailrepository/mailrepository-cassandra/src/main/java/org/apache/james/mailrepository/cassandra/CassandraMailRepository.java b/server/mailrepository/mailrepository-cassandra/src/main/java/org/apache/james/mailrepository/cassandra/CassandraMailRepository.java
index cf98198..134d969 100644
--- a/server/mailrepository/mailrepository-cassandra/src/main/java/org/apache/james/mailrepository/cassandra/CassandraMailRepository.java
+++ b/server/mailrepository/mailrepository-cassandra/src/main/java/org/apache/james/mailrepository/cassandra/CassandraMailRepository.java
@@ -19,6 +19,8 @@
package org.apache.james.mailrepository.cassandra;
+import static org.apache.james.util.ReactorUtils.publishIfPresent;
+
import java.util.Collection;
import java.util.Iterator;
@@ -93,7 +95,7 @@ public class CassandraMailRepository implements MailRepository {
@Override
public Mail retrieve(MailKey key) {
return mailDAO.read(url, key)
- .<MailDTO>handle((t, sink) -> t.ifPresent(sink::next))
+ .handle(publishIfPresent())
.flatMap(this::toMail)
.blockOptional()
.orElse(null);
diff --git a/server/protocols/jmap-draft/src/main/java/org/apache/james/jmap/draft/methods/GetMailboxesMethod.java b/server/protocols/jmap-draft/src/main/java/org/apache/james/jmap/draft/methods/GetMailboxesMethod.java
index f74b243..f5936a9 100644
--- a/server/protocols/jmap-draft/src/main/java/org/apache/james/jmap/draft/methods/GetMailboxesMethod.java
+++ b/server/protocols/jmap-draft/src/main/java/org/apache/james/jmap/draft/methods/GetMailboxesMethod.java
@@ -20,6 +20,7 @@
package org.apache.james.jmap.draft.methods;
import static org.apache.james.util.ReactorUtils.context;
+import static org.apache.james.util.ReactorUtils.publishIfPresent;
import java.util.Comparator;
import java.util.List;
@@ -144,7 +145,7 @@ public class GetMailboxesMethod implements Method {
.usingPreloadedMailboxesMetadata(NO_PRELOADED_METADATA)
.build())
.subscribeOn(Schedulers.elastic()))
- .handle((element, sink) -> element.ifPresent(sink::next));
+ .handle(publishIfPresent());
}
private Flux<Mailbox> retrieveAllMailboxes(MailboxSession mailboxSession) {
@@ -156,12 +157,13 @@ public class GetMailboxesMethod implements Method {
return userMailboxesMono.zipWith(quotaLoaderMono)
.flatMapMany(
tuple -> Flux.fromIterable(tuple.getT1())
- .flatMap(mailboxMetaData -> Mono.justOrEmpty(mailboxFactory.builder()
+ .map(mailboxMetaData -> mailboxFactory.builder()
.mailboxMetadata(mailboxMetaData)
.session(mailboxSession)
.usingPreloadedMailboxesMetadata(Optional.of(tuple.getT1()))
.quotaLoader(tuple.getT2())
- .build())));
+ .build())
+ .handle(publishIfPresent()));
}
private Flux<MailboxMetaData> getAllMailboxesMetaData(MailboxSession mailboxSession) {
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/configuration/EventsourcingConfigurationManagement.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/configuration/EventsourcingConfigurationManagement.java
index ba2c621..6cacd12 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/configuration/EventsourcingConfigurationManagement.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/configuration/EventsourcingConfigurationManagement.java
@@ -19,6 +19,8 @@
package org.apache.james.queue.rabbitmq.view.cassandra.configuration;
+import static org.apache.james.util.ReactorUtils.publishIfPresent;
+
import javax.inject.Inject;
import org.apache.james.eventsourcing.AggregateId;
@@ -53,9 +55,10 @@ public class EventsourcingConfigurationManagement {
@VisibleForTesting
Mono<CassandraMailQueueViewConfiguration> load() {
return Mono.from(eventStore.getEventsOfAggregate(CONFIGURATION_AGGREGATE_ID))
- .flatMap(history -> Mono.justOrEmpty(ConfigurationAggregate
+ .map(history -> ConfigurationAggregate
.load(CONFIGURATION_AGGREGATE_ID, history)
- .getCurrentConfiguration()));
+ .getCurrentConfiguration())
+ .handle(publishIfPresent());
}
public void registerConfiguration(CassandraMailQueueViewConfiguration newConfiguration) {
diff --git a/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQTerminationSubscriber.java b/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQTerminationSubscriber.java
index 174101a..2c1812f 100644
--- a/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQTerminationSubscriber.java
+++ b/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQTerminationSubscriber.java
@@ -20,6 +20,8 @@
package org.apache.james.task.eventsourcing.distributed;
+import static org.apache.james.util.ReactorUtils.publishIfPresent;
+
import java.io.Closeable;
import java.nio.charset.StandardCharsets;
import java.util.Optional;
@@ -91,7 +93,8 @@ public class RabbitMQTerminationSubscriber implements TerminationSubscriber, Sta
listenQueueHandle = listenerReceiver
.consumeAutoAck(queueName)
.subscribeOn(Schedulers.elastic())
- .<Event>handle((delivery, sink) -> toEvent(delivery).ifPresent(sink::next))
+ .map(this::toEvent)
+ .handle(publishIfPresent())
.subscribe(listener::onNext);
}
diff --git a/server/task/task-memory/src/main/java/org/apache/james/task/SerialTaskManagerWorker.java b/server/task/task-memory/src/main/java/org/apache/james/task/SerialTaskManagerWorker.java
index b024c4c..8341a16 100644
--- a/server/task/task-memory/src/main/java/org/apache/james/task/SerialTaskManagerWorker.java
+++ b/server/task/task-memory/src/main/java/org/apache/james/task/SerialTaskManagerWorker.java
@@ -18,6 +18,8 @@
****************************************************************/
package org.apache.james.task;
+import static org.apache.james.util.ReactorUtils.publishIfPresent;
+
import java.io.IOException;
import java.time.Duration;
import java.util.Optional;
@@ -91,7 +93,7 @@ public class SerialTaskManagerWorker implements TaskManagerWorker {
return Mono.fromCallable(() -> taskWithId.getTask().details())
.delayElement(pollingInterval, Schedulers.elastic())
.repeat()
- .<TaskExecutionDetails.AdditionalInformation>handle((maybeDetails, sink) -> maybeDetails.ifPresent(sink::next))
+ .handle(publishIfPresent())
.flatMap(information -> Mono.from(listener.updated(taskWithId.getId(), information)).thenReturn(information));
}
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 03/09: JAMES-3149 CassandraMailQueueBrowser caller
should choose on which scheduler he dispatch
Posted by rc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rcordier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 7795d0b779c0989858bdb1c576c218b466aa47d9
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Mon May 4 18:27:18 2020 +0700
JAMES-3149 CassandraMailQueueBrowser caller should choose on which scheduler he dispatch
This avoid enforcing the thread context switch upon browse,
optimizing thread usage.
Such an allocation is redundant when updating browse start.
---
.../queue/rabbitmq/view/cassandra/CassandraMailQueueBrowser.java | 4 +---
.../queue/rabbitmq/view/cassandra/CassandraMailQueueView.java | 8 +++++++-
2 files changed, 8 insertions(+), 4 deletions(-)
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueBrowser.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueBrowser.java
index 31e448c..91223ad 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueBrowser.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueBrowser.java
@@ -47,7 +47,6 @@ import com.google.common.base.Preconditions;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
-import reactor.core.scheduler.Schedulers;
public class CassandraMailQueueBrowser {
@@ -110,8 +109,7 @@ public class CassandraMailQueueBrowser {
Flux<EnqueuedItemWithSlicingContext> browseReferences(MailQueueName queueName) {
return browseStartDao.findBrowseStart(queueName)
.flatMapMany(this::allSlicesStartingAt)
- .flatMapSequential(slice -> browseSlice(queueName, slice))
- .subscribeOn(Schedulers.elastic());
+ .flatMapSequential(slice -> browseSlice(queueName, slice));
}
private Mono<Mail> toMailFuture(EnqueuedItemWithSlicingContext enqueuedItemWithSlicingContext) {
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueView.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueView.java
index aede929..b82de97 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueView.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueView.java
@@ -34,6 +34,7 @@ import org.apache.james.queue.rabbitmq.view.cassandra.configuration.Eventsourcin
import org.apache.james.queue.rabbitmq.view.cassandra.model.EnqueuedItemWithSlicingContext;
import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
public class CassandraMailQueueView implements MailQueueView {
@@ -91,13 +92,17 @@ public class CassandraMailQueueView implements MailQueueView {
public ManageableMailQueue.MailQueueIterator browse() {
return new CassandraMailQueueBrowser.CassandraMailQueueIterator(
cassandraMailQueueBrowser.browse(mailQueueName)
+ .subscribeOn(Schedulers.elastic())
.toIterable()
.iterator());
}
@Override
public long getSize() {
- return cassandraMailQueueBrowser.browseReferences(mailQueueName).count().block();
+ return cassandraMailQueueBrowser.browseReferences(mailQueueName)
+ .count()
+ .subscribeOn(Schedulers.elastic())
+ .block();
}
@Override
@@ -117,6 +122,7 @@ public class CassandraMailQueueView implements MailQueueView {
.flatMap(mailReference -> cassandraMailQueueMailDelete.considerDeleted(mailReference.getEnqueueId(), mailQueueName))
.count()
.doOnNext(ignored -> cassandraMailQueueMailDelete.updateBrowseStart(mailQueueName))
+ .subscribeOn(Schedulers.elastic())
.block();
}
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 05/09: [Refactoring] Reactor: use handle for
nullable transformations
Posted by rc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rcordier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit a4f75c27dea83aa6b7b9a5b232ebaf80ff67642a
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Mon May 4 11:12:26 2020 +0700
[Refactoring] Reactor: use handle for nullable transformations
https://github.com/reactor/reactor-core/issues/86
---
.../org/apache/james/backends/rabbitmq/SimpleConnectionPool.java | 4 +++-
.../james/mailbox/cassandra/quota/CassandraGlobalMaxQuotaDao.java | 5 +++--
.../james/mailbox/cassandra/quota/CassandraPerDomainMaxQuotaDao.java | 5 +++--
.../james/mailbox/cassandra/quota/CassandraPerUserMaxQuotaDao.java | 5 +++--
.../util/src/main/java/org/apache/james/util/ReactorUtils.java | 5 +++++
5 files changed, 17 insertions(+), 7 deletions(-)
diff --git a/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/SimpleConnectionPool.java b/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/SimpleConnectionPool.java
index b174795..97ebfbe 100644
--- a/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/SimpleConnectionPool.java
+++ b/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/SimpleConnectionPool.java
@@ -19,6 +19,8 @@
package org.apache.james.backends.rabbitmq;
+import static org.apache.james.util.ReactorUtils.transformAndPublishIfNotNull;
+
import java.io.IOException;
import java.time.Duration;
import java.util.Optional;
@@ -89,7 +91,7 @@ public class SimpleConnectionPool implements AutoCloseable {
public Mono<RabbitMQServerVersion> version() {
return getOpenConnection()
.map(Connection::getServerProperties)
- .flatMap(serverProperties -> Mono.justOrEmpty(serverProperties.get("version")))
+ .handle(transformAndPublishIfNotNull(serverProperties -> serverProperties.get("version")))
.map(Object::toString)
.map(RabbitMQServerVersion::of)
.timeout(Duration.ofSeconds(1))
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraGlobalMaxQuotaDao.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraGlobalMaxQuotaDao.java
index 9ee9159..03d4e54 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraGlobalMaxQuotaDao.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraGlobalMaxQuotaDao.java
@@ -25,6 +25,7 @@ import static com.datastax.driver.core.querybuilder.QueryBuilder.eq;
import static com.datastax.driver.core.querybuilder.QueryBuilder.insertInto;
import static com.datastax.driver.core.querybuilder.QueryBuilder.select;
import static org.apache.james.util.ReactorUtils.publishIfPresent;
+import static org.apache.james.util.ReactorUtils.transformAndPublishIfNotNull;
import javax.inject.Inject;
@@ -93,7 +94,7 @@ public class CassandraGlobalMaxQuotaDao {
Mono<QuotaSizeLimit> getGlobalMaxStorage() {
return queryExecutor.executeSingleRow(getGlobalMaxStatement.bind()
.setString(CassandraGlobalMaxQuota.TYPE, CassandraGlobalMaxQuota.STORAGE))
- .flatMap(row -> Mono.justOrEmpty(row.get(CassandraGlobalMaxQuota.VALUE, Long.class)))
+ .handle(transformAndPublishIfNotNull(row -> row.get(CassandraGlobalMaxQuota.VALUE, Long.class)))
.map(QuotaCodec::longToQuotaSize)
.handle(publishIfPresent());
}
@@ -101,7 +102,7 @@ public class CassandraGlobalMaxQuotaDao {
Mono<QuotaCountLimit> getGlobalMaxMessage() {
return queryExecutor.executeSingleRow(getGlobalMaxStatement.bind()
.setString(CassandraGlobalMaxQuota.TYPE, CassandraGlobalMaxQuota.MESSAGE))
- .flatMap(row -> Mono.justOrEmpty(row.get(CassandraGlobalMaxQuota.VALUE, Long.class)))
+ .handle(transformAndPublishIfNotNull(row -> row.get(CassandraGlobalMaxQuota.VALUE, Long.class)))
.map(QuotaCodec::longToQuotaCount)
.handle(publishIfPresent());
}
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraPerDomainMaxQuotaDao.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraPerDomainMaxQuotaDao.java
index 38161ef..102d793 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraPerDomainMaxQuotaDao.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraPerDomainMaxQuotaDao.java
@@ -25,6 +25,7 @@ import static com.datastax.driver.core.querybuilder.QueryBuilder.eq;
import static com.datastax.driver.core.querybuilder.QueryBuilder.insertInto;
import static com.datastax.driver.core.querybuilder.QueryBuilder.select;
import static org.apache.james.util.ReactorUtils.publishIfPresent;
+import static org.apache.james.util.ReactorUtils.transformAndPublishIfNotNull;
import javax.inject.Inject;
@@ -109,14 +110,14 @@ public class CassandraPerDomainMaxQuotaDao {
Mono<QuotaSizeLimit> getMaxStorage(Domain domain) {
return queryExecutor.executeSingleRow(getMaxStorageStatement.bind(domain.asString()))
- .flatMap(row -> Mono.justOrEmpty(row.get(CassandraDomainMaxQuota.STORAGE, Long.class)))
+ .handle(transformAndPublishIfNotNull(row -> row.get(CassandraDomainMaxQuota.STORAGE, Long.class)))
.map(QuotaCodec::longToQuotaSize)
.handle(publishIfPresent());
}
Mono<QuotaCountLimit> getMaxMessage(Domain domain) {
return queryExecutor.executeSingleRow(getMaxMessageStatement.bind(domain.asString()))
- .flatMap(row -> Mono.justOrEmpty(row.get(CassandraDomainMaxQuota.MESSAGE_COUNT, Long.class)))
+ .handle(transformAndPublishIfNotNull(row -> row.get(CassandraDomainMaxQuota.MESSAGE_COUNT, Long.class)))
.map(QuotaCodec::longToQuotaCount)
.handle(publishIfPresent());
}
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraPerUserMaxQuotaDao.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraPerUserMaxQuotaDao.java
index 2b4e21b..4abe876 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraPerUserMaxQuotaDao.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraPerUserMaxQuotaDao.java
@@ -25,6 +25,7 @@ import static com.datastax.driver.core.querybuilder.QueryBuilder.eq;
import static com.datastax.driver.core.querybuilder.QueryBuilder.insertInto;
import static com.datastax.driver.core.querybuilder.QueryBuilder.select;
import static org.apache.james.util.ReactorUtils.publishIfPresent;
+import static org.apache.james.util.ReactorUtils.transformAndPublishIfNotNull;
import javax.inject.Inject;
@@ -109,14 +110,14 @@ public class CassandraPerUserMaxQuotaDao {
Mono<QuotaSizeLimit> getMaxStorage(QuotaRoot quotaRoot) {
return queryExecutor.executeSingleRow(getMaxStorageStatement.bind(quotaRoot.getValue()))
- .flatMap(row -> Mono.justOrEmpty(row.get(CassandraMaxQuota.STORAGE, Long.class)))
+ .handle(transformAndPublishIfNotNull(row -> row.get(CassandraMaxQuota.STORAGE, Long.class)))
.map(QuotaCodec::longToQuotaSize)
.handle(publishIfPresent());
}
Mono<QuotaCountLimit> getMaxMessage(QuotaRoot quotaRoot) {
return queryExecutor.executeSingleRow(getMaxMessageStatement.bind(quotaRoot.getValue()))
- .flatMap(row -> Mono.justOrEmpty(row.get(CassandraMaxQuota.MESSAGE_COUNT, Long.class)))
+ .handle(transformAndPublishIfNotNull(row -> row.get(CassandraMaxQuota.MESSAGE_COUNT, Long.class)))
.map(QuotaCodec::longToQuotaCount)
.handle(publishIfPresent());
}
diff --git a/server/container/util/src/main/java/org/apache/james/util/ReactorUtils.java b/server/container/util/src/main/java/org/apache/james/util/ReactorUtils.java
index a477355..7fb5d5a 100644
--- a/server/container/util/src/main/java/org/apache/james/util/ReactorUtils.java
+++ b/server/container/util/src/main/java/org/apache/james/util/ReactorUtils.java
@@ -26,6 +26,7 @@ import java.util.Iterator;
import java.util.Optional;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
+import java.util.function.Function;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
@@ -45,6 +46,10 @@ public class ReactorUtils {
return (element, sink) -> element.ifPresent(sink::next);
}
+ public static <T, U> BiConsumer<U, SynchronousSink<T>> transformAndPublishIfNotNull(Function<U, T> mapper) {
+ return (element, sink) -> Optional.ofNullable(mapper.apply(element)).ifPresent(sink::next);
+ }
+
public static InputStream toInputStream(Flux<ByteBuffer> byteArrays) {
return new StreamInputStream(byteArrays.toIterable(1).iterator());
}
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 08/09: JAMES-3091 JSON serialization for Mailbox
POJO
Posted by rc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rcordier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit c01bf3a985407595fbfb2d50329718e47abf68c0
Author: Rene Cordier <rc...@linagora.com>
AuthorDate: Fri May 8 17:45:35 2020 +0700
JAMES-3091 JSON serialization for Mailbox POJO
---
server/protocols/jmap-rfc-8621/pom.xml | 5 +
.../org/apache/james/jmap/json/Serializer.scala | 73 ++++++++++-
.../scala/org/apache/james/jmap/mail/Mailbox.scala | 2 +-
.../scala/org/apache/james/jmap/mail/Quotas.scala | 19 ++-
.../james/jmap/json/MailboxSerializationTest.scala | 135 +++++++++++++++++++++
.../org/apache/james/jmap/mail/MailboxTest.scala | 12 +-
6 files changed, 233 insertions(+), 13 deletions(-)
diff --git a/server/protocols/jmap-rfc-8621/pom.xml b/server/protocols/jmap-rfc-8621/pom.xml
index bb6a83b..9f2d220 100644
--- a/server/protocols/jmap-rfc-8621/pom.xml
+++ b/server/protocols/jmap-rfc-8621/pom.xml
@@ -78,6 +78,11 @@
<scope>test</scope>
</dependency>
<dependency>
+ <groupId>net.javacrumbs.json-unit</groupId>
+ <artifactId>json-unit-assertj</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
<scope>test</scope>
diff --git a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/json/Serializer.scala b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/json/Serializer.scala
index 173a82d..87937a1 100644
--- a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/json/Serializer.scala
+++ b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/json/Serializer.scala
@@ -22,11 +22,14 @@ package org.apache.james.jmap.json
import java.io.InputStream
import java.net.URL
-import org.apache.james.core.Username
+import org.apache.james.core.{Domain, Username}
+import org.apache.james.jmap.mail.{DelegatedNamespace, IsSubscribed, Mailbox, MailboxNamespace, MailboxRights, MayAddItems, MayCreateChild, MayDelete, MayReadItems, MayRemoveItems, MayRename, MaySetKeywords, MaySetSeen, MaySubmit, PersonalNamespace, Quota, QuotaId, QuotaRoot, Quotas, Right, Rights, SortOrder, TotalEmails, TotalThreads, UnreadEmails, UnreadThreads, Value}
import org.apache.james.jmap.model
import org.apache.james.jmap.model.CapabilityIdentifier.CapabilityIdentifier
import org.apache.james.jmap.model.Invocation.{Arguments, MethodCallId, MethodName}
import org.apache.james.jmap.model.{Account, Invocation, Session, _}
+import org.apache.james.mailbox.Role
+import org.apache.james.mailbox.model.{MailboxACL, MailboxId}
import play.api.libs.functional.syntax._
import play.api.libs.json._
@@ -126,6 +129,70 @@ class Serializer {
private implicit val sessionWrites: Writes[Session] = Json.writes[Session]
+ private implicit val mailboxIdWrites: Writes[MailboxId] = mailboxId => JsString(mailboxId.serialize)
+ private implicit val roleWrites: Writes[Role] = role => JsString(role.serialize)
+ private implicit val sortOrderWrites: Writes[SortOrder] = Json.valueWrites[SortOrder]
+ private implicit val totalEmailsWrites: Writes[TotalEmails] = Json.valueWrites[TotalEmails]
+ private implicit val unreadEmailsWrites: Writes[UnreadEmails] = Json.valueWrites[UnreadEmails]
+ private implicit val totalThreadsWrites: Writes[TotalThreads] = Json.valueWrites[TotalThreads]
+ private implicit val unreadThreadsWrites: Writes[UnreadThreads] = Json.valueWrites[UnreadThreads]
+ private implicit val isSubscribedWrites: Writes[IsSubscribed] = Json.valueWrites[IsSubscribed]
+
+ private implicit val mayReadItemsWrites: Writes[MayReadItems] = Json.valueWrites[MayReadItems]
+ private implicit val mayAddItemsWrites: Writes[MayAddItems] = Json.valueWrites[MayAddItems]
+ private implicit val mayRemoveItemsWrites: Writes[MayRemoveItems] = Json.valueWrites[MayRemoveItems]
+ private implicit val maySetSeenWrites: Writes[MaySetSeen] = Json.valueWrites[MaySetSeen]
+ private implicit val maySetKeywordsWrites: Writes[MaySetKeywords] = Json.valueWrites[MaySetKeywords]
+ private implicit val mayCreateChildWrites: Writes[MayCreateChild] = Json.valueWrites[MayCreateChild]
+ private implicit val mayRenameWrites: Writes[MayRename] = Json.valueWrites[MayRename]
+ private implicit val mayDeleteWrites: Writes[MayDelete] = Json.valueWrites[MayDelete]
+ private implicit val maySubmitWrites: Writes[MaySubmit] = Json.valueWrites[MaySubmit]
+ private implicit val mailboxRightsWrites: Writes[MailboxRights] = Json.writes[MailboxRights]
+
+ private implicit val personalNamespaceWrites: Writes[PersonalNamespace] = namespace => JsString("Personal")
+ private implicit val delegatedNamespaceWrites: Writes[DelegatedNamespace] = namespace => JsString(s"Delegated[${namespace.owner.asString}]")
+ private implicit val mailboxNamespaceWrites: Writes[MailboxNamespace] = Json.writes[MailboxNamespace]
+
+ private implicit val mailboxACLWrites: Writes[MailboxACL.Right] = right => JsString(right.asCharacter.toString)
+
+ private implicit val rightWrites: Writes[Right] = Json.valueWrites[Right]
+ private implicit val rightsWrites: Writes[Rights] = Json.valueWrites[Rights]
+
+ private implicit def rightsMapWrites(implicit rightWriter: Writes[Seq[Right]]): Writes[Map[Username, Seq[Right]]] =
+ (m: Map[Username, Seq[Right]]) => {
+ m.foldLeft(JsObject.empty)((jsObject, kv) => {
+ val (username: Username, rights: Seq[Right]) = kv
+ jsObject.+(username.asString, rightWriter.writes(rights))
+ })
+ }
+
+ private implicit val domainWrites: Writes[Domain] = domain => JsString(domain.asString)
+ private implicit val quotaRootWrites: Writes[QuotaRoot] = Json.writes[QuotaRoot]
+ private implicit val quotaIdWrites: Writes[QuotaId] = Json.valueWrites[QuotaId]
+
+ private implicit val quotaValueWrites: Writes[Value] = Json.writes[Value]
+ private implicit val quotaWrites: Writes[Quota] = Json.valueWrites[Quota]
+
+ private implicit def quotaMapWrites(implicit valueWriter: Writes[Value]): Writes[Map[Quotas.Type, Value]] =
+ (m: Map[Quotas.Type, Value]) => {
+ m.foldLeft(JsObject.empty)((jsObject, kv) => {
+ val (quotaType: Quotas.Type, value: Value) = kv
+ jsObject.+(quotaType.toString, valueWriter.writes(value))
+ })
+ }
+
+ private implicit val quotasWrites: Writes[Quotas] = Json.valueWrites[Quotas]
+
+ private implicit def quotasMapWrites(implicit quotaWriter: Writes[Quota]): Writes[Map[QuotaId, Quota]] =
+ (m: Map[QuotaId, Quota]) => {
+ m.foldLeft(JsObject.empty)((jsObject, kv) => {
+ val (quotaId: QuotaId, quota: Quota) = kv
+ jsObject.+(quotaId.getName, quotaWriter.writes(quota))
+ })
+ }
+
+ private implicit val mailboxWrites: Writes[Mailbox] = Json.writes[Mailbox]
+
def serialize(session: Session): JsValue = {
Json.toJson(session)
}
@@ -138,6 +205,10 @@ class Serializer {
Json.toJson(responseObject)
}
+ def serialize(mailbox: Mailbox): JsValue = {
+ Json.toJson(mailbox)
+ }
+
def deserializeRequestObject(input: String): JsResult[RequestObject] = {
Json.parse(input).validate[RequestObject]
}
diff --git a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/mail/Mailbox.scala b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/mail/Mailbox.scala
index 09f881f..aaff40e 100644
--- a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/mail/Mailbox.scala
+++ b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/mail/Mailbox.scala
@@ -56,7 +56,7 @@ object MailboxNamespace {
sealed trait MailboxNamespace
-case object PersonalNamespace extends MailboxNamespace
+case class PersonalNamespace() extends MailboxNamespace
case class DelegatedNamespace(owner: Username) extends MailboxNamespace
diff --git a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/mail/Quotas.scala b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/mail/Quotas.scala
index be64e75..c9d4212 100644
--- a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/mail/Quotas.scala
+++ b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/mail/Quotas.scala
@@ -19,10 +19,19 @@
package org.apache.james.jmap.mail
-import java.util.Optional
-
+import org.apache.james.core.Domain
import org.apache.james.jmap.model.UnsignedInt.UnsignedInt
-import org.apache.james.mailbox.model.QuotaRoot
+import org.apache.james.mailbox.model.{QuotaRoot => JavaQuotaRoot}
+
+import scala.compat.java8.OptionConverters._
+
+object QuotaRoot{
+ def fromJava(quotaRoot: JavaQuotaRoot) = QuotaRoot(quotaRoot.getValue, quotaRoot.getDomain.asScala)
+}
+
+case class QuotaRoot(value: String, domain: Option[Domain]) {
+ def asJava: JavaQuotaRoot = JavaQuotaRoot.quotaRoot(value, domain.asJava)
+}
object Quotas {
sealed trait Type
@@ -39,11 +48,11 @@ object QuotaId {
}
case class QuotaId(quotaRoot: QuotaRoot) extends AnyVal {
- def getName: String = quotaRoot.getValue
+ def getName: String = quotaRoot.value
}
case class Quota(quota: Map[Quotas.Type, Value]) extends AnyVal
-case class Value(used: UnsignedInt, max: Optional[UnsignedInt])
+case class Value(used: UnsignedInt, max: Option[UnsignedInt])
case class Quotas(quotas: Map[QuotaId, Quota]) extends AnyVal
\ No newline at end of file
diff --git a/server/protocols/jmap-rfc-8621/src/test/scala/org/apache/james/jmap/json/MailboxSerializationTest.scala b/server/protocols/jmap-rfc-8621/src/test/scala/org/apache/james/jmap/json/MailboxSerializationTest.scala
new file mode 100644
index 0000000..5ab9690
--- /dev/null
+++ b/server/protocols/jmap-rfc-8621/src/test/scala/org/apache/james/jmap/json/MailboxSerializationTest.scala
@@ -0,0 +1,135 @@
+/****************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one *
+ * or more contributor license agreements. See the NOTICE file *
+ * distributed with this work for additional information *
+ * regarding copyright ownership. The ASF licenses this file *
+ * to you under the Apache License, Version 2.0 (the *
+ * "License"); you may not use this file except in compliance *
+ * with the License. You may obtain a copy of the License at *
+ * *
+ * http://www.apache.org/licenses/LICENSE-2.0 *
+ * *
+ * Unless required by applicable law or agreed to in writing, *
+ * software distributed under the License is distributed on an *
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY *
+ * KIND, either express or implied. See the License for the *
+ * specific language governing permissions and limitations *
+ * under the License. *
+ ****************************************************************/
+
+package org.apache.james.jmap.json
+
+import eu.timepit.refined.auto._
+import net.javacrumbs.jsonunit.assertj.JsonAssertions.assertThatJson
+import org.apache.james.core.{Domain, Username}
+import org.apache.james.jmap.json.MailboxSerializationTest.MAILBOX
+import org.apache.james.jmap.mail.Mailbox.MailboxName
+import org.apache.james.jmap.mail.{IsSubscribed, Mailbox, MailboxNamespace, MailboxRights, MayAddItems, MayCreateChild, MayDelete, MayReadItems, MayRemoveItems, MayRename, MaySetKeywords, MaySetSeen, MaySubmit, PersonalNamespace, Quota, QuotaId, QuotaRoot, Quotas, Right, Rights, SortOrder, TotalEmails, TotalThreads, UnreadEmails, UnreadThreads, Value}
+import org.apache.james.mailbox.Role
+import org.apache.james.mailbox.model.{MailboxId, TestId}
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
+import play.api.libs.json.Json
+
+object MailboxSerializationTest {
+ private val MAILBOX_ID: MailboxId = TestId.of(2)
+ private val MAILBOX_NAME: MailboxName = "inbox"
+ private val PARENT_ID: Option[MailboxId] = Option.apply(TestId.of(1))
+ private val ROLE: Option[Role] = Option.apply(Role.INBOX)
+ private val SORT_ORDER: SortOrder = SortOrder(10L)
+ private val TOTAL_EMAILS: TotalEmails = TotalEmails(1234L)
+ private val UNREAD_EMAILS: UnreadEmails = UnreadEmails(123L)
+ private val TOTAL_THREADS: TotalThreads = TotalThreads(58L)
+ private val UNREAD_THREADS: UnreadThreads = UnreadThreads(22L)
+ private val IS_SUBSCRIBED: IsSubscribed = IsSubscribed(true)
+ private val NAMESPACE: MailboxNamespace = PersonalNamespace()
+
+ private val MY_RIGHTS: MailboxRights = MailboxRights(
+ mayAddItems = MayAddItems(true),
+ mayReadItems = MayReadItems(false),
+ mayRemoveItems = MayRemoveItems(false),
+ maySetSeen = MaySetSeen(true),
+ maySetKeywords = MaySetKeywords(false),
+ mayCreateChild = MayCreateChild(true),
+ mayRename = MayRename(true),
+ mayDelete = MayDelete(false),
+ maySubmit = MaySubmit(false)
+ )
+
+ private val RIGHTS: Rights = Rights.of(Username.of("bob"), Seq(Right.Expunge, Right.Lookup))
+ .append(Username.of("alice"), Seq(Right.Read, Right.Write))
+
+ private val QUOTAS = Quotas(Map(
+ QuotaId(QuotaRoot("quotaRoot", None)) -> Quota(Map(
+ Quotas.Message -> Value(18L, Some(42L)),
+ Quotas.Storage -> Value(12L, None))),
+ QuotaId(QuotaRoot("quotaRoot2@localhost", Some(Domain.LOCALHOST))) -> Quota(Map(
+ Quotas.Message -> Value(14L, Some(43L)),
+ Quotas.Storage -> Value(19L, None)))))
+
+ private val MAILBOX: Mailbox = Mailbox(
+ id = MAILBOX_ID,
+ name = MAILBOX_NAME,
+ parentId = PARENT_ID,
+ role = ROLE,
+ sortOrder = SORT_ORDER,
+ totalEmails = TOTAL_EMAILS,
+ unreadEmails = UNREAD_EMAILS,
+ totalThreads = TOTAL_THREADS,
+ unreadThreads = UNREAD_THREADS,
+ myRights = MY_RIGHTS,
+ isSubscribed = IS_SUBSCRIBED,
+ namespace = NAMESPACE,
+ rights = RIGHTS,
+ quotas = QUOTAS
+ )
+}
+
+class MailboxSerializationTest extends AnyWordSpec with Matchers {
+ "Serialize Mailbox" should {
+ "succeed " in {
+
+ val expectedJson = Json.parse(
+ """{
+ | "id":"2",
+ | "name":"inbox",
+ | "parentId":"1",
+ | "role":"inbox",
+ | "sortOrder":10,
+ | "totalEmails":1234,
+ | "unreadEmails":123,
+ | "totalThreads":58,
+ | "unreadThreads":22,
+ | "myRights":{
+ | "mayReadItems":false,
+ | "mayAddItems":true,
+ | "mayRemoveItems":false,
+ | "maySetSeen":true,
+ | "maySetKeywords":false,
+ | "mayCreateChild":true,
+ | "mayRename":true,
+ | "mayDelete":false,
+ | "maySubmit":false
+ | },
+ | "isSubscribed":true,
+ | "namespace":"Personal",
+ | "rights":{
+ | "bob":["e","l"],
+ | "alice":["r","w"]
+ | },
+ | "quotas":{
+ | "quotaRoot":{
+ | "Message":{"used":18,"max":42},
+ | "Storage":{"used":12}
+ | },
+ | "quotaRoot2@localhost":{
+ | "Message":{"used":14,"max":43},
+ | "Storage":{"used":19}
+ | }
+ | }
+ |}""".stripMargin)
+
+ assertThatJson(new Serializer().serialize(MAILBOX)).isEqualTo(expectedJson)
+ }
+ }
+}
diff --git a/server/protocols/jmap-rfc-8621/src/test/scala/org/apache/james/jmap/mail/MailboxTest.scala b/server/protocols/jmap-rfc-8621/src/test/scala/org/apache/james/jmap/mail/MailboxTest.scala
index 018c2bb..47e8c95 100644
--- a/server/protocols/jmap-rfc-8621/src/test/scala/org/apache/james/jmap/mail/MailboxTest.scala
+++ b/server/protocols/jmap-rfc-8621/src/test/scala/org/apache/james/jmap/mail/MailboxTest.scala
@@ -58,7 +58,7 @@ class MailboxTest extends AnyWordSpec with Matchers {
MayDelete(true),
MaySubmit(true)),
IsSubscribed(true),
- MailboxNamespace.personal,
+ MailboxNamespace.personal(),
Rights.EMPTY,
Quotas(Map()))
.hasRole(Role.INBOX) must be(false)
@@ -84,7 +84,7 @@ class MailboxTest extends AnyWordSpec with Matchers {
MayDelete(true),
MaySubmit(true)),
IsSubscribed(true),
- MailboxNamespace.personal,
+ MailboxNamespace.personal(),
Rights.EMPTY,
Quotas(Map()))
.hasRole(Role.INBOX) must be(false)
@@ -110,7 +110,7 @@ class MailboxTest extends AnyWordSpec with Matchers {
MayDelete(true),
MaySubmit(true)),
IsSubscribed(true),
- MailboxNamespace.personal,
+ MailboxNamespace.personal(),
Rights.EMPTY,
Quotas(Map())).hasRole(Role.INBOX) must be(true)
}
@@ -138,7 +138,7 @@ class MailboxTest extends AnyWordSpec with Matchers {
MayDelete(true),
MaySubmit(true)),
IsSubscribed(true),
- MailboxNamespace.personal,
+ MailboxNamespace.personal(),
Rights.EMPTY,
Quotas(Map()))
.hasSystemRole must be(false)
@@ -164,7 +164,7 @@ class MailboxTest extends AnyWordSpec with Matchers {
MayDelete(true),
MaySubmit(true)),
IsSubscribed(true),
- MailboxNamespace.personal,
+ MailboxNamespace.personal(),
Rights.EMPTY,
Quotas(Map())).hasSystemRole must be(false)
}
@@ -189,7 +189,7 @@ class MailboxTest extends AnyWordSpec with Matchers {
MayDelete(true),
MaySubmit(true)),
IsSubscribed(true),
- MailboxNamespace.personal,
+ MailboxNamespace.personal(),
Rights.EMPTY,
Quotas(Map())).hasSystemRole must be(true)
}
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 06/09: [Refactoring] Reactor: extract .then() call
Posted by rc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rcordier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit db157e6016357caa09aba8329a8f9f5e356e9a61
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Mon May 4 11:26:29 2020 +0700
[Refactoring] Reactor: extract .then() call
---
.../org/apache/james/sieve/cassandra/CassandraSieveRepository.java | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
diff --git a/server/data/data-cassandra/src/main/java/org/apache/james/sieve/cassandra/CassandraSieveRepository.java b/server/data/data-cassandra/src/main/java/org/apache/james/sieve/cassandra/CassandraSieveRepository.java
index ec134e7..e3ca672 100644
--- a/server/data/data-cassandra/src/main/java/org/apache/james/sieve/cassandra/CassandraSieveRepository.java
+++ b/server/data/data-cassandra/src/main/java/org/apache/james/sieve/cassandra/CassandraSieveRepository.java
@@ -178,7 +178,8 @@ public class CassandraSieveRepository implements SieveRepository {
private Mono<Void> unactivateOldScript(Username username) {
return cassandraActiveScriptDAO.getActiveSctiptInfo(username)
- .flatMap(activeScriptInfo -> updateScriptActivation(username, activeScriptInfo.getName(), false).then());
+ .flatMap(activeScriptInfo -> updateScriptActivation(username, activeScriptInfo.getName(), false))
+ .then();
}
private Mono<Boolean> updateScriptActivation(Username username, ScriptName scriptName, boolean active) {
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 07/09: [Refactoring] Rely on publishIfPresent when
mapping null values
Posted by rc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rcordier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 9215920572977fff8f9614e6b495c4a3ff66dd70
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Fri May 8 06:21:14 2020 +0200
[Refactoring] Rely on publishIfPresent when mapping null values
---
.../org/apache/james/backends/rabbitmq/SimpleConnectionPool.java | 5 +++--
.../mailbox/cassandra/quota/CassandraGlobalMaxQuotaDao.java | 9 ++++++---
.../mailbox/cassandra/quota/CassandraPerDomainMaxQuotaDao.java | 9 ++++++---
.../mailbox/cassandra/quota/CassandraPerUserMaxQuotaDao.java | 9 ++++++---
.../util/src/main/java/org/apache/james/util/ReactorUtils.java | 5 -----
5 files changed, 21 insertions(+), 16 deletions(-)
diff --git a/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/SimpleConnectionPool.java b/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/SimpleConnectionPool.java
index 97ebfbe..d408666 100644
--- a/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/SimpleConnectionPool.java
+++ b/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/SimpleConnectionPool.java
@@ -19,7 +19,7 @@
package org.apache.james.backends.rabbitmq;
-import static org.apache.james.util.ReactorUtils.transformAndPublishIfNotNull;
+import static org.apache.james.util.ReactorUtils.publishIfPresent;
import java.io.IOException;
import java.time.Duration;
@@ -91,7 +91,8 @@ public class SimpleConnectionPool implements AutoCloseable {
public Mono<RabbitMQServerVersion> version() {
return getOpenConnection()
.map(Connection::getServerProperties)
- .handle(transformAndPublishIfNotNull(serverProperties -> serverProperties.get("version")))
+ .map(serverProperties -> Optional.ofNullable(serverProperties.get("version")))
+ .handle(publishIfPresent())
.map(Object::toString)
.map(RabbitMQServerVersion::of)
.timeout(Duration.ofSeconds(1))
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraGlobalMaxQuotaDao.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraGlobalMaxQuotaDao.java
index 03d4e54..ae8684e 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraGlobalMaxQuotaDao.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraGlobalMaxQuotaDao.java
@@ -25,7 +25,8 @@ import static com.datastax.driver.core.querybuilder.QueryBuilder.eq;
import static com.datastax.driver.core.querybuilder.QueryBuilder.insertInto;
import static com.datastax.driver.core.querybuilder.QueryBuilder.select;
import static org.apache.james.util.ReactorUtils.publishIfPresent;
-import static org.apache.james.util.ReactorUtils.transformAndPublishIfNotNull;
+
+import java.util.Optional;
import javax.inject.Inject;
@@ -94,7 +95,8 @@ public class CassandraGlobalMaxQuotaDao {
Mono<QuotaSizeLimit> getGlobalMaxStorage() {
return queryExecutor.executeSingleRow(getGlobalMaxStatement.bind()
.setString(CassandraGlobalMaxQuota.TYPE, CassandraGlobalMaxQuota.STORAGE))
- .handle(transformAndPublishIfNotNull(row -> row.get(CassandraGlobalMaxQuota.VALUE, Long.class)))
+ .map(row -> Optional.ofNullable(row.get(CassandraGlobalMaxQuota.VALUE, Long.class)))
+ .handle(publishIfPresent())
.map(QuotaCodec::longToQuotaSize)
.handle(publishIfPresent());
}
@@ -102,7 +104,8 @@ public class CassandraGlobalMaxQuotaDao {
Mono<QuotaCountLimit> getGlobalMaxMessage() {
return queryExecutor.executeSingleRow(getGlobalMaxStatement.bind()
.setString(CassandraGlobalMaxQuota.TYPE, CassandraGlobalMaxQuota.MESSAGE))
- .handle(transformAndPublishIfNotNull(row -> row.get(CassandraGlobalMaxQuota.VALUE, Long.class)))
+ .map(row -> Optional.ofNullable(row.get(CassandraGlobalMaxQuota.VALUE, Long.class)))
+ .handle(publishIfPresent())
.map(QuotaCodec::longToQuotaCount)
.handle(publishIfPresent());
}
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraPerDomainMaxQuotaDao.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraPerDomainMaxQuotaDao.java
index 102d793..bc88058 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraPerDomainMaxQuotaDao.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraPerDomainMaxQuotaDao.java
@@ -25,7 +25,8 @@ import static com.datastax.driver.core.querybuilder.QueryBuilder.eq;
import static com.datastax.driver.core.querybuilder.QueryBuilder.insertInto;
import static com.datastax.driver.core.querybuilder.QueryBuilder.select;
import static org.apache.james.util.ReactorUtils.publishIfPresent;
-import static org.apache.james.util.ReactorUtils.transformAndPublishIfNotNull;
+
+import java.util.Optional;
import javax.inject.Inject;
@@ -110,14 +111,16 @@ public class CassandraPerDomainMaxQuotaDao {
Mono<QuotaSizeLimit> getMaxStorage(Domain domain) {
return queryExecutor.executeSingleRow(getMaxStorageStatement.bind(domain.asString()))
- .handle(transformAndPublishIfNotNull(row -> row.get(CassandraDomainMaxQuota.STORAGE, Long.class)))
+ .map(row -> Optional.ofNullable(row.get(CassandraDomainMaxQuota.STORAGE, Long.class)))
+ .handle(publishIfPresent())
.map(QuotaCodec::longToQuotaSize)
.handle(publishIfPresent());
}
Mono<QuotaCountLimit> getMaxMessage(Domain domain) {
return queryExecutor.executeSingleRow(getMaxMessageStatement.bind(domain.asString()))
- .handle(transformAndPublishIfNotNull(row -> row.get(CassandraDomainMaxQuota.MESSAGE_COUNT, Long.class)))
+ .map(row -> Optional.ofNullable(row.get(CassandraDomainMaxQuota.MESSAGE_COUNT, Long.class)))
+ .handle(publishIfPresent())
.map(QuotaCodec::longToQuotaCount)
.handle(publishIfPresent());
}
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraPerUserMaxQuotaDao.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraPerUserMaxQuotaDao.java
index 4abe876..9085623 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraPerUserMaxQuotaDao.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/quota/CassandraPerUserMaxQuotaDao.java
@@ -25,7 +25,8 @@ import static com.datastax.driver.core.querybuilder.QueryBuilder.eq;
import static com.datastax.driver.core.querybuilder.QueryBuilder.insertInto;
import static com.datastax.driver.core.querybuilder.QueryBuilder.select;
import static org.apache.james.util.ReactorUtils.publishIfPresent;
-import static org.apache.james.util.ReactorUtils.transformAndPublishIfNotNull;
+
+import java.util.Optional;
import javax.inject.Inject;
@@ -110,14 +111,16 @@ public class CassandraPerUserMaxQuotaDao {
Mono<QuotaSizeLimit> getMaxStorage(QuotaRoot quotaRoot) {
return queryExecutor.executeSingleRow(getMaxStorageStatement.bind(quotaRoot.getValue()))
- .handle(transformAndPublishIfNotNull(row -> row.get(CassandraMaxQuota.STORAGE, Long.class)))
+ .map(row -> Optional.ofNullable(row.get(CassandraMaxQuota.STORAGE, Long.class)))
+ .handle(publishIfPresent())
.map(QuotaCodec::longToQuotaSize)
.handle(publishIfPresent());
}
Mono<QuotaCountLimit> getMaxMessage(QuotaRoot quotaRoot) {
return queryExecutor.executeSingleRow(getMaxMessageStatement.bind(quotaRoot.getValue()))
- .handle(transformAndPublishIfNotNull(row -> row.get(CassandraMaxQuota.MESSAGE_COUNT, Long.class)))
+ .map(row -> Optional.ofNullable(row.get(CassandraMaxQuota.MESSAGE_COUNT, Long.class)))
+ .handle(publishIfPresent())
.map(QuotaCodec::longToQuotaCount)
.handle(publishIfPresent());
}
diff --git a/server/container/util/src/main/java/org/apache/james/util/ReactorUtils.java b/server/container/util/src/main/java/org/apache/james/util/ReactorUtils.java
index 7fb5d5a..a477355 100644
--- a/server/container/util/src/main/java/org/apache/james/util/ReactorUtils.java
+++ b/server/container/util/src/main/java/org/apache/james/util/ReactorUtils.java
@@ -26,7 +26,6 @@ import java.util.Iterator;
import java.util.Optional;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
-import java.util.function.Function;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
@@ -46,10 +45,6 @@ public class ReactorUtils {
return (element, sink) -> element.ifPresent(sink::next);
}
- public static <T, U> BiConsumer<U, SynchronousSink<T>> transformAndPublishIfNotNull(Function<U, T> mapper) {
- return (element, sink) -> Optional.ofNullable(mapper.apply(element)).ifPresent(sink::next);
- }
-
public static InputStream toInputStream(Flux<ByteBuffer> byteArrays) {
return new StreamInputStream(byteArrays.toIterable(1).iterator());
}
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 01/09: JAMES-3149 ReactorElasticSearchClient should
not call subscribeOn
Posted by rc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rcordier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit d3ae4a1f1b2f8e6aad8ab7f247c8eb4cb594f703
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Mon May 4 18:03:02 2020 +0700
JAMES-3149 ReactorElasticSearchClient should not call subscribeOn
Do call `publishOn` instead like in `CassandraAsyncExecutor`
---
.../java/org/apache/james/backends/es/ReactorElasticSearchClient.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ReactorElasticSearchClient.java b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ReactorElasticSearchClient.java
index 4885b08..1636df6 100644
--- a/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ReactorElasticSearchClient.java
+++ b/backends-common/elasticsearch/src/main/java/org/apache/james/backends/es/ReactorElasticSearchClient.java
@@ -154,7 +154,7 @@ public class ReactorElasticSearchClient implements AutoCloseable {
private static <T> Mono<T> toReactor(Consumer<ActionListener<T>> async) {
return Mono.<T>create(sink -> async.accept(getListener(sink)))
- .subscribeOn(Schedulers.elastic());
+ .publishOn(Schedulers.elastic());
}
private static <T> ActionListener<T> getListener(MonoSink<T> sink) {
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 09/09: JAMES-3091 Mailbox/Get POJO
Posted by rc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rcordier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 88d05d912c647fe4cad7e06d8c68a9d3d60195e6
Author: Rene Cordier <rc...@linagora.com>
AuthorDate: Mon May 11 17:42:47 2020 +0700
JAMES-3091 Mailbox/Get POJO
---
.../org/apache/james/jmap/mail/MailboxGet.scala | 40 ++++++++++++++++++++++
1 file changed, 40 insertions(+)
diff --git a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/mail/MailboxGet.scala b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/mail/MailboxGet.scala
new file mode 100644
index 0000000..ac2cc8a
--- /dev/null
+++ b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/mail/MailboxGet.scala
@@ -0,0 +1,40 @@
+/****************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one *
+ * or more contributor license agreements. See the NOTICE file *
+ * distributed with this work for additional information *
+ * regarding copyright ownership. The ASF licenses this file *
+ * to you under the Apache License, Version 2.0 (the *
+ * "License"); you may not use this file except in compliance *
+ * with the License. You may obtain a copy of the License at *
+ * *
+ * http://www.apache.org/licenses/LICENSE-2.0 *
+ * *
+ * Unless required by applicable law or agreed to in writing, *
+ * software distributed under the License is distributed on an *
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY *
+ * KIND, either express or implied. See the License for the *
+ * specific language governing permissions and limitations *
+ * under the License. *
+ ****************************************************************/
+
+package org.apache.james.jmap.mail
+
+import eu.timepit.refined.types.string.NonEmptyString
+import org.apache.james.jmap.model.AccountId
+import org.apache.james.jmap.model.State.State
+import org.apache.james.mailbox.model.MailboxId
+
+case class Ids(value: List[MailboxId])
+
+case class Properties(value: List[NonEmptyString])
+
+case class MailboxGetRequest(accountId: AccountId,
+ ids: Ids,
+ properties: Option[Properties])
+
+case class NotFound(value: List[MailboxId])
+
+case class MailboxGetResponse(accountId: AccountId,
+ state: State,
+ list: List[Mailbox],
+ notFound: NotFound)
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 02/09: JAMES-3149 EventBus caller should choose on
which scheduler he dispatch
Posted by rc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rcordier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit f2fbe4f0cc378394254be7b7a7e36cda290009e6
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Mon May 4 18:25:21 2020 +0700
JAMES-3149 EventBus caller should choose on which scheduler he dispatch
This avoid enforcing the thread context switch upon dispatch,
optimizing thread usage.
---
.../apache/james/mailbox/events/GroupContract.java | 5 ++-
.../james/mailbox/events/EventDispatcher.java | 7 +---
.../james/mailbox/store/StoreMailboxManager.java | 49 ++++++++++++----------
.../james/mailbox/store/StoreMessageIdManager.java | 28 ++++++++-----
.../james/mailbox/store/StoreMessageManager.java | 28 +++++++------
.../james/mailbox/store/StoreRightManager.java | 3 ++
6 files changed, 69 insertions(+), 51 deletions(-)
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 83ef2cf..dfe31ac 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
@@ -56,6 +56,7 @@ import org.junit.jupiter.api.Test;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.ImmutableSortedMap;
+
import reactor.core.scheduler.Schedulers;
public interface GroupContract {
@@ -158,7 +159,9 @@ public interface GroupContract {
AtomicBoolean successfulRetry = new AtomicBoolean(false);
MailboxListener listener = event -> {
if (event.getEventId().equals(EVENT_ID)) {
- eventBus().dispatch(EVENT_2, NO_KEYS).block();
+ eventBus().dispatch(EVENT_2, NO_KEYS)
+ .subscribeOn(Schedulers.elastic())
+ .block();
successfulRetry.set(true);
}
};
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 3de1fe0..c150184 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
@@ -46,7 +46,6 @@ import com.rabbitmq.client.AMQP;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoProcessor;
-import reactor.core.scheduler.Schedulers;
import reactor.rabbitmq.ExchangeSpecification;
import reactor.rabbitmq.OutboundMessage;
import reactor.rabbitmq.Sender;
@@ -95,7 +94,6 @@ public class EventDispatcher {
.concat(
dispatchToLocalListeners(event, keys),
dispatchToRemoteListeners(event, keys))
- .subscribeOn(Schedulers.elastic())
.doOnError(throwable -> LOGGER.error("error while dispatching event", throwable))
.then()
.subscribeWith(MonoProcessor.create());
@@ -106,7 +104,7 @@ public class EventDispatcher {
.flatMap(key -> localListenerRegistry.getLocalMailboxListeners(key)
.map(listener -> Tuples.of(key, listener)))
.filter(pair -> pair.getT2().getExecutionMode() == MailboxListener.ExecutionMode.SYNCHRONOUS)
- .flatMap(pair -> executeListener(event, pair.getT2(), pair.getT1())).subscribeOn(Schedulers.elastic())
+ .flatMap(pair -> executeListener(event, pair.getT2(), pair.getT1()))
.then();
}
@@ -161,8 +159,7 @@ public class EventDispatcher {
if (routingKeys.isEmpty()) {
return Mono.empty();
}
- return sender.send(toMessages(serializedEvent, routingKeys))
- .subscribeOn(Schedulers.elastic());
+ return sender.send(toMessages(serializedEvent, routingKeys));
}
private Flux<OutboundMessage> toMessages(byte[] serializedEvent, Collection<RoutingKey> routingKeys) {
diff --git a/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreMailboxManager.java b/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreMailboxManager.java
index 1f72b2c..bb68c0d 100644
--- a/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreMailboxManager.java
+++ b/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreMailboxManager.java
@@ -93,6 +93,7 @@ import com.google.common.collect.Iterables;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
/**
* This base class of an {@link MailboxManager} implementation provides a high-level api for writing your own
@@ -382,6 +383,7 @@ public class StoreMailboxManager implements MailboxManager {
.mailbox(mailbox)
.build(),
new MailboxIdRegistrationKey(mailbox.getMailboxId()))
+ .subscribeOn(Schedulers.elastic())
.block();
}));
} catch (MailboxExistsException e) {
@@ -451,14 +453,15 @@ public class StoreMailboxManager implements MailboxManager {
Mailbox m = new Mailbox(mailbox);
mailboxMapper.delete(mailbox);
eventBus.dispatch(EventFactory.mailboxDeleted()
- .randomEventId()
- .mailboxSession(session)
- .mailbox(mailbox)
- .quotaRoot(quotaRoot)
- .quotaCount(QuotaCountUsage.count(messageCount))
- .quotaSize(QuotaSizeUsage.size(totalSize))
- .build(),
- new MailboxIdRegistrationKey(mailbox.getMailboxId()))
+ .randomEventId()
+ .mailboxSession(session)
+ .mailbox(mailbox)
+ .quotaRoot(quotaRoot)
+ .quotaCount(QuotaCountUsage.count(messageCount))
+ .quotaSize(QuotaSizeUsage.size(totalSize))
+ .build(),
+ new MailboxIdRegistrationKey(mailbox.getMailboxId()))
+ .subscribeOn(Schedulers.elastic())
.block();
return m;
}
@@ -519,13 +522,14 @@ public class StoreMailboxManager implements MailboxManager {
mapper.rename(mailbox);
eventBus.dispatch(EventFactory.mailboxRenamed()
- .randomEventId()
- .mailboxSession(session)
- .mailboxId(mailbox.getMailboxId())
- .oldPath(from)
- .newPath(newMailboxPath)
- .build(),
- new MailboxIdRegistrationKey(mailbox.getMailboxId()))
+ .randomEventId()
+ .mailboxSession(session)
+ .mailboxId(mailbox.getMailboxId())
+ .oldPath(from)
+ .newPath(newMailboxPath)
+ .build(),
+ new MailboxIdRegistrationKey(mailbox.getMailboxId()))
+ .subscribeOn(Schedulers.elastic())
.block();
// rename submailboxes
@@ -543,13 +547,14 @@ public class StoreMailboxManager implements MailboxManager {
sub.setName(subNewName);
mapper.rename(sub);
eventBus.dispatch(EventFactory.mailboxRenamed()
- .randomEventId()
- .mailboxSession(session)
- .mailboxId(sub.getMailboxId())
- .oldPath(fromPath)
- .newPath(sub.generateAssociatedPath())
- .build(),
- new MailboxIdRegistrationKey(sub.getMailboxId()))
+ .randomEventId()
+ .mailboxSession(session)
+ .mailboxId(sub.getMailboxId())
+ .oldPath(fromPath)
+ .newPath(sub.generateAssociatedPath())
+ .build(),
+ new MailboxIdRegistrationKey(sub.getMailboxId()))
+ .subscribeOn(Schedulers.elastic())
.block();
LOGGER.debug("Rename mailbox sub-mailbox {} to {}", subOriginalName, subNewName);
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 3f87736..17c0477 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
@@ -80,6 +80,7 @@ import com.google.common.collect.Sets;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
public class StoreMessageIdManager implements MessageIdManager {
@@ -235,6 +236,7 @@ public class StoreMessageIdManager implements MessageIdManager {
new MailboxIdRegistrationKey(metadataWithMailboxId.getMailboxId())))
.sneakyThrow())
.then()
+ .subscribeOn(Schedulers.elastic())
.block();
}
@@ -305,6 +307,7 @@ public class StoreMessageIdManager implements MessageIdManager {
messageMoves.impactedMailboxIds()
.map(MailboxIdRegistrationKey::new)
.collect(Guavate.toImmutableSet()))
+ .subscribeOn(Schedulers.elastic())
.block();
}
@@ -322,6 +325,7 @@ public class StoreMessageIdManager implements MessageIdManager {
.addMetaData(eventPayload)
.build(),
new MailboxIdRegistrationKey(mailboxId))
+ .subscribeOn(Schedulers.elastic())
.block();
}
}
@@ -331,12 +335,13 @@ public class StoreMessageIdManager implements MessageIdManager {
Mailbox mailbox = mailboxSessionMapperFactory.getMailboxMapper(mailboxSession).findMailboxById(mailboxId);
eventBus.dispatch(EventFactory.flagsUpdated()
- .randomEventId()
- .mailboxSession(mailboxSession)
- .mailbox(mailbox)
- .updatedFlags(updatedFlags)
- .build(),
+ .randomEventId()
+ .mailboxSession(mailboxSession)
+ .mailbox(mailbox)
+ .updatedFlags(updatedFlags)
+ .build(),
new MailboxIdRegistrationKey(mailboxId))
+ .subscribeOn(Schedulers.elastic())
.block();
}
}
@@ -393,12 +398,13 @@ public class StoreMessageIdManager implements MessageIdManager {
save(messageIdMapper, copy);
eventBus.dispatch(EventFactory.added()
- .randomEventId()
- .mailboxSession(mailboxSession)
- .mailbox(mailboxMapper.findMailboxById(mailboxId))
- .addMetaData(copy.metaData())
- .build(),
- new MailboxIdRegistrationKey(mailboxId))
+ .randomEventId()
+ .mailboxSession(mailboxSession)
+ .mailbox(mailboxMapper.findMailboxById(mailboxId))
+ .addMetaData(copy.metaData())
+ .build(),
+ new MailboxIdRegistrationKey(mailboxId))
+ .subscribeOn(Schedulers.elastic())
.block();
}
}
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 9fd3ad4..8708344 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
@@ -289,6 +289,7 @@ public class StoreMessageManager implements MessageManager {
.metaData(ImmutableSortedMap.copyOf(deletedMessages))
.build(),
new MailboxIdRegistrationKey(mailbox.getMailboxId()))
+ .subscribeOn(Schedulers.elastic())
.block();
}
@@ -437,12 +438,12 @@ public class StoreMessageManager implements MessageManager {
Mailbox mailbox = getMailboxEntity();
eventBus.dispatch(EventFactory.added()
- .randomEventId()
- .mailboxSession(mailboxSession)
- .mailbox(mailbox)
- .addMetaData(data.getLeft())
- .build(),
- new MailboxIdRegistrationKey(mailbox.getMailboxId()))
+ .randomEventId()
+ .mailboxSession(mailboxSession)
+ .mailbox(mailbox)
+ .addMetaData(data.getLeft())
+ .build(),
+ new MailboxIdRegistrationKey(mailbox.getMailboxId()))
.subscribeOn(Schedulers.elastic())
.block();
MessageMetaData messageMetaData = data.getLeft();
@@ -592,12 +593,13 @@ public class StoreMessageManager implements MessageManager {
List<UpdatedFlags> updatedFlags = Iterators.toStream(it).collect(Guavate.toImmutableList());
eventBus.dispatch(EventFactory.flagsUpdated()
- .randomEventId()
- .mailboxSession(mailboxSession)
- .mailbox(getMailboxEntity())
- .updatedFlags(updatedFlags)
- .build(),
- new MailboxIdRegistrationKey(mailbox.getMailboxId()))
+ .randomEventId()
+ .mailboxSession(mailboxSession)
+ .mailbox(getMailboxEntity())
+ .updatedFlags(updatedFlags)
+ .build(),
+ new MailboxIdRegistrationKey(mailbox.getMailboxId()))
+ .subscribeOn(Schedulers.elastic())
.block();
return updatedFlags.stream().collect(Guavate.toImmutableMap(
@@ -759,6 +761,7 @@ public class StoreMessageManager implements MessageManager {
.messageId(messageIds.build())
.build(),
messageMoves.impactedMailboxIds().map(MailboxIdRegistrationKey::new).collect(Guavate.toImmutableSet())))
+ .subscribeOn(Schedulers.elastic())
.blockLast();
return copiedUids;
@@ -800,6 +803,7 @@ public class StoreMessageManager implements MessageManager {
.session(session)
.build(),
messageMoves.impactedMailboxIds().map(MailboxIdRegistrationKey::new).collect(Guavate.toImmutableSet())))
+ .subscribeOn(Schedulers.elastic())
.blockLast();
return moveUids;
diff --git a/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreRightManager.java b/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreRightManager.java
index 20f4779..6b4c205 100644
--- a/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreRightManager.java
+++ b/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreRightManager.java
@@ -54,6 +54,7 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableMap;
import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
public class StoreRightManager implements RightManager {
public static final boolean GROUP_FOLDER = true;
@@ -156,6 +157,7 @@ public class StoreRightManager implements RightManager {
.aclDiff(aclDiff)
.build(),
new MailboxIdRegistrationKey(mailbox.getMailboxId()))
+ .subscribeOn(Schedulers.elastic())
.block();
}
@@ -241,6 +243,7 @@ public class StoreRightManager implements RightManager {
.aclDiff(aclDiff)
.build(),
new MailboxIdRegistrationKey(mailbox.getMailboxId()))
+ .subscribeOn(Schedulers.elastic())
.block();
}
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org