You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@james.apache.org by rc...@apache.org on 2021/05/17 02:21:43 UTC
[james-project] branch master updated (eb7fb92 -> 14b341e)
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 eb7fb92 [FIX] Solve build issue (TaskManagerContract::awaitShouldAwaitWaitingTask)
new 87ebd46 [REFACTORING] CassandraMessageIdMapper: Remove a useless Mono.defer call
new 35e136e [REFACTORING] Reactor: favor error suppliers
new 1ac0fe7 [REFACTORING] Reactor-Scala: raiseError is deprecated and error should be used
new 4f5d8c4 [REFACTORING] Reactive StoreMessageIdManager::validateQuota
new 59b9483 [REFACTORING] Reactive version for MessageIdMapper::copyInMailbox
new dac8150 [REFACTORING] Cassandra implementation should depend on interfaces for UidProvider and ModSeqProvider
new 5fa18f6 [REFACTORING] Reactive version for MessageIdMapper::copyInMailbox
new 07250a7 [REFACTORING] StoreMessageIdManager::setInMailboxes should not block
new 9e7ad28 [REFACTORING] Hide CassandraACLMapper behind an interface
new 91595e7 [REFACTORING] MailboxChangeListener was performing some blocking calls...
new 14b341e [REFACTORING] CassandraMailboxSessionMapperFactory should not instantiate one mapper per request
The 11 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:
.../CassandraMailboxSessionMapperFactory.java | 99 +++++++-------
.../mailbox/cassandra/DeleteMessageListener.java | 6 +-
.../james/mailbox/cassandra/mail/ACLMapper.java | 16 ++-
.../mailbox/cassandra/mail/CassandraACLMapper.java | 14 +-
.../cassandra/mail/CassandraMailboxMapper.java | 14 +-
.../cassandra/mail/CassandraMessageIdMapper.java | 17 ++-
.../cassandra/mail/CassandraMessageMapper.java | 16 ++-
.../cassandra/mail/CassandraModSeqProvider.java | 19 ++-
.../cassandra/mail/CassandraUidProvider.java | 11 +-
.../mail/task/MailboxMergingTaskRunner.java | 14 +-
.../mail/CassandraModSeqProviderTest.java | 4 +-
.../mailbox/cassandra/mail/utils/GuiceUtils.java | 6 +
.../mailbox/maildir/mail/MaildirMailboxMapper.java | 4 +-
.../james/vault/DeletedMessageVaultHook.java | 2 +-
.../james/mailbox/store/StoreMailboxManager.java | 2 +-
.../james/mailbox/store/StoreMessageIdManager.java | 142 ++++++++++++---------
.../james/mailbox/store/StoreRightManager.java | 1 +
.../james/mailbox/store/mail/MessageIdMapper.java | 5 +
.../james/mailbox/store/mail/ModSeqProvider.java | 6 +
.../james/mailbox/store/mail/UidProvider.java | 16 +++
.../AbstractMessageIdManagerSideEffectTest.java | 22 ++--
.../modules/mailbox/CassandraMailboxModule.java | 2 +
.../access/CassandraAccessTokenRepository.java | 2 +-
.../change/CassandraEmailChangeRepository.java | 4 +-
.../change/CassandraMailboxChangeRepository.java | 4 +-
.../james/jmap/api/change/MailboxChange.java | 6 +-
.../memory/change/MemoryEmailChangeRepository.java | 2 +-
.../change/MemoryMailboxChangeRepository.java | 2 +-
.../james/jmap/http/AuthenticationRoutes.java | 2 +-
.../james/jmap/change/MailboxChangeListener.scala | 85 ++++++------
.../james/jmap/http/MailboxesProvisioner.scala | 2 +-
.../apache/james/jmap/mail/MailboxFactory.scala | 6 +-
.../apache/james/jmap/method/EmailGetMethod.scala | 6 +-
.../james/jmap/method/EmailQueryMethod.scala | 7 +-
.../apache/james/jmap/method/MDNParseMethod.scala | 7 +-
.../apache/james/jmap/method/MDNSendMethod.scala | 8 +-
.../james/jmap/method/MailboxGetMethod.scala | 2 +-
.../james/jmap/method/MailboxQueryMethod.scala | 2 +-
.../jmap/method/MailboxSetDeletePerformer.scala | 2 +-
.../jmap/method/MailboxSetUpdatePerformer.scala | 2 +-
.../org/apache/james/jmap/method/Method.scala | 4 +-
.../apache/james/jmap/routes/DownloadRoutes.scala | 16 +--
.../org/apache/james/jmap/routes/JmapApi.scala | 4 +-
.../apache/james/jmap/routes/UploadRoutes.scala | 6 +-
.../james/jmap/routes/JMAPApiRoutesTest.scala | 2 +-
.../org/apache/james/jmap/http/Authenticator.java | 2 +-
46 files changed, 341 insertions(+), 282 deletions(-)
copy mpt/impl/imap-mailbox/core/src/main/java/org/apache/james/mpt/imapmailbox/GrantRightsOnHost.java => mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/ACLMapper.java (72%)
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 07/11: [REFACTORING] Reactive version for
MessageIdMapper::copyInMailbox
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 5fa18f63d83df1eadc87d767efd5e762798f6a52
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Thu May 13 16:29:53 2021 +0700
[REFACTORING] Reactive version for MessageIdMapper::copyInMailbox
---
.../main/java/org/apache/james/mailbox/store/mail/MessageIdMapper.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
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 0db04f8..02740cd 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
@@ -58,7 +58,7 @@ public interface MessageIdMapper {
void copyInMailbox(MailboxMessage mailboxMessage, Mailbox mailbox) throws MailboxException;
default Mono<Void> copyInMailboxReactive(MailboxMessage mailboxMessage, Mailbox mailbox) {
- return Mono.fromRunnable(Throwing.runnable(() -> copyInMailboxReactive(mailboxMessage, mailbox)).sneakyThrow());
+ return Mono.fromRunnable(Throwing.runnable(() -> copyInMailbox(mailboxMessage, mailbox)).sneakyThrow());
}
void delete(MessageId messageId);
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 03/11: [REFACTORING] Reactor-Scala: raiseError is
deprecated and error should be used
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 1ac0fe76744305ec983adf80613630b17e8b42b0
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Thu May 13 15:24:26 2021 +0700
[REFACTORING] Reactor-Scala: raiseError is deprecated and error should be used
Underneath the call effects are strictly the same.
---
.../apache/james/jmap/http/MailboxesProvisioner.scala | 2 +-
.../org/apache/james/jmap/mail/MailboxFactory.scala | 6 +++---
.../org/apache/james/jmap/method/EmailGetMethod.scala | 6 +++---
.../org/apache/james/jmap/method/EmailQueryMethod.scala | 7 +++----
.../org/apache/james/jmap/method/MDNParseMethod.scala | 7 ++++---
.../org/apache/james/jmap/method/MDNSendMethod.scala | 8 ++++----
.../org/apache/james/jmap/method/MailboxGetMethod.scala | 2 +-
.../apache/james/jmap/method/MailboxQueryMethod.scala | 2 +-
.../james/jmap/method/MailboxSetDeletePerformer.scala | 2 +-
.../james/jmap/method/MailboxSetUpdatePerformer.scala | 2 +-
.../main/scala/org/apache/james/jmap/method/Method.scala | 4 ++--
.../org/apache/james/jmap/routes/DownloadRoutes.scala | 16 ++++++++--------
.../scala/org/apache/james/jmap/routes/JmapApi.scala | 4 ++--
.../org/apache/james/jmap/routes/UploadRoutes.scala | 6 +++---
.../org/apache/james/jmap/routes/JMAPApiRoutesTest.scala | 2 +-
15 files changed, 38 insertions(+), 38 deletions(-)
diff --git a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/http/MailboxesProvisioner.scala b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/http/MailboxesProvisioner.scala
index 2690ee5..742105a 100644
--- a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/http/MailboxesProvisioner.scala
+++ b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/http/MailboxesProvisioner.scala
@@ -59,7 +59,7 @@ class MailboxesProvisioner @Inject() (mailboxManager: MailboxManager,
SMono(mailboxManager.mailboxExists(mailboxPath, session))
.map(exist => !exist)
} catch {
- case exception: MailboxException => SMono.raiseError(exception)
+ case exception: MailboxException => SMono.error(exception)
}
}
diff --git a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/mail/MailboxFactory.scala b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/mail/MailboxFactory.scala
index 4d092c3..50fff05 100644
--- a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/mail/MailboxFactory.scala
+++ b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/mail/MailboxFactory.scala
@@ -129,7 +129,7 @@ class MailboxFactory @Inject() (subscriptionManager: SubscriptionManager, mailbo
val sanitizedCounters: MailboxCounters = mailboxMetaData.getCounters.sanitize()
MailboxValidation.validate(mailboxMetaData.getPath, mailboxSession.getPathDelimiter, sanitizedCounters.getUnseen, sanitizedCounters.getUnseen, sanitizedCounters.getCount, sanitizedCounters.getCount) match {
- case Left(error) => SMono.raiseError(error)
+ case Left(error) => SMono.error(error)
case scala.Right(mailboxValidation) =>
SMono.fromPublisher(quotaLoader.getQuotas(mailboxMetaData.getPath))
.map(quotas => {
@@ -169,7 +169,7 @@ class MailboxFactory @Inject() (subscriptionManager: SubscriptionManager, mailbo
val sanitizedCounters: MailboxCounters = messageManager.getMailboxCounters(mailboxSession).sanitize()
MailboxValidation.validate(messageManager.getMailboxPath, mailboxSession.getPathDelimiter, sanitizedCounters.getUnseen, sanitizedCounters.getUnseen, sanitizedCounters.getCount, sanitizedCounters.getCount) match {
- case Left(error) => SMono.raiseError(error)
+ case Left(error) => SMono.error(error)
case scala.Right(mailboxValidation) =>
SMono.fromPublisher(quotaLoader.getQuotas(messageManager.getMailboxPath))
.map(quotas => {
@@ -206,7 +206,7 @@ class MailboxFactory @Inject() (subscriptionManager: SubscriptionManager, mailbo
isSubscribed = isSubscribed)})
}
} catch {
- case error: Exception => SMono.raiseError(error)
+ case error: Exception => SMono.error(error)
}
}
}
\ No newline at end of file
diff --git a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/EmailGetMethod.scala b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/EmailGetMethod.scala
index 188ab96..41cb0d0 100644
--- a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/EmailGetMethod.scala
+++ b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/EmailGetMethod.scala
@@ -87,7 +87,7 @@ class EmailGetMethod @Inject() (readerFactory: EmailViewReaderFactory,
override def doProcess(capabilities: Set[CapabilityIdentifier], invocation: InvocationWithContext, mailboxSession: MailboxSession, request: EmailGetRequest): SMono[InvocationWithContext] = {
computeResponseInvocation(request, invocation.invocation, mailboxSession).onErrorResume({
case e: IllegalArgumentException => SMono.just(Invocation.error(ErrorCode.InvalidArguments, e.getMessage, invocation.invocation.methodCallId))
- case e: Throwable => SMono.raiseError(e)
+ case e: Throwable => SMono.error(e)
}).map(invocationResult => InvocationWithContext(invocationResult, invocation.processingContext))
}
@@ -101,7 +101,7 @@ class EmailGetMethod @Inject() (readerFactory: EmailViewReaderFactory,
validateProperties(request)
.flatMap(properties => validateBodyProperties(request).map((properties, _)))
.fold(
- e => SMono.raiseError(e), {
+ e => SMono.error(e), {
case (properties, bodyProperties) => getEmails(request, mailboxSession)
.map(response => Invocation(
methodName = methodName,
@@ -141,7 +141,7 @@ class EmailGetMethod @Inject() (readerFactory: EmailViewReaderFactory,
private def getEmails(request: EmailGetRequest, mailboxSession: MailboxSession): SMono[EmailGetResponse] =
request.ids match {
- case None => SMono.raiseError(new IllegalArgumentException("ids can not be ommited for email/get"))
+ case None => SMono.error(new IllegalArgumentException("ids can not be ommited for email/get"))
case Some(ids) => getEmails(ids, mailboxSession, request)
.flatMap(result => SMono[JavaState](emailchangeRepository.getLatestState(JavaAccountId.fromUsername(mailboxSession.getUser)))
.map(state => EmailGetResponse(
diff --git a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/EmailQueryMethod.scala b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/EmailQueryMethod.scala
index 6f08d44..186e80c 100644
--- a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/EmailQueryMethod.scala
+++ b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/EmailQueryMethod.scala
@@ -42,7 +42,6 @@ import org.apache.james.metrics.api.MetricFactory
import org.apache.james.util.streams.{Limit => JavaLimit}
import play.api.libs.json.{JsError, JsSuccess}
import reactor.core.scala.publisher.{SFlux, SMono}
-import reactor.core.scheduler.Schedulers
import scala.jdk.CollectionConverters._
@@ -75,7 +74,7 @@ class EmailQueryMethod @Inject() (serializer: EmailQuerySerializer,
arguments = Arguments(serializer.serialize(response)),
methodCallId = invocation.methodCallId))
}
- validation.fold(SMono.raiseError, res => res)
+ validation.fold(SMono.error, res => res)
}
override def getRequest(mailboxSession: MailboxSession, invocation: Invocation): Either[Exception, EmailQueryRequest] =
@@ -115,7 +114,7 @@ class EmailQueryMethod @Inject() (serializer: EmailQuerySerializer,
.collectSeq())
.onErrorResume({
case _: MailboxNotFoundException => SMono.just[Seq[MessageId]](Seq())
- case e => SMono.raiseError[Seq[MessageId]](e)
+ case e => SMono.error[Seq[MessageId]](e)
})
}
@@ -129,7 +128,7 @@ class EmailQueryMethod @Inject() (serializer: EmailQuerySerializer,
.collectSeq())
.onErrorResume({
case _: MailboxNotFoundException => SMono.just[Seq[MessageId]](Seq())
- case e => SMono.raiseError[Seq[MessageId]](e)
+ case e => SMono.error[Seq[MessageId]](e)
})
}
diff --git a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/MDNParseMethod.scala b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/MDNParseMethod.scala
index bea693b..d0e52c5 100644
--- a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/MDNParseMethod.scala
+++ b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/MDNParseMethod.scala
@@ -19,7 +19,10 @@
package org.apache.james.jmap.method
+import java.io.InputStream
+
import eu.timepit.refined.auto._
+import javax.inject.Inject
import org.apache.james.jmap.core.CapabilityIdentifier.{CapabilityIdentifier, JMAP_CORE, JMAP_MAIL, JMAP_MDN}
import org.apache.james.jmap.core.Invocation
import org.apache.james.jmap.core.Invocation._
@@ -36,8 +39,6 @@ import org.apache.james.mime4j.message.DefaultMessageBuilder
import play.api.libs.json.{JsError, JsObject, JsSuccess, Json}
import reactor.core.scala.publisher.{SFlux, SMono}
-import java.io.InputStream
-import javax.inject.Inject
import scala.jdk.OptionConverters._
import scala.util.Try
@@ -113,7 +114,7 @@ class MDNParseMethod @Inject()(serializer: MDNSerializer,
} yield {
(mdn, message)
}
- maybeMdn.fold(_ => SMono.raiseError(BlobUnParsableException(blobId)), result => SMono.just(result))
+ maybeMdn.fold(_ => SMono.error(BlobUnParsableException(blobId)), result => SMono.just(result))
}
}
diff --git a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/MDNSendMethod.scala b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/MDNSendMethod.scala
index 261b6fb..aef5905 100644
--- a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/MDNSendMethod.scala
+++ b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/MDNSendMethod.scala
@@ -20,6 +20,9 @@
package org.apache.james.jmap.method
import eu.timepit.refined.auto._
+import javax.annotation.PreDestroy
+import javax.inject.Inject
+import javax.mail.internet.MimeMessage
import org.apache.james.jmap.core.CapabilityIdentifier.{CapabilityIdentifier, JMAP_CORE, JMAP_MAIL, JMAP_MDN}
import org.apache.james.jmap.core.Invocation
import org.apache.james.jmap.core.Invocation._
@@ -47,9 +50,6 @@ import play.api.libs.json.{JsError, JsObject, JsSuccess, Json}
import reactor.core.scala.publisher.{SFlux, SMono}
import reactor.core.scheduler.Schedulers
-import javax.annotation.PreDestroy
-import javax.inject.Inject
-import javax.mail.internet.MimeMessage
import scala.jdk.CollectionConverters._
import scala.jdk.OptionConverters._
import scala.util.Try
@@ -78,7 +78,7 @@ class MDNSendMethod @Inject()(serializer: MDNSerializer,
request: MDNSendRequest): SFlux[InvocationWithContext] =
identityResolver.resolveIdentityId(request.identityId, mailboxSession)
.flatMap(maybeIdentity => maybeIdentity.map(identity => create(identity, request, mailboxSession, invocation.processingContext))
- .getOrElse(SMono.raiseError(IdentityIdNotFoundException("The IdentityId cannot be found"))))
+ .getOrElse(SMono.error(IdentityIdNotFoundException("The IdentityId cannot be found"))))
.flatMapMany(createdResults => {
val explicitInvocation: InvocationWithContext = InvocationWithContext(
invocation = Invocation(
diff --git a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/MailboxGetMethod.scala b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/MailboxGetMethod.scala
index fa0940a..6dae60f 100644
--- a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/MailboxGetMethod.scala
+++ b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/MailboxGetMethod.scala
@@ -131,7 +131,7 @@ class MailboxGetMethod @Inject() (serializer: MailboxSerializer,
.map(mailbox => filterShared(capabilities, mailbox))
.onErrorResume {
case _: MailboxNotFoundException => SMono.just(MailboxGetResults.notFound(mailboxId))
- case error => SMono.raiseError(error)
+ case error => SMono.error(error)
})
.subscribeOn(Schedulers.elastic)
diff --git a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/MailboxQueryMethod.scala b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/MailboxQueryMethod.scala
index b19a8af..0483e18 100644
--- a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/MailboxQueryMethod.scala
+++ b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/MailboxQueryMethod.scala
@@ -46,7 +46,7 @@ class MailboxQueryMethod @Inject()(systemMailboxesProvider: SystemMailboxesProvi
errorCode = ErrorCode.InvalidArguments,
description = e.getMessage,
methodCallId = invocation.invocation.methodCallId))
- case e: Throwable => SMono.raiseError(e)
+ case e: Throwable => SMono.error(e)
}
.map(invocationResult => InvocationWithContext(invocationResult, invocation.processingContext))
}
diff --git a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/MailboxSetDeletePerformer.scala b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/MailboxSetDeletePerformer.scala
index 66704e1..c19c02e 100644
--- a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/MailboxSetDeletePerformer.scala
+++ b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/MailboxSetDeletePerformer.scala
@@ -74,7 +74,7 @@ class MailboxSetDeletePerformer @Inject()(mailboxManager: MailboxManager,
private def delete(mailboxSession: MailboxSession, id: UnparsedMailboxId, onDestroy: RemoveEmailsOnDestroy): SMono[MailboxDeletionResult] = {
MailboxGet.parse(mailboxIdFactory)(id)
- .fold(e => SMono.raiseError(e),
+ .fold(e => SMono.error(e),
id => SMono.fromCallable(() => doDelete(mailboxSession, id, onDestroy))
.subscribeOn(Schedulers.elastic())
.`then`(SMono.just[MailboxDeletionResult](MailboxDeletionSuccess(id))))
diff --git a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/MailboxSetUpdatePerformer.scala b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/MailboxSetUpdatePerformer.scala
index 0b41a86..fbf56fb 100644
--- a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/MailboxSetUpdatePerformer.scala
+++ b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/MailboxSetUpdatePerformer.scala
@@ -102,7 +102,7 @@ class MailboxSetUpdatePerformer @Inject()(serializer: MailboxSerializer,
patch: MailboxPatchObject,
capabilities: Set[CapabilityIdentifier]): SMono[MailboxUpdateResult] = {
patch.validate(mailboxIdFactory, serializer, capabilities, mailboxSession)
- .fold(e => SMono.raiseError(e), validatedPatch =>
+ .fold(e => SMono.error(e), validatedPatch =>
updateMailboxRights(mailboxId, validatedPatch, mailboxSession)
.`then`(updateSubscription(mailboxId, validatedPatch, mailboxSession))
.`then`(updateMailboxPath(mailboxId, unparsedMailboxId, validatedPatch, mailboxSession)))
diff --git a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/Method.scala b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/Method.scala
index 4e149d5..557ae2e 100644
--- a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/Method.scala
+++ b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/method/Method.scala
@@ -61,7 +61,7 @@ trait MethodRequiringAccountId[REQUEST <: WithAccountId] extends Method {
doProcess(capabilities, invocation, mailboxSession, request)
}
- val result: SFlux[InvocationWithContext] = SFlux.fromPublisher(either.fold(e => SFlux.raiseError[InvocationWithContext](e), r => r))
+ val result: SFlux[InvocationWithContext] = SFlux.fromPublisher(either.fold(e => SFlux.error[InvocationWithContext](e), r => r))
.onErrorResume[InvocationWithContext] {
case e: AccountNotFoundException => SFlux.just[InvocationWithContext] (InvocationWithContext(e.invocation, invocation.processingContext))
case e: UnsupportedRequestParameterException => SFlux.just[InvocationWithContext] (InvocationWithContext(Invocation.error(
@@ -85,7 +85,7 @@ trait MethodRequiringAccountId[REQUEST <: WithAccountId] extends Method {
case e: ChangeNotFoundException => SFlux.just[InvocationWithContext] (InvocationWithContext(Invocation.error(ErrorCode.CannotCalculateChanges, e.getMessage, invocation.invocation.methodCallId), invocation.processingContext))
case e: RequestTooLargeException => SFlux.just[InvocationWithContext] (InvocationWithContext(Invocation.error(ErrorCode.RequestTooLarge, e.description, invocation.invocation.methodCallId), invocation.processingContext))
case e: IdentityIdNotFoundException => SFlux.just[InvocationWithContext] (InvocationWithContext(Invocation.error(ErrorCode.InvalidArguments, e.description, invocation.invocation.methodCallId), invocation.processingContext))
- case e: Throwable => SFlux.raiseError[InvocationWithContext] (e)
+ case e: Throwable => SFlux.error[InvocationWithContext] (e)
}
metricFactory.decoratePublisherWithTimerMetricLogP99(JMAP_RFC8621_PREFIX + methodName.value, result)
diff --git a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/routes/DownloadRoutes.scala b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/routes/DownloadRoutes.scala
index 841f3e2..83c7c7b 100644
--- a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/routes/DownloadRoutes.scala
+++ b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/routes/DownloadRoutes.scala
@@ -130,7 +130,7 @@ class MessageBlobResolver @Inject()(val messageIdFactory: MessageId.Factory,
case Success(messageId) => Applicable(SMono.fromPublisher(
messageIdManager.getMessagesReactive(List(messageId).asJava, FetchGroup.FULL_CONTENT, mailboxSession))
.map[Blob](MessageBlob(blobId, _))
- .switchIfEmpty(SMono.raiseError(BlobNotFoundException(blobId))))
+ .switchIfEmpty(SMono.error(BlobNotFoundException(blobId))))
}
}
}
@@ -142,7 +142,7 @@ class AttachmentBlobResolver @Inject()(val attachmentManager: AttachmentManager)
Try(attachmentManager.getAttachment(attachmentId, mailboxSession)) match {
case Success(attachmentMetadata) => Applicable(
SMono.fromCallable(() => AttachmentBlob(attachmentMetadata, attachmentManager.load(attachmentMetadata, mailboxSession))))
- case Failure(_) => Applicable(SMono.raiseError(BlobNotFoundException(blobId)))
+ case Failure(_) => Applicable(SMono.error(BlobNotFoundException(blobId)))
}
case _ => NonApplicable()
@@ -180,7 +180,7 @@ class MessagePartBlobResolver @Inject()(val messageIdFactory: MessageId.Factory,
.fold(sink.error(BlobNotFoundException(blobId)))(part => sink.next(part))
}
.map[Blob](EmailBodyPartBlob(blobId, _))
- .switchIfEmpty(SMono.raiseError(BlobNotFoundException(blobId))))
+ .switchIfEmpty(SMono.error(BlobNotFoundException(blobId))))
}
}
}
@@ -193,7 +193,7 @@ class BlobResolvers @Inject()(val messageBlobResolver: MessageBlobResolver,
.resolve(blobId, mailboxSession).asOption
.orElse(messagePartBlobResolver.resolve(blobId, mailboxSession).asOption)
.orElse(attachmentBlobResolver.resolve(blobId, mailboxSession).asOption)
- .getOrElse(SMono.raiseError(BlobNotFoundException(blobId)))
+ .getOrElse(SMono.error(BlobNotFoundException(blobId)))
}
class DownloadRoutes @Inject()(@Named(InjectionKeys.RFC_8621) val authenticator: Authenticator,
@@ -244,7 +244,7 @@ class DownloadRoutes @Inject()(@Named(InjectionKeys.RFC_8621) val authenticator:
private def get(request: HttpServerRequest, response: HttpServerResponse, mailboxSession: MailboxSession): SMono[Unit] = {
BlobId.of(request.param(blobIdParam))
- .fold(e => SMono.raiseError(e),
+ .fold(e => SMono.error(e),
blobResolvers.resolve(_, mailboxSession))
.flatMap(blob => downloadBlob(
optionalName = queryParam(request, nameParam),
@@ -262,15 +262,15 @@ class DownloadRoutes @Inject()(@Named(InjectionKeys.RFC_8621) val authenticator:
val targetAccountId: AccountId = AccountId(id)
AccountId.from(mailboxSession.getUser).map(accountId => accountId.equals(targetAccountId))
.fold[SMono[Unit]](
- e => SMono.raiseError(e),
+ e => SMono.error(e),
value => if (value) {
get(request, response, mailboxSession)
} else {
- SMono.raiseError(ForbiddenException())
+ SMono.error(ForbiddenException())
})
}
- case Left(throwable: Throwable) => SMono.raiseError(throwable)
+ case Left(throwable: Throwable) => SMono.error(throwable)
}
}
diff --git a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/routes/JmapApi.scala b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/routes/JmapApi.scala
index 65a0320..3d19067 100644
--- a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/routes/JmapApi.scala
+++ b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/routes/JmapApi.scala
@@ -21,7 +21,7 @@ package org.apache.james.jmap.routes
import javax.inject.Inject
import org.apache.james.jmap.core.CapabilityIdentifier.CapabilityIdentifier
import org.apache.james.jmap.core.Invocation.MethodName
-import org.apache.james.jmap.core.{Capability, DefaultCapabilities, ErrorCode, Invocation, MissingCapabilityException, RequestObject, ResponseObject}
+import org.apache.james.jmap.core.{Capability, ErrorCode, Invocation, MissingCapabilityException, RequestObject, ResponseObject}
import org.apache.james.jmap.method.{InvocationWithContext, Method}
import org.apache.james.mailbox.MailboxSession
import org.slf4j.{Logger, LoggerFactory}
@@ -50,7 +50,7 @@ class JMAPApi (methods: Set[Method], defaultCapabilities: Set[Capability]) {
val capabilities: Set[CapabilityIdentifier] = requestObject.using.toSet
if (unsupportedCapabilities.nonEmpty) {
- SMono.raiseError(UnsupportedCapabilitiesException(unsupportedCapabilities))
+ SMono.error(UnsupportedCapabilitiesException(unsupportedCapabilities))
} else {
processSequentiallyAndUpdateContext(requestObject, mailboxSession, processingContext, capabilities)
.map(invocations => ResponseObject(ResponseObject.SESSION_STATE, invocations.map(_.invocation)))
diff --git a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/routes/UploadRoutes.scala b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/routes/UploadRoutes.scala
index bdc5931..8e7996b 100644
--- a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/routes/UploadRoutes.scala
+++ b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/routes/UploadRoutes.scala
@@ -136,7 +136,7 @@ class UploadRoutes @Inject()(@Named(InjectionKeys.RFC_8621) val authenticator: A
val targetAccountId: AccountId = AccountId(id)
AccountId.from(session.getUser).map(accountId => accountId.equals(targetAccountId))
.fold[SMono[Void]](
- e => SMono.raiseError(e),
+ e => SMono.error(e),
value => if (value) {
SMono.fromCallable(() => ReactorUtils.toInputStream(request.receive
// Unwrapping to byte array needed to solve data races and buffer reordering when using .asByteBuffer()
@@ -144,10 +144,10 @@ class UploadRoutes @Inject()(@Named(InjectionKeys.RFC_8621) val authenticator: A
.map(array => ByteBuffer.wrap(array))))
.flatMap(content => handle(targetAccountId, contentType, content, session, response))
} else {
- SMono.raiseError(ForbiddenException())
+ SMono.error(ForbiddenException())
})
- case Left(throwable: Throwable) => SMono.raiseError(throwable)
+ case Left(throwable: Throwable) => SMono.error(throwable)
}
}
diff --git a/server/protocols/jmap-rfc-8621/src/test/scala/org/apache/james/jmap/routes/JMAPApiRoutesTest.scala b/server/protocols/jmap-rfc-8621/src/test/scala/org/apache/james/jmap/routes/JMAPApiRoutesTest.scala
index ab43abe..da32739 100644
--- a/server/protocols/jmap-rfc-8621/src/test/scala/org/apache/james/jmap/routes/JMAPApiRoutesTest.scala
+++ b/server/protocols/jmap-rfc-8621/src/test/scala/org/apache/james/jmap/routes/JMAPApiRoutesTest.scala
@@ -433,7 +433,7 @@ class JMAPApiRoutesTest extends AnyFlatSpec with BeforeAndAfter with Matchers {
"RFC-8621 with random error when processing request " should "return 200, with serverFail error, others method call proceed normally" in {
val mockCoreEchoMethod = mock(classOf[CoreEchoMethod])
- doReturn(SFlux.raiseError(new RuntimeException("Unexpected Exception occur, the others method may proceed normally")))
+ doReturn(SFlux.error(new RuntimeException("Unexpected Exception occur, the others method may proceed normally")))
.doCallRealMethod()
.when(mockCoreEchoMethod)
.process(any[Set[CapabilityIdentifier]], any(), any())
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 04/11: [REFACTORING] Reactive
StoreMessageIdManager::validateQuota
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 4f5d8c43f4ae83a1847476dea698428b8d8788fd
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Thu May 13 15:39:52 2021 +0700
[REFACTORING] Reactive StoreMessageIdManager::validateQuota
This ensures using StoreMessageIdManager::setInMailboxesReactive do
not block for the purpose of quota validation for copy use cases.
---
.../james/mailbox/store/StoreMessageIdManager.java | 60 +++++++++++++---------
.../AbstractMessageIdManagerSideEffectTest.java | 22 ++++----
2 files changed, 49 insertions(+), 33 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 a7fa33b..b1ac262 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
@@ -46,6 +46,7 @@ import org.apache.james.mailbox.RightManager;
import org.apache.james.mailbox.events.MailboxIdRegistrationKey;
import org.apache.james.mailbox.exception.MailboxException;
import org.apache.james.mailbox.exception.MailboxNotFoundException;
+import org.apache.james.mailbox.exception.OverQuotaException;
import org.apache.james.mailbox.extension.PreDeletionHook;
import org.apache.james.mailbox.model.ComposedMessageIdWithMetaData;
import org.apache.james.mailbox.model.DeleteResult;
@@ -357,7 +358,7 @@ public class StoreMessageIdManager implements MessageIdManager {
.map(mailbox -> Pair.of(message, mailbox)))
.collect(Guavate.toImmutableList());
- return Mono.fromRunnable(Throwing.runnable(() -> validateQuota(messageMoves, mailboxMessage.get())).sneakyThrow())
+ return validateQuota(messageMoves, mailboxMessage.get())
.then(Mono.fromRunnable(Throwing.runnable(() ->
addMessageToMailboxes(mailboxMessage.get(), messageMoves.addedMailboxes(), mailboxSession)).sneakyThrow()))
.then(removeMessageFromMailboxes(mailboxMessage.get().getMessageId(), messagesToRemove, mailboxSession))
@@ -415,33 +416,44 @@ public class StoreMessageIdManager implements MessageIdManager {
return Mono.empty();
}
- private void validateQuota(MessageMovesWithMailbox messageMoves, MailboxMessage mailboxMessage) throws MailboxException {
+ private Mono<Void> validateQuota(MessageMovesWithMailbox messageMoves, MailboxMessage mailboxMessage) {
Map<QuotaRoot, Integer> messageCountByQuotaRoot = buildMapQuotaRoot(messageMoves);
- for (Map.Entry<QuotaRoot, Integer> entry : messageCountByQuotaRoot.entrySet()) {
- Integer additionalCopyCount = entry.getValue();
- if (additionalCopyCount > 0) {
- long additionalOccupiedSpace = additionalCopyCount * mailboxMessage.getFullContentOctets();
- new QuotaChecker(quotaManager.getQuotas(entry.getKey()), entry.getKey())
- .tryAddition(additionalCopyCount, additionalOccupiedSpace);
- }
- }
+
+ return Flux.fromIterable(messageCountByQuotaRoot.entrySet())
+ .filter(entry -> entry.getValue() > 0)
+ .flatMap(entry -> Mono.from(quotaManager.getQuotasReactive(entry.getKey()))
+ .map(quotas -> new QuotaChecker(quotas, entry.getKey()))
+ .handle((quotaChecker, sink) -> {
+ Integer additionalCopyCount = entry.getValue();
+ long additionalOccupiedSpace = additionalCopyCount * mailboxMessage.getFullContentOctets();
+ try {
+ quotaChecker.tryAddition(additionalCopyCount, additionalOccupiedSpace);
+ } catch (OverQuotaException e) {
+ sink.error(e);
+ }
+ }))
+ .then();
}
- private Map<QuotaRoot, Integer> buildMapQuotaRoot(MessageMovesWithMailbox messageMoves) throws MailboxException {
- Map<QuotaRoot, Integer> messageCountByQuotaRoot = new HashMap<>();
- for (Mailbox mailbox : messageMoves.addedMailboxes()) {
- QuotaRoot quotaRoot = quotaRootResolver.getQuotaRoot(mailbox);
- int currentCount = Optional.ofNullable(messageCountByQuotaRoot.get(quotaRoot))
- .orElse(0);
- messageCountByQuotaRoot.put(quotaRoot, currentCount + 1);
- }
- for (Mailbox mailbox : messageMoves.removedMailboxes()) {
- QuotaRoot quotaRoot = quotaRootResolver.getQuotaRoot(mailbox);
- int currentCount = Optional.ofNullable(messageCountByQuotaRoot.get(quotaRoot))
- .orElse(0);
- messageCountByQuotaRoot.put(quotaRoot, currentCount - 1);
+ private Map<QuotaRoot, Integer> buildMapQuotaRoot(MessageMovesWithMailbox messageMoves) {
+ try {
+ Map<QuotaRoot, Integer> messageCountByQuotaRoot = new HashMap<>();
+ for (Mailbox mailbox : messageMoves.addedMailboxes()) {
+ QuotaRoot quotaRoot = quotaRootResolver.getQuotaRoot(mailbox);
+ int currentCount = Optional.ofNullable(messageCountByQuotaRoot.get(quotaRoot))
+ .orElse(0);
+ messageCountByQuotaRoot.put(quotaRoot, currentCount + 1);
+ }
+ for (Mailbox mailbox : messageMoves.removedMailboxes()) {
+ QuotaRoot quotaRoot = quotaRootResolver.getQuotaRoot(mailbox);
+ int currentCount = Optional.ofNullable(messageCountByQuotaRoot.get(quotaRoot))
+ .orElse(0);
+ messageCountByQuotaRoot.put(quotaRoot, currentCount - 1);
+ }
+ return messageCountByQuotaRoot;
+ } catch (MailboxException e) {
+ throw new RuntimeException(e);
}
- return messageCountByQuotaRoot;
}
private void addMessageToMailboxes(MailboxMessage mailboxMessage, Set<Mailbox> mailboxes, MailboxSession mailboxSession) throws MailboxException {
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 a6011de..1c86c65 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
@@ -56,7 +56,6 @@ import org.apache.james.mailbox.events.MailboxEvents.Expunged;
import org.apache.james.mailbox.events.MailboxEvents.FlagsUpdated;
import org.apache.james.mailbox.events.MailboxEvents.MailboxEvent;
import org.apache.james.mailbox.events.MessageMoveEvent;
-import org.apache.james.mailbox.exception.MailboxException;
import org.apache.james.mailbox.exception.OverQuotaException;
import org.apache.james.mailbox.extension.PreDeletionHook;
import org.apache.james.mailbox.fixture.MailboxFixture;
@@ -115,6 +114,11 @@ public abstract class AbstractMessageIdManagerSideEffectTest {
eventCollector = new EventCollector();
quotaManager = mock(QuotaManager.class);
+ when(quotaManager.getQuotasReactive(any(QuotaRoot.class)))
+ .thenReturn(Mono.just(new QuotaManager.Quotas(
+ Quota.<QuotaCountLimit, QuotaCountUsage>builder().used(QuotaCountUsage.count(102)).computedLimit(QuotaCountLimit.unlimited()).build(),
+ Quota.<QuotaSizeLimit, QuotaSizeUsage>builder().used(QuotaSizeUsage.size(2)).computedLimit(QuotaSizeLimit.unlimited()).build())));
+
session = MailboxSessionUtil.create(ALICE);
setupMockForPreDeletionHooks();
testingData = createTestSystem(quotaManager, eventBus, ImmutableSet.of(preDeletionHook1, preDeletionHook2));
@@ -382,13 +386,13 @@ public abstract class AbstractMessageIdManagerSideEffectTest {
}
@Test
- void setInMailboxesShouldThrowExceptionWhenOverQuota() throws Exception {
+ void setInMailboxesShouldThrowExceptionWhenOverQuota() {
MessageId messageId = testingData.persist(mailbox1.getMailboxId(), messageUid1, FLAGS, session);
- when(quotaManager.getQuotas(any(QuotaRoot.class)))
- .thenReturn(new QuotaManager.Quotas(
+ when(quotaManager.getQuotasReactive(any(QuotaRoot.class)))
+ .thenReturn(Mono.just(new QuotaManager.Quotas(
OVER_QUOTA,
- Quota.<QuotaSizeLimit, QuotaSizeUsage>builder().used(QuotaSizeUsage.size(2)).computedLimit(QuotaSizeLimit.unlimited()).build()));
+ Quota.<QuotaSizeLimit, QuotaSizeUsage>builder().used(QuotaSizeUsage.size(2)).computedLimit(QuotaSizeLimit.unlimited()).build())));
assertThatThrownBy(() -> messageIdManager.setInMailboxes(messageId,
ImmutableList.of(mailbox1.getMailboxId(), mailbox2.getMailboxId()),
@@ -553,10 +557,10 @@ public abstract class AbstractMessageIdManagerSideEffectTest {
assertThat(eventCollector.getEvents()).isEmpty();
}
- protected void givenUnlimitedQuota() throws MailboxException {
- when(quotaManager.getQuotas(any(QuotaRoot.class)))
- .thenReturn(new QuotaManager.Quotas(
+ protected void givenUnlimitedQuota() {
+ when(quotaManager.getQuotasReactive(any(QuotaRoot.class)))
+ .thenReturn(Mono.just(new QuotaManager.Quotas(
Quota.<QuotaCountLimit, QuotaCountUsage>builder().used(QuotaCountUsage.count(2)).computedLimit(QuotaCountLimit.unlimited()).build(),
- Quota.<QuotaSizeLimit, QuotaSizeUsage>builder().used(QuotaSizeUsage.size(2)).computedLimit(QuotaSizeLimit.unlimited()).build()));
+ Quota.<QuotaSizeLimit, QuotaSizeUsage>builder().used(QuotaSizeUsage.size(2)).computedLimit(QuotaSizeLimit.unlimited()).build())));
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 08/11: [REFACTORING]
StoreMessageIdManager::setInMailboxes should not block
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 07250a7684737e2d6b8972a6a3e51ff3ba0d5b58
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Thu May 13 16:30:59 2021 +0700
[REFACTORING] StoreMessageIdManager::setInMailboxes should not block
---
.../james/mailbox/store/StoreMessageIdManager.java | 82 ++++++++++++----------
1 file changed, 43 insertions(+), 39 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 b1ac262..11ca8ff 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
@@ -39,9 +39,7 @@ import org.apache.james.events.EventBus;
import org.apache.james.mailbox.MailboxSession;
import org.apache.james.mailbox.MessageIdManager;
import org.apache.james.mailbox.MessageManager;
-import org.apache.james.mailbox.MessageUid;
import org.apache.james.mailbox.MetadataWithMailboxId;
-import org.apache.james.mailbox.ModSeq;
import org.apache.james.mailbox.RightManager;
import org.apache.james.mailbox.events.MailboxIdRegistrationKey;
import org.apache.james.mailbox.exception.MailboxException;
@@ -56,6 +54,7 @@ import org.apache.james.mailbox.model.MailboxACL;
import org.apache.james.mailbox.model.MailboxACL.Right;
import org.apache.james.mailbox.model.MailboxId;
import org.apache.james.mailbox.model.MessageId;
+import org.apache.james.mailbox.model.MessageMetaData;
import org.apache.james.mailbox.model.MessageMoves;
import org.apache.james.mailbox.model.MessageResult;
import org.apache.james.mailbox.model.QuotaRoot;
@@ -359,8 +358,7 @@ public class StoreMessageIdManager implements MessageIdManager {
.collect(Guavate.toImmutableList());
return validateQuota(messageMoves, mailboxMessage.get())
- .then(Mono.fromRunnable(Throwing.runnable(() ->
- addMessageToMailboxes(mailboxMessage.get(), messageMoves.addedMailboxes(), mailboxSession)).sneakyThrow()))
+ .then(addMessageToMailboxes(mailboxMessage.get(), messageMoves.addedMailboxes(), mailboxSession))
.then(removeMessageFromMailboxes(mailboxMessage.get().getMessageId(), messagesToRemove, mailboxSession))
.then(eventBus.dispatch(EventFactory.moved()
.session(mailboxSession)
@@ -456,45 +454,51 @@ public class StoreMessageIdManager implements MessageIdManager {
}
}
- private void addMessageToMailboxes(MailboxMessage mailboxMessage, Set<Mailbox> mailboxes, MailboxSession mailboxSession) throws MailboxException {
+ private Mono<Void> addMessageToMailboxes(MailboxMessage mailboxMessage, Set<Mailbox> mailboxes, MailboxSession mailboxSession) {
MessageIdMapper messageIdMapper = mailboxSessionMapperFactory.getMessageIdMapper(mailboxSession);
- for (Mailbox mailbox : mailboxes) {
- MailboxACL.Rfc4314Rights myRights = rightManager.myRights(mailbox, mailboxSession);
- boolean shouldPreserveFlags = myRights.contains(Right.Write);
- SimpleMailboxMessage copy =
- SimpleMailboxMessage.from(mailboxMessage)
- .mailboxId(mailbox.getMailboxId())
- .flags(
- FlagsFactory
- .builder()
- .flags(mailboxMessage.createFlags())
- .filteringFlags(
- FlagsFilter.builder()
- .systemFlagFilter(f -> shouldPreserveFlags)
- .userFlagFilter(f -> shouldPreserveFlags)
- .build())
- .build())
- .build();
- save(messageIdMapper, copy, mailbox);
-
- eventBus.dispatch(EventFactory.added()
- .randomEventId()
- .mailboxSession(mailboxSession)
- .mailbox(mailbox)
- .addMetaData(copy.metaData())
- .build(),
- new MailboxIdRegistrationKey(mailbox.getMailboxId()))
- .block();
- }
+ return Flux.fromIterable(mailboxes)
+ .flatMap(Throwing.<Mailbox, Mono<Void>>function(mailbox -> {
+ MailboxACL.Rfc4314Rights myRights = rightManager.myRights(mailbox, mailboxSession);
+ boolean shouldPreserveFlags = myRights.contains(Right.Write);
+ SimpleMailboxMessage copy =
+ SimpleMailboxMessage.from(mailboxMessage)
+ .mailboxId(mailbox.getMailboxId())
+ .flags(
+ FlagsFactory
+ .builder()
+ .flags(mailboxMessage.createFlags())
+ .filteringFlags(
+ FlagsFilter.builder()
+ .systemFlagFilter(f -> shouldPreserveFlags)
+ .userFlagFilter(f -> shouldPreserveFlags)
+ .build())
+ .build())
+ .build();
+
+ return save(messageIdMapper, copy, mailbox)
+ .flatMap(metadata -> eventBus.dispatch(EventFactory.added()
+ .randomEventId()
+ .mailboxSession(mailboxSession)
+ .mailbox(mailbox)
+ .addMetaData(metadata)
+ .build(),
+ new MailboxIdRegistrationKey(mailbox.getMailboxId())));
+ }).sneakyThrow())
+ .then();
}
- private void save(MessageIdMapper messageIdMapper, MailboxMessage mailboxMessage, Mailbox mailbox) throws MailboxException {
- ModSeq modSeq = mailboxSessionMapperFactory.getModSeqProvider().nextModSeq(mailbox.getMailboxId());
- MessageUid uid = mailboxSessionMapperFactory.getUidProvider().nextUid(mailbox.getMailboxId());
- mailboxMessage.setModSeq(modSeq);
- mailboxMessage.setUid(uid);
- messageIdMapper.copyInMailbox(mailboxMessage, mailbox);
+ private Mono<MessageMetaData> save(MessageIdMapper messageIdMapper, MailboxMessage mailboxMessage, Mailbox mailbox) {
+ return Mono.zip(
+ mailboxSessionMapperFactory.getModSeqProvider().nextModSeqReactive(mailbox.getMailboxId()),
+ mailboxSessionMapperFactory.getUidProvider().nextUidReactive(mailbox.getMailboxId()))
+ .flatMap(modSeqAndUid -> {
+ mailboxMessage.setModSeq(modSeqAndUid.getT1());
+ mailboxMessage.setUid(modSeqAndUid.getT2());
+
+ return messageIdMapper.copyInMailboxReactive(mailboxMessage, mailbox)
+ .thenReturn(mailboxMessage.metaData());
+ });
}
private ThrowingFunction<MailboxMessage, MessageResult> messageResultConverter(FetchGroup fetchGroup) {
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 01/11: [REFACTORING] CassandraMessageIdMapper:
Remove a useless Mono.defer 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 87ebd46f5c31bfd13208ad7ea37ea57086de7d26
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Thu May 13 15:15:58 2021 +0700
[REFACTORING] CassandraMessageIdMapper: Remove a useless Mono.defer call
Analise shows that the pipeline generated would always be the same.
Removing it do not prevents the mailbox/cassandra to pass.
Aggressive Mono.defer calls yield a little runtime overhead.
---
.../apache/james/mailbox/cassandra/mail/CassandraMessageIdMapper.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
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 e5db7eaf..f724466 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
@@ -263,7 +263,7 @@ public class CassandraMessageIdMapper implements MessageIdMapper {
}
private Flux<Pair<MailboxId, UpdatedFlags>> flagsUpdateWithRetry(Flags newState, MessageManager.FlagsUpdateMode updateMode, MailboxId mailboxId, MessageId messageId) {
- return Mono.defer(() -> updateFlags(mailboxId, messageId, newState, updateMode))
+ return updateFlags(mailboxId, messageId, newState, updateMode)
.retry(cassandraConfiguration.getFlagsUpdateMessageIdMaxRetry())
.onErrorResume(MailboxDeleteDuringUpdateException.class, e -> {
LOGGER.info("Mailbox {} was deleted during flag update", mailboxId);
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 10/11: [REFACTORING] MailboxChangeListener was
performing some blocking calls...
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 91595e7ecc19d9fe5a121c5c8c3963bd4414ecd9
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Thu May 13 17:37:05 2021 +0700
[REFACTORING] MailboxChangeListener was performing some blocking calls...
---
.../james/mailbox/store/StoreRightManager.java | 1 +
.../james/jmap/api/change/MailboxChange.java | 6 +-
.../james/jmap/change/MailboxChangeListener.scala | 85 ++++++++++------------
3 files changed, 44 insertions(+), 48 deletions(-)
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 43578ea..7274f0c 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
@@ -151,6 +151,7 @@ public class StoreRightManager implements RightManager {
return mailbox.getACL();
}
+ @Override
public MailboxACL listRights(MailboxId mailboxId, MailboxSession session) throws MailboxException {
MailboxMapper mapper = mailboxSessionMapperFactory.getMailboxMapper(session);
Mailbox mailbox = blockOptional(mapper.findMailboxById(mailboxId))
diff --git a/server/data/data-jmap/src/main/java/org/apache/james/jmap/api/change/MailboxChange.java b/server/data/data-jmap/src/main/java/org/apache/james/jmap/api/change/MailboxChange.java
index c1deef5..3da4de7 100644
--- a/server/data/data-jmap/src/main/java/org/apache/james/jmap/api/change/MailboxChange.java
+++ b/server/data/data-jmap/src/main/java/org/apache/james/jmap/api/change/MailboxChange.java
@@ -130,14 +130,14 @@ public class MailboxChange implements JmapChange {
this.stateFactory = stateFactory;
}
- public List<JmapChange> fromMailboxAdded(MailboxAdded mailboxAdded, ZonedDateTime now) {
- return ImmutableList.of(MailboxChange.builder()
+ public JmapChange fromMailboxAdded(MailboxAdded mailboxAdded, ZonedDateTime now) {
+ return MailboxChange.builder()
.accountId(AccountId.fromUsername(mailboxAdded.getUsername()))
.state(stateFactory.generate())
.date(now)
.isCountChange(false)
.created(ImmutableList.of(mailboxAdded.getMailboxId()))
- .build());
+ .build();
}
public List<JmapChange> fromMailboxRenamed(MailboxRenamed mailboxRenamed, ZonedDateTime now, List<AccountId> sharees) {
diff --git a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/change/MailboxChangeListener.scala b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/change/MailboxChangeListener.scala
index 84337db..b4c6aa9 100644
--- a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/change/MailboxChangeListener.scala
+++ b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/change/MailboxChangeListener.scala
@@ -31,10 +31,9 @@ import org.apache.james.jmap.api.change.{EmailChange, EmailChangeRepository, Jma
import org.apache.james.jmap.api.model.AccountId
import org.apache.james.jmap.change.MailboxChangeListener.LOGGER
import org.apache.james.jmap.core.UuidState
+import org.apache.james.mailbox.MailboxManager
import org.apache.james.mailbox.events.MailboxEvents.{Added, Expunged, FlagsUpdated, MailboxACLUpdated, MailboxAdded, MailboxDeletion, MailboxEvent, MailboxRenamed}
-import org.apache.james.mailbox.exception.MailboxException
import org.apache.james.mailbox.model.{MailboxACL, MailboxId}
-import org.apache.james.mailbox.{MailboxManager, MailboxSession}
import org.apache.james.util.ReactorUtils.DEFAULT_CONCURRENCY
import org.reactivestreams.Publisher
import org.slf4j.{Logger, LoggerFactory}
@@ -57,44 +56,43 @@ case class MailboxChangeListener @Inject() (@Named(InjectionKeys.JMAP) eventBus:
clock: Clock) extends ReactiveGroupEventListener {
override def reactiveEvent(event: Event): Publisher[Void] =
- handleEvent(event.asInstanceOf[MailboxEvent])
- .`then`(SMono.empty[Void])
- .asJava
+ jmapChanges(event.asInstanceOf[MailboxEvent])
+ .flatMap(saveChangeEvent, DEFAULT_CONCURRENCY)
+ .`then`()
override def getDefaultGroup: Group = MailboxChangeListenerGroup()
override def isHandling(event: Event): Boolean = event.isInstanceOf[MailboxEvent]
- private def handleEvent(mailboxEvent: MailboxEvent): SMono[Unit] = {
+ private def jmapChanges(mailboxEvent: MailboxEvent): SFlux[JmapChange] = {
val now: ZonedDateTime = ZonedDateTime.now(clock)
val mailboxId: MailboxId = mailboxEvent.getMailboxId
val username: Username = mailboxEvent.getUsername
- SFlux.fromIterable(
- mailboxEvent match {
- case mailboxAdded: MailboxAdded =>
- mailboxChangeFactory.fromMailboxAdded(mailboxAdded, now).asScala
- case mailboxRenamed: MailboxRenamed =>
- mailboxChangeFactory.fromMailboxRenamed(mailboxRenamed, now, getSharees(mailboxId, username).asJava).asScala
- case mailboxACLUpdated: MailboxACLUpdated =>
- mailboxChangeFactory.fromMailboxACLUpdated(mailboxACLUpdated, now, getSharees(mailboxId, username).asJava).asScala
- case mailboxDeletion: MailboxDeletion =>
- mailboxChangeFactory.fromMailboxDeletion(mailboxDeletion, now).asScala
- case added: Added =>
- val sharees = getSharees(mailboxId, username).asJava
- mailboxChangeFactory.fromAdded(added, now, sharees).asScala
- .concat(emailChangeFactory.fromAdded(added, now, sharees).asScala)
- case flagsUpdated: FlagsUpdated =>
- val sharees = getSharees(mailboxId, username).asJava
- mailboxChangeFactory.fromFlagsUpdated(flagsUpdated, now, sharees).asScala
- .concat(emailChangeFactory.fromFlagsUpdated(flagsUpdated, now, sharees).asScala)
- case expunged: Expunged =>
- val sharees = getSharees(mailboxId, username)
- mailboxChangeFactory.fromExpunged(expunged, now, sharees.asJava).asScala
- .concat(emailChangeFactory.fromExpunged(expunged, now, sharees.map(_.getIdentifier).map(Username.of).asJava).asScala)
- })
- .flatMap(saveChangeEvent, DEFAULT_CONCURRENCY)
- .`then`()
+ mailboxEvent match {
+ case mailboxAdded: MailboxAdded =>
+ SFlux.just(mailboxChangeFactory.fromMailboxAdded(mailboxAdded, now))
+ case mailboxRenamed: MailboxRenamed =>
+ getSharees(mailboxId, username)
+ .flatMapIterable(sharees => mailboxChangeFactory.fromMailboxRenamed(mailboxRenamed, now, sharees.asJava).asScala)
+ case mailboxACLUpdated: MailboxACLUpdated =>
+ getSharees(mailboxId, username)
+ .flatMapIterable(sharees => mailboxChangeFactory.fromMailboxACLUpdated(mailboxACLUpdated, now, sharees.asJava).asScala)
+ case mailboxDeletion: MailboxDeletion =>
+ SFlux.fromIterable(mailboxChangeFactory.fromMailboxDeletion(mailboxDeletion, now).asScala)
+ case added: Added =>
+ getSharees(mailboxId, username)
+ .flatMapIterable(sharees => mailboxChangeFactory.fromAdded(added, now, sharees.asJava).asScala
+ .concat(emailChangeFactory.fromAdded(added, now, sharees.asJava).asScala))
+ case flagsUpdated: FlagsUpdated =>
+ getSharees(mailboxId, username)
+ .flatMapIterable(sharees => mailboxChangeFactory.fromFlagsUpdated(flagsUpdated, now, sharees.asJava).asScala
+ .concat(emailChangeFactory.fromFlagsUpdated(flagsUpdated, now, sharees.asJava).asScala))
+ case expunged: Expunged =>
+ getSharees(mailboxId, username)
+ .flatMapIterable(sharees => mailboxChangeFactory.fromExpunged(expunged, now, sharees.asJava).asScala
+ .concat(emailChangeFactory.fromExpunged(expunged, now, sharees.map(_.getIdentifier).map(Username.of).asJava).asScala))
+ }
}
private def saveChangeEvent(jmapChange: JmapChange): Publisher[Void] =
@@ -103,23 +101,21 @@ case class MailboxChangeListener @Inject() (@Named(InjectionKeys.JMAP) eventBus:
case emailChange: EmailChange => emailChangeRepository.save(emailChange)
}).`then`(SMono(eventBus.dispatch(toStateChangeEvent(jmapChange), AccountIdRegistrationKey(jmapChange.getAccountId))))
-
- private def getSharees(mailboxId: MailboxId, username: Username): List[AccountId] = {
- val mailboxSession: MailboxSession = mailboxManager.createSystemSession(username)
- try {
- val mailboxACL = mailboxManager.listRights(mailboxId, mailboxSession)
- mailboxACL.getEntries.keySet
+ private def getSharees(mailboxId: MailboxId, username: Username): SMono[List[AccountId]] = {
+ val session = mailboxManager.createSystemSession(username)
+ SMono(mailboxManager.getMailboxReactive(mailboxId, session))
+ .map(mailbox => mailbox.getResolvedAcl(session))
+ .map(mailboxACL => mailboxACL.getEntries.keySet
.asScala
.filter(!_.isNegative)
.filter(_.getNameType == MailboxACL.NameType.user)
.map(_.getName)
.map(AccountId.fromString)
- .toList
- } catch {
- case e: MailboxException =>
- LOGGER.warn("Could not get sharees for mailbox [%s] when listening to change events", mailboxId)
- List.empty
- }
+ .toList)
+ .onErrorResume(e => {
+ LOGGER.warn("Could not get sharees for mailbox [%s] when listening to change events", mailboxId, e)
+ SMono.just(List.empty)
+ })
}
private def toStateChangeEvent(jmapChange: JmapChange): StateChangeEvent = jmapChange match {
@@ -130,8 +126,7 @@ case class MailboxChangeListener @Inject() (@Named(InjectionKeys.JMAP) eventBus:
Some(UuidState.fromJava(emailChange.getState))
.filter(_ => !emailChange.getCreated.isEmpty)
.map(emailDeliveryState => Map(EmailDeliveryTypeName -> emailDeliveryState))
- .getOrElse(Map())).toMap
- )
+ .getOrElse(Map())).toMap)
case mailboxChange: MailboxChange => StateChangeEvent(
eventId = EventId.random(),
username = Username.of(mailboxChange.getAccountId.getIdentifier),
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 09/11: [REFACTORING] Hide CassandraACLMapper behind
an interface
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 9e7ad28ca351c7dbdedb61dddc8d1c78f9a48592
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Thu May 13 16:47:42 2021 +0700
[REFACTORING] Hide CassandraACLMapper behind an interface
This enables tailor made James server for non collaborative
environments to specify noop implementations for the ACLMapper.
For my workloads at Linagora ACL read requests proved to be one of my
top queries even if the backing table is empty.
---
.../CassandraMailboxSessionMapperFactory.java | 6 ++--
.../mailbox/cassandra/DeleteMessageListener.java | 6 ++--
.../james/mailbox/cassandra/mail/ACLMapper.java | 36 ++++++++++++++++++++++
.../mailbox/cassandra/mail/CassandraACLMapper.java | 6 +++-
.../cassandra/mail/CassandraMailboxMapper.java | 14 ++++-----
.../mail/task/MailboxMergingTaskRunner.java | 14 ++++-----
.../mailbox/cassandra/mail/utils/GuiceUtils.java | 3 ++
.../modules/mailbox/CassandraMailboxModule.java | 2 ++
8 files changed, 66 insertions(+), 21 deletions(-)
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/CassandraMailboxSessionMapperFactory.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/CassandraMailboxSessionMapperFactory.java
index 4194924..5a809f2 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/CassandraMailboxSessionMapperFactory.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/CassandraMailboxSessionMapperFactory.java
@@ -26,7 +26,7 @@ import org.apache.james.backends.cassandra.utils.CassandraUtils;
import org.apache.james.backends.cassandra.versions.CassandraSchemaVersionManager;
import org.apache.james.blob.api.BlobStore;
import org.apache.james.mailbox.MailboxSession;
-import org.apache.james.mailbox.cassandra.mail.CassandraACLMapper;
+import org.apache.james.mailbox.cassandra.mail.ACLMapper;
import org.apache.james.mailbox.cassandra.mail.CassandraAnnotationMapper;
import org.apache.james.mailbox.cassandra.mail.CassandraApplicableFlagDAO;
import org.apache.james.mailbox.cassandra.mail.CassandraAttachmentDAOV2;
@@ -92,7 +92,7 @@ public class CassandraMailboxSessionMapperFactory extends MailboxSessionMapperFa
private final BlobStore blobStore;
private final CassandraAttachmentMessageIdDAO attachmentMessageIdDAO;
private final CassandraAttachmentOwnerDAO ownerDAO;
- private final CassandraACLMapper aclMapper;
+ private final ACLMapper aclMapper;
private final CassandraUserMailboxRightsDAO userMailboxRightsDAO;
private final CassandraSchemaVersionManager versionManager;
private final CassandraUtils cassandraUtils;
@@ -107,7 +107,7 @@ public class CassandraMailboxSessionMapperFactory extends MailboxSessionMapperFa
CassandraMailboxPathDAOImpl mailboxPathDAO, CassandraMailboxPathV2DAO mailboxPathV2DAO, CassandraMailboxPathV3DAO mailboxPathV3DAO, CassandraFirstUnseenDAO firstUnseenDAO, CassandraApplicableFlagDAO applicableFlagDAO,
CassandraAttachmentDAOV2 attachmentDAOV2, CassandraDeletedMessageDAO deletedMessageDAO,
BlobStore blobStore, CassandraAttachmentMessageIdDAO attachmentMessageIdDAO,
- CassandraAttachmentOwnerDAO ownerDAO, CassandraACLMapper aclMapper,
+ CassandraAttachmentOwnerDAO ownerDAO, ACLMapper aclMapper,
CassandraUserMailboxRightsDAO userMailboxRightsDAO,
CassandraSchemaVersionManager versionManager,
RecomputeMailboxCountersService recomputeMailboxCountersService,
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/DeleteMessageListener.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/DeleteMessageListener.java
index 25af4c3..bcde273 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/DeleteMessageListener.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/DeleteMessageListener.java
@@ -38,7 +38,7 @@ import org.apache.james.events.Group;
import org.apache.james.mailbox.acl.ACLDiff;
import org.apache.james.mailbox.cassandra.ids.CassandraId;
import org.apache.james.mailbox.cassandra.ids.CassandraMessageId;
-import org.apache.james.mailbox.cassandra.mail.CassandraACLMapper;
+import org.apache.james.mailbox.cassandra.mail.ACLMapper;
import org.apache.james.mailbox.cassandra.mail.CassandraApplicableFlagDAO;
import org.apache.james.mailbox.cassandra.mail.CassandraAttachmentDAOV2;
import org.apache.james.mailbox.cassandra.mail.CassandraAttachmentMessageIdDAO;
@@ -92,7 +92,7 @@ public class DeleteMessageListener implements EventListener.ReactiveGroupEventLi
private final CassandraAttachmentDAOV2 attachmentDAO;
private final CassandraAttachmentOwnerDAO ownerDAO;
private final CassandraAttachmentMessageIdDAO attachmentMessageIdDAO;
- private final CassandraACLMapper aclMapper;
+ private final ACLMapper aclMapper;
private final CassandraUserMailboxRightsDAO rightsDAO;
private final CassandraApplicableFlagDAO applicableFlagDAO;
private final CassandraFirstUnseenDAO firstUnseenDAO;
@@ -105,7 +105,7 @@ public class DeleteMessageListener implements EventListener.ReactiveGroupEventLi
@Inject
public DeleteMessageListener(CassandraMessageIdToImapUidDAO imapUidDAO, CassandraMessageIdDAO messageIdDAO, CassandraMessageDAO messageDAO,
CassandraMessageDAOV3 messageDAOV3, CassandraAttachmentDAOV2 attachmentDAO, CassandraAttachmentOwnerDAO ownerDAO,
- CassandraAttachmentMessageIdDAO attachmentMessageIdDAO, CassandraACLMapper aclMapper,
+ CassandraAttachmentMessageIdDAO attachmentMessageIdDAO, ACLMapper aclMapper,
CassandraUserMailboxRightsDAO rightsDAO, CassandraApplicableFlagDAO applicableFlagDAO,
CassandraFirstUnseenDAO firstUnseenDAO, CassandraDeletedMessageDAO deletedMessageDAO,
CassandraMailboxCounterDAO counterDAO, CassandraMailboxRecentsDAO recentsDAO, BlobStore blobStore,
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/ACLMapper.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/ACLMapper.java
new file mode 100644
index 0000000..aaa9290
--- /dev/null
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/ACLMapper.java
@@ -0,0 +1,36 @@
+/****************************************************************
+ * 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.mailbox.cassandra.mail;
+
+import org.apache.james.mailbox.acl.ACLDiff;
+import org.apache.james.mailbox.cassandra.ids.CassandraId;
+import org.apache.james.mailbox.model.MailboxACL;
+
+import reactor.core.publisher.Mono;
+
+public interface ACLMapper {
+ Mono<MailboxACL> getACL(CassandraId cassandraId);
+
+ Mono<ACLDiff> updateACL(CassandraId cassandraId, MailboxACL.ACLCommand command);
+
+ Mono<ACLDiff> setACL(CassandraId cassandraId, MailboxACL mailboxACL);
+
+ Mono<Void> delete(CassandraId cassandraId);
+}
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraACLMapper.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraACLMapper.java
index 97012c3..da83a9a 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraACLMapper.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraACLMapper.java
@@ -46,7 +46,7 @@ import com.google.common.collect.ImmutableSet;
import reactor.core.publisher.Mono;
-public class CassandraACLMapper {
+public class CassandraACLMapper implements ACLMapper {
public interface Store {
Mono<MailboxACL> getACL(CassandraId cassandraId);
@@ -168,18 +168,22 @@ public class CassandraACLMapper {
});
}
+ @Override
public Mono<MailboxACL> getACL(CassandraId cassandraId) {
return store().flatMap(store -> store.getACL(cassandraId));
}
+ @Override
public Mono<ACLDiff> updateACL(CassandraId cassandraId, MailboxACL.ACLCommand command) {
return store().flatMap(store -> store.updateACL(cassandraId, command));
}
+ @Override
public Mono<ACLDiff> setACL(CassandraId cassandraId, MailboxACL mailboxACL) {
return store().flatMap(store -> store.setACL(cassandraId, mailboxACL));
}
+ @Override
public Mono<Void> delete(CassandraId cassandraId) {
return store().flatMap(store -> store.delete(cassandraId));
}
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMailboxMapper.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMailboxMapper.java
index 17fcae8..3454619 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMailboxMapper.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMailboxMapper.java
@@ -70,7 +70,7 @@ public class CassandraMailboxMapper implements MailboxMapper {
private final CassandraMailboxPathDAOImpl mailboxPathDAO;
private final CassandraMailboxPathV2DAO mailboxPathV2DAO;
private final CassandraMailboxPathV3DAO mailboxPathV3DAO;
- private final CassandraACLMapper cassandraACLMapper;
+ private final ACLMapper aclMapper;
private final CassandraUserMailboxRightsDAO userMailboxRightsDAO;
private final CassandraSchemaVersionManager versionManager;
private final CassandraConfiguration cassandraConfiguration;
@@ -82,7 +82,7 @@ public class CassandraMailboxMapper implements MailboxMapper {
CassandraMailboxPathV2DAO mailboxPathV2DAO,
CassandraMailboxPathV3DAO mailboxPathV3DAO,
CassandraUserMailboxRightsDAO userMailboxRightsDAO,
- CassandraACLMapper aclMapper,
+ ACLMapper aclMapper,
CassandraSchemaVersionManager versionManager,
CassandraConfiguration cassandraConfiguration) {
this.mailboxDAO = mailboxDAO;
@@ -90,7 +90,7 @@ public class CassandraMailboxMapper implements MailboxMapper {
this.mailboxPathV2DAO = mailboxPathV2DAO;
this.mailboxPathV3DAO = mailboxPathV3DAO;
this.userMailboxRightsDAO = userMailboxRightsDAO;
- this.cassandraACLMapper = aclMapper;
+ this.aclMapper = aclMapper;
this.versionManager = versionManager;
this.cassandraConfiguration = cassandraConfiguration;
this.secureRandom = new SecureRandom();
@@ -182,7 +182,7 @@ public class CassandraMailboxMapper implements MailboxMapper {
private Mono<Mailbox> addAcl(Mailbox mailbox) {
CassandraId mailboxId = (CassandraId) mailbox.getMailboxId();
- return cassandraACLMapper.getACL(mailboxId)
+ return aclMapper.getACL(mailboxId)
.map(acl -> {
mailbox.setACL(acl);
return mailbox;
@@ -236,7 +236,7 @@ public class CassandraMailboxMapper implements MailboxMapper {
}
private Mono<MailboxACL> retrieveAcl(CassandraId mailboxId) {
- return cassandraACLMapper.getACL(mailboxId)
+ return aclMapper.getACL(mailboxId)
.defaultIfEmpty(MailboxACL.EMPTY);
}
@@ -341,13 +341,13 @@ public class CassandraMailboxMapper implements MailboxMapper {
@Override
public Mono<ACLDiff> updateACL(Mailbox mailbox, MailboxACL.ACLCommand mailboxACLCommand) {
CassandraId cassandraId = (CassandraId) mailbox.getMailboxId();
- return cassandraACLMapper.updateACL(cassandraId, mailboxACLCommand);
+ return aclMapper.updateACL(cassandraId, mailboxACLCommand);
}
@Override
public Mono<ACLDiff> setACL(Mailbox mailbox, MailboxACL mailboxACL) {
CassandraId cassandraId = (CassandraId) mailbox.getMailboxId();
- return cassandraACLMapper.setACL(cassandraId, mailboxACL);
+ return aclMapper.setACL(cassandraId, mailboxACL);
}
private Mono<Mailbox> toMailboxWithAcl(Mailbox mailbox) {
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/task/MailboxMergingTaskRunner.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/task/MailboxMergingTaskRunner.java
index 4231d50..f7be117 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/task/MailboxMergingTaskRunner.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/task/MailboxMergingTaskRunner.java
@@ -25,7 +25,7 @@ import org.apache.james.core.Username;
import org.apache.james.mailbox.MailboxManager;
import org.apache.james.mailbox.MailboxSession;
import org.apache.james.mailbox.cassandra.ids.CassandraId;
-import org.apache.james.mailbox.cassandra.mail.CassandraACLMapper;
+import org.apache.james.mailbox.cassandra.mail.ACLMapper;
import org.apache.james.mailbox.cassandra.mail.CassandraMailboxDAO;
import org.apache.james.mailbox.cassandra.mail.CassandraMessageIdDAO;
import org.apache.james.mailbox.exception.MailboxException;
@@ -50,16 +50,16 @@ public class MailboxMergingTaskRunner {
private final StoreMessageIdManager messageIdManager;
private final CassandraMessageIdDAO cassandraMessageIdDAO;
private final CassandraMailboxDAO mailboxDAO;
- private final CassandraACLMapper cassandraACLMapper;
+ private final ACLMapper aclMapper;
private final MailboxSession mailboxSession;
@Inject
- public MailboxMergingTaskRunner(MailboxManager mailboxManager, StoreMessageIdManager messageIdManager, CassandraMessageIdDAO cassandraMessageIdDAO, CassandraMailboxDAO mailboxDAO, CassandraACLMapper cassandraACLMapper) {
+ public MailboxMergingTaskRunner(MailboxManager mailboxManager, StoreMessageIdManager messageIdManager, CassandraMessageIdDAO cassandraMessageIdDAO, CassandraMailboxDAO mailboxDAO, ACLMapper aclMapper) {
this.mailboxSession = mailboxManager.createSystemSession(Username.of("task"));
this.messageIdManager = messageIdManager;
this.cassandraMessageIdDAO = cassandraMessageIdDAO;
this.mailboxDAO = mailboxDAO;
- this.cassandraACLMapper = cassandraACLMapper;
+ this.aclMapper = aclMapper;
}
public Task.Result run(CassandraId oldMailboxId, CassandraId newMailboxId, MailboxMergingTask.Context context) {
@@ -91,10 +91,10 @@ public class MailboxMergingTaskRunner {
private Mono<Void> mergeRights(CassandraId oldMailboxId, CassandraId newMailboxId) {
return Flux.concat(
- cassandraACLMapper.getACL(oldMailboxId),
- cassandraACLMapper.getACL(newMailboxId))
+ aclMapper.getACL(oldMailboxId),
+ aclMapper.getACL(newMailboxId))
.reduce(Throwing.biFunction(MailboxACL::union))
- .flatMap(union -> cassandraACLMapper.setACL(newMailboxId, union))
+ .flatMap(union -> aclMapper.setACL(newMailboxId, union))
.then();
}
}
diff --git a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/utils/GuiceUtils.java b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/utils/GuiceUtils.java
index c1c9ba1..eb3f20b 100644
--- a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/utils/GuiceUtils.java
+++ b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/utils/GuiceUtils.java
@@ -38,6 +38,8 @@ import org.apache.james.eventsourcing.eventstore.cassandra.dto.EventDTOModule;
import org.apache.james.json.DTO;
import org.apache.james.json.DTOModule;
import org.apache.james.mailbox.cassandra.ids.CassandraMessageId;
+import org.apache.james.mailbox.cassandra.mail.ACLMapper;
+import org.apache.james.mailbox.cassandra.mail.CassandraACLMapper;
import org.apache.james.mailbox.cassandra.mail.CassandraUidProvider;
import org.apache.james.mailbox.cassandra.mail.eventsourcing.acl.ACLModule;
import org.apache.james.mailbox.model.MessageId;
@@ -76,6 +78,7 @@ public class GuiceUtils {
return Modules.combine(
binder -> binder.bind(MessageId.Factory.class).toInstance(messageIdFactory),
binder -> binder.bind(UidProvider.class).to(CassandraUidProvider.class),
+ binder -> binder.bind(ACLMapper.class).to(CassandraACLMapper.class),
binder -> binder.bind(BlobId.Factory.class).toInstance(new HashBlobId.Factory()),
binder -> binder.bind(BlobStore.class).toProvider(() -> CassandraBlobStoreFactory.forTesting(session).passthrough()),
binder -> binder.bind(Session.class).toInstance(session),
diff --git a/server/container/guice/cassandra/src/main/java/org/apache/james/modules/mailbox/CassandraMailboxModule.java b/server/container/guice/cassandra/src/main/java/org/apache/james/modules/mailbox/CassandraMailboxModule.java
index 01e47ca..8a83afe 100644
--- a/server/container/guice/cassandra/src/main/java/org/apache/james/modules/mailbox/CassandraMailboxModule.java
+++ b/server/container/guice/cassandra/src/main/java/org/apache/james/modules/mailbox/CassandraMailboxModule.java
@@ -54,6 +54,7 @@ import org.apache.james.mailbox.cassandra.CassandraMailboxSessionMapperFactory;
import org.apache.james.mailbox.cassandra.DeleteMessageListener;
import org.apache.james.mailbox.cassandra.ids.CassandraId;
import org.apache.james.mailbox.cassandra.ids.CassandraMessageId;
+import org.apache.james.mailbox.cassandra.mail.ACLMapper;
import org.apache.james.mailbox.cassandra.mail.CassandraACLDAOV1;
import org.apache.james.mailbox.cassandra.mail.CassandraACLDAOV2;
import org.apache.james.mailbox.cassandra.mail.CassandraACLMapper;
@@ -181,6 +182,7 @@ public class CassandraMailboxModule extends AbstractModule {
bind(MailboxSessionMapperFactory.class).to(CassandraMailboxSessionMapperFactory.class);
bind(SubscriptionMapperFactory.class).to(CassandraMailboxSessionMapperFactory.class);
+ bind(ACLMapper.class).to(CassandraACLMapper.class);
bind(ModSeqProvider.class).to(CassandraModSeqProvider.class);
bind(UidProvider.class).to(CassandraUidProvider.class);
bind(SubscriptionManager.class).to(StoreSubscriptionManager.class);
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 11/11: [REFACTORING]
CassandraMailboxSessionMapperFactory should not instantiate one mapper per
request
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 14b341ead35bdc1f024c944586be36f09ba03c43
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Thu May 13 22:55:24 2021 +0700
[REFACTORING] CassandraMailboxSessionMapperFactory should not instantiate one mapper per request
Some resources like SecureRandom can be expensive to instantiate
and result in CPU resources waste. This had been identified by
profiling and generating a flame graph with async_profiler.
---
.../CassandraMailboxSessionMapperFactory.java | 86 ++++++++++------------
1 file changed, 37 insertions(+), 49 deletions(-)
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/CassandraMailboxSessionMapperFactory.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/CassandraMailboxSessionMapperFactory.java
index 5a809f2..327aeca 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/CassandraMailboxSessionMapperFactory.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/CassandraMailboxSessionMapperFactory.java
@@ -69,9 +69,6 @@ import com.datastax.driver.core.Session;
* Cassandra implementation of {@link MailboxSessionMapperFactory}
*/
public class CassandraMailboxSessionMapperFactory extends MailboxSessionMapperFactory implements AttachmentMapperFactory {
- protected static final String ATTACHMENTMAPPER = "ATTACHMENTMAPPER";
-
- private final Session session;
private final UidProvider uidProvider;
private final ModSeqProvider modSeqProvider;
private final CassandraMessageDAO messageDAO;
@@ -80,11 +77,6 @@ public class CassandraMailboxSessionMapperFactory extends MailboxSessionMapperFa
private final CassandraMessageIdToImapUidDAO imapUidDAO;
private final CassandraMailboxCounterDAO mailboxCounterDAO;
private final CassandraMailboxRecentsDAO mailboxRecentsDAO;
- private final CassandraIndexTableHandler indexTableHandler;
- private final CassandraMailboxDAO mailboxDAO;
- private final CassandraMailboxPathDAOImpl mailboxPathDAO;
- private final CassandraMailboxPathV2DAO mailboxPathV2DAO;
- private final CassandraMailboxPathV3DAO mailboxPathV3DAO;
private final CassandraFirstUnseenDAO firstUnseenDAO;
private final CassandraApplicableFlagDAO applicableFlagDAO;
private final CassandraAttachmentDAOV2 attachmentDAOV2;
@@ -94,10 +86,13 @@ public class CassandraMailboxSessionMapperFactory extends MailboxSessionMapperFa
private final CassandraAttachmentOwnerDAO ownerDAO;
private final ACLMapper aclMapper;
private final CassandraUserMailboxRightsDAO userMailboxRightsDAO;
- private final CassandraSchemaVersionManager versionManager;
- private final CassandraUtils cassandraUtils;
- private final RecomputeMailboxCountersService recomputeMailboxCountersService;
private final CassandraConfiguration cassandraConfiguration;
+ private final CassandraMailboxMapper cassandraMailboxMapper;
+ private final CassandraSubscriptionMapper cassandraSubscriptionMapper;
+ private final CassandraAttachmentMapper cassandraAttachmentMapper;
+ private final CassandraMessageMapper cassandraMessageMapper;
+ private final CassandraMessageIdMapper cassandraMessageIdMapper;
+ private final CassandraAnnotationMapper cassandraAnnotationMapper;
@Inject
public CassandraMailboxSessionMapperFactory(UidProvider uidProvider, CassandraModSeqProvider modSeqProvider, Session session,
@@ -114,17 +109,12 @@ public class CassandraMailboxSessionMapperFactory extends MailboxSessionMapperFa
CassandraUtils cassandraUtils, CassandraConfiguration cassandraConfiguration) {
this.uidProvider = uidProvider;
this.modSeqProvider = modSeqProvider;
- this.session = session;
this.messageDAO = messageDAO;
this.messageDAOV3 = messageDAOV3;
this.messageIdDAO = messageIdDAO;
this.imapUidDAO = imapUidDAO;
this.mailboxCounterDAO = mailboxCounterDAO;
this.mailboxRecentsDAO = mailboxRecentsDAO;
- this.mailboxDAO = mailboxDAO;
- this.mailboxPathDAO = mailboxPathDAO;
- this.mailboxPathV2DAO = mailboxPathV2DAO;
- this.mailboxPathV3DAO = mailboxPathV3DAO;
this.firstUnseenDAO = firstUnseenDAO;
this.attachmentDAOV2 = attachmentDAOV2;
this.deletedMessageDAO = deletedMessageDAO;
@@ -133,59 +123,62 @@ public class CassandraMailboxSessionMapperFactory extends MailboxSessionMapperFa
this.attachmentMessageIdDAO = attachmentMessageIdDAO;
this.aclMapper = aclMapper;
this.userMailboxRightsDAO = userMailboxRightsDAO;
- this.versionManager = versionManager;
- this.cassandraUtils = cassandraUtils;
this.ownerDAO = ownerDAO;
- this.recomputeMailboxCountersService = recomputeMailboxCountersService;
this.cassandraConfiguration = cassandraConfiguration;
- this.indexTableHandler = new CassandraIndexTableHandler(
+ CassandraIndexTableHandler indexTableHandler = new CassandraIndexTableHandler(
mailboxRecentsDAO,
mailboxCounterDAO,
firstUnseenDAO,
applicableFlagDAO,
deletedMessageDAO);
+ this.cassandraMailboxMapper = new CassandraMailboxMapper(mailboxDAO, mailboxPathDAO, mailboxPathV2DAO, mailboxPathV3DAO, userMailboxRightsDAO, aclMapper, versionManager, cassandraConfiguration);
+ this.cassandraSubscriptionMapper = new CassandraSubscriptionMapper(session, cassandraUtils);
+ this.cassandraAttachmentMapper = new CassandraAttachmentMapper(attachmentDAOV2, blobStore, attachmentMessageIdDAO, ownerDAO);
+ this.cassandraMessageMapper = new CassandraMessageMapper(
+ uidProvider,
+ modSeqProvider,
+ cassandraAttachmentMapper,
+ messageDAO,
+ messageDAOV3,
+ messageIdDAO,
+ imapUidDAO,
+ mailboxCounterDAO,
+ mailboxRecentsDAO,
+ applicableFlagDAO,
+ indexTableHandler,
+ firstUnseenDAO,
+ deletedMessageDAO,
+ cassandraConfiguration, recomputeMailboxCountersService);
+ this.cassandraMessageIdMapper = new CassandraMessageIdMapper(cassandraMailboxMapper, mailboxDAO,
+ cassandraAttachmentMapper,
+ imapUidDAO, messageIdDAO, messageDAO, messageDAOV3, indexTableHandler, modSeqProvider,
+ cassandraConfiguration);
+ this.cassandraAnnotationMapper = new CassandraAnnotationMapper(session, cassandraUtils);
}
@Override
public CassandraMessageMapper createMessageMapper(MailboxSession mailboxSession) {
- return new CassandraMessageMapper(
- uidProvider,
- modSeqProvider,
- createAttachmentMapper(mailboxSession),
- messageDAO,
- messageDAOV3,
- messageIdDAO,
- imapUidDAO,
- mailboxCounterDAO,
- mailboxRecentsDAO,
- applicableFlagDAO,
- indexTableHandler,
- firstUnseenDAO,
- deletedMessageDAO,
- cassandraConfiguration, recomputeMailboxCountersService);
+ return cassandraMessageMapper;
}
@Override
public MessageIdMapper createMessageIdMapper(MailboxSession mailboxSession) {
- return new CassandraMessageIdMapper(getMailboxMapper(mailboxSession), mailboxDAO,
- createAttachmentMapper(mailboxSession),
- imapUidDAO, messageIdDAO, messageDAO, messageDAOV3, indexTableHandler, modSeqProvider,
- cassandraConfiguration);
+ return cassandraMessageIdMapper;
}
@Override
public MailboxMapper createMailboxMapper(MailboxSession mailboxSession) {
- return new CassandraMailboxMapper(mailboxDAO, mailboxPathDAO, mailboxPathV2DAO, mailboxPathV3DAO, userMailboxRightsDAO, aclMapper, versionManager, cassandraConfiguration);
+ return cassandraMailboxMapper;
}
@Override
public CassandraAttachmentMapper createAttachmentMapper(MailboxSession mailboxSession) {
- return new CassandraAttachmentMapper(attachmentDAOV2, blobStore, attachmentMessageIdDAO, ownerDAO);
+ return cassandraAttachmentMapper;
}
@Override
public SubscriptionMapper createSubscriptionMapper(MailboxSession mailboxSession) {
- return new CassandraSubscriptionMapper(session, cassandraUtils);
+ return cassandraSubscriptionMapper;
}
@Override
@@ -200,17 +193,12 @@ public class CassandraMailboxSessionMapperFactory extends MailboxSessionMapperFa
@Override
public AnnotationMapper createAnnotationMapper(MailboxSession mailboxSession) {
- return new CassandraAnnotationMapper(session, cassandraUtils);
+ return cassandraAnnotationMapper;
}
@Override
public AttachmentMapper getAttachmentMapper(MailboxSession session) {
- AttachmentMapper mapper = (AttachmentMapper) session.getAttributes().get(ATTACHMENTMAPPER);
- if (mapper == null) {
- mapper = createAttachmentMapper(session);
- session.getAttributes().put(ATTACHMENTMAPPER, mapper);
- }
- return mapper;
+ return cassandraAttachmentMapper;
}
public DeleteMessageListener deleteMessageListener() {
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 02/11: [REFACTORING] Reactor: favor error suppliers
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 35e136ea9a8b27d853d098774d950e077d03b4c9
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Thu May 13 15:18:34 2021 +0700
[REFACTORING] Reactor: favor error suppliers
This prevents the exception to be evaluated upon the planning phase of the
pipeline, as executions not entering the error clause do not need to
evaluate it.
Creating exception triggers a stack capture which, if repeated frequently
can negatively impact performance.
---
.../apache/james/mailbox/cassandra/mail/CassandraACLMapper.java | 8 ++++----
.../apache/james/mailbox/maildir/mail/MaildirMailboxMapper.java | 4 ++--
.../main/java/org/apache/james/vault/DeletedMessageVaultHook.java | 2 +-
.../java/org/apache/james/mailbox/store/StoreMailboxManager.java | 2 +-
.../jmap/cassandra/access/CassandraAccessTokenRepository.java | 2 +-
.../jmap/cassandra/change/CassandraEmailChangeRepository.java | 4 ++--
.../jmap/cassandra/change/CassandraMailboxChangeRepository.java | 4 ++--
.../james/jmap/memory/change/MemoryEmailChangeRepository.java | 2 +-
.../james/jmap/memory/change/MemoryMailboxChangeRepository.java | 2 +-
.../java/org/apache/james/jmap/http/AuthenticationRoutes.java | 2 +-
.../src/main/java/org/apache/james/jmap/http/Authenticator.java | 2 +-
11 files changed, 17 insertions(+), 17 deletions(-)
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraACLMapper.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraACLMapper.java
index e91d2aa..97012c3 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraACLMapper.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraACLMapper.java
@@ -77,7 +77,7 @@ public class CassandraACLMapper {
return cassandraACLDAOV1.updateACL(cassandraId, command)
.flatMap(aclDiff -> userMailboxRightsDAO.update(cassandraId, aclDiff)
.thenReturn(aclDiff))
- .switchIfEmpty(Mono.error(new MailboxException("Unable to update ACL")));
+ .switchIfEmpty(Mono.error(() -> new MailboxException("Unable to update ACL")));
}
@Override
@@ -85,7 +85,7 @@ public class CassandraACLMapper {
return cassandraACLDAOV1.setACL(cassandraId, mailboxACL)
.flatMap(aclDiff -> userMailboxRightsDAO.update(cassandraId, aclDiff)
.thenReturn(aclDiff))
- .switchIfEmpty(Mono.defer(() -> Mono.error(new MailboxException("Unable to update ACL"))));
+ .switchIfEmpty(Mono.error(() -> new MailboxException("Unable to update ACL")));
}
public Mono<Void> delete(CassandraId cassandraId) {
@@ -125,7 +125,7 @@ public class CassandraACLMapper {
.map(ACLUpdated.class::cast)
.map(ACLUpdated::getAclDiff)
.next()
- .switchIfEmpty(Mono.defer(() -> Mono.error(new MailboxException("Unable to update ACL"))));
+ .switchIfEmpty(Mono.error(() -> new MailboxException("Unable to update ACL")));
}
@Override
@@ -136,7 +136,7 @@ public class CassandraACLMapper {
.map(ACLUpdated.class::cast)
.map(ACLUpdated::getAclDiff)
.next()
- .switchIfEmpty(Mono.defer(() -> Mono.error(new MailboxException("Unable to set ACL"))));
+ .switchIfEmpty(Mono.error(() -> new MailboxException("Unable to set ACL")));
}
@Override
diff --git a/mailbox/maildir/src/main/java/org/apache/james/mailbox/maildir/mail/MaildirMailboxMapper.java b/mailbox/maildir/src/main/java/org/apache/james/mailbox/maildir/mail/MaildirMailboxMapper.java
index edadc5c..a18c3c5 100644
--- a/mailbox/maildir/src/main/java/org/apache/james/mailbox/maildir/mail/MaildirMailboxMapper.java
+++ b/mailbox/maildir/src/main/java/org/apache/james/mailbox/maildir/mail/MaildirMailboxMapper.java
@@ -95,7 +95,7 @@ public class MaildirMailboxMapper extends NonTransactionalMapper implements Mail
}
return true;
})
- .switchIfEmpty(Mono.error(new MailboxNotFoundException(mailbox.generateAssociatedPath())))
+ .switchIfEmpty(Mono.error(() -> new MailboxNotFoundException(mailbox.generateAssociatedPath())))
.then();
}
@@ -130,7 +130,7 @@ public class MaildirMailboxMapper extends NonTransactionalMapper implements Mail
return list()
.filter(mailbox -> mailbox.getMailboxId().equals(id))
.next()
- .switchIfEmpty(Mono.error(new MailboxNotFoundException(id)));
+ .switchIfEmpty(Mono.error(() -> new MailboxNotFoundException(id)));
}
@Override
diff --git a/mailbox/plugin/deleted-messages-vault/src/main/java/org/apache/james/vault/DeletedMessageVaultHook.java b/mailbox/plugin/deleted-messages-vault/src/main/java/org/apache/james/vault/DeletedMessageVaultHook.java
index 30c03e2..7e5bbc5 100644
--- a/mailbox/plugin/deleted-messages-vault/src/main/java/org/apache/james/vault/DeletedMessageVaultHook.java
+++ b/mailbox/plugin/deleted-messages-vault/src/main/java/org/apache/james/vault/DeletedMessageVaultHook.java
@@ -123,7 +123,7 @@ public class DeletedMessageVaultHook implements PreDeletionHook {
return mapperFactory.getMessageIdMapper(session)
.findReactive(ImmutableList.of(deletedMessageMailboxContext.getMessageId()), MessageMapper.FetchType.Full)
.next()
- .switchIfEmpty(Mono.error(new RuntimeException("Cannot find " + deletedMessageMailboxContext.getMessageId())))
+ .switchIfEmpty(Mono.error(() -> new RuntimeException("Cannot find " + deletedMessageMailboxContext.getMessageId())))
.flatMap(mailboxMessage -> Mono.fromCallable(() -> Pair.of(mailboxMessage,
deletedMessageConverter.convert(deletedMessageMailboxContext, mailboxMessage,
ZonedDateTime.ofInstant(clock.instant(), ZoneOffset.UTC)))))
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 6837272..5de99f6 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
@@ -427,7 +427,7 @@ public class StoreMailboxManager implements MailboxManager {
mailboxMapper.execute(() -> block(mailboxMapper.findMailboxByPath(mailboxPath)
.flatMap(mailbox -> doDeleteMailbox(mailboxMapper, mailbox, session))
- .switchIfEmpty(Mono.error(new MailboxNotFoundException(mailboxPath)))));
+ .switchIfEmpty(Mono.error(() -> new MailboxNotFoundException(mailboxPath)))));
}
@Override
diff --git a/server/data/data-jmap-cassandra/src/main/java/org/apache/james/jmap/cassandra/access/CassandraAccessTokenRepository.java b/server/data/data-jmap-cassandra/src/main/java/org/apache/james/jmap/cassandra/access/CassandraAccessTokenRepository.java
index 81c29c6..2a19d5a 100644
--- a/server/data/data-jmap-cassandra/src/main/java/org/apache/james/jmap/cassandra/access/CassandraAccessTokenRepository.java
+++ b/server/data/data-jmap-cassandra/src/main/java/org/apache/james/jmap/cassandra/access/CassandraAccessTokenRepository.java
@@ -59,6 +59,6 @@ public class CassandraAccessTokenRepository implements AccessTokenRepository {
Preconditions.checkNotNull(accessToken);
return cassandraAccessTokenDAO.getUsernameFromToken(accessToken)
- .switchIfEmpty(Mono.error(new InvalidAccessToken(accessToken)));
+ .switchIfEmpty(Mono.error(() -> new InvalidAccessToken(accessToken)));
}
}
diff --git a/server/data/data-jmap-cassandra/src/main/java/org/apache/james/jmap/cassandra/change/CassandraEmailChangeRepository.java b/server/data/data-jmap-cassandra/src/main/java/org/apache/james/jmap/cassandra/change/CassandraEmailChangeRepository.java
index deb4eba..13d3c7c 100644
--- a/server/data/data-jmap-cassandra/src/main/java/org/apache/james/jmap/cassandra/change/CassandraEmailChangeRepository.java
+++ b/server/data/data-jmap-cassandra/src/main/java/org/apache/james/jmap/cassandra/change/CassandraEmailChangeRepository.java
@@ -67,7 +67,7 @@ public class CassandraEmailChangeRepository implements EmailChangeRepository {
}
return emailChangeRepositoryDAO.getChangesSince(accountId, state)
- .switchIfEmpty(Flux.error(new ChangeNotFoundException(state, String.format("State '%s' could not be found", state.getValue()))))
+ .switchIfEmpty(Flux.error(() -> new ChangeNotFoundException(state, String.format("State '%s' could not be found", state.getValue()))))
.filter(change -> !change.isDelegated())
.filter(change -> !change.getState().equals(state))
.collect(new EmailChanges.Builder.EmailChangeCollector(state, maxChanges.orElse(defaultLimit)));
@@ -85,7 +85,7 @@ public class CassandraEmailChangeRepository implements EmailChangeRepository {
}
return emailChangeRepositoryDAO.getChangesSince(accountId, state)
- .switchIfEmpty(Flux.error(new ChangeNotFoundException(state, String.format("State '%s' could not be found", state.getValue()))))
+ .switchIfEmpty(Flux.error(() -> new ChangeNotFoundException(state, String.format("State '%s' could not be found", state.getValue()))))
.filter(change -> !change.getState().equals(state))
.collect(new EmailChanges.Builder.EmailChangeCollector(state, maxChanges.orElse(defaultLimit)));
}
diff --git a/server/data/data-jmap-cassandra/src/main/java/org/apache/james/jmap/cassandra/change/CassandraMailboxChangeRepository.java b/server/data/data-jmap-cassandra/src/main/java/org/apache/james/jmap/cassandra/change/CassandraMailboxChangeRepository.java
index ce2b5b9..98c5046 100644
--- a/server/data/data-jmap-cassandra/src/main/java/org/apache/james/jmap/cassandra/change/CassandraMailboxChangeRepository.java
+++ b/server/data/data-jmap-cassandra/src/main/java/org/apache/james/jmap/cassandra/change/CassandraMailboxChangeRepository.java
@@ -67,7 +67,7 @@ public class CassandraMailboxChangeRepository implements MailboxChangeRepository
}
return mailboxChangeRepositoryDAO.getChangesSince(accountId, state)
- .switchIfEmpty(Flux.error(new ChangeNotFoundException(state, String.format("State '%s' could not be found", state.getValue()))))
+ .switchIfEmpty(Flux.error(() -> new ChangeNotFoundException(state, String.format("State '%s' could not be found", state.getValue()))))
.filter(change -> !change.isDelegated())
.filter(change -> !change.getState().equals(state))
.collect(new MailboxChanges.MailboxChangesBuilder.MailboxChangeCollector(state, maxChanges.orElse(defaultLimit)));
@@ -85,7 +85,7 @@ public class CassandraMailboxChangeRepository implements MailboxChangeRepository
}
return mailboxChangeRepositoryDAO.getChangesSince(accountId, state)
- .switchIfEmpty(Flux.error(new ChangeNotFoundException(state, String.format("State '%s' could not be found", state.getValue()))))
+ .switchIfEmpty(Flux.error(() -> new ChangeNotFoundException(state, String.format("State '%s' could not be found", state.getValue()))))
.filter(change -> !change.getState().equals(state))
.collect(new MailboxChanges.MailboxChangesBuilder.MailboxChangeCollector(state, maxChanges.orElse(defaultLimit)));
}
diff --git a/server/data/data-jmap/src/main/java/org/apache/james/jmap/memory/change/MemoryEmailChangeRepository.java b/server/data/data-jmap/src/main/java/org/apache/james/jmap/memory/change/MemoryEmailChangeRepository.java
index 58d595b..9636693 100644
--- a/server/data/data-jmap/src/main/java/org/apache/james/jmap/memory/change/MemoryEmailChangeRepository.java
+++ b/server/data/data-jmap/src/main/java/org/apache/james/jmap/memory/change/MemoryEmailChangeRepository.java
@@ -120,7 +120,7 @@ public class MemoryEmailChangeRepository implements EmailChangeRepository {
private Mono<EmailChange> findByState(AccountId accountId, State state) {
return Flux.fromIterable(emailChangeMap.get(accountId))
.filter(change -> change.getState().equals(state))
- .switchIfEmpty(Mono.error(new ChangeNotFoundException(state, String.format("State '%s' could not be found", state.getValue()))))
+ .switchIfEmpty(Mono.error(() -> new ChangeNotFoundException(state, String.format("State '%s' could not be found", state.getValue()))))
.single();
}
}
diff --git a/server/data/data-jmap/src/main/java/org/apache/james/jmap/memory/change/MemoryMailboxChangeRepository.java b/server/data/data-jmap/src/main/java/org/apache/james/jmap/memory/change/MemoryMailboxChangeRepository.java
index b034b9b..9198922 100644
--- a/server/data/data-jmap/src/main/java/org/apache/james/jmap/memory/change/MemoryMailboxChangeRepository.java
+++ b/server/data/data-jmap/src/main/java/org/apache/james/jmap/memory/change/MemoryMailboxChangeRepository.java
@@ -101,7 +101,7 @@ public class MemoryMailboxChangeRepository implements MailboxChangeRepository {
private Mono<MailboxChange> findByState(AccountId accountId, State state) {
return Flux.fromIterable(mailboxChangeMap.get(accountId))
.filter(change -> change.getState().equals(state))
- .switchIfEmpty(Mono.error(new ChangeNotFoundException(state, String.format("State '%s' could not be found", state.getValue()))))
+ .switchIfEmpty(Mono.error(() -> new ChangeNotFoundException(state, String.format("State '%s' could not be found", state.getValue()))))
.single();
}
diff --git a/server/protocols/jmap-draft/src/main/java/org/apache/james/jmap/http/AuthenticationRoutes.java b/server/protocols/jmap-draft/src/main/java/org/apache/james/jmap/http/AuthenticationRoutes.java
index b939cb8..0ae3ce9 100644
--- a/server/protocols/jmap-draft/src/main/java/org/apache/james/jmap/http/AuthenticationRoutes.java
+++ b/server/protocols/jmap-draft/src/main/java/org/apache/james/jmap/http/AuthenticationRoutes.java
@@ -213,7 +213,7 @@ public class AuthenticationRoutes implements JMAPRoutes {
throw new BadRequestException("Request can't be deserialized", e);
}
})
- .switchIfEmpty(Mono.error(new BadRequestException("Empty body")));
+ .switchIfEmpty(Mono.error(() -> new BadRequestException("Empty body")));
}
private Mono<Void> handleContinuationTokenRequest(ContinuationTokenRequest request, HttpServerResponse resp) {
diff --git a/server/protocols/jmap/src/main/java/org/apache/james/jmap/http/Authenticator.java b/server/protocols/jmap/src/main/java/org/apache/james/jmap/http/Authenticator.java
index cfd1bcb..cb751ee 100644
--- a/server/protocols/jmap/src/main/java/org/apache/james/jmap/http/Authenticator.java
+++ b/server/protocols/jmap/src/main/java/org/apache/james/jmap/http/Authenticator.java
@@ -56,7 +56,7 @@ public class Authenticator {
Flux.fromIterable(authMethods)
.concatMap(auth -> auth.createMailboxSession(request))
.next()
- .switchIfEmpty(Mono.error(noAuthSupplied()))));
+ .switchIfEmpty(Mono.error(this::noAuthSupplied))));
}
private NoAuthorizationSuppliedException noAuthSupplied() {
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 06/11: [REFACTORING] Cassandra implementation
should depend on interfaces for UidProvider and ModSeqProvider
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 dac815063fb735c69d2b22e39979ba420eb52cea
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Thu May 13 15:59:47 2021 +0700
[REFACTORING] Cassandra implementation should depend on interfaces for UidProvider and ModSeqProvider
This enables people to inject their own generation logic, that might rely
on stronger mechanisms than LWTs.
Candidates might be Atomix, Zookeeper, Consul, Etcd, etc...
---
.../CassandraMailboxSessionMapperFactory.java | 7 +++----
.../cassandra/mail/CassandraMessageIdMapper.java | 7 ++++---
.../cassandra/mail/CassandraMessageMapper.java | 16 +++++++++-------
.../cassandra/mail/CassandraModSeqProvider.java | 19 +++++++++----------
.../mailbox/cassandra/mail/CassandraUidProvider.java | 11 ++++++++---
.../cassandra/mail/CassandraModSeqProviderTest.java | 4 ++--
.../mailbox/cassandra/mail/utils/GuiceUtils.java | 3 +++
.../james/mailbox/store/mail/ModSeqProvider.java | 6 ++++++
.../apache/james/mailbox/store/mail/UidProvider.java | 16 ++++++++++++++++
9 files changed, 60 insertions(+), 29 deletions(-)
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/CassandraMailboxSessionMapperFactory.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/CassandraMailboxSessionMapperFactory.java
index 671b894..4194924 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/CassandraMailboxSessionMapperFactory.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/CassandraMailboxSessionMapperFactory.java
@@ -50,7 +50,6 @@ import org.apache.james.mailbox.cassandra.mail.CassandraMessageIdMapper;
import org.apache.james.mailbox.cassandra.mail.CassandraMessageIdToImapUidDAO;
import org.apache.james.mailbox.cassandra.mail.CassandraMessageMapper;
import org.apache.james.mailbox.cassandra.mail.CassandraModSeqProvider;
-import org.apache.james.mailbox.cassandra.mail.CassandraUidProvider;
import org.apache.james.mailbox.cassandra.mail.CassandraUserMailboxRightsDAO;
import org.apache.james.mailbox.cassandra.mail.task.RecomputeMailboxCountersService;
import org.apache.james.mailbox.cassandra.user.CassandraSubscriptionMapper;
@@ -73,8 +72,8 @@ public class CassandraMailboxSessionMapperFactory extends MailboxSessionMapperFa
protected static final String ATTACHMENTMAPPER = "ATTACHMENTMAPPER";
private final Session session;
- private final CassandraUidProvider uidProvider;
- private final CassandraModSeqProvider modSeqProvider;
+ private final UidProvider uidProvider;
+ private final ModSeqProvider modSeqProvider;
private final CassandraMessageDAO messageDAO;
private final CassandraMessageDAOV3 messageDAOV3;
private final CassandraMessageIdDAO messageIdDAO;
@@ -101,7 +100,7 @@ public class CassandraMailboxSessionMapperFactory extends MailboxSessionMapperFa
private final CassandraConfiguration cassandraConfiguration;
@Inject
- public CassandraMailboxSessionMapperFactory(CassandraUidProvider uidProvider, CassandraModSeqProvider modSeqProvider, Session session,
+ public CassandraMailboxSessionMapperFactory(UidProvider uidProvider, CassandraModSeqProvider modSeqProvider, Session session,
CassandraMessageDAO messageDAO,
CassandraMessageDAOV3 messageDAOV3, CassandraMessageIdDAO messageIdDAO, CassandraMessageIdToImapUidDAO imapUidDAO,
CassandraMailboxCounterDAO mailboxCounterDAO, CassandraMailboxRecentsDAO mailboxRecentsDAO, CassandraMailboxDAO mailboxDAO,
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 9ac0b67..84407d8 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
@@ -49,6 +49,7 @@ import org.apache.james.mailbox.store.MailboxReactorUtils;
import org.apache.james.mailbox.store.mail.MailboxMapper;
import org.apache.james.mailbox.store.mail.MessageIdMapper;
import org.apache.james.mailbox.store.mail.MessageMapper.FetchType;
+import org.apache.james.mailbox.store.mail.ModSeqProvider;
import org.apache.james.mailbox.store.mail.model.MailboxMessage;
import org.apache.james.util.FunctionalUtils;
import org.apache.james.util.ReactorUtils;
@@ -79,14 +80,14 @@ public class CassandraMessageIdMapper implements MessageIdMapper {
private final CassandraMessageDAO messageDAO;
private final CassandraMessageDAOV3 messageDAOV3;
private final CassandraIndexTableHandler indexTableHandler;
- private final CassandraModSeqProvider modSeqProvider;
+ private final ModSeqProvider modSeqProvider;
private final AttachmentLoader attachmentLoader;
private final CassandraConfiguration cassandraConfiguration;
public CassandraMessageIdMapper(MailboxMapper mailboxMapper, CassandraMailboxDAO mailboxDAO, CassandraAttachmentMapper attachmentMapper,
CassandraMessageIdToImapUidDAO imapUidDAO, CassandraMessageIdDAO messageIdDAO,
CassandraMessageDAO messageDAO, CassandraMessageDAOV3 messageDAOV3, CassandraIndexTableHandler indexTableHandler,
- CassandraModSeqProvider modSeqProvider, CassandraConfiguration cassandraConfiguration) {
+ ModSeqProvider modSeqProvider, CassandraConfiguration cassandraConfiguration) {
this.mailboxMapper = mailboxMapper;
this.mailboxDAO = mailboxDAO;
@@ -308,7 +309,7 @@ public class CassandraMessageIdMapper implements MessageIdMapper {
if (identicalFlags(oldComposedId, newFlags)) {
return Mono.just(Pair.of(oldComposedId.getFlags(), oldComposedId));
} else {
- return modSeqProvider.nextModSeq(cassandraId)
+ return modSeqProvider.nextModSeqReactive(cassandraId)
.map(modSeq -> new ComposedMessageIdWithMetaData(
oldComposedId.getComposedMessageId(),
newFlags,
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 f929b83..2a1f903 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
@@ -58,6 +58,8 @@ import org.apache.james.mailbox.model.MessageRange;
import org.apache.james.mailbox.model.UpdatedFlags;
import org.apache.james.mailbox.store.FlagsUpdateCalculator;
import org.apache.james.mailbox.store.mail.MessageMapper;
+import org.apache.james.mailbox.store.mail.ModSeqProvider;
+import org.apache.james.mailbox.store.mail.UidProvider;
import org.apache.james.mailbox.store.mail.model.MailboxMessage;
import org.apache.james.mailbox.store.mail.model.impl.SimpleMailboxMessage;
import org.apache.james.task.Task;
@@ -82,8 +84,8 @@ public class CassandraMessageMapper implements MessageMapper {
private static final Duration MIN_RETRY_BACKOFF = Duration.ofMillis(10);
private static final Duration MAX_RETRY_BACKOFF = Duration.ofMillis(1000);
- private final CassandraModSeqProvider modSeqProvider;
- private final CassandraUidProvider uidProvider;
+ private final ModSeqProvider modSeqProvider;
+ private final UidProvider uidProvider;
private final CassandraMessageDAO messageDAO;
private final CassandraMessageDAOV3 messageDAOV3;
private final CassandraMessageIdDAO messageIdDAO;
@@ -99,7 +101,7 @@ public class CassandraMessageMapper implements MessageMapper {
private final RecomputeMailboxCountersService recomputeMailboxCountersService;
private final SecureRandom secureRandom;
- public CassandraMessageMapper(CassandraUidProvider uidProvider, CassandraModSeqProvider modSeqProvider,
+ public CassandraMessageMapper(UidProvider uidProvider, ModSeqProvider modSeqProvider,
CassandraAttachmentMapper attachmentMapper,
CassandraMessageDAO messageDAO, CassandraMessageDAOV3 messageDAOV3, CassandraMessageIdDAO messageIdDAO,
CassandraMessageIdToImapUidDAO imapUidDAO, CassandraMailboxCounterDAO mailboxCounterDAO,
@@ -343,10 +345,10 @@ public class CassandraMessageMapper implements MessageMapper {
private Mono<MailboxMessage> addUidAndModseq(MailboxMessage message, CassandraId mailboxId) {
Mono<MessageUid> messageUidMono = uidProvider
- .nextUids(mailboxId)
+ .nextUidReactive(mailboxId)
.switchIfEmpty(Mono.error(() -> new MailboxException("Can not find a UID to save " + message.getMessageId() + " in " + mailboxId)));
- Mono<ModSeq> nextModSeqMono = modSeqProvider.nextModSeq(mailboxId)
+ Mono<ModSeq> nextModSeqMono = modSeqProvider.nextModSeqReactive(mailboxId)
.switchIfEmpty(Mono.error(() -> new MailboxException("Can not find a MODSEQ to save " + message.getMessageId() + " in " + mailboxId)));
return Mono.zip(messageUidMono, nextModSeqMono)
@@ -438,7 +440,7 @@ public class CassandraMessageMapper implements MessageMapper {
}
private Mono<ModSeq> computeNewModSeq(CassandraId mailboxId) {
- return modSeqProvider.nextModSeq(mailboxId)
+ return modSeqProvider.nextModSeqReactive(mailboxId)
.switchIfEmpty(ReactorUtils.executeAndEmpty(() -> new RuntimeException("ModSeq generation failed for mailbox " + mailboxId.asUuid())));
}
@@ -492,7 +494,7 @@ public class CassandraMessageMapper implements MessageMapper {
CassandraId mailboxId = (CassandraId) mailbox.getMailboxId();
Mono<List<MessageUid>> uids = uidProvider.nextUids(mailboxId, messages.size());
- Mono<ModSeq> nextModSeq = modSeqProvider.nextModSeq(mailboxId);
+ Mono<ModSeq> nextModSeq = modSeqProvider.nextModSeqReactive(mailboxId);
Mono<List<MailboxMessage>> messagesWithUidAndModSeq = nextModSeq.flatMap(modSeq -> uids.map(uidList -> Pair.of(uidList, modSeq)))
.map(pair -> pair.getKey().stream()
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 e41df05..65896d2 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
@@ -121,19 +121,16 @@ public class CassandraModSeqProvider implements ModSeqProvider {
.where(eq(MAILBOX_ID, bindMarker(MAILBOX_ID))));
}
-
-
@Override
public ModSeq nextModSeq(Mailbox mailbox) throws MailboxException {
- CassandraId mailboxId = (CassandraId) mailbox.getMailboxId();
- return nextModSeq(mailboxId)
+ return nextModSeqReactive(mailbox.getMailboxId())
.blockOptional()
- .orElseThrow(() -> new MailboxException("Can not retrieve modseq for " + mailboxId));
+ .orElseThrow(() -> new MailboxException("Can not retrieve modseq for " + mailbox.getMailboxId()));
}
@Override
public ModSeq nextModSeq(MailboxId mailboxId) throws MailboxException {
- return nextModSeq((CassandraId) mailboxId)
+ return nextModSeqReactive(mailboxId)
.blockOptional()
.orElseThrow(() -> new MailboxException("Can not retrieve modseq for " + mailboxId));
}
@@ -184,13 +181,15 @@ public class CassandraModSeqProvider implements ModSeqProvider {
return Optional.empty();
}
- public Mono<ModSeq> nextModSeq(CassandraId mailboxId) {
+ @Override
+ public Mono<ModSeq> nextModSeqReactive(MailboxId mailboxId) {
+ CassandraId cassandraId = (CassandraId) mailboxId;
Duration firstBackoff = Duration.ofMillis(10);
- return findHighestModSeq(mailboxId)
+ return findHighestModSeq(cassandraId)
.flatMap(maybeHighestModSeq -> maybeHighestModSeq
- .map(highestModSeq -> tryUpdateModSeq(mailboxId, highestModSeq))
- .orElseGet(() -> tryInsertModSeq(mailboxId, ModSeq.first())))
+ .map(highestModSeq -> tryUpdateModSeq(cassandraId, highestModSeq))
+ .orElseGet(() -> tryInsertModSeq(cassandraId, ModSeq.first())))
.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 afc7111..877c15c 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
@@ -105,12 +105,14 @@ public class CassandraUidProvider implements UidProvider {
@Override
public MessageUid nextUid(MailboxId mailboxId) throws MailboxException {
CassandraId cassandraId = (CassandraId) mailboxId;
- return nextUids(cassandraId)
+ return nextUidReactive(cassandraId)
.blockOptional()
.orElseThrow(() -> new MailboxException("Error during Uid update"));
}
- public Mono<MessageUid> nextUids(CassandraId cassandraId) {
+ @Override
+ public Mono<MessageUid> nextUidReactive(MailboxId mailboxId) {
+ CassandraId cassandraId = (CassandraId) mailboxId;
Mono<MessageUid> updateUid = findHighestUid(cassandraId)
.flatMap(messageUid -> tryUpdateUid(cassandraId, messageUid));
@@ -122,7 +124,10 @@ public class CassandraUidProvider implements UidProvider {
.retryWhen(Retry.backoff(maxUidRetries, firstBackoff).scheduler(Schedulers.elastic()));
}
- public Mono<List<MessageUid>> nextUids(CassandraId cassandraId, int count) {
+ @Override
+ public Mono<List<MessageUid>> nextUids(MailboxId mailboxId, int count) {
+ CassandraId cassandraId = (CassandraId) mailboxId;
+
Mono<List<MessageUid>> updateUid = findHighestUid(cassandraId)
.flatMap(messageUid -> tryUpdateUid(cassandraId, messageUid, count)
.map(highest -> range(messageUid, highest)));
diff --git a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraModSeqProviderTest.java b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraModSeqProviderTest.java
index 8773221..8ad2b1e 100644
--- a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraModSeqProviderTest.java
+++ b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraModSeqProviderTest.java
@@ -115,10 +115,10 @@ class CassandraModSeqProviderTest {
.times(1)
.whenQueryStartsWith("SELECT nextModseq FROM modseq WHERE mailboxId=:mailboxId;"));
- CompletableFuture<ModSeq> operation1 = modSeqProvider.nextModSeq(CASSANDRA_ID)
+ CompletableFuture<ModSeq> operation1 = modSeqProvider.nextModSeqReactive(CASSANDRA_ID)
.subscribeOn(Schedulers.elastic())
.toFuture();
- CompletableFuture<ModSeq> operation2 = modSeqProvider.nextModSeq(CASSANDRA_ID)
+ CompletableFuture<ModSeq> operation2 = modSeqProvider.nextModSeqReactive(CASSANDRA_ID)
.subscribeOn(Schedulers.elastic())
.toFuture();
diff --git a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/utils/GuiceUtils.java b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/utils/GuiceUtils.java
index 1c6481a..c1c9ba1 100644
--- a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/utils/GuiceUtils.java
+++ b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/utils/GuiceUtils.java
@@ -38,8 +38,10 @@ import org.apache.james.eventsourcing.eventstore.cassandra.dto.EventDTOModule;
import org.apache.james.json.DTO;
import org.apache.james.json.DTOModule;
import org.apache.james.mailbox.cassandra.ids.CassandraMessageId;
+import org.apache.james.mailbox.cassandra.mail.CassandraUidProvider;
import org.apache.james.mailbox.cassandra.mail.eventsourcing.acl.ACLModule;
import org.apache.james.mailbox.model.MessageId;
+import org.apache.james.mailbox.store.mail.UidProvider;
import com.datastax.driver.core.Session;
import com.google.common.collect.ImmutableSet;
@@ -73,6 +75,7 @@ public class GuiceUtils {
CassandraConfiguration configuration) {
return Modules.combine(
binder -> binder.bind(MessageId.Factory.class).toInstance(messageIdFactory),
+ binder -> binder.bind(UidProvider.class).to(CassandraUidProvider.class),
binder -> binder.bind(BlobId.Factory.class).toInstance(new HashBlobId.Factory()),
binder -> binder.bind(BlobStore.class).toProvider(() -> CassandraBlobStoreFactory.forTesting(session).passthrough()),
binder -> binder.bind(Session.class).toInstance(session),
diff --git a/mailbox/store/src/main/java/org/apache/james/mailbox/store/mail/ModSeqProvider.java b/mailbox/store/src/main/java/org/apache/james/mailbox/store/mail/ModSeqProvider.java
index 7c96c74..a363401 100644
--- a/mailbox/store/src/main/java/org/apache/james/mailbox/store/mail/ModSeqProvider.java
+++ b/mailbox/store/src/main/java/org/apache/james/mailbox/store/mail/ModSeqProvider.java
@@ -23,6 +23,8 @@ import org.apache.james.mailbox.exception.MailboxException;
import org.apache.james.mailbox.model.Mailbox;
import org.apache.james.mailbox.model.MailboxId;
+import reactor.core.publisher.Mono;
+
/**
* Take care of provide mod-seqences for a given {@link Mailbox}. Be aware that implementations
* need to be thread-safe!
@@ -56,4 +58,8 @@ public interface ModSeqProvider {
* Return the highest mod-sequence which were used for the {@link Mailbox}
*/
ModSeq highestModSeq(MailboxId mailboxId) throws MailboxException;
+
+ default Mono<ModSeq> nextModSeqReactive(MailboxId mailboxId) {
+ return Mono.fromCallable(() -> nextModSeq(mailboxId));
+ }
}
diff --git a/mailbox/store/src/main/java/org/apache/james/mailbox/store/mail/UidProvider.java b/mailbox/store/src/main/java/org/apache/james/mailbox/store/mail/UidProvider.java
index 57c3f8c..d4e6b38 100644
--- a/mailbox/store/src/main/java/org/apache/james/mailbox/store/mail/UidProvider.java
+++ b/mailbox/store/src/main/java/org/apache/james/mailbox/store/mail/UidProvider.java
@@ -18,6 +18,7 @@
****************************************************************/
package org.apache.james.mailbox.store.mail;
+import java.util.List;
import java.util.Optional;
import org.apache.james.mailbox.MessageUid;
@@ -25,6 +26,11 @@ import org.apache.james.mailbox.exception.MailboxException;
import org.apache.james.mailbox.model.Mailbox;
import org.apache.james.mailbox.model.MailboxId;
+import com.github.steveash.guavate.Guavate;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
/**
* Take care of provide uids for a given {@link Mailbox}. Be aware that implementations
* need to be thread-safe!
@@ -44,4 +50,14 @@ public interface UidProvider {
Optional<MessageUid> lastUid(Mailbox mailbox) throws MailboxException;
MessageUid nextUid(MailboxId mailboxId) throws MailboxException;
+
+ default Mono<MessageUid> nextUidReactive(MailboxId mailboxId) {
+ return Mono.fromCallable(() -> nextUid(mailboxId));
+ }
+
+ default Mono<List<MessageUid>> nextUids(MailboxId mailboxId, int count) {
+ return Flux.range(0, count)
+ .flatMap(i -> nextUidReactive(mailboxId))
+ .collect(Guavate.toImmutableList());
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 05/11: [REFACTORING] Reactive version for
MessageIdMapper::copyInMailbox
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 59b94833262bfead2211d1d3278936a5f4bce352
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Thu May 13 15:43:23 2021 +0700
[REFACTORING] Reactive version for MessageIdMapper::copyInMailbox
---
.../james/mailbox/cassandra/mail/CassandraMessageIdMapper.java | 8 ++++++--
.../java/org/apache/james/mailbox/store/mail/MessageIdMapper.java | 5 +++++
2 files changed, 11 insertions(+), 2 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 f724466..9ac0b67 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
@@ -172,9 +172,13 @@ public class CassandraMessageIdMapper implements MessageIdMapper {
@Override
public void copyInMailbox(MailboxMessage mailboxMessage, Mailbox mailbox) throws MailboxException {
- CassandraId mailboxId = (CassandraId) mailbox.getMailboxId();
+ MailboxReactorUtils.block(copyInMailboxReactive(mailboxMessage, mailbox));
+ }
- MailboxReactorUtils.block(saveMessageMetadata(mailboxMessage, mailboxId));
+ @Override
+ public Mono<Void> copyInMailboxReactive(MailboxMessage mailboxMessage, Mailbox mailbox) {
+ CassandraId mailboxId = (CassandraId) mailbox.getMailboxId();
+ return saveMessageMetadata(mailboxMessage, mailboxId);
}
private Mono<Void> saveMessageMetadata(MailboxMessage mailboxMessage, CassandraId mailboxId) {
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 02e1726..0db04f8 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
@@ -35,6 +35,7 @@ import org.apache.james.mailbox.store.mail.MessageMapper.FetchType;
import org.apache.james.mailbox.store.mail.model.MailboxMessage;
import org.reactivestreams.Publisher;
+import com.github.fge.lambdas.Throwing;
import com.google.common.collect.Multimap;
import reactor.core.publisher.Flux;
@@ -56,6 +57,10 @@ public interface MessageIdMapper {
void copyInMailbox(MailboxMessage mailboxMessage, Mailbox mailbox) throws MailboxException;
+ default Mono<Void> copyInMailboxReactive(MailboxMessage mailboxMessage, Mailbox mailbox) {
+ return Mono.fromRunnable(Throwing.runnable(() -> copyInMailboxReactive(mailboxMessage, mailbox)).sneakyThrow());
+ }
+
void delete(MessageId messageId);
void delete(MessageId messageId, Collection<MailboxId> mailboxIds);
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org