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 2020/11/09 11:45:34 UTC
[james-project] 06/09: JAMES-3434 Allow Methods to produce several
responses
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 18462afb1684000037872bad9d6edfb760639d01
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Fri Nov 6 10:52:28 2020 +0700
JAMES-3434 Allow Methods to produce several responses
---
.../org/apache/james/jmap/method/Method.scala | 22 +++++++++--------
.../apache/james/jmap/routes/JMAPApiRoutes.scala | 28 +++++++++++-----------
2 files changed, 26 insertions(+), 24 deletions(-)
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 bcb7edb..7a4cdff 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
@@ -29,7 +29,9 @@ import org.apache.james.metrics.api.MetricFactory
import org.reactivestreams.Publisher
import reactor.core.scala.publisher.SMono
-case class InvocationWithContext(invocation: Invocation, processingContext: ProcessingContext)
+case class InvocationWithContext(invocation: Invocation, processingContext: ProcessingContext) {
+ def recordInvocation: InvocationWithContext = InvocationWithContext(invocation, processingContext.recordInvocation(invocation))
+}
trait Method {
val JMAP_RFC8621_PREFIX: String = "JMAP-RFC8621-"
@@ -49,13 +51,14 @@ trait MethodRequiringAccountId[REQUEST <: WithAccountId] extends Method {
def sessionSupplier: SessionSupplier
override def process(capabilities: Set[CapabilityIdentifier], invocation: InvocationWithContext, mailboxSession: MailboxSession): Publisher[InvocationWithContext] = {
- val result = (for {
- request <- getRequest(mailboxSession, invocation.invocation)
- response <- validateAccountId(request.accountId, mailboxSession, sessionSupplier, invocation.invocation).flatMap{
- case Right(_) => doProcess(capabilities, invocation, mailboxSession, request)
- case Left(errorInvocation) => SMono.just(InvocationWithContext(errorInvocation, invocation.processingContext))
- }
- } yield response)
+ val result = getRequest(mailboxSession, invocation.invocation)
+ .flatMapMany(request => {
+ validateAccountId(request.accountId, mailboxSession, sessionSupplier, invocation.invocation)
+ .flatMapMany {
+ case Right(_) => doProcess(capabilities, invocation, mailboxSession, request)
+ case Left(errorInvocation) => SMono.just(InvocationWithContext(errorInvocation, invocation.processingContext))
+ }
+ })
.onErrorResume {
case e: UnsupportedRequestParameterException => SMono.just(InvocationWithContext(Invocation.error(
ErrorCode.InvalidArguments,
@@ -78,7 +81,6 @@ trait MethodRequiringAccountId[REQUEST <: WithAccountId] extends Method {
case e: Throwable => SMono.raiseError(e)
}
-
metricFactory.decoratePublisherWithTimerMetricLogP99(JMAP_RFC8621_PREFIX + methodName.value, result)
}
@@ -89,7 +91,7 @@ trait MethodRequiringAccountId[REQUEST <: WithAccountId] extends Method {
.switchIfEmpty(SMono.just(Left[Invocation, Session](Invocation.error(ErrorCode.AccountNotFound, invocation.methodCallId))))
}
- def doProcess(capabilities: Set[CapabilityIdentifier], invocation: InvocationWithContext, mailboxSession: MailboxSession, request: REQUEST): SMono[InvocationWithContext]
+ def doProcess(capabilities: Set[CapabilityIdentifier], invocation: InvocationWithContext, mailboxSession: MailboxSession, request: REQUEST): Publisher[InvocationWithContext]
def getRequest(mailboxSession: MailboxSession, invocation: Invocation): SMono[REQUEST]
}
diff --git a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/routes/JMAPApiRoutes.scala b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/routes/JMAPApiRoutes.scala
index 42efd8f..e7d2c8f 100644
--- a/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/routes/JMAPApiRoutes.scala
+++ b/server/protocols/jmap-rfc-8621/src/main/scala/org/apache/james/jmap/routes/JMAPApiRoutes.scala
@@ -44,7 +44,7 @@ import org.apache.james.jmap.{Endpoint, JMAPRoute, JMAPRoutes}
import org.apache.james.mailbox.MailboxSession
import org.slf4j.{Logger, LoggerFactory}
import play.api.libs.json.{JsError, JsSuccess}
-import reactor.core.publisher.Mono
+import reactor.core.publisher.{Flux, Mono}
import reactor.core.scala.publisher.{SFlux, SMono}
import reactor.core.scheduler.Schedulers
import reactor.netty.http.server.{HttpServerRequest, HttpServerResponse}
@@ -134,11 +134,11 @@ class JMAPApiRoutes (val authenticator: Authenticator,
private def processSequentiallyAndUpdateContext(requestObject: RequestObject, mailboxSession: MailboxSession, processingContext: ProcessingContext, capabilities: Set[CapabilityIdentifier]): SMono[Seq[(InvocationWithContext)]] = {
SFlux.fromIterable(requestObject.methodCalls)
- .foldLeft(List[SMono[InvocationWithContext]]())((acc, elem) => {
+ .foldLeft(List[SFlux[InvocationWithContext]]())((acc, elem) => {
val lastProcessingContext: SMono[ProcessingContext] = acc.headOption
- .map(last => last.map(_.processingContext))
+ .map(last => SMono.fromPublisher(Flux.from(last.map(_.processingContext)).last()))
.getOrElse(SMono.just(processingContext))
- val invocation: SMono[InvocationWithContext] = lastProcessingContext.flatMap(context => process(capabilities, mailboxSession, InvocationWithContext(elem, context)))
+ val invocation: SFlux[InvocationWithContext] = lastProcessingContext.flatMapMany(context => process(capabilities, mailboxSession, InvocationWithContext(elem, context)))
invocation.cache() :: acc
})
.map(_.reverse)
@@ -147,26 +147,26 @@ class JMAPApiRoutes (val authenticator: Authenticator,
.collectSeq())
}
- private def process(capabilities: Set[CapabilityIdentifier], mailboxSession: MailboxSession, invocation: InvocationWithContext) : SMono[InvocationWithContext] = {
- SMono.defer(() => {
+ private def process(capabilities: Set[CapabilityIdentifier], mailboxSession: MailboxSession, invocation: InvocationWithContext) : SFlux[InvocationWithContext] = {
+ SFlux.fromPublisher(Flux.defer(() => {
invocation.processingContext.resolveBackReferences(invocation.invocation) match {
- case Left(e) => SMono.just(InvocationWithContext(Invocation.error(
+ case Left(e) => SFlux.just[InvocationWithContext](InvocationWithContext(Invocation.error(
errorCode = ErrorCode.InvalidResultReference,
description = s"Failed resolving back-reference: ${e.message}",
methodCallId = invocation.invocation.methodCallId), invocation.processingContext))
case Right(resolvedInvocation) => processMethodWithMatchName(capabilities, InvocationWithContext(resolvedInvocation, invocation.processingContext), mailboxSession)
- .map((invocationWithContext: InvocationWithContext) => InvocationWithContext(invocationWithContext.invocation, invocationWithContext.processingContext.recordInvocation(invocationWithContext.invocation)))
+ .map(_.recordInvocation)
}
- })
+ }))
}
- private def processMethodWithMatchName(capabilities: Set[CapabilityIdentifier], invocation: InvocationWithContext, mailboxSession: MailboxSession): SMono[InvocationWithContext] =
+ private def processMethodWithMatchName(capabilities: Set[CapabilityIdentifier], invocation: InvocationWithContext, mailboxSession: MailboxSession): SFlux[InvocationWithContext] =
SMono.justOrEmpty(methodsByName.get(invocation.invocation.methodName))
- .flatMap(method => validateCapabilities(capabilities, method.requiredCapabilities)
- .fold(e => SMono.just(InvocationWithContext(Invocation.error(ErrorCode.UnknownMethod, e.description, invocation.invocation.methodCallId), invocation.processingContext)),
- _ => SMono.fromPublisher(method.process(capabilities, invocation, mailboxSession))))
+ .flatMapMany(method => validateCapabilities(capabilities, method.requiredCapabilities)
+ .fold(e => SFlux.just(InvocationWithContext(Invocation.error(ErrorCode.UnknownMethod, e.description, invocation.invocation.methodCallId), invocation.processingContext)),
+ _ => SFlux.fromPublisher(method.process(capabilities, invocation, mailboxSession))))
.onErrorResume(throwable => SMono.just(InvocationWithContext(Invocation.error(ErrorCode.ServerFail, throwable.getMessage, invocation.invocation.methodCallId), invocation.processingContext)))
- .switchIfEmpty(SMono.just(InvocationWithContext(Invocation.error(ErrorCode.UnknownMethod, invocation.invocation.methodCallId), invocation.processingContext)))
+ .switchIfEmpty(SFlux.just(InvocationWithContext(Invocation.error(ErrorCode.UnknownMethod, invocation.invocation.methodCallId), invocation.processingContext)))
private def validateCapabilities(capabilities: Set[CapabilityIdentifier], requiredCapabilities: Set[CapabilityIdentifier]): Either[MissingCapabilityException, Unit] = {
val missingCapabilities = requiredCapabilities -- capabilities
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org