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 6c8dc2ba Encode response data frames through GrpcProtocolWriter with
deferred writes (#909)
6c8dc2ba is described below
commit 6c8dc2baf9d9c689c92e8d907e8af9489134e856
Author: PJ Fanning <[email protected]>
AuthorDate: Tue Oct 20 17:19:26 2026 +0100
Encode response data frames through GrpcProtocolWriter with deferred writes
(#909)
* Generalise unary response optimisation.
Encapsulate optimisation of unary responses into a
GrpcProtocol/GrpcProtocolWriter concern, allowing it to be applied to other
variants.
This removes the need for the frame encoder, as this is just a map with
encodeFrame.
* Keep the frame serializer types internal.
Motivation:
The deferred data frame change made ProtobufFrameSerializer a public
@ApiMayChange trait, which would expose the internal DeferredDataWriter
to anyone extending the public serializers.
Modification:
- Keep ProtobufFrameSerializer private[grpc], as it is on main.
- Make DeferredDataWriter's methods, and the ScalaPB and Google
serializers' implementations of them, private[grpc].
- Add the MiMa filters for the Scala 3 positional accessors of
GrpcProtocolReader and GrpcProtocolWriter, and rename the filter file
after the change.
- Remove debug println output from the serializer specs.
Result:
No new public API. Binary compatibility passes on Scala 2.13 and 3.
Tests:
- sbt runtime/test: 180 passed
- sbt +runtime/mimaReportBinaryIssues: passed
References:
Refs #905
* Add a benchmark for encoding response data frames.
Motivation:
The deferred data frame change affects how response messages are
written into gRPC frames, but no benchmark covered gRPC-Web responses.
Modification:
Add FrameWriteBenchmark, which marshals unary and streamed responses
for the native and gRPC-Web protocols without compression, with 16 B,
1 KiB and 64 KiB bytes payloads.
Result:
Frame encoding can be compared with main.
Tests:
- sbt benchmarks/Jmh/compile: passed
References:
Refs #905
* Fix FrameWriteBenchmark formatting and Scala 3.9 warnings.
Replace the deprecated = _ initializers, which fail the next build under
-Werror, and apply scalafmt.
* Use the pekko import prefix for DeferredDataFrame imports.
* Fix the frane parameter name typo in GrpcProtocolWeb.
* Use OutboundFrame in GrpcProtocolWeb encodeFrame and encodeFrameToBytes,
matching GrpcProtocolNative.
---------
Co-authored-by: Tim Whittington <[email protected]>
Co-authored-by: PJ Fanning <[email protected]>
---
.../apache/pekko/grpc/FrameWriteBenchmark.scala | 79 ++++++++++++++++++++++
.../deferred-data-frame.excludes | 43 ++++++++++++
.../scala/org/apache/pekko/grpc/GrpcProtocol.scala | 52 +++++++++++---
.../org/apache/pekko/grpc/ProtobufSerializer.scala | 12 +++-
.../pekko/grpc/internal/AbstractGrpcProtocol.scala | 37 ++++++----
.../pekko/grpc/internal/GrpcEntityHelpers.scala | 27 +++-----
.../pekko/grpc/internal/GrpcProtocolNative.scala | 15 ++--
.../pekko/grpc/internal/GrpcProtocolWeb.scala | 19 ++++--
.../pekko/grpc/internal/GrpcResponseHelpers.scala | 38 +----------
.../grpc/javadsl/GoogleProtobufSerializer.scala | 14 ++--
.../grpc/scaladsl/ScalapbProtobufSerializer.scala | 18 +++--
.../javadsl/GoogleProtobufSerializerSpec.scala | 14 ++++
.../scaladsl/ScalapbProtobufSerializerSpec.scala | 15 ++++
13 files changed, 276 insertions(+), 107 deletions(-)
diff --git
a/benchmarks/src/main/scala/org/apache/pekko/grpc/FrameWriteBenchmark.scala
b/benchmarks/src/main/scala/org/apache/pekko/grpc/FrameWriteBenchmark.scala
new file mode 100644
index 00000000..e939f196
--- /dev/null
+++ b/benchmarks/src/main/scala/org/apache/pekko/grpc/FrameWriteBenchmark.scala
@@ -0,0 +1,79 @@
+/*
+ * 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
+
+import scala.concurrent.Await
+import scala.concurrent.duration.Duration
+
+import org.apache.pekko
+import pekko.actor.ActorSystem
+import pekko.grpc.internal.{ GrpcProtocolNative, GrpcProtocolWeb, Identity }
+import pekko.grpc.scaladsl.{ GrpcMarshalling, ScalapbProtobufSerializer }
+import pekko.http.scaladsl.model.HttpEntity
+import pekko.stream.{ Materializer, SystemMaterializer }
+import pekko.stream.scaladsl.{ Sink, Source }
+import pekko.util.ByteString
+import com.google.protobuf.{ ByteString => ProtobufByteString }
+import com.google.protobuf.any.{ Any => ScalapbAny }
+import org.openjdk.jmh.annotations._
+import org.openjdk.jmh.infra.Blackhole
+
+// Encoding of response data frames, for the native and gRPC-Web protocols
without compression.
+class FrameWriteBenchmark extends CommonBenchmark {
+ @Param(Array("native", "web"))
+ var protocol: String = "native"
+
+ @Param(Array("16", "1024", "65536"))
+ var size: Int = 0
+
+ implicit val system: ActorSystem = ActorSystem("bench")
+ implicit val mat: Materializer = SystemMaterializer(system).materializer
+ implicit val serializer: ScalapbProtobufSerializer[ScalapbAny] = new
ScalapbProtobufSerializer(ScalapbAny)
+
+ implicit var writer: GrpcProtocol.GrpcProtocolWriter =
GrpcProtocolNative.newWriter(Identity)
+ var message: ScalapbAny = ScalapbAny.defaultInstance
+
+ @Setup
+ def setup(): Unit = {
+ writer = protocol match {
+ case "native" => GrpcProtocolNative.newWriter(Identity)
+ case "web" => GrpcProtocolWeb.newWriter(Identity)
+ }
+ message = ScalapbAny("benchmark", ProtobufByteString.copyFrom(new
Array[Byte](size)))
+ }
+
+ @TearDown
+ def tearDown(): Unit = system.terminate()
+
+ @Benchmark
+ def unary(blackhole: Blackhole): Unit =
+ consume(GrpcMarshalling.marshal(message).entity, blackhole)
+
+ @Benchmark
+ @OperationsPerInvocation(100)
+ def stream(blackhole: Blackhole): Unit =
+
consume(GrpcMarshalling.marshalStream(Source.repeat(message).take(100)).entity,
blackhole)
+
+ private def consume(entity: HttpEntity, blackhole: Blackhole): Unit =
+ entity match {
+ case HttpEntity.Strict(_, data) => blackhole.consume(data)
+ case _ =>
+ blackhole.consume(
+ Await.result(entity.dataBytes.runWith(Sink.fold(0)((n, b:
ByteString) => n + b.length)), Duration.Inf))
+ }
+}
diff --git
a/runtime/src/main/mima-filters/2.0.x.backwards.excludes/deferred-data-frame.excludes
b/runtime/src/main/mima-filters/2.0.x.backwards.excludes/deferred-data-frame.excludes
new file mode 100644
index 00000000..40a6dd58
--- /dev/null
+++
b/runtime/src/main/mima-filters/2.0.x.backwards.excludes/deferred-data-frame.excludes
@@ -0,0 +1,43 @@
+# 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.
+
+# encode data frames through GrpcProtocolWriter, with deferred writes into the
frame buffer
+ProblemFilters.exclude[IncompatibleSignatureProblem]("org.apache.pekko.grpc.GrpcProtocol#GrpcProtocolReader.frameDecoder")
+ProblemFilters.exclude[IncompatibleSignatureProblem]("org.apache.pekko.grpc.GrpcProtocol#GrpcProtocolReader.copy")
+ProblemFilters.exclude[IncompatibleSignatureProblem]("org.apache.pekko.grpc.GrpcProtocol#GrpcProtocolReader.copy$default$3")
+ProblemFilters.exclude[IncompatibleSignatureProblem]("org.apache.pekko.grpc.GrpcProtocol#GrpcProtocolReader.this")
+ProblemFilters.exclude[IncompatibleSignatureProblem]("org.apache.pekko.grpc.GrpcProtocol#GrpcProtocolReader.apply")
+ProblemFilters.exclude[IncompatibleSignatureProblem]("org.apache.pekko.grpc.GrpcProtocol#GrpcProtocolReader.unapply")
+ProblemFilters.exclude[IncompatibleSignatureProblem]("org.apache.pekko.grpc.GrpcProtocol#GrpcProtocolWriter.encodeFrame")
+ProblemFilters.exclude[IncompatibleSignatureProblem]("org.apache.pekko.grpc.GrpcProtocol#GrpcProtocolWriter.encodeDataToResponse")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.grpc.GrpcProtocol#GrpcProtocolWriter.frameEncoder")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.grpc.GrpcProtocol#GrpcProtocolWriter.copy")
+ProblemFilters.exclude[IncompatibleSignatureProblem]("org.apache.pekko.grpc.GrpcProtocol#GrpcProtocolWriter.copy$default$3")
+ProblemFilters.exclude[IncompatibleSignatureProblem]("org.apache.pekko.grpc.GrpcProtocol#GrpcProtocolWriter.copy$default$4")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.grpc.GrpcProtocol#GrpcProtocolWriter.copy$default$5")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.grpc.GrpcProtocol#GrpcProtocolWriter.this")
+ProblemFilters.exclude[MissingTypesProblem]("org.apache.pekko.grpc.GrpcProtocol$GrpcProtocolWriter$")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.grpc.GrpcProtocol#GrpcProtocolWriter.apply")
+ProblemFilters.exclude[IncompatibleSignatureProblem]("org.apache.pekko.grpc.GrpcProtocol#GrpcProtocolWriter.unapply")
+ProblemFilters.exclude[IncompatibleSignatureProblem]("org.apache.pekko.grpc.internal.AbstractGrpcProtocol.writer")
+ProblemFilters.exclude[IncompatibleClassSignatureProblem]("org.apache.pekko.grpc.GrpcProtocol$GrpcProtocolReader$")
+ProblemFilters.exclude[IncompatibleClassSignatureProblem]("org.apache.pekko.grpc.internal.AbstractGrpcProtocol$GrpcFramingDecoderStage")
+# Scala 3 positional accessors of the changed GrpcProtocolReader and
GrpcProtocolWriter fields
+ProblemFilters.exclude[IncompatibleSignatureProblem]("org.apache.pekko.grpc.GrpcProtocol#GrpcProtocolReader._3")
+ProblemFilters.exclude[IncompatibleSignatureProblem]("org.apache.pekko.grpc.GrpcProtocol#GrpcProtocolWriter._3")
+ProblemFilters.exclude[IncompatibleSignatureProblem]("org.apache.pekko.grpc.GrpcProtocol#GrpcProtocolWriter._4")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.grpc.GrpcProtocol#GrpcProtocolWriter._5")
diff --git a/runtime/src/main/scala/org/apache/pekko/grpc/GrpcProtocol.scala
b/runtime/src/main/scala/org/apache/pekko/grpc/GrpcProtocol.scala
index 22c7823e..b3ea8b1b 100644
--- a/runtime/src/main/scala/org/apache/pekko/grpc/GrpcProtocol.scala
+++ b/runtime/src/main/scala/org/apache/pekko/grpc/GrpcProtocol.scala
@@ -18,6 +18,7 @@ import pekko.NotUsed
import pekko.annotation.InternalApi
import pekko.annotation.InternalStableApi
import pekko.grpc.GrpcProtocol.{ GrpcProtocolReader, GrpcProtocolWriter }
+import pekko.grpc.GrpcProtocol.DeferredDataFrame.DeferredDataWriter
import pekko.grpc.internal.{
AbstractGrpcProtocol,
Codec,
@@ -87,11 +88,48 @@ object GrpcProtocol {
/** A frame in a logical gRPC protocol stream */
sealed trait Frame
- /** A data (or message) frame in a gRPC protocol stream. */
- case class DataFrame(data: ByteString) extends Frame
+ /** An outbound data frame to be encoded in a gRPC protocol stream */
+ sealed trait OutboundFrame extends Frame
+
+ /** A frame received from a peer system. */
+ sealed trait InboundFrame extends Frame
+
+ /** A data (or message) frame in a gRPC protocol stream */
+ case class DataFrame(data: ByteString) extends InboundFrame with
OutboundFrame
/** A trailer (status headers) frame in a gRPC protocol stream */
- case class TrailerFrame(trailers: List[HttpHeader]) extends Frame
+ case class TrailerFrame(trailers: List[HttpHeader]) extends InboundFrame
with OutboundFrame
+
+ /** An outbound write of a data frame, deferred to allow optimized write
into a pre-allocated buffer */
+ case class DeferredDataFrame[T](element: T, writer: DeferredDataWriter[T])
extends OutboundFrame {
+ type Element = T
+
+ /** Fallback to DataFrame style write, when optimized framed write is not
possible. */
+ def data: ByteString = {
+ val data = new Array[Byte](writer.serializedSize(element))
+ writer.serializeTo(element, data, 0)
+ ByteString.fromArrayUnsafe(data)
+ }
+
+ }
+
+ object DeferredDataFrame {
+ trait DeferredDataWriter[T] {
+
+ /**
+ * Compute the size of the serialized form of the given element.
+ */
+ private[grpc] def serializedSize(t: T): Int
+
+ /**
+ * Serialize the given element into the given frame, starting at the
given offset.
+ * @param t the element to serialize.
+ * @param frame a preallocated frame buffer, which will be at least of
size offset + serializedSize(t)
+ * @param offset the offset to place the serialized data of the element
at.
+ */
+ private[grpc] def serializeTo(t: T, frame: Array[Byte], offset: Int):
Unit
+ }
+ }
/**
* Implements the encoding of a stream of gRPC Frames into a
physical/transport layer.
@@ -104,11 +142,9 @@ object GrpcProtocol {
/** The compression codec to be used for data frame bodies */
messageEncoding: Codec,
/** Encodes a frame as a part in a chunk stream. */
- encodeFrame: Frame => ChunkStreamPart,
+ encodeFrame: OutboundFrame => ChunkStreamPart,
/** A shortcut to encode a data frame directly into a Response */
- encodeDataToResponse: (ByteString, Seq[HttpHeader], Trailer) =>
HttpResponse,
- /** A Flow over a stream of Frame using this frame encoding */
- frameEncoder: Flow[Frame, ChunkStreamPart, NotUsed])
+ encodeDataToResponse: (OutboundFrame, Seq[HttpHeader], Trailer) =>
HttpResponse)
/**
* Implements the decoding of the gRPC framing from a physical/transport
layer.
@@ -118,7 +154,7 @@ object GrpcProtocol {
messageEncoding: Codec,
decodeSingleFrame: ByteString => ByteString,
/** A Flow of Frames over a stream of messages encoded in gRPC framing.
*/
- frameDecoder: Flow[ByteString, Frame, NotUsed]) {
+ frameDecoder: Flow[ByteString, InboundFrame, NotUsed]) {
/**
* A Flow of Frames over a stream of messages encoded in gRPC framing that
only
diff --git
a/runtime/src/main/scala/org/apache/pekko/grpc/ProtobufSerializer.scala
b/runtime/src/main/scala/org/apache/pekko/grpc/ProtobufSerializer.scala
index e5392960..688b698d 100644
--- a/runtime/src/main/scala/org/apache/pekko/grpc/ProtobufSerializer.scala
+++ b/runtime/src/main/scala/org/apache/pekko/grpc/ProtobufSerializer.scala
@@ -14,6 +14,7 @@
package org.apache.pekko.grpc
import org.apache.pekko
+import pekko.grpc.GrpcProtocol.DeferredDataFrame
import pekko.grpc.internal.ByteStringUtils
import pekko.util.ByteString
@@ -21,10 +22,15 @@ import java.io.InputStream
trait ProtobufSerializer[T] {
def serialize(t: T): ByteString
+
def deserialize(bytes: ByteString): T
+
def deserialize(stream: InputStream): T =
deserialize(ByteStringUtils.fromInputStream(stream))
}
-private[grpc] trait ProtobufFrameSerializer[T] extends ProtobufSerializer[T] {
- private[grpc] def serializeDataFrame(t: T): ByteString
-}
+/**
+ * Optional optimisation of ProtobufSerializer, which allows for more
efficient serialization into a frame where the
+ * serialized size of the encoded element can be determined in advance.
+ */
+private[grpc] trait ProtobufFrameSerializer[T] extends ProtobufSerializer[T]
+ with DeferredDataFrame.DeferredDataWriter[T]
diff --git
a/runtime/src/main/scala/org/apache/pekko/grpc/internal/AbstractGrpcProtocol.scala
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/AbstractGrpcProtocol.scala
index 7aa57f88..0b79158b 100644
---
a/runtime/src/main/scala/org/apache/pekko/grpc/internal/AbstractGrpcProtocol.scala
+++
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/AbstractGrpcProtocol.scala
@@ -15,7 +15,8 @@ package org.apache.pekko.grpc.internal
import org.apache.pekko
import pekko.NotUsed
import pekko.grpc.GrpcProtocol
-import pekko.grpc.GrpcProtocol.{ Frame, GrpcProtocolReader, GrpcProtocolWriter
}
+import pekko.grpc.GrpcProtocol.{ GrpcProtocolReader, GrpcProtocolWriter,
InboundFrame, OutboundFrame }
+import pekko.grpc.GrpcProtocol.DeferredDataFrame.DeferredDataWriter
import pekko.http.javadsl.{ model => jmodel }
import pekko.http.scaladsl.model.HttpEntity.ChunkStreamPart
import pekko.http.scaladsl.model.{ ContentType, HttpHeader, HttpResponse,
MediaType, Trailer }
@@ -113,6 +114,15 @@ object AbstractGrpcProtocol {
})
.toContentType
+ def encodeFrameData[T](element: T, elementDataWriter: DeferredDataWriter[T],
isCompressed: Boolean,
+ isTrailer: Boolean): ByteString = {
+ val dataLength = elementDataWriter.serializedSize(element)
+ val frame = new Array[Byte](AbstractGrpcProtocol.FrameHeaderSize +
dataLength)
+ writeFrameHeader(frame, 0, dataLength, isCompressed, isTrailer)
+ elementDataWriter.serializeTo(element, frame,
AbstractGrpcProtocol.FrameHeaderSize)
+ ByteString.fromArrayUnsafe(frame)
+ }
+
def encodeFrameData(data: ByteString, isCompressed: Boolean, isTrailer:
Boolean): ByteString = {
val header = new Array[Byte](FrameHeaderSize)
writeFrameHeader(header, 0, data.length, isCompressed, isTrailer)
@@ -122,14 +132,13 @@ object AbstractGrpcProtocol {
def writer(
protocol: GrpcProtocol,
codec: Codec,
- encodeFrame: Frame => ChunkStreamPart,
- encodeDataToResponse: (ByteString, Seq[HttpHeader], Trailer) =>
HttpResponse): GrpcProtocolWriter =
+ encodeFrame: OutboundFrame => ChunkStreamPart,
+ encodeDataToResponse: (OutboundFrame, Seq[HttpHeader], Trailer) =>
HttpResponse): GrpcProtocolWriter =
GrpcProtocolWriter(
adjustCompressibility(protocol.contentType, codec),
codec,
encodeFrame,
- encodeDataToResponse,
- Flow[Frame].map(encodeFrame))
+ encodeDataToResponse)
/**
* The default maximum inbound message size (4 MiB), matching grpc-java's
default.
@@ -138,12 +147,12 @@ object AbstractGrpcProtocol {
def reader(
codec: Codec,
- decodeFrame: (Int, ByteString) => Frame,
+ decodeFrame: (Int, ByteString) => InboundFrame,
preDecodeStrict: ByteString => ByteString = null,
preDecodeFlow: Flow[ByteString, ByteString, NotUsed] = null,
maxInboundMessageSize: Int = DefaultMaxInboundMessageSize):
GrpcProtocolReader = {
val strictAdapter: ByteString => ByteString = if (preDecodeStrict eq null)
identity else preDecodeStrict
- val adapter: Flow[ByteString, Frame, NotUsed] => Flow[ByteString, Frame,
NotUsed] =
+ val adapter: Flow[ByteString, InboundFrame, NotUsed] => Flow[ByteString,
InboundFrame, NotUsed] =
if (preDecodeFlow eq null) identity
else x => Flow[ByteString].via(preDecodeFlow).via(x)
@@ -171,23 +180,23 @@ object AbstractGrpcProtocol {
adapter(Flow.fromGraph(new GrpcFramingDecoderStage(codec, decodeFrame,
maxInboundMessageSize))))
}
- class GrpcFramingDecoderStage(codec: Codec, deframe: (Int, ByteString) =>
Frame, maxInboundMessageSize: Int)
- extends ByteStringParser[Frame] {
+ class GrpcFramingDecoderStage(codec: Codec, deframe: (Int, ByteString) =>
InboundFrame, maxInboundMessageSize: Int)
+ extends ByteStringParser[InboundFrame] {
override def createLogic(inheritedAttributes: Attributes): GraphStageLogic
=
new ParsingLogic {
startWith(ReadFrameHeader)
- trait Step extends ParseStep[Frame]
+ trait Step extends ParseStep[InboundFrame]
// handle explicitly to avoid noisy log: a peer sending oversized or
malformed frame
// headers should not cost us a stack trace per attempt
- private def failWith(status: Status): ParseResult[Frame] = {
+ private def failWith(status: Status): ParseResult[InboundFrame] = {
failStage(new StatusException(status))
ParseResult(None, Failed)
}
object ReadFrameHeader extends Step {
- override def parse(reader: ByteReader): ParseResult[Frame] = {
+ override def parse(reader: ByteReader): ParseResult[InboundFrame] = {
val frameType = reader.readByte()
val length = reader.readIntBE()
@@ -205,7 +214,7 @@ object AbstractGrpcProtocol {
sealed case class ReadFrame(frameType: Int, length: Int) extends Step {
private val compression = (frameType & 0x01) == 1
- override def parse(reader: ByteReader): ParseResult[Frame] =
+ override def parse(reader: ByteReader): ParseResult[InboundFrame] =
try ParseResult(
Some(deframe(frameType, codec.uncompress(compression,
reader.take(length), maxInboundMessageSize))),
ReadFrameHeader)
@@ -217,7 +226,7 @@ object AbstractGrpcProtocol {
}
case object Failed extends Step {
- override def parse(reader: ByteReader): ParseResult[Frame] =
ParseResult(None, Failed)
+ override def parse(reader: ByteReader): ParseResult[InboundFrame] =
ParseResult(None, Failed)
}
}
}
diff --git
a/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcEntityHelpers.scala
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcEntityHelpers.scala
index d39f0364..d3aa213a 100644
---
a/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcEntityHelpers.scala
+++
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcEntityHelpers.scala
@@ -18,9 +18,9 @@ import pekko.NotUsed
import pekko.actor.{ ActorSystem, ClassicActorSystemProvider }
import pekko.annotation.InternalApi
import pekko.grpc.{ GrpcServiceException, ProtobufFrameSerializer,
ProtobufSerializer, Trailers }
-import pekko.grpc.GrpcProtocol.{ DataFrame, Frame, GrpcProtocolWriter,
TrailerFrame }
+import pekko.grpc.GrpcProtocol.{ DataFrame, DeferredDataFrame,
GrpcProtocolWriter, OutboundFrame, TrailerFrame }
import pekko.grpc.scaladsl.{ headers, BytesEntry, Metadata, MetadataEntry,
StringEntry }
-import pekko.http.scaladsl.model.HttpEntity.{ Chunk, ChunkStreamPart }
+import pekko.http.scaladsl.model.HttpEntity.ChunkStreamPart
import pekko.http.scaladsl.model.HttpHeader
import pekko.http.scaladsl.model.headers.RawHeader
import pekko.stream.scaladsl.Source
@@ -65,24 +65,17 @@ object GrpcEntityHelpers {
FlowShape(merge.in(0), merge.out)
}
- private def chunks[T](e: Source[T, NotUsed], trail: Source[Frame, NotUsed])(
+ private def chunks[T](e: Source[T, NotUsed], trail: Source[OutboundFrame,
NotUsed])(
implicit m: ProtobufSerializer[T],
writer: GrpcProtocolWriter): Source[ChunkStreamPart, NotUsed] =
- if ((writer.messageEncoding eq Identity) && writer.contentType ==
GrpcProtocolNative.contentType) {
- m match {
- case frameSerializer: ProtobufFrameSerializer[T @unchecked] =>
- e.map { msg => Chunk(frameSerializer.serializeDataFrame(msg)):
ChunkStreamPart }
- .via(concatCheap(trail.map(writer.encodeFrame)))
- case _ => frameEncodedChunks(e, trail)
- }
- } else {
- frameEncodedChunks(e, trail)
- }
+ e.map(msg =>
writer.encodeFrame(outboundDataFrame(msg))).via(concatCheap(trail.map(writer.encodeFrame)))
- private def frameEncodedChunks[T](e: Source[T, NotUsed], trail:
Source[Frame, NotUsed])(
- implicit m: ProtobufSerializer[T],
- writer: GrpcProtocolWriter): Source[ChunkStreamPart, NotUsed] =
- e.map { msg => DataFrame(m.serialize(msg))
}.via(concatCheap(trail)).via(writer.frameEncoder)
+ def outboundDataFrame[T](t: T)(implicit m: ProtobufSerializer[T]):
OutboundFrame = {
+ m match {
+ case fm: ProtobufFrameSerializer[T] => DeferredDataFrame(t, fm)
+ case _ => DataFrame(m.serialize(t))
+ }
+ }
def trailer(status: Status): TrailerFrame =
TrailerFrame(trailers = statusHeaders(status))
diff --git
a/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcProtocolNative.scala
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcProtocolNative.scala
index 97d3f16f..1ad3166f 100644
---
a/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcProtocolNative.scala
+++
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcProtocolNative.scala
@@ -49,21 +49,26 @@ object GrpcProtocolNative extends
AbstractGrpcProtocol("grpc") {
private def decodeFrame(@nowarn("msg=is never used") frameType: Int, data:
ByteString) = DataFrame(data)
@inline
- private def encodeFrame(codec: Codec, frame: Frame): ChunkStreamPart =
+ private def encodeFrame(codec: Codec, frame: OutboundFrame): ChunkStreamPart
=
frame match {
+ case ef @ DeferredDataFrame(element, dataWriter) =>
+ if (codec eq Identity)
+ Chunk(AbstractGrpcProtocol.encodeFrameData(element, dataWriter,
codec.isCompressed, isTrailer = false))
+ else
+ Chunk(AbstractGrpcProtocol.encodeFrameData(codec.compress(ef.data),
codec.isCompressed, isTrailer = false))
case DataFrame(data) =>
Chunk(AbstractGrpcProtocol.encodeFrameData(codec.compress(data),
codec.isCompressed, isTrailer = false))
case TrailerFrame(headers) => LastChunk(trailer = headers)
}
+
+ @inline
private def encodeDataToResponse(
- codec: Codec)(data: ByteString, headers: Seq[HttpHeader], trailer:
Trailer): HttpResponse =
+ codec: Codec)(frame: OutboundFrame, headers: Seq[HttpHeader], trailer:
Trailer): HttpResponse =
new HttpResponse(
status = StatusCodes.OK,
headers = headers,
- entity = HttpEntity(contentType, encodeDataToFrameBytes(codec, data)),
+ entity = HttpEntity(contentType, encodeFrame(codec, frame).data),
protocol = HttpProtocols.`HTTP/1.1`,
attributes = Map.empty[AttributeKey[?],
Any].updated(AttributeKeys.trailer, trailer))
- private def encodeDataToFrameBytes(codec: Codec, data: ByteString):
ByteString =
- AbstractGrpcProtocol.encodeFrameData(codec.compress(data),
codec.isCompressed, isTrailer = false)
}
diff --git
a/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcProtocolWeb.scala
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcProtocolWeb.scala
index a0f0a758..64d29d1a 100644
---
a/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcProtocolWeb.scala
+++
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcProtocolWeb.scala
@@ -34,26 +34,31 @@ abstract class GrpcProtocolWebBase(subType: String) extends
AbstractGrpcProtocol
override protected def reader(codec: Codec, maxInboundMessageSize: Int):
GrpcProtocolReader =
AbstractGrpcProtocol.reader(codec, decodeFrame, preDecodeStrict,
preDecodeFlow, maxInboundMessageSize)
- private def encodeFrame(codec: Codec, frame: Frame): ChunkStreamPart =
+ private def encodeFrame(codec: Codec, frame: OutboundFrame): ChunkStreamPart
=
Chunk(postEncode(encodeFrameToBytes(codec, frame)))
private def encodeDataToResponse(
- codec: Codec)(data: ByteString, headers: Seq[HttpHeader], trailer:
Trailer): HttpResponse =
+ codec: Codec)(frame: OutboundFrame, headers: Seq[HttpHeader], trailer:
Trailer): HttpResponse =
HttpResponse(
status = StatusCodes.OK,
headers = headers,
- entity = HttpEntity(contentType, encodeDataToFrameBytes(codec, data,
trailer)),
+ entity = HttpEntity(contentType, encodeDataToFrameBytes(codec, frame,
trailer)),
protocol = HttpProtocols.`HTTP/1.1`)
- private def encodeDataToFrameBytes(codec: Codec, data: ByteString, trailer:
Trailer): ByteString = {
+ private def encodeDataToFrameBytes(codec: Codec, frame: OutboundFrame,
trailer: Trailer): ByteString = {
val trailerData = encodeTrailerHeaders(trailer.headers.iterator)
val trailerFrame =
AbstractGrpcProtocol.encodeFrameData(codec.compress(trailerData),
codec.isCompressed, isTrailer = true)
- postEncode(encodeFrameToBytes(codec, DataFrame(data)) ++ trailerFrame)
+ postEncode(encodeFrameToBytes(codec, frame) ++ trailerFrame)
}
- private def encodeFrameToBytes(codec: Codec, frame: Frame): ByteString =
+ private def encodeFrameToBytes(codec: Codec, frame: OutboundFrame):
ByteString =
frame match {
+ case ef @ DeferredDataFrame(element, dataWriter) =>
+ if (codec eq Identity)
+ AbstractGrpcProtocol.encodeFrameData(element, dataWriter,
codec.isCompressed, isTrailer = false)
+ else
+ AbstractGrpcProtocol.encodeFrameData(codec.compress(ef.data),
codec.isCompressed, isTrailer = false)
case DataFrame(data) =>
AbstractGrpcProtocol.encodeFrameData(codec.compress(data),
codec.isCompressed, isTrailer = false)
case TrailerFrame(trailer) =>
@@ -63,7 +68,7 @@ abstract class GrpcProtocolWebBase(subType: String) extends
AbstractGrpcProtocol
isTrailer = true)
}
- private final def decodeFrame(frameHeader: Int, data: ByteString): Frame = {
+ private final def decodeFrame(frameHeader: Int, data: ByteString):
InboundFrame = {
(frameHeader & 0x80) match {
case 0 => DataFrame(data)
case 0x80 => TrailerFrame(decodeTrailer(data))
diff --git
a/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcResponseHelpers.scala
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcResponseHelpers.scala
index 0c5bfd4e..9da29529 100644
---
a/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcResponseHelpers.scala
+++
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcResponseHelpers.scala
@@ -19,22 +19,12 @@ import pekko.actor.{ ActorSystem,
ClassicActorSystemProvider }
import pekko.annotation.InternalApi
import pekko.grpc.GrpcProtocol.{ GrpcProtocolWriter, TrailerFrame }
import pekko.grpc.scaladsl.{ headers, GrpcExceptionHandler }
-import pekko.grpc.{ ProtobufFrameSerializer, ProtobufSerializer, Trailers }
+import pekko.grpc.{ ProtobufSerializer, Trailers }
import pekko.http.scaladsl.model.headers.{ Trailer => TrailerHeader }
import pekko.http.scaladsl.model.HttpEntity.ChunkStreamPart
-import pekko.http.scaladsl.model.{
- AttributeKey,
- AttributeKeys,
- HttpEntity,
- HttpHeader,
- HttpProtocols,
- HttpResponse,
- StatusCodes,
- Trailer
-}
+import pekko.http.scaladsl.model.{ HttpEntity, HttpHeader, HttpResponse,
Trailer }
import pekko.stream.Materializer
import pekko.stream.scaladsl.Source
-import pekko.util.ByteString
import io.grpc.Status
import scala.concurrent.{ ExecutionContext, Future }
@@ -49,8 +39,6 @@ import scala.util.control.NonFatal
object GrpcResponseHelpers {
private val TrailerOk = GrpcEntityHelpers.trailer(Status.OK)
private val TrailerOkAttribute = Trailer(TrailerOk.trailers)
- private val TrailerOkAttributes =
- Map.empty[AttributeKey[?], Any].updated(AttributeKeys.trailer,
TrailerOkAttribute)
private val IdentityResponseHeaders: Seq[HttpHeader] =
headers.`Message-Encoding`(Identity.name) :: Nil
@@ -72,16 +60,7 @@ object GrpcResponseHelpers {
system: ClassicActorSystemProvider): HttpResponse = {
val responseHeaders = responseHeadersFor(writer)
try {
- if ((writer.messageEncoding eq Identity) && writer.contentType ==
GrpcProtocolNative.contentType) {
- m match {
- case frameSerializer: ProtobufFrameSerializer[T @unchecked] =>
- nativeResponse(writer, frameSerializer.serializeDataFrame(e),
responseHeaders)
- case _ =>
- writer.encodeDataToResponse(m.serialize(e), responseHeaders,
TrailerOkAttribute)
- }
- } else {
- writer.encodeDataToResponse(m.serialize(e), responseHeaders,
TrailerOkAttribute)
- }
+ writer.encodeDataToResponse(GrpcEntityHelpers.outboundDataFrame(e),
responseHeaders, TrailerOkAttribute)
} catch {
case NonFatal(ex) => status(GrpcEntityHelpers.handleException(ex,
eHandler))
}
@@ -91,17 +70,6 @@ object GrpcResponseHelpers {
if (writer.messageEncoding eq Identity) IdentityResponseHeaders
else headers.`Message-Encoding`(writer.messageEncoding.name) :: Nil
- private def nativeResponse(
- writer: GrpcProtocolWriter,
- encodedData: ByteString,
- responseHeaders: Seq[HttpHeader]): HttpResponse =
- new HttpResponse(
- status = StatusCodes.OK,
- headers = responseHeaders,
- entity = HttpEntity(writer.contentType, encodedData),
- protocol = HttpProtocols.`HTTP/1.1`,
- attributes = TrailerOkAttributes)
-
def apply[T](e: Source[T, NotUsed], status: Future[Status])(
implicit m: ProtobufSerializer[T],
mat: Materializer,
diff --git
a/runtime/src/main/scala/org/apache/pekko/grpc/javadsl/GoogleProtobufSerializer.scala
b/runtime/src/main/scala/org/apache/pekko/grpc/javadsl/GoogleProtobufSerializer.scala
index c71b6ed1..60b46b90 100644
---
a/runtime/src/main/scala/org/apache/pekko/grpc/javadsl/GoogleProtobufSerializer.scala
+++
b/runtime/src/main/scala/org/apache/pekko/grpc/javadsl/GoogleProtobufSerializer.scala
@@ -16,7 +16,6 @@ package org.apache.pekko.grpc.javadsl
import org.apache.pekko
import pekko.annotation.ApiMayChange
import pekko.grpc.ProtobufFrameSerializer
-import pekko.grpc.internal.AbstractGrpcProtocol
import pekko.util.ByteString
import com.google.protobuf.CodedOutputStream
import com.google.protobuf.Parser
@@ -28,17 +27,16 @@ class GoogleProtobufSerializer[T <:
com.google.protobuf.Message](parser: Parser[
override def serialize(t: T): ByteString =
ByteString.fromArrayUnsafe(t.toByteArray)
- override private[grpc] def serializeDataFrame(t: T): ByteString = {
- val dataLength = t.getSerializedSize
- val frame = new Array[Byte](AbstractGrpcProtocol.FrameHeaderSize +
dataLength)
- AbstractGrpcProtocol.writeFrameHeader(frame, 0, dataLength, isCompressed =
false, isTrailer = false)
- val output = CodedOutputStream.newInstance(frame,
AbstractGrpcProtocol.FrameHeaderSize, dataLength)
+ override private[grpc] def serializedSize(t: T): Int = t.getSerializedSize
+
+ override private[grpc] def serializeTo(t: T, frame: Array[Byte], offset:
Int): Unit = {
+ val dataLength = t.getSerializedSize
+ val output = CodedOutputStream.newInstance(frame, offset, dataLength)
t.writeTo(output)
output.checkNoSpaceLeft()
-
- ByteString.fromArrayUnsafe(frame)
}
+
override def deserialize(bytes: ByteString): T = {
val inputStream = bytes.asInputStream
try parser.parseFrom(inputStream)
diff --git
a/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/ScalapbProtobufSerializer.scala
b/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/ScalapbProtobufSerializer.scala
index a9884852..142f1ca8 100644
---
a/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/ScalapbProtobufSerializer.scala
+++
b/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/ScalapbProtobufSerializer.scala
@@ -16,7 +16,6 @@ package org.apache.pekko.grpc.scaladsl
import org.apache.pekko
import pekko.annotation.ApiMayChange
import pekko.grpc.ProtobufFrameSerializer
-import pekko.grpc.internal.AbstractGrpcProtocol
import pekko.util.ByteString
import com.google.protobuf.CodedOutputStream
import com.google.protobuf.CodedInputStream
@@ -25,21 +24,20 @@ import scalapb.{ GeneratedMessage,
GeneratedMessageCompanion }
import java.io.InputStream
@ApiMayChange
-class ScalapbProtobufSerializer[T <: GeneratedMessage](companion:
GeneratedMessageCompanion[T])
- extends ProtobufFrameSerializer[T] {
+class ScalapbProtobufSerializer[T <: GeneratedMessage](
+ companion: GeneratedMessageCompanion[T]) extends
ProtobufFrameSerializer[T] {
override def serialize(t: T): ByteString =
ByteString.fromArrayUnsafe(t.toByteArray)
- override private[grpc] def serializeDataFrame(t: T): ByteString = {
- val dataLength = t.serializedSize
- val frame = new Array[Byte](AbstractGrpcProtocol.FrameHeaderSize +
dataLength)
- AbstractGrpcProtocol.writeFrameHeader(frame, 0, dataLength, isCompressed =
false, isTrailer = false)
- val output = CodedOutputStream.newInstance(frame,
AbstractGrpcProtocol.FrameHeaderSize, dataLength)
+ override private[grpc] def serializedSize(t: T): Int = t.serializedSize
+
+ override private[grpc] def serializeTo(t: T, frame: Array[Byte], offset:
Int): Unit = {
+ val dataLength = t.serializedSize
+ val output = CodedOutputStream.newInstance(frame, offset, dataLength)
t.writeTo(output)
output.checkNoSpaceLeft()
-
- ByteString.fromArrayUnsafe(frame)
}
+
override def deserialize(bytes: ByteString): T =
companion.parseFrom(CodedInputStream.newInstance(bytes.asByteBuffer))
override def deserialize(data: InputStream): T =
diff --git
a/runtime/src/test/scala/org/apache/pekko/grpc/javadsl/GoogleProtobufSerializerSpec.scala
b/runtime/src/test/scala/org/apache/pekko/grpc/javadsl/GoogleProtobufSerializerSpec.scala
index 212c0824..f5c9386b 100644
---
a/runtime/src/test/scala/org/apache/pekko/grpc/javadsl/GoogleProtobufSerializerSpec.scala
+++
b/runtime/src/test/scala/org/apache/pekko/grpc/javadsl/GoogleProtobufSerializerSpec.scala
@@ -27,5 +27,19 @@ class GoogleProtobufSerializerSpec extends AnyWordSpec with
Matchers {
val deserialized = anySerializer.deserialize(serialized)
deserialized should be(obj)
}
+
+ "serialize into buffer at offset" in {
+ val anySerializer = new GoogleProtobufSerializer(ProtobufAny.parser())
+
+ val obj =
ProtobufAny.newBuilder().setTypeUrl("asdf").setValue(ByteString.copyFromUtf8("ASDF")).build()
+ val serialized = anySerializer.serialize(obj).toByteBuffer.array()
+
+ val prefixLength = 27
+ val frame = new Array[Byte](serialized.length + prefixLength)
+ anySerializer.serializeTo(obj, frame, prefixLength)
+
+ (0 until prefixLength).foreach(frame(_) shouldBe 0)
+ (prefixLength until frame.length).foreach(i => frame(i) shouldBe
serialized(i - prefixLength))
+ }
}
}
diff --git
a/runtime/src/test/scala/org/apache/pekko/grpc/scaladsl/ScalapbProtobufSerializerSpec.scala
b/runtime/src/test/scala/org/apache/pekko/grpc/scaladsl/ScalapbProtobufSerializerSpec.scala
index 95150340..acf4c74b 100644
---
a/runtime/src/test/scala/org/apache/pekko/grpc/scaladsl/ScalapbProtobufSerializerSpec.scala
+++
b/runtime/src/test/scala/org/apache/pekko/grpc/scaladsl/ScalapbProtobufSerializerSpec.scala
@@ -28,5 +28,20 @@ class ScalapbProtobufSerializerSpec extends AnyWordSpec with
Matchers {
val deserialized = anySerializer.deserialize(serialized)
deserialized should be(obj)
}
+
+ "serialize into buffer at offset" in {
+ val anySerializer = new ScalapbProtobufSerializer(ScalapbAny)
+
+ val obj = ScalapbAny("asdf", ByteString.copyFromUtf8("ASDF"))
+ val serialized = anySerializer.serialize(obj).toByteBuffer.array()
+
+ val prefixLength = 27
+ val frame = new Array[Byte](serialized.length + prefixLength)
+ anySerializer.serializeTo(obj, frame, prefixLength)
+
+ (0 until prefixLength).foreach(frame(_) shouldBe 0)
+ (prefixLength until frame.length).foreach(i => frame(i) shouldBe
serialized(i - prefixLength))
+ }
+
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
