This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-grpc.git
The following commit(s) were added to refs/heads/main by this push:
new 2629e0af Use FastFuture for already completed marshalling futures
(#863)
2629e0af is described below
commit 2629e0af13344bd82bee60114d4c0b0de8d0b5ce
Author: PJ Fanning <[email protected]>
AuthorDate: Tue Sep 1 11:20:59 2026 +0100
Use FastFuture for already completed marshalling futures (#863)
* Use FastFuture for completed futures.
Use FastFuture.apply and FastFuture.successful/failed for trivially
completed futures to optimise subsequent transforms.
This speeds up a lot of service invocations in existing generated code that
depend on these, especially GrpcMarshalling.unmarshal on strict request
entities.
* Benchmark and test FastFuture completion of marshalling futures
Motivation:
The FastFuture change is only observable as a threading/scheduling
property: the returned futures are already completed, so transforms
chained onto them run directly instead of being scheduled on an
ExecutionContext. Nothing asserted or measured that.
Modification:
Add scaladsl GrpcMarshallingSpec, which asserts that unmarshal of a
strict entity (both success and failure), unmarshalStream,
GrpcExceptionHandler.from and the ServiceHandler notFound /
unsupportedMediaType responses are already completed, and that a
transform chained onto each still completes when given an
ExecutionContext that throws if anything is scheduled on it.
Add GrpcMarshallingBenchmark.unmarshallStrictChained, covering the
unmarshal-then-transform chain that generated unary handlers use, and
ScalaUnaryHandlerBenchmark.oldStyleUnaryStrictRequestProcessingFastImplementation,
a variant whose service implementation returns an already completed
FastFuture so the whole handler chain can run directly.
Result:
The intent of the change is pinned by tests, and the benchmarks show
where the saved dispatches actually come from.
Tests:
- sbt runtime/test - 139 succeeded, 0 failed
- sbt checkCodeStyle - success
- sbt "benchmarks/Jmh/run -f 2 -wi 5 -i 5 ..." - see PR body
References:
Refs #862
---------
Co-authored-by: Tim Whittington <[email protected]>
---
.../pekko/grpc/GrpcMarshallingBenchmark.scala | 14 +++
.../pekko/grpc/ScalaUnaryHandlerBenchmark.scala | 20 +++-
.../grpc/internal/HardcodedServiceDiscovery.scala | 3 +-
.../pekko/grpc/internal/PekkoHttpClientUtils.scala | 11 +-
.../pekko/grpc/internal/RequestBuilderImpl.scala | 3 +-
.../pekko/grpc/scaladsl/GrpcExceptionHandler.scala | 4 +-
.../pekko/grpc/scaladsl/GrpcMarshalling.scala | 6 +-
.../pekko/grpc/scaladsl/ServiceHandler.scala | 5 +-
.../pekko/grpc/scaladsl/GrpcMarshallingSpec.scala | 113 +++++++++++++++++++++
9 files changed, 162 insertions(+), 17 deletions(-)
diff --git
a/benchmarks/src/main/scala/org/apache/pekko/grpc/GrpcMarshallingBenchmark.scala
b/benchmarks/src/main/scala/org/apache/pekko/grpc/GrpcMarshallingBenchmark.scala
index 4304cc13..deebe886 100644
---
a/benchmarks/src/main/scala/org/apache/pekko/grpc/GrpcMarshallingBenchmark.scala
+++
b/benchmarks/src/main/scala/org/apache/pekko/grpc/GrpcMarshallingBenchmark.scala
@@ -14,6 +14,8 @@
package org.apache.pekko.grpc
import scala.concurrent.Await
+import scala.concurrent.ExecutionContext
+import scala.concurrent.Future
import scala.concurrent.duration.Duration
import com.google.protobuf.{ Any => JavaAny, ByteString => JavaByteString }
@@ -52,6 +54,7 @@ class GrpcMarshallingBenchmark extends CommonBenchmark {
isTrailer = false))
val mat = SystemMaterializer(system).materializer
+ implicit val ec: ExecutionContext = mat.executionContext
@Benchmark
def marshall(): HttpResponse = {
@@ -68,6 +71,17 @@ class GrpcMarshallingBenchmark extends CommonBenchmark {
Await.result(GrpcMarshalling.unmarshal(entity), Duration.Inf)
}
+ // Unmarshalling a strict entity and then chaining the transforms a
generated handler applies to it.
+ // This is the shape of the call chain in generated (and hand written) unary
handlers.
+ @Benchmark
+ def unmarshallStrictChained(): ServerReflectionRequest = {
+ val result = GrpcMarshalling
+ .unmarshal(entity)
+ .flatMap(req => Future.successful(req))
+ .map(identity)
+ Await.result(result, Duration.Inf)
+ }
+
@Benchmark
def unmarshallJavaStrict(): JavaAny = {
pekko.grpc.javadsl.GrpcMarshalling.unmarshal(javaEntity, javaSerializer,
mat, reader).toCompletableFuture.get()
diff --git
a/benchmarks/src/main/scala/org/apache/pekko/grpc/ScalaUnaryHandlerBenchmark.scala
b/benchmarks/src/main/scala/org/apache/pekko/grpc/ScalaUnaryHandlerBenchmark.scala
index 1384ecce..a4215c4d 100644
---
a/benchmarks/src/main/scala/org/apache/pekko/grpc/ScalaUnaryHandlerBenchmark.scala
+++
b/benchmarks/src/main/scala/org/apache/pekko/grpc/ScalaUnaryHandlerBenchmark.scala
@@ -48,6 +48,7 @@ import pekko.http.scaladsl.model.HttpResponse
import pekko.http.scaladsl.model.StatusCodes
import pekko.http.scaladsl.model.TransferEncodings
import pekko.http.scaladsl.model.Uri
+import pekko.http.scaladsl.util.FastFuture
import pekko.stream.Materializer
import pekko.stream.SystemMaterializer
import pekko.stream.scaladsl.Sink
@@ -66,7 +67,11 @@ class ScalaUnaryHandlerBenchmark extends CommonBenchmark {
private val writer = GrpcProtocolNative.newWriter(Identity)
private val requestMessage = HelloRequest("Alice")
private val responseMessage = HelloReply("Hello, Alice")
- private val implementation = new BenchmarkGreeterService(responseMessage)
+ // A typical service implementation, returning an ordinary completed Future.
+ private val implementation = new
BenchmarkGreeterService(Future.successful(responseMessage))
+ // A service implementation that returns an already completed FastFuture, so
that every transform
+ // the handler chains onto it can also run directly instead of being
scheduled.
+ private val fastImplementation = new
BenchmarkGreeterService(FastFuture.successful(responseMessage))
private val request: HttpRequest = {
val data =
@@ -88,7 +93,7 @@ class ScalaUnaryHandlerBenchmark extends CommonBenchmark {
private val generatedHandler: HttpRequest => Future[HttpResponse] =
GreeterServiceHandler(implementation)
- private val oldStyleHandler: HttpRequest => Future[HttpResponse] = {
+ private def oldStyleHandlerFor(implementation: GreeterService): HttpRequest
=> Future[HttpResponse] = {
val notFound = Future.successful(HttpResponse(StatusCodes.NotFound))
val unsupportedMediaType =
Future.successful(HttpResponse(StatusCodes.UnsupportedMediaType))
val spi = TelemetryExtension(system).spi
@@ -117,6 +122,9 @@ class ScalaUnaryHandlerBenchmark extends CommonBenchmark {
}
}
+ private val oldStyleHandler: HttpRequest => Future[HttpResponse] =
oldStyleHandlerFor(implementation)
+ private val fastOldStyleHandler: HttpRequest => Future[HttpResponse] =
oldStyleHandlerFor(fastImplementation)
+
@Benchmark
def generatedUnaryStrictRequestProcessing(blackhole: Blackhole): Unit =
consumeResponse(Await.result(generatedHandler(request), Duration.Inf),
blackhole)
@@ -125,6 +133,10 @@ class ScalaUnaryHandlerBenchmark extends CommonBenchmark {
def oldStyleUnaryStrictRequestProcessing(blackhole: Blackhole): Unit =
consumeResponse(Await.result(oldStyleHandler(request), Duration.Inf),
blackhole)
+ @Benchmark
+ def oldStyleUnaryStrictRequestProcessingFastImplementation(blackhole:
Blackhole): Unit =
+ consumeResponse(Await.result(fastOldStyleHandler(request), Duration.Inf),
blackhole)
+
private def consumeResponse(response: HttpResponse, blackhole: Blackhole):
Unit = {
blackhole.consume(response.status)
response.entity match {
@@ -139,9 +151,9 @@ class ScalaUnaryHandlerBenchmark extends CommonBenchmark {
def tearDown(): Unit =
system.terminate()
- private final class BenchmarkGreeterService(response: HelloReply) extends
GreeterService {
+ private final class BenchmarkGreeterService(response: Future[HelloReply])
extends GreeterService {
override def sayHello(in: HelloRequest): Future[HelloReply] =
- Future.successful(response)
+ response
override def itKeepsTalking(in: Source[HelloRequest, NotUsed]):
Future[HelloReply] =
throw new UnsupportedOperationException("itKeepsTalking")
diff --git
a/runtime/src/main/scala/org/apache/pekko/grpc/internal/HardcodedServiceDiscovery.scala
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/HardcodedServiceDiscovery.scala
index 4054faa6..7ceda5e3 100644
---
a/runtime/src/main/scala/org/apache/pekko/grpc/internal/HardcodedServiceDiscovery.scala
+++
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/HardcodedServiceDiscovery.scala
@@ -16,11 +16,12 @@ package org.apache.pekko.grpc.internal
import org.apache.pekko
import pekko.discovery.{ Lookup, ServiceDiscovery }
import pekko.discovery.ServiceDiscovery.Resolved
+import pekko.http.scaladsl.util.FastFuture
import scala.concurrent.Future
import scala.concurrent.duration.FiniteDuration
class HardcodedServiceDiscovery(resolved: Resolved) extends ServiceDiscovery {
override def lookup(lookup: Lookup, resolveTimeout: FiniteDuration):
Future[Resolved] =
- Future.successful(resolved)
+ FastFuture.successful(resolved)
}
diff --git
a/runtime/src/main/scala/org/apache/pekko/grpc/internal/PekkoHttpClientUtils.scala
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/PekkoHttpClientUtils.scala
index 43fe647f..f513f72d 100644
---
a/runtime/src/main/scala/org/apache/pekko/grpc/internal/PekkoHttpClientUtils.scala
+++
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/PekkoHttpClientUtils.scala
@@ -30,6 +30,7 @@ import pekko.http.scaladsl.{ ClientTransport,
ConnectionContext, Http, HttpsConn
import pekko.http.scaladsl.model._
import pekko.http.scaladsl.model.headers.RawHeader
import pekko.http.scaladsl.settings.ClientConnectionSettings
+import pekko.http.scaladsl.util.FastFuture
import pekko.stream.{ Materializer, QueueOfferResult }
import pekko.stream.scaladsl.{ Keep, Sink, Source }
import pekko.util.ByteString
@@ -317,7 +318,7 @@ object PekkoHttpClientUtils {
if (response.status != StatusCodes.OK) {
response.entity.discardBytes()
val failure = mapToStatusException(response, immutable.Seq.empty)
- Source.failed(failure).mapMaterializedValue(_ =>
Future.failed(failure))
+ Source.failed(failure).mapMaterializedValue(_ =>
FastFuture.failed(failure))
} else {
Codecs.detect(response) match {
case Success(codec) =>
@@ -362,7 +363,7 @@ object PekkoHttpClientUtils {
.via(reader.dataFrameDecoder)
.map(deserializer.deserialize)
.mapMaterializedValue(_ =>
- Future.successful(new GrpcResponseMetadata() {
+ FastFuture.successful(new GrpcResponseMetadata() {
override def headers: pekko.grpc.scaladsl.Metadata =
new HeaderMetadataImpl(response.headers)
@@ -379,7 +380,7 @@ object PekkoHttpClientUtils {
.asJava
}))
case Failure(e) =>
- Source.failed[O](e).mapMaterializedValue(_ => Future.failed(e))
+ Source.failed[O](e).mapMaterializedValue(_ =>
FastFuture.failed(e))
}
}
}
@@ -391,9 +392,9 @@ object PekkoHttpClientUtils {
val allHeaders = response.headers ++ trailers
allHeaders.find(_.name == "grpc-status").map(_.value) match {
case Some("0") =>
- Future.successful(())
+ FastFuture.successful(())
case _ =>
- Future.failed(mapToStatusException(response, trailers))
+ FastFuture.failed(mapToStatusException(response, trailers))
}
}
diff --git
a/runtime/src/main/scala/org/apache/pekko/grpc/internal/RequestBuilderImpl.scala
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/RequestBuilderImpl.scala
index 648c2021..ea1f980a 100644
---
a/runtime/src/main/scala/org/apache/pekko/grpc/internal/RequestBuilderImpl.scala
+++
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/RequestBuilderImpl.scala
@@ -19,6 +19,7 @@ import org.apache.pekko
import pekko.NotUsed
import pekko.annotation.{ InternalApi, InternalStableApi }
import pekko.grpc.{ GrpcClientSettings, GrpcResponseMetadata,
GrpcServiceException, GrpcSingleResponse }
+import pekko.http.scaladsl.util.FastFuture
import pekko.stream.{ Graph, Materializer, SourceShape }
import pekko.stream.javadsl.{ Source => JavaSource }
import pekko.stream.scaladsl.{ Keep, Sink, Source }
@@ -349,7 +350,7 @@ object RequestBuilderImpl {
}
def richError[U]: PartialFunction[Throwable, Future[U]] = {
- case item => Future.failed(RequestBuilderImpl.lift(item))
+ case item => FastFuture.failed(RequestBuilderImpl.lift(item))
}
def lift(item: Throwable): scala.Throwable = item match {
diff --git
a/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/GrpcExceptionHandler.scala
b/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/GrpcExceptionHandler.scala
index ab60d298..a0c5d47c 100644
---
a/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/GrpcExceptionHandler.scala
+++
b/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/GrpcExceptionHandler.scala
@@ -21,6 +21,7 @@ import pekko.grpc.{ GrpcServiceException, Trailers }
import pekko.grpc.GrpcProtocol.GrpcProtocolWriter
import pekko.grpc.internal.{ GrpcMetadataImpl, GrpcResponseHelpers,
MissingParameterException }
import pekko.http.scaladsl.model.HttpResponse
+import pekko.http.scaladsl.util.FastFuture
import io.grpc.{ Status, StatusRuntimeException }
import org.apache.pekko.http.scaladsl.model.http2.PeerClosedStreamException
@@ -71,6 +72,7 @@ object GrpcExceptionHandler {
def from(mapper: PartialFunction[Throwable, Trailers])(
implicit system: ClassicActorSystemProvider,
writer: GrpcProtocolWriter): PartialFunction[Throwable,
Future[HttpResponse]] =
- mapper.orElse(defaultMapper(system.classicSystem)).andThen(s =>
Future.successful(GrpcResponseHelpers.status(s)))
+ mapper.orElse(defaultMapper(system.classicSystem)).andThen(s =>
+ FastFuture.successful(GrpcResponseHelpers.status(s)))
}
diff --git
a/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/GrpcMarshalling.scala
b/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/GrpcMarshalling.scala
index dd316325..b781fb85 100644
---
a/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/GrpcMarshalling.scala
+++
b/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/GrpcMarshalling.scala
@@ -57,7 +57,7 @@ object GrpcMarshalling {
def negotiated[T](req: HttpRequest, f: (GrpcProtocolReader,
GrpcProtocolWriter) => Future[T]): Option[Future[T]] =
GrpcProtocol.negotiate(req).map {
case (Success(reader), writer) => f(reader, writer)
- case (Failure(ex), _) => Future.failed(ex)
+ case (Failure(ex), _) => FastFuture.failed(ex)
}
/**
@@ -87,7 +87,7 @@ object GrpcMarshalling {
def unmarshal[T](
entity: HttpEntity)(implicit u: ProtobufSerializer[T], mat:
Materializer, reader: GrpcProtocolReader): Future[T] =
entity match {
- case HttpEntity.Strict(_, data) =>
Future.fromTry(Try(u.deserialize(reader.decodeSingleFrame(data))))
+ case HttpEntity.Strict(_, data) =>
FastFuture(Try(u.deserialize(reader.decodeSingleFrame(data))))
case _ => unmarshal(entity.dataBytes)
}
@@ -95,7 +95,7 @@ object GrpcMarshalling {
implicit u: ProtobufSerializer[T],
@nowarn("msg=is never used") mat: Materializer,
reader: GrpcProtocolReader): Future[Source[T, NotUsed]] = {
- Future.successful(
+ FastFuture.successful(
data
.mapMaterializedValue(_ => NotUsed)
.via(reader.dataFrameDecoder)
diff --git
a/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/ServiceHandler.scala
b/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/ServiceHandler.scala
index b020474a..970b2da7 100644
--- a/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/ServiceHandler.scala
+++ b/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/ServiceHandler.scala
@@ -19,16 +19,17 @@ import pekko.grpc.GrpcProtocol
import pekko.grpc.internal.{ GrpcProtocolWeb, GrpcProtocolWebText }
import pekko.http.javadsl.{ model => jmodel }
import pekko.http.scaladsl.model.{ HttpRequest, HttpResponse, StatusCodes }
+import pekko.http.scaladsl.util.FastFuture
import scala.concurrent.Future
@ApiMayChange
object ServiceHandler {
- private[scaladsl] val notFound: Future[HttpResponse] =
Future.successful(HttpResponse(StatusCodes.NotFound))
+ private[scaladsl] val notFound: Future[HttpResponse] =
FastFuture.successful(HttpResponse(StatusCodes.NotFound))
private[scaladsl] val unsupportedMediaType: Future[HttpResponse] =
- Future.successful(HttpResponse(StatusCodes.UnsupportedMediaType))
+ FastFuture.successful(HttpResponse(StatusCodes.UnsupportedMediaType))
private def matchesVariant(variants: Set[GrpcProtocol])(request:
jmodel.HttpRequest) =
variants.exists(_.mediaTypes.contains(request.entity.getContentType.mediaType))
diff --git
a/runtime/src/test/scala/org/apache/pekko/grpc/scaladsl/GrpcMarshallingSpec.scala
b/runtime/src/test/scala/org/apache/pekko/grpc/scaladsl/GrpcMarshallingSpec.scala
new file mode 100644
index 00000000..550a343e
--- /dev/null
+++
b/runtime/src/test/scala/org/apache/pekko/grpc/scaladsl/GrpcMarshallingSpec.scala
@@ -0,0 +1,113 @@
+/*
+ * 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.pekko.grpc.scaladsl
+
+import scala.concurrent.{ Await, ExecutionContext }
+import scala.concurrent.duration._
+import scala.util.Success
+
+import com.google.protobuf.ByteString
+import com.google.protobuf.any.{ Any => ScalapbAny }
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
+
+import org.apache.pekko
+import pekko.actor.ActorSystem
+import pekko.grpc.GrpcProtocol.{ GrpcProtocolReader, GrpcProtocolWriter }
+import pekko.grpc.internal.{ AbstractGrpcProtocol, GrpcProtocolNative,
Identity }
+import pekko.http.scaladsl.model.HttpEntity
+import pekko.stream.{ Materializer, SystemMaterializer }
+import pekko.util.{ ByteString => PekkoByteString }
+
+class GrpcMarshallingSpec extends AnyWordSpec with Matchers with
BeforeAndAfterAll {
+
+ private implicit val system: ActorSystem = ActorSystem("GrpcMarshallingSpec")
+ private implicit val mat: Materializer =
SystemMaterializer(system).materializer
+ private implicit val serializer: ScalapbProtobufSerializer[ScalapbAny] =
+ new ScalapbProtobufSerializer(ScalapbAny)
+ private implicit val reader: GrpcProtocolReader =
GrpcProtocolNative.newReader(Identity)
+ private implicit val writer: GrpcProtocolWriter =
GrpcProtocolNative.newWriter(Identity)
+
+ /**
+ * An ExecutionContext that refuses to run anything, so that a transform
which needs to be
+ * scheduled fails loudly instead of silently completing on another thread.
+ */
+ private object NoDispatch extends ExecutionContext {
+ override def execute(runnable: Runnable): Unit =
+ throw new AssertionError("transform was scheduled on an ExecutionContext
instead of completing directly")
+ override def reportFailure(cause: Throwable): Unit = throw cause
+ }
+
+ private val message = ScalapbAny("type.googleapis.com/test",
ByteString.copyFromUtf8("payload"))
+
+ private def strictEntity(data: PekkoByteString): HttpEntity.Strict =
+ HttpEntity.Strict(GrpcProtocolNative.contentType, data)
+
+ private val validEntity =
+
strictEntity(AbstractGrpcProtocol.encodeFrameData(serializer.serialize(message),
isCompressed = false,
+ isTrailer = false))
+
+ // A single zero byte is an invalid protobuf tag, so deserialization of this
frame fails.
+ private val corruptEntity =
+ strictEntity(
+ AbstractGrpcProtocol.encodeFrameData(PekkoByteString(0), isCompressed =
false, isTrailer = false))
+
+ override protected def afterAll(): Unit = Await.result(system.terminate(),
10.seconds)
+
+ "The scaladsl GrpcMarshalling" should {
+
+ "complete unmarshal of a strict entity directly, without scheduling" in {
+ val unmarshalled = GrpcMarshalling.unmarshal[ScalapbAny](validEntity)
+
+ unmarshalled.value should be(Some(Success(message)))
+ // A generated handler chains further transforms onto this future; they
must not need a dispatch either.
+ unmarshalled.flatMap(a =>
GrpcMarshalling.unmarshal[ScalapbAny](validEntity).map(_ => a)(NoDispatch))(
+ NoDispatch).value should be(Some(Success(message)))
+ }
+
+ "complete a failed unmarshal of a strict entity directly, without
scheduling" in {
+ val unmarshalled = GrpcMarshalling.unmarshal[ScalapbAny](corruptEntity)
+
+ unmarshalled.value.map(_.isFailure) should be(Some(true))
+ unmarshalled.recover { case _ => message }(NoDispatch).value should
be(Some(Success(message)))
+ }
+
+ "complete unmarshalStream directly, without scheduling" in {
+ val unmarshalled =
GrpcMarshalling.unmarshalStream[ScalapbAny](validEntity)
+
+ unmarshalled.value.map(_.isSuccess) should be(Some(true))
+ unmarshalled.map(_ => message)(NoDispatch).value should
be(Some(Success(message)))
+ }
+
+ "complete the exception handler response directly, without scheduling" in {
+ val handled =
+
GrpcExceptionHandler.from(GrpcExceptionHandler.defaultMapper(system))(system,
writer)(
+ new RuntimeException("boom"))
+
+ handled.value.map(_.isSuccess) should be(Some(true))
+ handled.map(_.status.intValue)(NoDispatch).value should
be(Some(Success(200)))
+ handled.value.get.get.getHeader("grpc-status").get.value should be("13")
+ }
+
+ "complete the not-found and unsupported-media-type responses directly,
without scheduling" in {
+ ServiceHandler.notFound.map(_.status.intValue)(NoDispatch).value should
be(Some(Success(404)))
+
ServiceHandler.unsupportedMediaType.map(_.status.intValue)(NoDispatch).value
should be(Some(Success(415)))
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
