You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@james.apache.org by bt...@apache.org on 2021/05/08 00:40:00 UTC
[james-project] branch master updated (b8577d3 -> c2ca11b)
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 b8577d3 JAMES-2813 Add a missing log upon Migration
new 20ca538 JAMES-3575 Make message deletion slightly more reactive
new da19351 JAMES-3575 MessageIdManager::delete should be reactive
new 0763ff05 JAMES-3575 MessageIdMapper::setFlags can easily be reactive
new d464711 JAMES-3575 Rewrite StoreMessageIdManager::setFlags in a more reactive way
new 42fc513 JAMES-3575 Reactive single flag update for JMAP RFC-8621
new bde8a85 JAMES-3575 Reactive single message move for JMAP RFC-8621
new eb9bd9c JAMES-3575 [REFACTORING] Flatten EmailSubmissionSet for/yield
new c2ca11b JAMES-3575 Reactify EmailSubmission/set
The 8 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:
.../org/apache/james/mailbox/MessageIdManager.java | 12 +-
.../cassandra/mail/CassandraMessageIdMapper.java | 16 +-
.../mail/CassandraMessageIdMapperTest.java | 3 +-
.../inmemory/mail/InMemoryMessageIdMapper.java | 9 +-
.../james/vault/DeletedMessageVaultHookTest.java | 6 +-
.../james/mailbox/store/StoreMessageIdManager.java | 162 +++++++++++----------
.../james/mailbox/store/mail/MessageIdMapper.java | 6 +-
.../store/AbstractCombinationManagerTest.java | 6 +-
.../AbstractMessageIdManagerSideEffectTest.java | 7 +-
.../store/AbstractMessageIdManagerStorageTest.java | 2 +-
.../store/mail/model/MessageIdMapperTest.java | 57 ++++----
.../SetMessagesOutboxFlagUpdateTest.java | 9 ++
.../methods/SetMessagesDestructionProcessor.java | 7 +-
.../jmap/method/EmailSetDeletePerformer.scala | 2 +-
.../jmap/method/EmailSetUpdatePerformer.scala | 9 +-
.../jmap/method/EmailSubmissionSetMethod.scala | 86 +++++------
.../java/org/apache/james/queue/api/MailQueue.java | 2 +
.../james/queue/file/FileCacheableMailQueue.java | 6 +
.../james/queue/jms/JMSCacheableMailQueue.java | 6 +
.../james/queue/memory/MemoryMailQueueFactory.java | 6 +
.../org/apache/james/queue/rabbitmq/Enqueuer.java | 7 +-
.../james/queue/rabbitmq/RabbitMQMailQueue.java | 15 +-
22 files changed, 264 insertions(+), 177 deletions(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 08/08: JAMES-3575 Reactify EmailSubmission/set
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 c2ca11b5eabfc23cea02685c1e2d70e9b96e23c8
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Sun May 2 07:38:13 2021 +0700
JAMES-3575 Reactify EmailSubmission/set
---
.../SetMessagesOutboxFlagUpdateTest.java | 9 ++++
.../jmap/method/EmailSubmissionSetMethod.scala | 60 ++++++++++++----------
.../java/org/apache/james/queue/api/MailQueue.java | 2 +
.../james/queue/file/FileCacheableMailQueue.java | 6 +++
.../james/queue/jms/JMSCacheableMailQueue.java | 6 +++
.../james/queue/memory/MemoryMailQueueFactory.java | 6 +++
.../org/apache/james/queue/rabbitmq/Enqueuer.java | 7 ++-
.../james/queue/rabbitmq/RabbitMQMailQueue.java | 15 ++++--
8 files changed, 76 insertions(+), 35 deletions(-)
diff --git a/server/protocols/jmap-draft-integration-testing/jmap-draft-integration-testing-common/src/test/java/org/apache/james/jmap/draft/methods/integration/SetMessagesOutboxFlagUpdateTest.java b/server/protocols/jmap-draft-integration-testing/jmap-draft-integration-testing-common/src/test/java/org/apache/james/jmap/draft/methods/integration/SetMessagesOutboxFlagUpdateTest.java
index bf21d1e..1dd258b 100644
--- a/server/protocols/jmap-draft-integration-testing/jmap-draft-integration-testing-common/src/test/java/org/apache/james/jmap/draft/methods/integration/SetMessagesOutboxFlagUpdateTest.java
+++ b/server/protocols/jmap-draft-integration-testing/jmap-draft-integration-testing-common/src/test/java/org/apache/james/jmap/draft/methods/integration/SetMessagesOutboxFlagUpdateTest.java
@@ -55,12 +55,16 @@ import org.apache.mailet.Mail;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
+import org.reactivestreams.Publisher;
+
+import com.github.fge.lambdas.Throwing;
import io.restassured.RestAssured;
import io.restassured.builder.RequestSpecBuilder;
import io.restassured.http.ContentType;
import io.restassured.parsing.Parser;
import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
public abstract class SetMessagesOutboxFlagUpdateTest {
private static final Username USERNAME = Username.of("username@" + DOMAIN);
@@ -94,6 +98,11 @@ public abstract class SetMessagesOutboxFlagUpdateTest {
}
@Override
+ public Publisher<Void> enqueueReactive(Mail mail) {
+ return Mono.fromRunnable(Throwing.runnable(() -> enQueue(mail)).sneakyThrow());
+ }
+
+ @Override
public void enQueue(Mail mail) {
}
diff --git a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/EmailSubmissionSetMethod.scala b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/EmailSubmissionSetMethod.scala
index 4c9edd3..f7c03a5 100644
--- a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/EmailSubmissionSetMethod.scala
+++ b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/EmailSubmissionSetMethod.scala
@@ -199,30 +199,38 @@ class EmailSubmissionSetMethod @Inject()(serializer: EmailSubmissionSetSerialize
private def create(request: EmailSubmissionSetRequest,
session: MailboxSession,
processingContext: ProcessingContext): SMono[(CreationResults, ProcessingContext)] =
- SFlux.fromIterable(request.create
- .getOrElse(Map.empty)
- .view)
- .fold((CreationResults(Nil), processingContext)) {
- (acc : (CreationResults, ProcessingContext), elem: (EmailSubmissionCreationId, JsObject)) => {
+ SFlux.fromIterable(
+ request.create
+ .getOrElse(Map.empty)
+ .view)
+ .fold[SMono[(CreationResults, ProcessingContext)]](SMono.just((CreationResults(Nil), processingContext))) {
+ (acc: SMono[(CreationResults, ProcessingContext)], elem: (EmailSubmissionCreationId, JsObject)) => {
val (emailSubmissionCreationId, jsObject) = elem
- val (creationResult, updatedProcessingContext) = createSubmission(session, emailSubmissionCreationId, jsObject, acc._2)
- (CreationResults(acc._1.created :+ creationResult), updatedProcessingContext)
+ acc.flatMap {
+ case (creationResults, processingContext) =>
+ createSubmission(session, emailSubmissionCreationId, jsObject, processingContext)
+ .map {
+ case (created, updatedProcessingContext) => CreationResults(creationResults.created :+ created) -> updatedProcessingContext
+ }
+ .switchIfEmpty(SMono.error(new RuntimeException("I should not be empty")))
+ }.cache()
}
}
+ .flatMap(x => x)
.subscribeOn(Schedulers.elastic())
private def createSubmission(mailboxSession: MailboxSession,
emailSubmissionCreationId: EmailSubmissionCreationId,
jsObject: JsObject,
- processingContext: ProcessingContext): (CreationResult, ProcessingContext) =
+ processingContext: ProcessingContext): SMono[(CreationResult, ProcessingContext)] =
parseCreate(jsObject)
- .flatMap(emailSubmissionCreationRequest => sendEmail(mailboxSession, emailSubmissionCreationRequest))
+ .fold(e => SMono.error(e), sendEmail(mailboxSession, _))
.map {
case (creationResponse, messageId) =>
- (creationResponse, messageId, recordCreationIdInProcessingContext(emailSubmissionCreationId, processingContext, creationResponse.id))
+ CreationSuccess(emailSubmissionCreationId, creationResponse, messageId) ->
+ recordCreationIdInProcessingContext(emailSubmissionCreationId, processingContext, creationResponse.id)
}
- .fold(e => (CreationFailure(emailSubmissionCreationId, e), processingContext),
- creation => CreationSuccess(emailSubmissionCreationId, creation._1, creation._2) -> creation._3)
+ .onErrorResume(e => SMono.just((CreationFailure(emailSubmissionCreationId, e), processingContext)))
private def parseCreate(jsObject: JsObject): Either[EmailSubmissionCreationParseException, EmailSubmissionCreationRequest] =
EmailSubmissionCreationRequest.validateProperties(jsObject)
@@ -240,18 +248,16 @@ class EmailSubmissionSetMethod @Inject()(serializer: EmailSubmissionSetSerialize
}
private def sendEmail(mailboxSession: MailboxSession,
- request: EmailSubmissionCreationRequest): Either[Throwable, (EmailSubmissionCreationResponse, MessageId)] = {
- val result = for {
- message <- messageIdManager.getMessage(request.emailId, FetchGroup.FULL_CONTENT, mailboxSession)
- .asScala
- .toList
- .headOption
- .toRight(MessageNotFoundException(request.emailId))
+ request: EmailSubmissionCreationRequest): SMono[(EmailSubmissionCreationResponse, MessageId)] =
+ for {
+ message <- SFlux(messageIdManager.getMessagesReactive(List(request.emailId).asJava, FetchGroup.FULL_CONTENT, mailboxSession))
+ .next
+ .switchIfEmpty(SMono.error(MessageNotFoundException(request.emailId)))
submissionId = EmailSubmissionId.generate
- message <- toMimeMessage(submissionId.value, message).toEither
- envelope <- resolveEnvelope(message, request.envelope).toEither
- validation <- validate(mailboxSession)(message, envelope).toEither
- mail <- Try({
+ message <- SMono.fromTry(toMimeMessage(submissionId.value, message))
+ envelope <- SMono.fromTry(resolveEnvelope(message, request.envelope))
+ validation <- SMono.fromTry(validate(mailboxSession)(message, envelope))
+ mail <- SMono.fromCallable(() => {
val mailImpl = MailImpl.builder()
.name(submissionId.value)
.addRecipients(envelope.rcptTo.map(_.email).asJava)
@@ -260,13 +266,11 @@ class EmailSubmissionSetMethod @Inject()(serializer: EmailSubmissionSetSerialize
.build()
mailImpl.setMessageNoCopy(message)
mailImpl
- }).toEither
- enqueue <- Try(queue.enQueue(mail)).toEither
+ })
+ enqueue <- SMono(queue.enqueueReactive(mail)).`then`(SMono.just(submissionId))
} yield {
- EmailSubmissionCreationResponse(submissionId)
+ EmailSubmissionCreationResponse(submissionId) -> request.emailId
}
- result.map(response => (response, request.emailId))
- }
private def toMimeMessage(name: String, message: MessageResult): Try[MimeMessageWrapper] = {
val source = MessageMimeMessageSource(name, message)
diff --git a/server/queue/queue-api/src/main/java/org/apache/james/queue/api/MailQueue.java b/server/queue/queue-api/src/main/java/org/apache/james/queue/api/MailQueue.java
index 38b8c3f..a39990c 100644
--- a/server/queue/queue-api/src/main/java/org/apache/james/queue/api/MailQueue.java
+++ b/server/queue/queue-api/src/main/java/org/apache/james/queue/api/MailQueue.java
@@ -105,6 +105,8 @@ public interface MailQueue extends Closeable {
*/
void enQueue(Mail mail) throws MailQueueException;
+ Publisher<Void> enqueueReactive(Mail mail);
+
/**
* Dequeue the next ready-to-process Mail of the queue. This method will
* block until a Mail is ready and then process the operation.
diff --git a/server/queue/queue-file/src/main/java/org/apache/james/queue/file/FileCacheableMailQueue.java b/server/queue/queue-file/src/main/java/org/apache/james/queue/file/FileCacheableMailQueue.java
index d8b0676..146ddb2 100644
--- a/server/queue/queue-file/src/main/java/org/apache/james/queue/file/FileCacheableMailQueue.java
+++ b/server/queue/queue-file/src/main/java/org/apache/james/queue/file/FileCacheableMailQueue.java
@@ -61,6 +61,7 @@ import org.apache.mailet.AttributeName;
import org.apache.mailet.AttributeUtils;
import org.apache.mailet.AttributeValue;
import org.apache.mailet.Mail;
+import org.reactivestreams.Publisher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -121,6 +122,11 @@ public class FileCacheableMailQueue implements ManageableMailQueue {
return queueName;
}
+ @Override
+ public Publisher<Void> enqueueReactive(Mail mail) {
+ return Mono.fromRunnable(Throwing.runnable(() -> enQueue(mail)).sneakyThrow());
+ }
+
private void init() throws IOException {
for (int i = 1; i <= SPLITCOUNT; i++) {
diff --git a/server/queue/queue-jms/src/main/java/org/apache/james/queue/jms/JMSCacheableMailQueue.java b/server/queue/queue-jms/src/main/java/org/apache/james/queue/jms/JMSCacheableMailQueue.java
index 56b374d..e60e9fa 100644
--- a/server/queue/queue-jms/src/main/java/org/apache/james/queue/jms/JMSCacheableMailQueue.java
+++ b/server/queue/queue-jms/src/main/java/org/apache/james/queue/jms/JMSCacheableMailQueue.java
@@ -74,6 +74,7 @@ import org.apache.mailet.AttributeUtils;
import org.apache.mailet.AttributeValue;
import org.apache.mailet.Mail;
import org.apache.mailet.PerRecipientHeaders;
+import org.reactivestreams.Publisher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -305,6 +306,11 @@ public class JMSCacheableMailQueue implements ManageableMailQueue, JMSSupport, M
enQueue(mail, NO_DELAY, TimeUnit.MILLISECONDS);
}
+ @Override
+ public Publisher<Void> enqueueReactive(Mail mail) {
+ return Mono.fromRunnable(Throwing.runnable(() -> enQueue(mail)).sneakyThrow());
+ }
+
/**
* Produce the mail to the JMS Queue
*/
diff --git a/server/queue/queue-memory/src/main/java/org/apache/james/queue/memory/MemoryMailQueueFactory.java b/server/queue/queue-memory/src/main/java/org/apache/james/queue/memory/MemoryMailQueueFactory.java
index 025a024..a941b91 100644
--- a/server/queue/queue-memory/src/main/java/org/apache/james/queue/memory/MemoryMailQueueFactory.java
+++ b/server/queue/queue-memory/src/main/java/org/apache/james/queue/memory/MemoryMailQueueFactory.java
@@ -48,6 +48,7 @@ import org.apache.james.queue.api.MailQueueName;
import org.apache.james.queue.api.ManageableMailQueue;
import org.apache.james.server.core.MailImpl;
import org.apache.mailet.Mail;
+import org.reactivestreams.Publisher;
import org.threeten.extra.Temporals;
import com.github.fge.lambdas.Throwing;
@@ -127,6 +128,11 @@ public class MemoryMailQueueFactory implements MailQueueFactory<MemoryMailQueueF
}
}
+ @Override
+ public Publisher<Void> enqueueReactive(Mail mail) {
+ return Mono.fromRunnable(Throwing.runnable(() -> enQueue(mail)).sneakyThrow());
+ }
+
private ZonedDateTime calculateNextDelivery(Duration delay) {
if (!delay.isNegative()) {
try {
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/Enqueuer.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/Enqueuer.java
index 13db251..82fbc18 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/Enqueuer.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/Enqueuer.java
@@ -68,9 +68,9 @@ class Enqueuer {
this.enqueueMetric = metricFactory.generate(ENQUEUED_METRIC_NAME_PREFIX + name.asString());
}
- void enQueue(Mail mail) throws MailQueue.MailQueueException {
+ Mono<Void> enQueue(Mail mail) throws MailQueue.MailQueueException {
EnqueueId enqueueId = EnqueueId.generate();
- saveMail(mail)
+ return saveMail(mail)
.map(partIds -> new MailReference(enqueueId, mail, partIds))
.flatMap(Throwing.<MailReference, Mono<Void>>function(mailReference -> {
EnqueuedItem enqueuedItem = toEnqueuedItems(mailReference);
@@ -79,8 +79,7 @@ class Enqueuer {
publishReferenceToRabbit(mailReference))
.then();
}).sneakyThrow())
- .thenEmpty(Mono.fromRunnable(enqueueMetric::increment))
- .block();
+ .thenEmpty(Mono.fromRunnable(enqueueMetric::increment));
}
Mono<Void> reQueue(CassandraMailQueueBrowser.CassandraMailQueueItemView item) {
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueue.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueue.java
index 52c8660..cd233fb 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueue.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueue.java
@@ -31,10 +31,10 @@ import org.apache.james.queue.rabbitmq.view.api.DeleteCondition;
import org.apache.james.queue.rabbitmq.view.api.MailQueueView;
import org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueBrowser;
import org.apache.mailet.Mail;
+import org.reactivestreams.Publisher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.github.fge.lambdas.Throwing;
import com.google.common.base.Function;
import com.google.common.base.MoreObjects;
@@ -83,8 +83,17 @@ public class RabbitMQMailQueue implements ManageableMailQueue {
@Override
public void enQueue(Mail mail) {
- metricFactory.runPublishingTimerMetric(ENQUEUED_TIMER_METRIC_NAME_PREFIX + name.asString(),
- Throwing.runnable(() -> enqueuer.enQueue(mail)).sneakyThrow());
+ Mono.from(enqueueReactive(mail)).block();
+ }
+
+ @Override
+ public Publisher<Void> enqueueReactive(Mail mail) {
+ try {
+ return metricFactory.decoratePublisherWithTimerMetric(ENQUEUED_TIMER_METRIC_NAME_PREFIX + name.asString(),
+ enqueuer.enQueue(mail));
+ } catch (MailQueueException e) {
+ return Mono.error(e);
+ }
}
@Override
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 03/08: JAMES-3575 MessageIdMapper::setFlags can
easily be reactive
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 0763ff05e653d815522eede47f2b54e9eacf6cb8
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Sat May 1 22:26:41 2021 +0700
JAMES-3575 MessageIdMapper::setFlags can easily be reactive
---
.../cassandra/mail/CassandraMessageIdMapper.java | 5 +-
.../mail/CassandraMessageIdMapperTest.java | 3 +-
.../inmemory/mail/InMemoryMessageIdMapper.java | 9 ++--
.../james/mailbox/store/StoreMessageIdManager.java | 4 +-
.../james/mailbox/store/mail/MessageIdMapper.java | 2 +-
.../store/mail/model/MessageIdMapperTest.java | 57 ++++++++++++----------
6 files changed, 44 insertions(+), 36 deletions(-)
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdMapper.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdMapper.java
index 5d5972e..59ea987 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdMapper.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdMapper.java
@@ -254,14 +254,13 @@ public class CassandraMessageIdMapper implements MessageIdMapper {
}
@Override
- public Multimap<MailboxId, UpdatedFlags> setFlags(MessageId messageId, List<MailboxId> mailboxIds, Flags newState, MessageManager.FlagsUpdateMode updateMode) throws MailboxException {
+ public Mono<Multimap<MailboxId, UpdatedFlags>> setFlags(MessageId messageId, List<MailboxId> mailboxIds, Flags newState, MessageManager.FlagsUpdateMode updateMode) throws MailboxException {
return Flux.fromIterable(mailboxIds)
.distinct()
.map(mailboxId -> (CassandraId) mailboxId)
.concatMap(mailboxId -> flagsUpdateWithRetry(newState, updateMode, mailboxId, messageId))
.flatMap(this::updateCounts, ReactorUtils.DEFAULT_CONCURRENCY)
- .collect(Guavate.toImmutableListMultimap(Pair::getLeft, Pair::getRight))
- .block();
+ .collect(Guavate.toImmutableListMultimap(Pair::getLeft, Pair::getRight));
}
private Flux<Pair<MailboxId, UpdatedFlags>> flagsUpdateWithRetry(Flags newState, MessageManager.FlagsUpdateMode updateMode, MailboxId mailboxId, MessageId messageId) {
diff --git a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdMapperTest.java b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdMapperTest.java
index 85dd03d..5261cc2 100644
--- a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdMapperTest.java
+++ b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdMapperTest.java
@@ -118,7 +118,8 @@ class CassandraMessageIdMapperTest extends MessageIdMapperTest {
mapperFactory.getMessageIdMapper(MAILBOX_SESSION).setFlags(message1.getMessageId(),
ImmutableList.of(message1.getMailboxId()),
new Flags(Flags.Flag.DELETED),
- MessageManager.FlagsUpdateMode.REPLACE);
+ MessageManager.FlagsUpdateMode.REPLACE)
+ .block();
assertThat(statementRecorder.listExecutedStatements(
StatementRecorder.Selector.preparedStatementStartingWith("SELECT messageId,mailboxId,uid,modSeq,flagAnswered,flagDeleted," +
diff --git a/mailbox/memory/src/main/java/org/apache/james/mailbox/inmemory/mail/InMemoryMessageIdMapper.java b/mailbox/memory/src/main/java/org/apache/james/mailbox/inmemory/mail/InMemoryMessageIdMapper.java
index 626640a..8be1835 100644
--- a/mailbox/memory/src/main/java/org/apache/james/mailbox/inmemory/mail/InMemoryMessageIdMapper.java
+++ b/mailbox/memory/src/main/java/org/apache/james/mailbox/inmemory/mail/InMemoryMessageIdMapper.java
@@ -52,6 +52,7 @@ import com.google.common.collect.ImmutableList;
import com.google.common.collect.Multimap;
import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
public class InMemoryMessageIdMapper implements MessageIdMapper {
private final MailboxMapper mailboxMapper;
@@ -132,16 +133,16 @@ public class InMemoryMessageIdMapper implements MessageIdMapper {
}
@Override
- public Multimap<MailboxId, UpdatedFlags> setFlags(MessageId messageId, List<MailboxId> mailboxIds,
- Flags newState, FlagsUpdateMode updateMode) {
- return find(ImmutableList.of(messageId), MessageMapper.FetchType.Metadata)
+ public Mono<Multimap<MailboxId, UpdatedFlags>> setFlags(MessageId messageId, List<MailboxId> mailboxIds,
+ Flags newState, FlagsUpdateMode updateMode) {
+ return Mono.fromCallable(() -> find(ImmutableList.of(messageId), MessageMapper.FetchType.Metadata)
.stream()
.filter(message -> mailboxIds.contains(message.getMailboxId()))
.map(updateMessage(newState, updateMode))
.distinct()
.collect(Guavate.toImmutableListMultimap(
Pair::getKey,
- Pair::getValue));
+ Pair::getValue)));
}
private Function<MailboxMessage, Pair<MailboxId, UpdatedFlags>> updateMessage(Flags newState, FlagsUpdateMode updateMode) {
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 116e91b..6ab1ffc 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
@@ -133,7 +133,9 @@ public class StoreMessageIdManager implements MessageIdManager {
assertRightsOnMailboxes(targetMailboxes, mailboxSession, Right.Write);
- Multimap<MailboxId, UpdatedFlags> updatedFlags = messageIdMapper.setFlags(messageId, mailboxIds, newState, replace);
+ Multimap<MailboxId, UpdatedFlags> updatedFlags = messageIdMapper.setFlags(messageId, mailboxIds, newState, replace)
+ .subscribeOn(Schedulers.elastic())
+ .block();
for (Map.Entry<MailboxId, Collection<UpdatedFlags>> entry : updatedFlags.asMap().entrySet()) {
dispatchFlagsChange(mailboxSession, entry.getKey(), ImmutableList.copyOf(entry.getValue()), targetMailboxes);
}
diff --git a/mailbox/store/src/main/java/org/apache/james/mailbox/store/mail/MessageIdMapper.java b/mailbox/store/src/main/java/org/apache/james/mailbox/store/mail/MessageIdMapper.java
index 7bdafbc..02e1726 100644
--- a/mailbox/store/src/main/java/org/apache/james/mailbox/store/mail/MessageIdMapper.java
+++ b/mailbox/store/src/main/java/org/apache/james/mailbox/store/mail/MessageIdMapper.java
@@ -82,5 +82,5 @@ public interface MessageIdMapper {
* @return Metadata of the update, indexed by mailboxIds.
* @throws MailboxException
*/
- Multimap<MailboxId, UpdatedFlags> setFlags(MessageId messageId, List<MailboxId> mailboxIds, Flags newState, MessageManager.FlagsUpdateMode updateMode) throws MailboxException;
+ Mono<Multimap<MailboxId, UpdatedFlags>> setFlags(MessageId messageId, List<MailboxId> mailboxIds, Flags newState, MessageManager.FlagsUpdateMode updateMode) throws MailboxException;
}
diff --git a/mailbox/store/src/test/java/org/apache/james/mailbox/store/mail/model/MessageIdMapperTest.java b/mailbox/store/src/test/java/org/apache/james/mailbox/store/mail/model/MessageIdMapperTest.java
index e30a2ac..0c61eb0 100644
--- a/mailbox/store/src/test/java/org/apache/james/mailbox/store/mail/model/MessageIdMapperTest.java
+++ b/mailbox/store/src/test/java/org/apache/james/mailbox/store/mail/model/MessageIdMapperTest.java
@@ -350,7 +350,7 @@ public abstract class MessageIdMapperTest {
MessageId messageId = message1.getMessageId();
Flags newFlags = new Flags(Flag.ANSWERED);
- Multimap<MailboxId, UpdatedFlags> flags = sut.setFlags(messageId, ImmutableList.of(message1.getMailboxId()), newFlags, FlagsUpdateMode.ADD);
+ Multimap<MailboxId, UpdatedFlags> flags = sut.setFlags(messageId, ImmutableList.of(message1.getMailboxId()), newFlags, FlagsUpdateMode.ADD).block();
ModSeq modSeq = mapperProvider.highestModSeq(benwaInboxMailbox);
UpdatedFlags expectedUpdatedFlags = UpdatedFlags.builder()
@@ -378,7 +378,7 @@ public abstract class MessageIdMapperTest {
.add("userflag")
.build();
- Multimap<MailboxId, UpdatedFlags> flags = sut.setFlags(messageId, ImmutableList.of(message1.getMailboxId()), newFlags, FlagsUpdateMode.REPLACE);
+ Multimap<MailboxId, UpdatedFlags> flags = sut.setFlags(messageId, ImmutableList.of(message1.getMailboxId()), newFlags, FlagsUpdateMode.REPLACE).block();
ModSeq modSeq = mapperProvider.highestModSeq(benwaInboxMailbox);
UpdatedFlags expectedUpdatedFlags = UpdatedFlags.builder()
@@ -408,7 +408,7 @@ public abstract class MessageIdMapperTest {
.add("userflag")
.build();
- Multimap<MailboxId, UpdatedFlags> flags = sut.setFlags(messageId, ImmutableList.of(message1.getMailboxId()), newFlags, FlagsUpdateMode.REMOVE);
+ Multimap<MailboxId, UpdatedFlags> flags = sut.setFlags(messageId, ImmutableList.of(message1.getMailboxId()), newFlags, FlagsUpdateMode.REMOVE).block();
ModSeq modSeq = mapperProvider.highestModSeq(benwaInboxMailbox);
UpdatedFlags expectedUpdatedFlags = UpdatedFlags.builder()
@@ -438,7 +438,7 @@ public abstract class MessageIdMapperTest {
.add("userflag")
.build();
- sut.setFlags(messageId, ImmutableList.of(message1.getMailboxId()), newFlags, FlagsUpdateMode.REMOVE);
+ sut.setFlags(messageId, ImmutableList.of(message1.getMailboxId()), newFlags, FlagsUpdateMode.REMOVE).block();
List<MailboxMessage> messages = sut.find(ImmutableList.of(messageId), MessageMapper.FetchType.Body);
assertThat(messages).hasSize(1);
@@ -454,7 +454,7 @@ public abstract class MessageIdMapperTest {
MessageId messageId = message1.getMessageId();
Flags newFlags = new Flags(Flag.ANSWERED);
- Multimap<MailboxId, UpdatedFlags> flags = sut.setFlags(messageId, ImmutableList.of(), newFlags, FlagsUpdateMode.REMOVE);
+ Multimap<MailboxId, UpdatedFlags> flags = sut.setFlags(messageId, ImmutableList.of(), newFlags, FlagsUpdateMode.REMOVE).block();
assertThat(flags.asMap()).isEmpty();
}
@@ -462,7 +462,7 @@ public abstract class MessageIdMapperTest {
@Test
void setFlagsShouldReturnEmptyWhenMessageIdDoesntExist() throws Exception {
MessageId unknownMessageId = mapperProvider.generateMessageId();
- Multimap<MailboxId, UpdatedFlags> flags = sut.setFlags(unknownMessageId, ImmutableList.of(message1.getMailboxId()), new Flags(Flag.RECENT), FlagsUpdateMode.REMOVE);
+ Multimap<MailboxId, UpdatedFlags> flags = sut.setFlags(unknownMessageId, ImmutableList.of(message1.getMailboxId()), new Flags(Flag.RECENT), FlagsUpdateMode.REMOVE).block();
assertThat(flags.asMap()).isEmpty();
}
@@ -477,7 +477,7 @@ public abstract class MessageIdMapperTest {
MessageId messageId = message1.getMessageId();
- Multimap<MailboxId, UpdatedFlags> flags = sut.setFlags(messageId, ImmutableList.of(message1.getMailboxId()), new Flags(Flag.ANSWERED), FlagsUpdateMode.ADD);
+ Multimap<MailboxId, UpdatedFlags> flags = sut.setFlags(messageId, ImmutableList.of(message1.getMailboxId()), new Flags(Flag.ANSWERED), FlagsUpdateMode.ADD).block();
Flags newFlags = new FlagsBuilder()
.add(Flag.RECENT)
@@ -508,7 +508,7 @@ public abstract class MessageIdMapperTest {
MessageId messageId = message1.getMessageId();
Flags newFlags = new Flags(Flag.ANSWERED);
- Multimap<MailboxId, UpdatedFlags> flags = sut.setFlags(messageId, ImmutableList.of(message1.getMailboxId(), message1InOtherMailbox.getMailboxId()), newFlags, FlagsUpdateMode.ADD);
+ Multimap<MailboxId, UpdatedFlags> flags = sut.setFlags(messageId, ImmutableList.of(message1.getMailboxId(), message1InOtherMailbox.getMailboxId()), newFlags, FlagsUpdateMode.ADD).block();
ModSeq modSeqBenwaInboxMailbox = mapperProvider.highestModSeq(benwaInboxMailbox);
ModSeq modSeqBenwaWorkMailbox = mapperProvider.highestModSeq(benwaWorkMailbox);
@@ -538,7 +538,7 @@ public abstract class MessageIdMapperTest {
sut.save(message1);
MessageId messageId = message1.getMessageId();
- sut.setFlags(messageId, ImmutableList.of(message1.getMailboxId()), new Flags(Flag.ANSWERED), FlagsUpdateMode.ADD);
+ sut.setFlags(messageId, ImmutableList.of(message1.getMailboxId()), new Flags(Flag.ANSWERED), FlagsUpdateMode.ADD).block();
List<MailboxMessage> messages = sut.find(ImmutableList.of(messageId), MessageMapper.FetchType.Body);
assertThat(messages).hasSize(1);
@@ -554,7 +554,7 @@ public abstract class MessageIdMapperTest {
MessageId messageId = message1.getMessageId();
Flags newFlags = new Flags(Flag.ANSWERED);
- sut.setFlags(messageId, ImmutableList.of(), newFlags, FlagsUpdateMode.REMOVE);
+ sut.setFlags(messageId, ImmutableList.of(), newFlags, FlagsUpdateMode.REMOVE).block();
List<MailboxMessage> messages = sut.find(ImmutableList.of(messageId), MessageMapper.FetchType.Body);
assertThat(messages).hasSize(1);
@@ -569,7 +569,7 @@ public abstract class MessageIdMapperTest {
sut.save(message1);
MessageId messageId = message1.getMessageId();
- sut.setFlags(messageId, ImmutableList.of(message1.getMailboxId()), new Flags(Flag.ANSWERED), FlagsUpdateMode.ADD);
+ sut.setFlags(messageId, ImmutableList.of(message1.getMailboxId()), new Flags(Flag.ANSWERED), FlagsUpdateMode.ADD).block();
List<MailboxMessage> messages = sut.find(ImmutableList.of(messageId), MessageMapper.FetchType.Body);
assertThat(messages).hasSize(1);
@@ -587,7 +587,7 @@ public abstract class MessageIdMapperTest {
MessageId messageId = message1.getMessageId();
Flags newFlags = new Flags(Flag.ANSWERED);
- sut.setFlags(messageId, ImmutableList.of(), newFlags, FlagsUpdateMode.REMOVE);
+ sut.setFlags(messageId, ImmutableList.of(), newFlags, FlagsUpdateMode.REMOVE).block();
List<MailboxMessage> messages = sut.find(ImmutableList.of(messageId), MessageMapper.FetchType.Body);
assertThat(messages).hasSize(1);
@@ -606,7 +606,7 @@ public abstract class MessageIdMapperTest {
sut.save(message1InOtherMailbox);
MessageId messageId = message1.getMessageId();
- sut.setFlags(messageId, ImmutableList.of(message1.getMailboxId(), message1InOtherMailbox.getMailboxId()), new Flags(Flag.ANSWERED), FlagsUpdateMode.ADD);
+ sut.setFlags(messageId, ImmutableList.of(message1.getMailboxId(), message1InOtherMailbox.getMailboxId()), new Flags(Flag.ANSWERED), FlagsUpdateMode.ADD).block();
List<MailboxMessage> messages = sut.find(ImmutableList.of(messageId), MessageMapper.FetchType.Body);
assertThat(messages).hasSize(2);
@@ -630,7 +630,7 @@ public abstract class MessageIdMapperTest {
sut.save(message4);
MessageId messageId = message1.getMessageId();
- sut.setFlags(messageId, ImmutableList.of(message1.getMailboxId()), new Flags(Flag.ANSWERED), FlagsUpdateMode.ADD);
+ sut.setFlags(messageId, ImmutableList.of(message1.getMailboxId()), new Flags(Flag.ANSWERED), FlagsUpdateMode.ADD).block();
List<MailboxMessage> messages = sut.find(ImmutableList.of(messageId), MessageMapper.FetchType.Body);
assertThat(messages).hasSize(1);
@@ -653,7 +653,7 @@ public abstract class MessageIdMapperTest {
sut.save(message4);
MessageId messageId = message1.getMessageId();
- sut.setFlags(messageId, ImmutableList.of(message1.getMailboxId(), message1.getMailboxId()), new Flags(Flag.ANSWERED), FlagsUpdateMode.ADD);
+ sut.setFlags(messageId, ImmutableList.of(message1.getMailboxId(), message1.getMailboxId()), new Flags(Flag.ANSWERED), FlagsUpdateMode.ADD).block();
List<MailboxMessage> messages = sut.find(ImmutableList.of(messageId), MessageMapper.FetchType.Body);
assertThat(messages).hasSize(1);
@@ -673,7 +673,7 @@ public abstract class MessageIdMapperTest {
.operation((threadNumber, step) -> sut.setFlags(message1.getMessageId(),
ImmutableList.of(message1.getMailboxId()),
new Flags("custom-" + threadNumber + "-" + step),
- FlagsUpdateMode.ADD))
+ FlagsUpdateMode.ADD).block())
.threadCount(threadCount)
.operationCount(updateCount)
.runSuccessfullyWithin(Duration.ofMinutes(1));
@@ -698,12 +698,14 @@ public abstract class MessageIdMapperTest {
sut.setFlags(message1.getMessageId(),
ImmutableList.of(message1.getMailboxId()),
new Flags("custom-" + threadNumber + "-" + step),
- FlagsUpdateMode.ADD);
+ FlagsUpdateMode.ADD)
+ .block();
} else {
sut.setFlags(message1.getMessageId(),
ImmutableList.of(message1.getMailboxId()),
new Flags("custom-" + threadNumber + "-" + (updateCount - step - 1)),
- FlagsUpdateMode.REMOVE);
+ FlagsUpdateMode.REMOVE)
+ .block();
}
})
.threadCount(threadCount)
@@ -787,7 +789,7 @@ public abstract class MessageIdMapperTest {
message1.setModSeq(mapperProvider.generateModSeq(benwaInboxMailbox));
sut.save(message1);
- sut.setFlags(message1.getMessageId(), ImmutableList.of(message1.getMailboxId()), new Flags(Flag.SEEN), FlagsUpdateMode.ADD);
+ sut.setFlags(message1.getMessageId(), ImmutableList.of(message1.getMailboxId()), new Flags(Flag.SEEN), FlagsUpdateMode.ADD).block();
assertThat(messageMapper.getMailboxCounters(benwaInboxMailbox).getUnseen()).isEqualTo(0);
}
@@ -799,7 +801,7 @@ public abstract class MessageIdMapperTest {
message1.setFlags(new Flags(Flag.SEEN));
sut.save(message1);
- sut.setFlags(message1.getMessageId(), ImmutableList.of(message1.getMailboxId()), new Flags(Flag.SEEN), FlagsUpdateMode.REMOVE);
+ sut.setFlags(message1.getMessageId(), ImmutableList.of(message1.getMailboxId()), new Flags(Flag.SEEN), FlagsUpdateMode.REMOVE).block();
assertThat(messageMapper.getMailboxCounters(benwaInboxMailbox).getUnseen()).isEqualTo(1);
}
@@ -815,7 +817,8 @@ public abstract class MessageIdMapperTest {
sut.setFlags(message1.getMessageId(),
ImmutableList.of(message1.getMailboxId()),
new Flags(Flag.SEEN),
- FlagsUpdateMode.ADD);
+ FlagsUpdateMode.ADD)
+ .block();
List<MailboxMessage> messages = sut.find(ImmutableList.of(message1.getMessageId()), MessageMapper.FetchType.Body);
assertThat(messages).hasSize(1);
@@ -834,7 +837,8 @@ public abstract class MessageIdMapperTest {
sut.setFlags(message1.getMessageId(),
ImmutableList.of(message1.getMailboxId()),
flags,
- FlagsUpdateMode.ADD);
+ FlagsUpdateMode.ADD)
+ .block();
List<MailboxMessage> messages = sut.find(ImmutableList.of(message1.getMessageId()), MessageMapper.FetchType.Body);
assertThat(messages).hasSize(1);
@@ -853,7 +857,8 @@ public abstract class MessageIdMapperTest {
Multimap<MailboxId, UpdatedFlags> mailboxIdUpdatedFlagsMap = sut.setFlags(message1.getMessageId(),
ImmutableList.of(message1.getMailboxId()),
flags,
- FlagsUpdateMode.ADD);
+ FlagsUpdateMode.ADD)
+ .block();
assertThat(mailboxIdUpdatedFlagsMap.asMap())
.containsOnly(MapEntry.entry(message1.getMailboxId(),
@@ -873,7 +878,7 @@ public abstract class MessageIdMapperTest {
message1.setFlags(new Flags(Flag.RECENT));
sut.save(message1);
- sut.setFlags(message1.getMessageId(), ImmutableList.of(message1.getMailboxId()), new Flags(Flag.ANSWERED), FlagsUpdateMode.REMOVE);
+ sut.setFlags(message1.getMessageId(), ImmutableList.of(message1.getMailboxId()), new Flags(Flag.ANSWERED), FlagsUpdateMode.REMOVE).block();
assertThat(messageMapper.getMailboxCounters(benwaInboxMailbox).getUnseen()).isEqualTo(1);
}
@@ -927,7 +932,7 @@ public abstract class MessageIdMapperTest {
addMessageAndSetModSeq(benwaInboxMailbox, message1);
addMessageAndSetModSeq(benwaInboxMailbox, message1);
- sut.setFlags(message1.getMessageId(), ImmutableList.of(message1.getMailboxId()), new Flags(Flag.ANSWERED), FlagsUpdateMode.ADD);
+ sut.setFlags(message1.getMessageId(), ImmutableList.of(message1.getMailboxId()), new Flags(Flag.ANSWERED), FlagsUpdateMode.ADD).block();
assertThat(sut.find(ImmutableList.of(message1.getMessageId()), FetchType.Metadata))
.extracting(MailboxMessage::createFlags)
@@ -941,7 +946,7 @@ public abstract class MessageIdMapperTest {
addMessageAndSetModSeq(benwaInboxMailbox, message1);
addMessageAndSetModSeq(benwaInboxMailbox, message1);
- Multimap<MailboxId, UpdatedFlags> map = sut.setFlags(message1.getMessageId(), ImmutableList.of(message1.getMailboxId()), new Flags(Flag.ANSWERED), FlagsUpdateMode.ADD);
+ Multimap<MailboxId, UpdatedFlags> map = sut.setFlags(message1.getMessageId(), ImmutableList.of(message1.getMailboxId()), new Flags(Flag.ANSWERED), FlagsUpdateMode.ADD).block();
assertThat(map.values()).hasSize(2);
assertThat(map.asMap()).hasSize(1);
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 01/08: JAMES-3575 Make message deletion slightly
more reactive
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 20ca53821b0342b0bc12d466c1450534e3b197e1
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Sat May 1 22:06:42 2021 +0700
JAMES-3575 Make message deletion slightly more reactive
---
.../cassandra/mail/CassandraMessageIdMapper.java | 11 +++--
.../james/mailbox/store/StoreMessageIdManager.java | 50 +++++++++++-----------
.../james/mailbox/store/mail/MessageIdMapper.java | 4 ++
3 files changed, 37 insertions(+), 28 deletions(-)
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdMapper.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdMapper.java
index 1224932..5d5972e 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdMapper.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdMapper.java
@@ -217,13 +217,18 @@ public class CassandraMessageIdMapper implements MessageIdMapper {
@Override
public void delete(Multimap<MessageId, MailboxId> ids) {
- Flux.fromIterable(ids.asMap()
+ deleteReactive(ids)
+ .block();
+ }
+
+
+ public Mono<Void> deleteReactive(Multimap<MessageId, MailboxId> ids) {
+ return Flux.fromIterable(ids.asMap()
.entrySet())
.publishOn(Schedulers.elastic())
.flatMap(entry -> deleteReactive(entry.getKey(), entry.getValue()), cassandraConfiguration.getExpungeChunkSize(),
DEFAULT_CONCURRENCY)
- .then()
- .block();
+ .then();
}
private Mono<Void> retrieveAndDeleteIndices(CassandraMessageId messageId, Optional<CassandraId> mailboxId) {
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 e40d42d..d34b4a7 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
@@ -206,7 +206,9 @@ public class StoreMessageIdManager implements MessageIdManager {
.collect(Guavate.toImmutableList());
if (!messageList.isEmpty()) {
- deleteWithPreHooks(messageIdMapper, messageList, mailboxSession);
+ deleteWithPreHooks(messageIdMapper, messageList, mailboxSession)
+ .subscribeOn(Schedulers.elastic())
+ .block();
return DeleteResult.destroyed(messageId);
}
return DeleteResult.notFound(messageId);
@@ -230,7 +232,9 @@ public class StoreMessageIdManager implements MessageIdManager {
.collect(Guavate.toImmutableSet());
Sets.SetView<MessageId> nonAccessibleMessages = Sets.difference(ImmutableSet.copyOf(messageIds), accessibleMessageIds);
- deleteWithPreHooks(messageIdMapper, accessibleMessages, mailboxSession);
+ deleteWithPreHooks(messageIdMapper, accessibleMessages, mailboxSession)
+ .subscribeOn(Schedulers.elastic())
+ .block();
return DeleteResult.builder()
.addDestroyed(accessibleMessageIds)
@@ -238,38 +242,34 @@ public class StoreMessageIdManager implements MessageIdManager {
.build();
}
- private void deleteWithPreHooks(MessageIdMapper messageIdMapper, List<MailboxMessage> messageList, MailboxSession mailboxSession) throws MailboxException {
+ private Mono<Void> deleteWithPreHooks(MessageIdMapper messageIdMapper, List<MailboxMessage> messageList, MailboxSession mailboxSession) {
ImmutableList<MetadataWithMailboxId> metadataWithMailbox = messageList.stream()
.map(mailboxMessage -> MetadataWithMailboxId.from(mailboxMessage.metaData(), mailboxMessage.getMailboxId()))
.collect(Guavate.toImmutableList());
- preDeletionHooks.runHooks(PreDeletionHook.DeleteOperation.from(metadataWithMailbox))
- .then(Mono.fromRunnable(Throwing.runnable(
- () -> delete(messageIdMapper, messageList, mailboxSession, metadataWithMailbox))))
- .block();
+ return preDeletionHooks.runHooks(PreDeletionHook.DeleteOperation.from(metadataWithMailbox))
+ .then(delete(messageIdMapper, messageList, mailboxSession, metadataWithMailbox));
}
- private void delete(MessageIdMapper messageIdMapper, List<MailboxMessage> messageList, MailboxSession mailboxSession,
- ImmutableList<MetadataWithMailboxId> metadataWithMailbox) throws MailboxException {
- messageIdMapper.delete(
+ private Mono<Void> delete(MessageIdMapper messageIdMapper, List<MailboxMessage> messageList, MailboxSession mailboxSession, ImmutableList<MetadataWithMailboxId> metadataWithMailbox) {
+ MailboxMapper mailboxMapper = mailboxSessionMapperFactory.getMailboxMapper(mailboxSession);
+
+ return messageIdMapper.deleteReactive(
messageList.stream()
.collect(Guavate.toImmutableListMultimap(
Message::getMessageId,
- MailboxMessage::getMailboxId)));
-
- MailboxMapper mailboxMapper = mailboxSessionMapperFactory.getMailboxMapper(mailboxSession);
- Flux.fromIterable(metadataWithMailbox)
- .flatMap(metadataWithMailboxId -> mailboxMapper.findMailboxById(metadataWithMailboxId.getMailboxId())
- .flatMap(mailbox -> eventBus.dispatch(EventFactory.expunged()
- .randomEventId()
- .mailboxSession(mailboxSession)
- .mailbox(mailbox)
- .addMetaData(metadataWithMailboxId.getMessageMetaData())
- .build(),
- new MailboxIdRegistrationKey(metadataWithMailboxId.getMailboxId()))), DEFAULT_CONCURRENCY)
- .then()
- .subscribeOn(Schedulers.elastic())
- .block();
+ MailboxMessage::getMailboxId)))
+ .then(
+ Flux.fromIterable(metadataWithMailbox)
+ .flatMap(metadataWithMailboxId -> mailboxMapper.findMailboxById(metadataWithMailboxId.getMailboxId())
+ .flatMap(mailbox -> eventBus.dispatch(EventFactory.expunged()
+ .randomEventId()
+ .mailboxSession(mailboxSession)
+ .mailbox(mailbox)
+ .addMetaData(metadataWithMailboxId.getMessageMetaData())
+ .build(),
+ new MailboxIdRegistrationKey(metadataWithMailboxId.getMailboxId()))), DEFAULT_CONCURRENCY)
+ .then());
}
@Override
diff --git a/mailbox/store/src/main/java/org/apache/james/mailbox/store/mail/MessageIdMapper.java b/mailbox/store/src/main/java/org/apache/james/mailbox/store/mail/MessageIdMapper.java
index 5191435..7bdafbc 100644
--- a/mailbox/store/src/main/java/org/apache/james/mailbox/store/mail/MessageIdMapper.java
+++ b/mailbox/store/src/main/java/org/apache/james/mailbox/store/mail/MessageIdMapper.java
@@ -69,6 +69,10 @@ public interface MessageIdMapper {
.forEach(this::delete);
}
+ default Mono<Void> deleteReactive(Multimap<MessageId, MailboxId> ids) {
+ return Mono.fromRunnable(() -> delete(ids));
+ }
+
/**
* Updates the flags of the messages with the given MessageId in the supplied mailboxes
*
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 05/08: JAMES-3575 Reactive single flag update for
JMAP RFC-8621
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 42fc513c5be8547d139b85210268b52f5941dfa2
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Sat May 1 22:38:48 2021 +0700
JAMES-3575 Reactive single flag update for JMAP RFC-8621
---
.../main/java/org/apache/james/mailbox/MessageIdManager.java | 2 ++
.../org/apache/james/mailbox/store/StoreMessageIdManager.java | 10 +++++++---
.../org/apache/james/jmap/method/EmailSetUpdatePerformer.scala | 5 ++---
3 files changed, 11 insertions(+), 6 deletions(-)
diff --git a/mailbox/api/src/main/java/org/apache/james/mailbox/MessageIdManager.java b/mailbox/api/src/main/java/org/apache/james/mailbox/MessageIdManager.java
index 6fef5d4..abb0f8f 100644
--- a/mailbox/api/src/main/java/org/apache/james/mailbox/MessageIdManager.java
+++ b/mailbox/api/src/main/java/org/apache/james/mailbox/MessageIdManager.java
@@ -52,6 +52,8 @@ public interface MessageIdManager {
void setFlags(Flags newState, FlagsUpdateMode replace, MessageId messageId, List<MailboxId> mailboxIds, MailboxSession mailboxSession) throws MailboxException;
+ Publisher<Void> setFlagsReactive(Flags newState, FlagsUpdateMode replace, MessageId messageId, List<MailboxId> mailboxIds, MailboxSession mailboxSession) throws MailboxException;
+
List<MessageResult> getMessages(Collection<MessageId> messageIds, FetchGroup minimal, MailboxSession mailboxSession) throws MailboxException;
default Publisher<MessageResult> getMessagesReactive(Collection<MessageId> messageIds, FetchGroup minimal, MailboxSession mailboxSession) {
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 5031573..7dfc8ae 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
@@ -120,12 +120,17 @@ public class StoreMessageIdManager implements MessageIdManager {
@Override
public void setFlags(Flags newState, MessageManager.FlagsUpdateMode replace, MessageId messageId, List<MailboxId> mailboxIds, MailboxSession mailboxSession) throws MailboxException {
+ MailboxReactorUtils.block(setFlagsReactive(newState, replace, messageId, mailboxIds, mailboxSession));
+ }
+
+ @Override
+ public Mono<Void> setFlagsReactive(Flags newState, MessageManager.FlagsUpdateMode replace, MessageId messageId, List<MailboxId> mailboxIds, MailboxSession mailboxSession) {
MessageIdMapper messageIdMapper = mailboxSessionMapperFactory.getMessageIdMapper(mailboxSession);
MailboxMapper mailboxMapper = mailboxSessionMapperFactory.getMailboxMapper(mailboxSession);
int concurrency = 4;
- MailboxReactorUtils.block(Flux.fromIterable(mailboxIds)
+ return Flux.fromIterable(mailboxIds)
.flatMap(mailboxMapper::findMailboxById, concurrency)
.collect(Guavate.toImmutableList())
.flatMap(Throwing.<List<Mailbox>, Mono<Void>>function(targetMailboxes -> {
@@ -135,8 +140,7 @@ public class StoreMessageIdManager implements MessageIdManager {
.flatMapIterable(updatedFlags -> updatedFlags.asMap().entrySet())
.concatMap(entry -> dispatchFlagsChange(mailboxSession, entry.getKey(), ImmutableList.copyOf(entry.getValue()), targetMailboxes))
.then();
- }).sneakyThrow())
- .subscribeOn(Schedulers.elastic()));
+ }).sneakyThrow());
}
@Override
diff --git a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/EmailSetUpdatePerformer.scala b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/EmailSetUpdatePerformer.scala
index 3e42e69..c69517f 100644
--- a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/EmailSetUpdatePerformer.scala
+++ b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/EmailSetUpdatePerformer.scala
@@ -235,10 +235,9 @@ class EmailSetUpdatePerformer @Inject() (serializer: EmailSetSerializer,
if (newFlags.equals(originalFlags)) {
SMono.just[EmailUpdateResult](EmailUpdateSuccess(messageId))
} else {
- SMono.fromCallable(() =>
- messageIdManager.setFlags(newFlags, FlagsUpdateMode.REPLACE, messageId, ImmutableList.copyOf(mailboxIds.value.asJavaCollection), session))
- .subscribeOn(Schedulers.elastic())
+ SMono(messageIdManager.setFlagsReactive(newFlags, FlagsUpdateMode.REPLACE, messageId, ImmutableList.copyOf(mailboxIds.value.asJavaCollection), session))
.`then`(SMono.just[EmailUpdateResult](EmailUpdateSuccess(messageId)))
+ .subscribeOn(Schedulers.elastic())
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 07/08: JAMES-3575 [REFACTORING] Flatten
EmailSubmissionSet for/yield
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 eb9bd9c77ebfce6e439d7308235c34c70ea229e3
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Sun May 2 06:46:03 2021 +0700
JAMES-3575 [REFACTORING] Flatten EmailSubmissionSet for/yield
---
.../jmap/method/EmailSubmissionSetMethod.scala | 54 ++++++++++------------
1 file changed, 25 insertions(+), 29 deletions(-)
diff --git a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/EmailSubmissionSetMethod.scala b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/EmailSubmissionSetMethod.scala
index e5f3bb8..4c9edd3 100644
--- a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/EmailSubmissionSetMethod.scala
+++ b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/EmailSubmissionSetMethod.scala
@@ -241,35 +241,31 @@ class EmailSubmissionSetMethod @Inject()(serializer: EmailSubmissionSetSerialize
private def sendEmail(mailboxSession: MailboxSession,
request: EmailSubmissionCreationRequest): Either[Throwable, (EmailSubmissionCreationResponse, MessageId)] = {
- val message: Either[Exception, MessageResult] = messageIdManager.getMessage(request.emailId, FetchGroup.FULL_CONTENT, mailboxSession)
- .asScala
- .toList
- .headOption
- .toRight(MessageNotFoundException(request.emailId))
-
- message.flatMap(m => {
- val submissionId = EmailSubmissionId.generate
-
- val result: Try[EmailSubmissionCreationResponse] = for {
- message <- toMimeMessage(submissionId.value, m)
- envelope <- resolveEnvelope(message, request.envelope)
- validation <- validate(mailboxSession)(message, envelope)
- mail <- Try({
- val mailImpl = MailImpl.builder()
- .name(submissionId.value)
- .addRecipients(envelope.rcptTo.map(_.email).asJava)
- .sender(envelope.mailFrom.email)
- .addAttribute(new Attribute(MAIL_METADATA_USERNAME_ATTRIBUTE, AttributeValue.of(mailboxSession.getUser.asString())))
- .build()
- mailImpl.setMessageNoCopy(message)
- mailImpl
- })
- enqueue <- Try(queue.enQueue(mail))
- } yield {
- EmailSubmissionCreationResponse(submissionId)
- }
- result.toEither.map(response => (response, request.emailId))
- })
+ val result = for {
+ message <- messageIdManager.getMessage(request.emailId, FetchGroup.FULL_CONTENT, mailboxSession)
+ .asScala
+ .toList
+ .headOption
+ .toRight(MessageNotFoundException(request.emailId))
+ submissionId = EmailSubmissionId.generate
+ message <- toMimeMessage(submissionId.value, message).toEither
+ envelope <- resolveEnvelope(message, request.envelope).toEither
+ validation <- validate(mailboxSession)(message, envelope).toEither
+ mail <- Try({
+ val mailImpl = MailImpl.builder()
+ .name(submissionId.value)
+ .addRecipients(envelope.rcptTo.map(_.email).asJava)
+ .sender(envelope.mailFrom.email)
+ .addAttribute(new Attribute(MAIL_METADATA_USERNAME_ATTRIBUTE, AttributeValue.of(mailboxSession.getUser.asString())))
+ .build()
+ mailImpl.setMessageNoCopy(message)
+ mailImpl
+ }).toEither
+ enqueue <- Try(queue.enQueue(mail)).toEither
+ } yield {
+ EmailSubmissionCreationResponse(submissionId)
+ }
+ result.map(response => (response, request.emailId))
}
private def toMimeMessage(name: String, message: MessageResult): Try[MimeMessageWrapper] = {
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 02/08: JAMES-3575 MessageIdManager::delete should
be reactive
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 da19351ec92eecb9d2eedc0b80d5539c46ecd5fa
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Sat May 1 22:19:23 2021 +0700
JAMES-3575 MessageIdManager::delete should be reactive
End to end reactive chain for JMAP RFC-8621
deletes...
---
.../org/apache/james/mailbox/MessageIdManager.java | 8 +++--
.../james/vault/DeletedMessageVaultHookTest.java | 6 ++--
.../james/mailbox/store/StoreMessageIdManager.java | 38 +++++++++++++---------
.../store/AbstractCombinationManagerTest.java | 6 +++-
.../AbstractMessageIdManagerSideEffectTest.java | 7 ++--
.../store/AbstractMessageIdManagerStorageTest.java | 2 +-
.../methods/SetMessagesDestructionProcessor.java | 7 +++-
.../jmap/method/EmailSetDeletePerformer.scala | 2 +-
8 files changed, 51 insertions(+), 25 deletions(-)
diff --git a/mailbox/api/src/main/java/org/apache/james/mailbox/MessageIdManager.java b/mailbox/api/src/main/java/org/apache/james/mailbox/MessageIdManager.java
index 26a53cb..6fef5d4 100644
--- a/mailbox/api/src/main/java/org/apache/james/mailbox/MessageIdManager.java
+++ b/mailbox/api/src/main/java/org/apache/james/mailbox/MessageIdManager.java
@@ -38,6 +38,8 @@ import org.reactivestreams.Publisher;
import com.google.common.collect.ImmutableList;
import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
public interface MessageIdManager {
default Publisher<ComposedMessageIdWithMetaData> messageMetadata(MessageId id, MailboxSession session) {
@@ -62,7 +64,7 @@ public interface MessageIdManager {
DeleteResult delete(MessageId messageId, List<MailboxId> mailboxIds, MailboxSession mailboxSession) throws MailboxException;
- DeleteResult delete(List<MessageId> messageId, MailboxSession mailboxSession) throws MailboxException;
+ Publisher<DeleteResult> delete(List<MessageId> messageId, MailboxSession mailboxSession) throws MailboxException;
void setInMailboxes(MessageId messageId, Collection<MailboxId> mailboxIds, MailboxSession mailboxSession) throws MailboxException;
@@ -71,7 +73,9 @@ public interface MessageIdManager {
}
default DeleteResult delete(MessageId messageId, MailboxSession mailboxSession) throws MailboxException {
- return delete(ImmutableList.of(messageId), mailboxSession);
+ return Mono.from(delete(ImmutableList.of(messageId), mailboxSession))
+ .subscribeOn(Schedulers.elastic())
+ .block();
}
}
diff --git a/mailbox/plugin/deleted-messages-vault/src/test/java/org/apache/james/vault/DeletedMessageVaultHookTest.java b/mailbox/plugin/deleted-messages-vault/src/test/java/org/apache/james/vault/DeletedMessageVaultHookTest.java
index 7251a98..79e919d 100644
--- a/mailbox/plugin/deleted-messages-vault/src/test/java/org/apache/james/vault/DeletedMessageVaultHookTest.java
+++ b/mailbox/plugin/deleted-messages-vault/src/test/java/org/apache/james/vault/DeletedMessageVaultHookTest.java
@@ -65,6 +65,8 @@ import com.github.steveash.guavate.Guavate;
import com.google.common.collect.ImmutableList;
import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
class DeletedMessageVaultHookTest {
@@ -159,7 +161,7 @@ class DeletedMessageVaultHookTest {
MessageId messageId = composedId.getMessageId();
long messageSize = messageSize(messageManager, composedId);
- messageIdManager.delete(ImmutableList.of(messageId), aliceSession);
+ messageIdManager.delete(messageId, aliceSession);
DeletedMessage deletedMessage = buildDeletedMessage(ImmutableList.of(aliceMailbox), messageId, ALICE, messageSize);
assertThat(Flux.from(messageVault.search(ALICE, Query.ALL)).blockFirst())
@@ -175,7 +177,7 @@ class DeletedMessageVaultHookTest {
.mapToObj(Throwing.intFunction(i -> appendMessage(messageManager).getMessageId()))
.collect(Guavate.toImmutableList());
- assertThatCode(() -> messageIdManager.delete(ids, aliceSession))
+ assertThatCode(() -> Mono.from(messageIdManager.delete(ids, aliceSession)).subscribeOn(Schedulers.elastic()).block())
.doesNotThrowAnyException();
}
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 d34b4a7..116e91b 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
@@ -187,10 +187,14 @@ public class StoreMessageIdManager implements MessageIdManager {
}
private ImmutableSet<MailboxId> getAllowedMailboxIds(MailboxSession mailboxSession, Stream<MailboxId> idList, Right... rights) throws MailboxException {
- return MailboxReactorUtils.block(Flux.fromStream(idList)
+ return MailboxReactorUtils.block(getAllowedMailboxIdsReactive(mailboxSession, idList, rights));
+ }
+
+ private Mono<ImmutableSet<MailboxId>> getAllowedMailboxIdsReactive(MailboxSession mailboxSession, Stream<MailboxId> idList, Right... rights) {
+ return Flux.fromStream(idList)
.distinct()
.filterWhen(hasRightsOnMailboxReactive(mailboxSession, rights), DEFAULT_CONCURRENCY)
- .collect(Guavate.toImmutableSet()));
+ .collect(Guavate.toImmutableSet());
}
@Override
@@ -215,14 +219,21 @@ public class StoreMessageIdManager implements MessageIdManager {
}
@Override
- public DeleteResult delete(List<MessageId> messageIds, MailboxSession mailboxSession) throws MailboxException {
+ public Mono<DeleteResult> delete(List<MessageId> messageIds, MailboxSession mailboxSession) throws MailboxException {
MessageIdMapper messageIdMapper = mailboxSessionMapperFactory.getMessageIdMapper(mailboxSession);
- List<MailboxMessage> messageList = messageIdMapper.find(messageIds, MessageMapper.FetchType.Metadata);
- ImmutableSet<MailboxId> allowedMailboxIds = getAllowedMailboxIds(mailboxSession, messageList
- .stream()
- .map(MailboxMessage::getMailboxId), Right.DeleteMessages);
+ return messageIdMapper.findReactive(messageIds, MessageMapper.FetchType.Metadata)
+ .collectList()
+ .flatMap(messageList ->
+ getAllowedMailboxIdsReactive(mailboxSession,
+ messageList
+ .stream()
+ .map(MailboxMessage::getMailboxId),
+ Right.DeleteMessages)
+ .flatMap(allowedMailboxIds -> deleteInAllowedMailboxes(messageIds, mailboxSession, messageIdMapper, messageList, allowedMailboxIds)));
+ }
+ private Mono<DeleteResult> deleteInAllowedMailboxes(List<MessageId> messageIds, MailboxSession mailboxSession, MessageIdMapper messageIdMapper, List<MailboxMessage> messageList, ImmutableSet<MailboxId> allowedMailboxIds) {
List<MailboxMessage> accessibleMessages = messageList.stream()
.filter(message -> allowedMailboxIds.contains(message.getMailboxId()))
.collect(Guavate.toImmutableList());
@@ -232,14 +243,11 @@ public class StoreMessageIdManager implements MessageIdManager {
.collect(Guavate.toImmutableSet());
Sets.SetView<MessageId> nonAccessibleMessages = Sets.difference(ImmutableSet.copyOf(messageIds), accessibleMessageIds);
- deleteWithPreHooks(messageIdMapper, accessibleMessages, mailboxSession)
- .subscribeOn(Schedulers.elastic())
- .block();
-
- return DeleteResult.builder()
- .addDestroyed(accessibleMessageIds)
- .addNotFound(nonAccessibleMessages)
- .build();
+ return deleteWithPreHooks(messageIdMapper, accessibleMessages, mailboxSession)
+ .thenReturn(DeleteResult.builder()
+ .addDestroyed(accessibleMessageIds)
+ .addNotFound(nonAccessibleMessages)
+ .build());
}
private Mono<Void> deleteWithPreHooks(MessageIdMapper messageIdMapper, List<MailboxMessage> messageList, MailboxSession mailboxSession) {
diff --git a/mailbox/store/src/test/java/org/apache/james/mailbox/store/AbstractCombinationManagerTest.java b/mailbox/store/src/test/java/org/apache/james/mailbox/store/AbstractCombinationManagerTest.java
index bdddff1..56c1ba3 100644
--- a/mailbox/store/src/test/java/org/apache/james/mailbox/store/AbstractCombinationManagerTest.java
+++ b/mailbox/store/src/test/java/org/apache/james/mailbox/store/AbstractCombinationManagerTest.java
@@ -53,6 +53,8 @@ import org.junit.jupiter.api.Test;
import com.google.common.collect.ImmutableList;
import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
public abstract class AbstractCombinationManagerTest {
@@ -534,7 +536,9 @@ public abstract class AbstractCombinationManagerTest {
.appendMessage(MessageManager.AppendCommand.from(mailContent), session)
.getId().getMessageId();
- messageIdManager.delete(ImmutableList.of(messageId1, messageId2), session);
+ Mono.from(messageIdManager.delete(ImmutableList.of(messageId1, messageId2), session))
+ .subscribeOn(Schedulers.elastic())
+ .block();
SearchQuery searchQuery = SearchQuery.of(SearchQuery.all());
assertThat(Flux.from(messageManager1.search(searchQuery, session)).toStream()).isEmpty();
diff --git a/mailbox/store/src/test/java/org/apache/james/mailbox/store/AbstractMessageIdManagerSideEffectTest.java b/mailbox/store/src/test/java/org/apache/james/mailbox/store/AbstractMessageIdManagerSideEffectTest.java
index 7a1d087..a6011de 100644
--- a/mailbox/store/src/test/java/org/apache/james/mailbox/store/AbstractMessageIdManagerSideEffectTest.java
+++ b/mailbox/store/src/test/java/org/apache/james/mailbox/store/AbstractMessageIdManagerSideEffectTest.java
@@ -83,6 +83,7 @@ import com.google.common.collect.ImmutableSet;
import com.google.common.collect.ImmutableSortedMap;
import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
public abstract class AbstractMessageIdManagerSideEffectTest {
private static final Quota<QuotaCountLimit, QuotaCountUsage> OVER_QUOTA = Quota.<QuotaCountLimit, QuotaCountUsage>builder()
@@ -169,7 +170,9 @@ public abstract class AbstractMessageIdManagerSideEffectTest {
MessageMetaData simpleMessageMetaData2 = messageResult2.messageMetaData();
eventBus.register(eventCollector);
- messageIdManager.delete(ImmutableList.of(messageId1, messageId2), session);
+ Mono.from(messageIdManager.delete(ImmutableList.of(messageId1, messageId2), session))
+ .subscribeOn(Schedulers.elastic())
+ .block();
AbstractListAssert<?, List<? extends Expunged>, Expunged, ObjectAssert<Expunged>> events =
assertThat(eventCollector.getEvents())
@@ -523,7 +526,7 @@ public abstract class AbstractMessageIdManagerSideEffectTest {
MessageId messageId = testingData.createNotUsedMessageId();
eventBus.register(eventCollector);
- messageIdManager.delete(ImmutableList.of(messageId), session);
+ messageIdManager.delete(messageId, session);
assertThat(eventCollector.getEvents()).isEmpty();
}
diff --git a/mailbox/store/src/test/java/org/apache/james/mailbox/store/AbstractMessageIdManagerStorageTest.java b/mailbox/store/src/test/java/org/apache/james/mailbox/store/AbstractMessageIdManagerStorageTest.java
index d9beb5b..a154ae9 100644
--- a/mailbox/store/src/test/java/org/apache/james/mailbox/store/AbstractMessageIdManagerStorageTest.java
+++ b/mailbox/store/src/test/java/org/apache/james/mailbox/store/AbstractMessageIdManagerStorageTest.java
@@ -363,7 +363,7 @@ public abstract class AbstractMessageIdManagerStorageTest {
void deleteAllShouldReturnNotDeleteWhenDeletingOnOtherSession() throws Exception {
MessageId messageId = testingData.persist(bobMailbox1.getMailboxId(), messageUid1, FLAGS, bobSession);
- messageIdManager.delete(ImmutableList.of(messageId), aliceSession);
+ messageIdManager.delete(messageId, aliceSession);
assertThat(messageIdManager.getMessage(messageId, FetchGroup.MINIMAL, bobSession)).hasSize(1);
}
diff --git a/server/protocols/jmap-draft/src/main/java/org/apache/james/jmap/draft/methods/SetMessagesDestructionProcessor.java b/server/protocols/jmap-draft/src/main/java/org/apache/james/jmap/draft/methods/SetMessagesDestructionProcessor.java
index a961a73..5a90a6b 100644
--- a/server/protocols/jmap-draft/src/main/java/org/apache/james/jmap/draft/methods/SetMessagesDestructionProcessor.java
+++ b/server/protocols/jmap-draft/src/main/java/org/apache/james/jmap/draft/methods/SetMessagesDestructionProcessor.java
@@ -40,6 +40,9 @@ import org.slf4j.LoggerFactory;
import com.google.common.annotations.VisibleForTesting;
+import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
+
public class SetMessagesDestructionProcessor implements SetMessagesProcessor {
private static final Logger LOGGER = LoggerFactory.getLogger(SetMessagesCreationProcessor.class);
@@ -70,7 +73,9 @@ public class SetMessagesDestructionProcessor implements SetMessagesProcessor {
if (toBeDestroyed.isEmpty()) {
return Stream.empty();
}
- DeleteResult deleteResult = messageIdManager.delete(toBeDestroyed, mailboxSession);
+ DeleteResult deleteResult = Mono.from(messageIdManager.delete(toBeDestroyed, mailboxSession))
+ .subscribeOn(Schedulers.elastic())
+ .block();
Stream<SetMessagesResponse> destroyed = deleteResult.getDestroyed().stream()
.map(messageId -> SetMessagesResponse.builder().destroyed(messageId).build());
diff --git a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/EmailSetDeletePerformer.scala b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/EmailSetDeletePerformer.scala
index 14f106a..6908f7c 100644
--- a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/EmailSetDeletePerformer.scala
+++ b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/EmailSetDeletePerformer.scala
@@ -85,7 +85,7 @@ class EmailSetDeletePerformer @Inject()(messageIdManager: MessageIdManager,
case _ => None
}
- SMono.fromCallable(() => messageIdManager.delete(messageIds.toList.asJava, mailboxSession))
+ SMono(messageIdManager.delete(messageIds.toList.asJava, mailboxSession))
.map(DestroyResult.from)
.subscribeOn(Schedulers.elastic())
.onErrorResume(e => SMono.just(messageIds.map(id => DestroyFailure(EmailSet.asUnparsed(id), e))))
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 04/08: JAMES-3575 Rewrite
StoreMessageIdManager::setFlags in a more reactive way
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 d4647112f6e71186f8f039c95e94d3425cfc4962
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Sat May 1 22:34:04 2021 +0700
JAMES-3575 Rewrite StoreMessageIdManager::setFlags in a more reactive way
---
.../james/mailbox/store/StoreMessageIdManager.java | 32 ++++++++++------------
1 file changed, 14 insertions(+), 18 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 6ab1ffc..5031573 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
@@ -82,7 +82,6 @@ import com.github.fge.lambdas.functions.ThrowingFunction;
import com.github.steveash.guavate.Guavate;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
-import com.google.common.collect.Multimap;
import com.google.common.collect.Sets;
import reactor.core.publisher.Flux;
@@ -125,20 +124,19 @@ public class StoreMessageIdManager implements MessageIdManager {
MailboxMapper mailboxMapper = mailboxSessionMapperFactory.getMailboxMapper(mailboxSession);
int concurrency = 4;
- List<Mailbox> targetMailboxes = Flux.fromIterable(mailboxIds)
+
+ MailboxReactorUtils.block(Flux.fromIterable(mailboxIds)
.flatMap(mailboxMapper::findMailboxById, concurrency)
.collect(Guavate.toImmutableList())
- .subscribeOn(Schedulers.elastic())
- .block();
-
- assertRightsOnMailboxes(targetMailboxes, mailboxSession, Right.Write);
+ .flatMap(Throwing.<List<Mailbox>, Mono<Void>>function(targetMailboxes -> {
+ assertRightsOnMailboxes(targetMailboxes, mailboxSession, Right.Write);
- Multimap<MailboxId, UpdatedFlags> updatedFlags = messageIdMapper.setFlags(messageId, mailboxIds, newState, replace)
- .subscribeOn(Schedulers.elastic())
- .block();
- for (Map.Entry<MailboxId, Collection<UpdatedFlags>> entry : updatedFlags.asMap().entrySet()) {
- dispatchFlagsChange(mailboxSession, entry.getKey(), ImmutableList.copyOf(entry.getValue()), targetMailboxes);
- }
+ return messageIdMapper.setFlags(messageId, mailboxIds, newState, replace)
+ .flatMapIterable(updatedFlags -> updatedFlags.asMap().entrySet())
+ .concatMap(entry -> dispatchFlagsChange(mailboxSession, entry.getKey(), ImmutableList.copyOf(entry.getValue()), targetMailboxes))
+ .then();
+ }).sneakyThrow())
+ .subscribeOn(Schedulers.elastic()));
}
@Override
@@ -389,8 +387,7 @@ public class StoreMessageIdManager implements MessageIdManager {
.then());
}
- private void dispatchFlagsChange(MailboxSession mailboxSession, MailboxId mailboxId, ImmutableList<UpdatedFlags> updatedFlags,
- List<Mailbox> knownMailboxes) throws MailboxException {
+ private Mono<Void> dispatchFlagsChange(MailboxSession mailboxSession, MailboxId mailboxId, ImmutableList<UpdatedFlags> updatedFlags, List<Mailbox> knownMailboxes) {
if (updatedFlags.stream().anyMatch(UpdatedFlags::flagsChanged)) {
Mailbox mailbox = knownMailboxes.stream()
.filter(knownMailbox -> knownMailbox.getMailboxId().equals(mailboxId))
@@ -399,16 +396,15 @@ public class StoreMessageIdManager implements MessageIdManager {
.findMailboxById(mailboxId)
.subscribeOn(Schedulers.elastic())))
.sneakyThrow());
- eventBus.dispatch(EventFactory.flagsUpdated()
+ return eventBus.dispatch(EventFactory.flagsUpdated()
.randomEventId()
.mailboxSession(mailboxSession)
.mailbox(mailbox)
.updatedFlags(updatedFlags)
.build(),
- new MailboxIdRegistrationKey(mailboxId))
- .subscribeOn(Schedulers.elastic())
- .block();
+ new MailboxIdRegistrationKey(mailboxId));
}
+ return Mono.empty();
}
private void validateQuota(MessageMovesWithMailbox messageMoves, MailboxMessage mailboxMessage) throws MailboxException {
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 06/08: JAMES-3575 Reactive single message move for
JMAP RFC-8621
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 bde8a850b1536d992b296d51a855f3e6e5bb73e2
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Sat May 1 22:48:36 2021 +0700
JAMES-3575 Reactive single message move for JMAP RFC-8621
---
.../org/apache/james/mailbox/MessageIdManager.java | 2 +
.../james/mailbox/store/StoreMessageIdManager.java | 44 ++++++++++++----------
.../jmap/method/EmailSetUpdatePerformer.scala | 4 +-
3 files changed, 28 insertions(+), 22 deletions(-)
diff --git a/mailbox/api/src/main/java/org/apache/james/mailbox/MessageIdManager.java b/mailbox/api/src/main/java/org/apache/james/mailbox/MessageIdManager.java
index abb0f8f..218819a 100644
--- a/mailbox/api/src/main/java/org/apache/james/mailbox/MessageIdManager.java
+++ b/mailbox/api/src/main/java/org/apache/james/mailbox/MessageIdManager.java
@@ -70,6 +70,8 @@ public interface MessageIdManager {
void setInMailboxes(MessageId messageId, Collection<MailboxId> mailboxIds, MailboxSession mailboxSession) throws MailboxException;
+ Publisher<Void> setInMailboxesReactive(MessageId messageId, Collection<MailboxId> mailboxIds, MailboxSession mailboxSession);
+
default List<MessageResult> getMessage(MessageId messageId, FetchGroup fetchGroup, MailboxSession mailboxSession) throws MailboxException {
return getMessages(ImmutableList.of(messageId), fetchGroup, mailboxSession);
}
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 7dfc8ae..a7fa33b 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
@@ -286,25 +286,29 @@ public class StoreMessageIdManager implements MessageIdManager {
@Override
public void setInMailboxes(MessageId messageId, Collection<MailboxId> targetMailboxIds, MailboxSession mailboxSession) throws MailboxException {
- List<MailboxMessage> currentMailboxMessages = findRelatedMailboxMessages(messageId, mailboxSession);
-
- MailboxReactorUtils.block(messageMovesWithMailbox(MessageMoves.builder()
- .targetMailboxIds(targetMailboxIds)
- .previousMailboxIds(toMailboxIds(currentMailboxMessages))
- .build(), mailboxSession)
- .flatMap(Throwing.<MessageMovesWithMailbox, Mono<Void>>function(messageMove -> {
- MessageMovesWithMailbox refined = messageMove.filterPrevious(hasRightsOnMailbox(mailboxSession, Right.Read));
+ MailboxReactorUtils.block(setInMailboxesReactive(messageId, targetMailboxIds, mailboxSession)
+ .subscribeOn(Schedulers.elastic()));
+ }
- if (messageMove.getPreviousMailboxes().isEmpty()) {
- LOGGER.info("Tried to access {} not accessible for {}", messageId, mailboxSession.getUser().asString());
+ @Override
+ public Mono<Void> setInMailboxesReactive(MessageId messageId, Collection<MailboxId> targetMailboxIds, MailboxSession mailboxSession) {
+ return findRelatedMailboxMessages(messageId, mailboxSession)
+ .flatMap(currentMailboxMessages -> messageMovesWithMailbox(MessageMoves.builder()
+ .targetMailboxIds(targetMailboxIds)
+ .previousMailboxIds(toMailboxIds(currentMailboxMessages))
+ .build(), mailboxSession)
+ .flatMap(Throwing.<MessageMovesWithMailbox, Mono<Void>>function(messageMove -> {
+ MessageMovesWithMailbox refined = messageMove.filterPrevious(hasRightsOnMailbox(mailboxSession, Right.Read));
+
+ if (messageMove.getPreviousMailboxes().isEmpty()) {
+ LOGGER.info("Tried to access {} not accessible for {}", messageId, mailboxSession.getUser().asString());
+ return Mono.empty();
+ }
+ if (refined.isChange()) {
+ return applyMessageMoves(mailboxSession, currentMailboxMessages, refined);
+ }
return Mono.empty();
- }
- if (refined.isChange()) {
- return applyMessageMoves(mailboxSession, currentMailboxMessages, refined);
- }
- return Mono.empty();
- }).sneakyThrow())
- .subscribeOn(Schedulers.elastic()));
+ }).sneakyThrow()));
}
public void setInMailboxesNoCheck(MessageId messageId, MailboxId targetMailboxId, MailboxSession mailboxSession) throws MailboxException {
@@ -325,11 +329,11 @@ public class StoreMessageIdManager implements MessageIdManager {
.subscribeOn(Schedulers.elastic()));
}
- private List<MailboxMessage> findRelatedMailboxMessages(MessageId messageId, MailboxSession mailboxSession) throws MailboxException {
+ private Mono<List<MailboxMessage>> findRelatedMailboxMessages(MessageId messageId, MailboxSession mailboxSession) {
MessageIdMapper messageIdMapper = mailboxSessionMapperFactory.getMessageIdMapper(mailboxSession);
- return MailboxReactorUtils.block(messageIdMapper.findReactive(ImmutableList.of(messageId), MessageMapper.FetchType.Metadata)
- .collect(Guavate.toImmutableList()));
+ return messageIdMapper.findReactive(ImmutableList.of(messageId), MessageMapper.FetchType.Metadata)
+ .collect(Guavate.toImmutableList());
}
private Mono<Void> applyMessageMoves(MailboxSession mailboxSession, List<MailboxMessage> currentMailboxMessages, MessageMovesWithMailbox messageMoves) throws MailboxNotFoundException {
diff --git a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/EmailSetUpdatePerformer.scala b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/EmailSetUpdatePerformer.scala
index c69517f..409ecaf 100644
--- a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/EmailSetUpdatePerformer.scala
+++ b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/EmailSetUpdatePerformer.scala
@@ -219,11 +219,11 @@ class EmailSetUpdatePerformer @Inject() (serializer: EmailSetSerializer,
if (targetIds.equals(mailboxIds)) {
SMono.just[EmailUpdateResult](EmailUpdateSuccess(messageId))
} else {
- SMono.fromCallable(() => messageIdManager.setInMailboxes(messageId, targetIds.value.asJava, session))
- .subscribeOn(Schedulers.elastic())
+ SMono(messageIdManager.setInMailboxesReactive(messageId, targetIds.value.asJava, session))
.`then`(SMono.just[EmailUpdateResult](EmailUpdateSuccess(messageId)))
.onErrorResume(e => SMono.just[EmailUpdateResult](EmailUpdateFailure(EmailSet.asUnparsed(messageId), e)))
.switchIfEmpty(SMono.just[EmailUpdateResult](EmailUpdateSuccess(messageId)))
+ .subscribeOn(Schedulers.elastic())
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org