(pekko-grpc) branch main updated: pre-announce trailers in server streaming calls (#456)

[fa...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

fanningpj 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 3c51f0c0 pre-announce trailers in server streaming calls (#456)
3c51f0c0 is described below

commit 3c51f0c009f6b07470e5c1ecb06951e73b749572
Author: navid <na...@moia.io>
AuthorDate: Wed May 28 10:36:41 2025 +0200

    pre-announce trailers in server streaming calls (#456)
    
    * preannounce trailers in server streaming calls
    
    * use ArraySeq.unsafeWrapArray
    
    * add test for checking `trailer` is part of metadata, make `trailer` header private to scaladsl, add comment linking to rfc
    
    * make trailer visible to `grpc` package, make test simpler, add comment on top of test
    
    * extend comment on top of response fn
    
    * move test into runtime module
    
    * copy paste header from somewhere else
    
    * generate header with sbt instead
---
 .../apache/pekko/grpc/scaladsl/PowerApiSpec.scala  |  1 -
 .../pekko/grpc/internal/GrpcResponseHelpers.scala  |  6 ++-
 .../pekko/grpc/scaladsl/headers/headers.scala      | 27 +++++++++++
 .../grpc/internal/GrpcResponseHelpersSpec.scala    | 53 ++++++++++++++++++++++
 4 files changed, 85 insertions(+), 2 deletions(-)

diff --git a/interop-tests/src/test/scala/org/apache/pekko/grpc/scaladsl/PowerApiSpec.scala b/interop-tests/src/test/scala/org/apache/pekko/grpc/scaladsl/PowerApiSpec.scala
index 9a38b1bf..c1560c4a 100644
--- a/interop-tests/src/test/scala/org/apache/pekko/grpc/scaladsl/PowerApiSpec.scala
+++ b/interop-tests/src/test/scala/org/apache/pekko/grpc/scaladsl/PowerApiSpec.scala
@@ -183,5 +183,4 @@ abstract class PowerApiSpec(backend: String)
       metadataServer.terminate(3.seconds)
     }
   }
-
 }
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 848fc629..58fef5a6 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
@@ -97,7 +97,11 @@ object GrpcResponseHelpers {
 
   private def response(entity: Source[ChunkStreamPart, NotUsed])(implicit writer: GrpcProtocolWriter) = {
     HttpResponse(
-      headers = immutable.Seq(headers.`Message-Encoding`(writer.messageEncoding.name)),
+      headers = immutable.Seq(
+        headers.`Message-Encoding`(writer.messageEncoding.name),
+        // Pre-announcing trailers: See https://www.rfc-editor.org/rfc/rfc7230 #Section 4.4
+        headers.`Trailer`(headers.`Status`.name)
+      ),
       entity = HttpEntity.Chunked(writer.contentType, entity))
   }
 
diff --git a/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/headers/headers.scala b/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/headers/headers.scala
index 86c09741..d607b943 100644
--- a/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/headers/headers.scala
+++ b/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/headers/headers.scala
@@ -19,6 +19,7 @@ import pekko.http.scaladsl.model.HttpHeader
 import pekko.http.scaladsl.model.headers.{ ModeledCustomHeader, ModeledCustomHeaderCompanion }
 import pekko.http.javadsl.{ model => jm }
 
+import scala.collection.compat.immutable.ArraySeq
 import scala.collection.immutable
 import scala.util.Try
 
@@ -111,3 +112,29 @@ object `Status-Message` extends ModeledCustomHeaderCompanion[`Status-Message`] {
   def findIn(headers: immutable.Seq[HttpHeader]): Option[String] =
     headers.collectFirst { case h if h.is(name) => h.value() }
 }
+
+private[grpc] final class `Trailer` private (values: immutable.Seq[String]) extends ModeledCustomHeader[`Trailer`] {
+
+  override def companion: ModeledCustomHeaderCompanion[`Trailer`] = `Trailer`
+
+  override def value(): String = values.mkString(", ")
+
+  override def renderInRequests(): Boolean = true
+
+  override def renderInResponses(): Boolean = true
+}
+
+private[grpc] object `Trailer` extends ModeledCustomHeaderCompanion[`Trailer`] {
+  def apply(values: immutable.Seq[String]): `Trailer` = new `Trailer`(values.map(_.trim))
+
+  override val name = "trailer"
+
+  override val lowercaseName: String = super.lowercaseName
+
+  override def parse(value: String): Try[`Trailer`] = Try(`Trailer`(ArraySeq.unsafeWrapArray(value.split(','))))
+
+  def findIn(headers: immutable.Seq[HttpHeader]): Option[immutable.Seq[String]] =
+    headers.collectFirst {
+      case header if header.is(name) => ArraySeq.unsafeWrapArray(header.value().split(',').map(_.trim))
+    }
+}
diff --git a/runtime/src/test/scala/org/apache/pekko/grpc/internal/GrpcResponseHelpersSpec.scala b/runtime/src/test/scala/org/apache/pekko/grpc/internal/GrpcResponseHelpersSpec.scala
new file mode 100644
index 00000000..e45b3a50
--- /dev/null
+++ b/runtime/src/test/scala/org/apache/pekko/grpc/internal/GrpcResponseHelpersSpec.scala
@@ -0,0 +1,53 @@
+/*
+ * 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.internal
+
+import grpc.reflection.v1alpha.reflection.{ ServerReflection, ServerReflectionResponse }
+import io.grpc.Status
+import org.apache.pekko.actor.ActorSystem
+import org.apache.pekko.grpc.GrpcProtocol
+import org.apache.pekko.grpc.scaladsl.{ headers, ScalapbProtobufSerializer }
+import org.apache.pekko.stream.scaladsl.Source
+import org.apache.pekko.testkit.TestKit
+import org.scalatest.matchers.should.Matchers._
+import org.scalatest.wordspec.AnyWordSpecLike
+
+final class GrpcResponseHelpersSpec extends TestKit(ActorSystem("GrpcResponseHelpersSpec")) with AnyWordSpecLike {
+  "GrpcResponseHelpers" should {
+
+    /**
+     * Pre-announcing trailers: See https://www.rfc-editor.org/rfc/rfc7230 #Section 4.4
+     * Certain reverse proxies (e.g. tyk) will not behave correctly if trailers are not pre-announced and
+     * errors are not reported with trailer-only optimization.
+     */
+    "pre-announce trailers in the headers" in {
+      implicit val serializer: ScalapbProtobufSerializer[ServerReflectionResponse] =
+        ServerReflection.Serializers.ServerReflectionResponseSerializer
+      implicit val writer: GrpcProtocol.GrpcProtocolWriter = GrpcProtocolNative.newWriter(Identity)
+      val response =
+        GrpcResponseHelpers(
+          e = Source.single(ServerReflectionResponse()),
+          trail = Source.single(GrpcEntityHelpers.trailer(Status.OK))
+        )
+
+      val preAnnouncedTrailers = headers.`Trailer`.findIn(response.headers)
+
+      preAnnouncedTrailers shouldBe Some(Seq(headers.`Status`.name))
+    }
+  }
+}


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org