feat: add adaptive gzip compression for gRPC responses

codegen/src/main/scala/org/apache/pekko/grpc/gen/scaladsl/Method.scala

@@ -66,6 +66,8 @@ case class Method(
     else if (ReservedMethodNames.contains(name)) s"$name$ReservedMethodNameSuffix"
     else name
 
+  val handlerName: String = s"${name}Handler"
+
 }
 
 object Method {

codegen/src/main/twirl/templates/ScalaServer/Handler.scala.txt

@@ -107,29 +107,46 @@ object @{serviceName}Handler {
           pekko.grpc.scaladsl.ServerReflection.partial(List(@{service.name})))
 }
 
-    private def methodName(request: model.HttpRequest, prefix: String): String =
+    private def methodName(request: model.HttpRequest, prefix: String): String = {
+      // Use Uri.Path structure matching instead of toString to avoid
+      // renderPath + StringBuilder + String allocations per request.
+      // gRPC path format: /{service}/{method}
       request.uri.path match {
-        case model.Uri.Path.Slash(model.Uri.Path.Segment(`prefix`, model.Uri.Path.Slash(model.Uri.Path.Segment(method, model.Uri.Path.Empty)))) =>
-          method
-        case _ =>
-          null
+        case model.Uri.Path.Slash(
+          model.Uri.Path.Segment(serviceName,
+            model.Uri.Path.Slash(
+              model.Uri.Path.Segment(method, model.Uri.Path.Empty))))
+          if serviceName == prefix => method
+        case _ => null
       }
+    }
 
     private def handler(implementation: @serviceName, prefix: String, eHandler: ActorSystem => PartialFunction[Throwable, Trailers])(implicit system: ClassicActorSystemProvider): model.HttpRequest => scala.concurrent.Future[model.HttpResponse] = {
       implicit val mat: Materializer = SystemMaterializer(system).materializer
       implicit val ec: ExecutionContext = mat.executionContext
       val spi = TelemetryExtension(system).spi
+      val compressionThreshold = system.classicSystem.settings.config.getInt("pekko.grpc.server.compression-threshold")
 
       import @{service.name}.Serializers.@{service.scalaCompatConstants.WildcardImport}
 
+      @if(!powerApis) {
+      // Pre-create lambdas for unary methods to avoid per-request allocation
+      @for(method <- service.methods) {
+      @if(method.methodType == org.apache.pekko.grpc.gen.Unary) {
+      val @{method.handlerName}: @method.parameterType => scala.concurrent.Future[@method.outputTypeUnboxed] =
+        (e: @method.parameterType) => implementation.@{method.nameSafe}(e@{if(powerApis) { ", metadata" } else { "" }})
+      }
+      }
+      }
+
       def handle(request: model.HttpRequest, method: String): scala.concurrent.Future[model.HttpResponse] =
         GrpcMarshalling.negotiated(request, (reader, writer) =>
           method match {
             @for(method <- service.methods) {
             case "@method.grpcName" =>
                 @{if(powerApis) { "val metadata = MetadataBuilder.fromHeaders(request.headers)" } else { "" }}
                 @if(method.methodType == org.apache.pekko.grpc.gen.Unary) {
-                GrpcMarshalling.handleUnary(request.entity, (e: @method.parameterType) => implementation.@{method.nameSafe}(e@{if(powerApis) { ", metadata" } else { "" }}), eHandler)(
+                GrpcMarshalling.handleUnary(request.entity, @{if(powerApis) { s"(e: ${method.parameterType}) => implementation.${method.nameSafe}(e, metadata)" } else method.handlerName}, eHandler)(
                   @{service.scalaCompatConstants.ImplicitUsing}@method.deserializer.name, @method.serializer.name, mat, reader, writer, system, ec)
                 } else {
                 @{method.unmarshal}(request.entity)(@{service.scalaCompatConstants.ImplicitUsing}@method.deserializer.name, mat, reader)
@@ -140,7 +157,7 @@ object @{serviceName}Handler {
             }
             case m => GrpcExceptionHandler.from(eHandler(system.classicSystem))(system, writer)(new NotImplementedError(s"Not implemented: $m"))
           }
-      ).getOrElse(unsupportedMediaType)
+      , compressionThreshold).getOrElse(unsupportedMediaType)
 
       request => {
         val method = methodName(request, prefix)
@@ -162,17 +179,28 @@ object @{serviceName}Handler {
       implicit val mat: Materializer = SystemMaterializer(system).materializer
       implicit val ec: ExecutionContext = mat.executionContext
       val spi = TelemetryExtension(system).spi
+      val compressionThreshold = system.classicSystem.settings.config.getInt("pekko.grpc.server.compression-threshold")
 
       import @{service.name}.Serializers.@{service.scalaCompatConstants.WildcardImport}
 
+      @if(!powerApis) {
+      // Pre-create lambdas for unary methods to avoid per-request allocation
+      @for(method <- service.methods) {
+      @if(method.methodType == org.apache.pekko.grpc.gen.Unary) {
+      val @{method.handlerName}: @method.parameterType => scala.concurrent.Future[@method.outputTypeUnboxed] =
+        (e: @method.parameterType) => implementation.@{method.nameSafe}(e@{if(powerApis) { ", metadata" } else { "" }})
+      }
+      }
+      }
+
       def handle(request: model.HttpRequest, method: String): scala.concurrent.Future[model.HttpResponse] =
         GrpcMarshalling.negotiated(request, (reader, writer) =>
           method match {
             @for(method <- service.methods) {
             case "@method.grpcName" =>
                 @{if(powerApis) { "val metadata = MetadataBuilder.fromHeaders(request.headers)" } else { "" }}
                 @if(method.methodType == org.apache.pekko.grpc.gen.Unary) {
-                GrpcMarshalling.handleUnary(request.entity, (e: @method.parameterType) => implementation.@{method.nameSafe}(e@{if(powerApis) { ", metadata" } else { "" }}), eHandler)(
+                GrpcMarshalling.handleUnary(request.entity, @{if(powerApis) { s"(e: ${method.parameterType}) => implementation.${method.nameSafe}(e, metadata)" } else method.handlerName}, eHandler)(
                   @{service.scalaCompatConstants.ImplicitUsing}@method.deserializer.name, @method.serializer.name, mat, reader, writer, system, ec)
                 } else {
                 @{method.unmarshal}(request.entity)(@{service.scalaCompatConstants.ImplicitUsing}@method.deserializer.name, mat, reader)
@@ -183,7 +211,7 @@ object @{serviceName}Handler {
             }
             case m => GrpcExceptionHandler.from(eHandler(system.classicSystem))(system, writer)(new NotImplementedError(s"Not implemented: $m"))
           }
-      ).getOrElse(unsupportedMediaType)
+      , compressionThreshold).getOrElse(unsupportedMediaType)
 
       Function.unlift((req: model.HttpRequest) => {
         val method = methodName(req, prefix)

runtime/src/main/resources/reference.conf

@@ -1,56 +1,69 @@
 # SPDX-License-Identifier: Apache-2.0
 
 //#defaults
-pekko.grpc.client."*" {
-  # netty or pekko-http (experimental)
-  backend = "netty"
-
-  # Host to use if service-discovery-mechanism is set to static or grpc-dns
-  host = ""
-
-  service-discovery {
-    mechanism = "static"
-    # Service name to use if a service-discovery.mechanism other than static or grpc-dns
-    service-name = ""
-    # See https://pekko.apache.org/docs/pekko-management/current/discovery/index.html for meanings for each mechanism
-    # if blank then not passed to the lookup
-    port-name = ""
-    protocol = ""
-
-    # timeout for service discovery resolving
-    resolve-timeout = 1s
+pekko.grpc {
+  # Server-side configuration for gRPC services
+  server {
+    # Adaptive compression: messages smaller than this threshold (in bytes)
+    # are sent uncompressed, while larger messages are compressed with gzip.
+    # This avoids the CPU overhead of gzip for small messages where the
+    # bandwidth savings are negligible.
+    # Set to 0 to always compress (every message goes through gzip).
+    # Set to a very large value (e.g. 2147483647) to effectively disable compression.
+    compression-threshold = 1024
   }
 
-  # port to use if service-discovery-mechanism is static or service discovery does not return a port
-  port = 0
-
-  # Experimental in grpc-java https://github.com/grpc/grpc-java/issues/1771
-  # pick_first or round_robin
-  load-balancing-policy = ""
-
-  deadline = infinite
-  override-authority = ""
-  user-agent = ""
-  # Location on the classpath of CA PEM to trust
-  trusted = ""
-  use-tls = true
-  # SSL provider to use:
-  # leave empty to auto-detect, or configure 'jdk' or 'openssl'.
-  ssl-provider = ""
-
-  # TODO: Enforce HTTP/2 TLS restrictions: https://tools.ietf.org/html/draft-ietf-httpbis-http2-17#section-9.2
-
-  # The number of times to try connecting before giving up.
-  # '-1': means retry indefinitely, '0' is invalid, '1' means fail
-  # after the first failed attempt.
-  # When load balancing we don't count individual connection
-  # failures, so in that case any value larger than '1' is also
-  # interpreted as retrying 'indefinitely'.
-  connection-attempts = 20
-
-  # Service discovery mechanism to use. The default is to use a static host
-  # and port that will be resolved via DNS.
-  # Any of the mechanisms described in https://pekko.apache.org/docs/pekko-management/current/discovery/index.html can be used
-  # including Kubernetes, Consul, AWS API
+  client."*" {
+    # netty or pekko-http (experimental)
+    backend = "netty"
+
+    # Host to use if service-discovery-mechanism is set to static or grpc-dns
+    host = ""
+
+    service-discovery {
+      mechanism = "static"
+      # Service name to use if a service-discovery.mechanism other than static or grpc-dns
+      service-name = ""
+      # See https://pekko.apache.org/docs/pekko-management/current/discovery/index.html for meanings for each mechanism
+      # if blank then not passed to the lookup
+      port-name = ""
+      protocol = ""
+
+      # timeout for service discovery resolving
+      resolve-timeout = 1s
+    }
+
+    # port to use if service-discovery-mechanism is static or service discovery does not return a port
+    port = 0
+
+    # Experimental in grpc-java https://github.com/grpc/grpc-java/issues/1771
+    # pick_first or round_robin
+    load-balancing-policy = ""
+
+    deadline = infinite
+    override-authority = ""
+    user-agent = ""
+    # Location on the classpath of CA PEM to trust
+    trusted = ""
+    use-tls = true
+    # SSL provider to use:
+    # leave empty to auto-detect, or configure 'jdk' or 'openssl'.
+    ssl-provider = ""
+
+    # TODO: Enforce HTTP/2 TLS restrictions: https://tools.ietf.org/html/draft-ietf-httpbis-http2-17#section-9.2
+
+    # The number of times to try connecting before giving up.
+    # '-1': means retry indefinitely, '0' is invalid, '1' means fail
+    # after the first failed attempt.
+    # When load balancing we don't count individual connection
+    # failures, so in that case any value larger than '1' is also
+    # interpreted as retrying 'indefinitely'.
+    connection-attempts = 20
+
+    # Service discovery mechanism to use. The default is to use a static host
+    # and port that will be resolved via DNS.
+    # Any of the mechanisms described in https://pekko.apache.org/docs/pekko-management/current/discovery/index.html can be used
+    # including Kubernetes, Consul, AWS API
+  }
 }
 //#defaults

runtime/src/main/scala/org/apache/pekko/grpc/GrpcProtocol.scala

@@ -18,7 +18,16 @@ import pekko.NotUsed
 import pekko.annotation.InternalApi
 import pekko.annotation.InternalStableApi
 import pekko.grpc.GrpcProtocol.{ GrpcProtocolReader, GrpcProtocolWriter }
-import pekko.grpc.internal.{ Codec, Codecs, GrpcProtocolNative, GrpcProtocolWeb, GrpcProtocolWebText }
+import pekko.grpc.internal.{
+  AdaptiveGzip,
+  Codec,
+  Codecs,
+  GrpcProtocolNative,
+  GrpcProtocolWeb,
+  GrpcProtocolWebText,
+  Gzip,
+  Identity
+}
 import pekko.http.javadsl.{ model => jmodel }
 import pekko.http.scaladsl.model.{ ContentType, HttpHeader, HttpResponse, Trailer }
 import pekko.http.scaladsl.model.HttpEntity.ChunkStreamPart
@@ -141,15 +150,99 @@ object GrpcProtocol {
     case _                                       => None
   }
 
+  // Pre-computed negotiation result for Identity reader + Identity writer (no gzip support)
+  private val NativeIdentityIdentity: (Try[GrpcProtocolReader], GrpcProtocolWriter) =
+    (scala.util.Success(GrpcProtocolNative.newReader(Identity)), GrpcProtocolNative.newWriter(Identity))
+
+  // Pre-computed negotiation result for Identity reader + AdaptiveGzip writer (default threshold)
+  // This is the most common case: client accepts gzip, server uses adaptive compression.
+  private val NativeIdentityAdaptiveGzip: (Try[GrpcProtocolReader], GrpcProtocolWriter) =
+    (scala.util.Success(GrpcProtocolNative.newReader(Identity)),
+      GrpcProtocolNative.newWriter(AdaptiveGzip(AdaptiveGzip.DefaultCompressionThreshold)))
+
   /**
    * Calculates the gRPC protocol encoding to use for an interaction with a gRPC client.
+   * Optimized with a fast path for native gRPC that does a single header scan.
+   *
+   * For response encoding, uses adaptive gzip: messages above a configurable threshold
+   * (default 1KB) are compressed with gzip, while smaller messages are sent uncompressed.
+   * The gRPC per-frame compression flag (bit 0 of the 5-byte frame header) tells the
+   * client whether each individual frame is compressed.
    *
    * @param request the client request to respond to.
    * @return the protocol reader for the request, and a protocol writer for the response.
    */
   def negotiate(request: jmodel.HttpRequest): Option[(Try[GrpcProtocolReader], GrpcProtocolWriter)] =
-    detect(request).map { variant =>
-      (Codecs.detect(request).map(variant.newReader), variant.newWriter(Codecs.negotiate(request)))
+    negotiate(request, AdaptiveGzip.DefaultCompressionThreshold)
+
+  /**
+   * Calculates the gRPC protocol encoding to use for an interaction with a gRPC client.
+   * Optimized with a fast path for native gRPC that does a single header scan.
+   *
+   * @param request the client request to respond to.
+   * @param compressionThreshold minimum message size in bytes to trigger gzip compression.
+   *                             Messages below this threshold are sent uncompressed.
+   * @return the protocol reader for the request, and a protocol writer for the response.
+   */
+  def negotiate(
+      request: jmodel.HttpRequest,
+      compressionThreshold: Int): Option[(Try[GrpcProtocolReader], GrpcProtocolWriter)] = {
+    val mediaType = request.entity.getContentType.mediaType
+    val subType = mediaType.subType
+    val isNative = (subType == "grpc+proto") || (subType == "grpc")
+
+    if (isNative) {
+      // Single-pass header scan for native gRPC
+      var requestEncoding: String = null
+      var acceptEncoding: String = null
+      request match {
+        case sReq: pekko.http.scaladsl.model.HttpMessage =>
+          val headers = sReq.headers
+          var i = 0
+          while (i < headers.size) {
+            val h = headers(i)
+            val name = h.lowercaseName
+            if (name == "grpc-encoding") requestEncoding = h.value
+            else if (name == "grpc-accept-encoding") acceptEncoding = h.value
+            i += 1
+          }
+        case _ =>
+      }
+      // Determine reader codec (request encoding)
+      val readerCodec: Codec =
+        if (requestEncoding eq null) Identity
+        else if (requestEncoding == "identity") Identity
+        else if (requestEncoding == "gzip") Gzip
+        else return slowNegotiateOpt(request, subType)
+      // Determine writer codec (response encoding)
+      // Adaptive compression: use gzip for messages above threshold, identity for smaller ones.
+      // Only activate when client advertises gzip support via grpc-accept-encoding.
+      val clientAcceptsGzip = acceptEncoding != null && acceptEncoding.contains("gzip")
+      if (!clientAcceptsGzip && (readerCodec eq Identity)) return Some(NativeIdentityIdentity)
+      if (clientAcceptsGzip && (readerCodec eq Identity) &&
+        compressionThreshold == AdaptiveGzip.DefaultCompressionThreshold)
+        return Some(NativeIdentityAdaptiveGzip)
+      val writerCodec: Codec =
+        if (clientAcceptsGzip) AdaptiveGzip(compressionThreshold)
+        else Identity
+      return Some(
+        (scala.util.Success(GrpcProtocolNative.newReader(readerCodec)),
+          GrpcProtocolNative.newWriter(writerCodec)))
+    }
+
+    // Non-native protocols - slow path
+    slowNegotiateOpt(request, subType)
+  }
+
+  private def slowNegotiateOpt(request: jmodel.HttpRequest, subType: String)
+      : Option[(Try[GrpcProtocolReader], GrpcProtocolWriter)] = {
+    val variant = subType match {
+      case "grpc-web" | "grpc-web+proto"           => Some(GrpcProtocolWeb)
+      case "grpc-web-text" | "grpc-web-text+proto" => Some(GrpcProtocolWebText)
+      case "grpc" | "grpc+proto"                   => Some(GrpcProtocolNative)
+      case _                                       => None
     }
+    variant.map(v => (Codecs.detect(request).map(v.newReader), v.newWriter(Codecs.negotiate(request))))
+  }
 
 }

runtime/src/main/scala/org/apache/pekko/grpc/ProtobufSerializer.scala

@@ -27,4 +27,22 @@ trait ProtobufSerializer[T] {
 
 private[grpc] trait ProtobufFrameSerializer[T] extends ProtobufSerializer[T] {
   private[grpc] def serializeDataFrame(t: T): ByteString
+
+  /**
+   * Deserialize a protobuf message from a ByteString starting at the given offset.
+   * Avoids ByteString.slice allocation by passing the offset directly to the parser.
+   * Default implementation falls back to slice + deserialize.
+   */
+  private[grpc] def deserialize(data: ByteString, offset: Int, length: Int): T =
+    deserialize(data.slice(offset, offset + length))
+
+  /**
+   * Returns the serialized size of the message without actually serializing it.
+   * Used by adaptive compression to decide whether to compress small messages.
+   *
+   * Implementations should override this with an efficient method (e.g.
+   * `GeneratedMessage.serializedSize` for ScalaPB or `Message.getSerializedSize`
+   * for Google protobuf) to avoid the overhead of full serialization.
+   */
+  private[grpc] def serializedDataSize(t: T): Int = serialize(t).length
 }

runtime/src/main/scala/org/apache/pekko/grpc/internal/AbstractGrpcProtocol.scala

@@ -100,10 +100,19 @@ object AbstractGrpcProtocol {
       })
       .toContentType
 
+  private val SmallMessageThreshold = 4096
+
   def encodeFrameData(data: ByteString, isCompressed: Boolean, isTrailer: Boolean): ByteString = {
-    val header = new Array[Byte](FrameHeaderSize)
-    writeFrameHeader(header, 0, data.length, isCompressed, isTrailer)
-    ByteString.fromArrayUnsafe(header, 0, FrameHeaderSize) ++ data
+    if (data.length <= SmallMessageThreshold) {
+      val combined = new Array[Byte](FrameHeaderSize + data.length)
+      writeFrameHeader(combined, 0, data.length, isCompressed, isTrailer)
+      data.copyToArray(combined, FrameHeaderSize)
+      ByteString.fromArrayUnsafe(combined)
+    } else {
+      val header = new Array[Byte](FrameHeaderSize)
+      writeFrameHeader(header, 0, data.length, isCompressed, isTrailer)
+      ByteString.fromArrayUnsafe(header, 0, FrameHeaderSize) ++ data
+    }
   }
 
   def writer(