Introduce separate type for Any's in error details. Split codec classes into files

Author: igor-vovk

conformance/src/main/scala/org/ivovk/connect_rpc_scala/conformance/ConformanceServiceImpl.scala

@@ -69,7 +69,7 @@ class ConformanceServiceImpl[F[_] : Async] extends ConformanceServiceFs2GrpcTrai
         val status = Status.fromCodeValue(code.value)
           .withDescription(message.orNull)
           .augmentDescription(
-            TextFormat.printToSingleLineUnicodeString(requestInfo.toProtoErrorAny)
+            TextFormat.printToSingleLineUnicodeString(requestInfo.toProtoErrorDetailsAny)
           )
 
         throw new StatusRuntimeException(status)

conformance/src/main/scala/org/ivovk/connect_rpc_scala/conformance/Main.scala

@@ -5,7 +5,7 @@ import com.comcast.ip4s.{Port, host, port}
 import connectrpc.conformance.v1.{ConformanceServiceFs2GrpcTrailers, ServerCompatRequest, ServerCompatResponse}
 import org.http4s.ember.server.EmberServerBuilder
 import org.ivovk.connect_rpc_scala.ConnectRouteBuilder
-import scalapb.json4s.TypeRegistry
+import org.ivovk.connect_rpc_scala.http.codec.JsonMessageCodecBuilder
 
 import java.io.InputStream
 import java.nio.ByteBuffer
@@ -36,19 +36,21 @@ object Main extends IOApp.Simple {
         ConformanceServiceImpl[IO]()
       )
 
-      app <- ConnectRouteBuilder.forService[IO](service)
+      jsonCodec = JsonMessageCodecBuilder[IO]()
         // Registering message types in TypeRegistry is required to pass com.google.protobuf.any.Any
         // JSON-serialization conformance tests
-        .withJsonPrinterConfigurator { p =>
-          p.withTypeRegistry(
-            TypeRegistry.default
-              .addMessage[connectrpc.conformance.v1.UnaryRequest]
-              .addMessage[connectrpc.conformance.v1.IdempotentUnaryRequest]
-              .addMessage[connectrpc.conformance.v1.ConformancePayload.RequestInfo]
-          )
+        .withTypeRegistryConfigurator { tp =>
+          tp
+            .addMessage[connectrpc.conformance.v1.UnaryRequest]
+            .addMessage[connectrpc.conformance.v1.IdempotentUnaryRequest]
+            .addMessage[connectrpc.conformance.v1.ConformancePayload.RequestInfo]
         }
         .build
 
+      app <- ConnectRouteBuilder.forService[IO](service)
+        .withJsonCodec(jsonCodec)
+        .build
+
       server <- EmberServerBuilder.default[IO]
         .withHost(host"127.0.0.1")
         .withPort(port"0") // random port

core/src/main/protobuf/connectrpc/error.proto

@@ -16,7 +16,15 @@ syntax = "proto3";
 
 package connectrpc;
 
-import "google/protobuf/any.proto";
+// This message is similar to the google.protobuf.Any message.
+//
+// Separate type was needed to introduce a separate JSON serializer for this message, since Any in error details
+// has different JSON serialization rules compared to a generic google.protobuf.Any.
+// See https://github.com/connectrpc/conformance/issues/948#issuecomment-2511130448 for some details.
+message ErrorDetailsAny {
+  string type = 1;
+  bytes value = 2;
+}
 
 // An error definition used for specifying a desired error response
 message Error {
@@ -30,5 +38,5 @@ message Error {
   optional string message = 2;
   // Errors in Connect and gRPC protocols can have arbitrary messages
   // attached to them, which are known as error details.
-  repeated google.protobuf.Any details = 3;
+  repeated ErrorDetailsAny details = 3;
 }

core/src/main/scala/org/ivovk/connect_rpc_scala/ConnectHandler.scala

@@ -11,8 +11,9 @@ import org.http4s.{Header, MediaType, MessageFailure, Method, Response}
 import org.ivovk.connect_rpc_scala.Mappings.*
 import org.ivovk.connect_rpc_scala.grpc.{MethodName, MethodRegistry}
 import org.ivovk.connect_rpc_scala.http.Headers.`X-Test-Case-Name`
-import org.ivovk.connect_rpc_scala.http.MessageCodec.given
-import org.ivovk.connect_rpc_scala.http.{MediaTypes, MessageCodec, MessageCodecRegistry, RequestEntity}
+import org.ivovk.connect_rpc_scala.http.codec.MessageCodec.given
+import org.ivovk.connect_rpc_scala.http.codec.{MessageCodec, MessageCodecRegistry}
+import org.ivovk.connect_rpc_scala.http.{MediaTypes, RequestEntity}
 import org.slf4j.{Logger, LoggerFactory}
 import scalapb.grpc.ClientCalls
 import scalapb.{GeneratedMessage, GeneratedMessageCompanion, TextFormat}
@@ -149,16 +150,16 @@ class ConnectHandler[F[_] : Async](
 
         val messageWithDetails = rawMessage
           .map(
-            _.split("\n").partition(m => !m.startsWith("type_url: "))
+            _.split("\n").partition(m => !m.startsWith("type: "))
           )
           .map((messageParts, additionalDetails) =>
             val details = additionalDetails
-              .map(TextFormat.fromAscii(com.google.protobuf.any.Any, _) match {
+              .map(TextFormat.fromAscii(connectrpc.ErrorDetailsAny, _) match {
                 case Right(details) => details
                 case Left(e) =>
                   logger.warn(s"Failed to parse additional details", e)
 
-                  com.google.protobuf.wrappers.StringValue(e.msg).toProtoAny
+                  com.google.protobuf.wrappers.StringValue(e.msg).toProtoErrorDetailsAny
               })
               .toSeq
 

core/src/main/scala/org/ivovk/connect_rpc_scala/ConnectRouteBuilder.scala

@@ -9,13 +9,11 @@ import org.http4s.{HttpApp, HttpRoutes, Method, Uri}
 import org.ivovk.connect_rpc_scala.grpc.*
 import org.ivovk.connect_rpc_scala.http.*
 import org.ivovk.connect_rpc_scala.http.QueryParams.*
-import org.ivovk.connect_rpc_scala.http.json.ConnectJsonRegistry
-import scalapb.json4s.{JsonFormat, Printer}
+import org.ivovk.connect_rpc_scala.http.codec.{JsonMessageCodec, JsonMessageCodecBuilder, MessageCodecRegistry, ProtoMessageCodec}
 
 import java.util.concurrent.Executor
 import scala.concurrent.ExecutionContext
 import scala.concurrent.duration.*
-import scala.util.chaining.*
 
 object ConnectRouteBuilder {
 
@@ -32,9 +30,9 @@ object ConnectRouteBuilder {
 
 case class ConnectRouteBuilder[F[_] : Async] private(
   services: Seq[ServerServiceDefinition],
-  jsonPrinterConfigurator: Endo[Printer] = identity,
   serverConfigurator: Endo[ServerBuilder[_]] = identity,
   channelConfigurator: Endo[ManagedChannelBuilder[_]] = identity,
+  customJsonCodec: Option[JsonMessageCodec[F]] = None,
   pathPrefix: Uri.Path = Uri.Path.Root,
   executor: Executor = ExecutionContext.global,
   waitForShutdown: Duration = 5.seconds,
@@ -43,8 +41,8 @@ case class ConnectRouteBuilder[F[_] : Async] private(
 
   import Mappings.*
 
-  def withJsonPrinterConfigurator(method: Endo[Printer]): ConnectRouteBuilder[F] =
-    copy(jsonPrinterConfigurator = method)
+  def withJsonCodec(codec: JsonMessageCodec[F]): ConnectRouteBuilder[F] =
+    copy(customJsonCodec = Some(codec))
 
   def withServerConfigurator(method: Endo[ServerBuilder[_]]): ConnectRouteBuilder[F] =
     copy(serverConfigurator = method)
@@ -80,14 +78,9 @@ case class ConnectRouteBuilder[F[_] : Async] private(
     val httpDsl = Http4sDsl[F]
     import httpDsl.*
 
-    val compressor  = Compressor[F]
-    val jsonPrinter = JsonFormat.printer
-      .withFormatRegistry(ConnectJsonRegistry.default)
-      .pipe(jsonPrinterConfigurator)
-
     val codecRegistry = MessageCodecRegistry[F](
-      JsonMessageCodec[F](compressor, jsonPrinter),
-      ProtoMessageCodec[F](compressor)
+      customJsonCodec.getOrElse(JsonMessageCodecBuilder[F]().build),
+      ProtoMessageCodec[F](),
     )
 
     val methodRegistry = MethodRegistry(services)

core/src/main/scala/org/ivovk/connect_rpc_scala/Mappings.scala

@@ -106,15 +106,18 @@ trait StatusCodeMappings {
 trait ProtoMappings {
 
   extension [T <: GeneratedMessage](t: T) {
-    private def any(typeUrlPrefix: String = "type.googleapis.com/"): com.google.protobuf.any.Any =
+    def toProtoAny: com.google.protobuf.any.Any = {
       com.google.protobuf.any.Any(
-        typeUrl = typeUrlPrefix + t.companion.scalaDescriptor.fullName,
+        typeUrl = "type.googleapis.com/" + t.companion.scalaDescriptor.fullName,
         value = t.toByteString
       )
+    }
 
-    def toProtoAny: com.google.protobuf.any.Any = any()
-
-    def toProtoErrorAny: com.google.protobuf.any.Any = any("")
+    def toProtoErrorDetailsAny: connectrpc.ErrorDetailsAny =
+      connectrpc.ErrorDetailsAny(
+        `type` = t.companion.scalaDescriptor.fullName,
+        value = t.toByteString
+      )
 
   }
 

core/src/main/scala/org/ivovk/connect_rpc_scala/http/MessageCodec.scala

@@ -5,6 +5,7 @@ import fs2.Stream
 import org.http4s.headers.{`Content-Encoding`, `Content-Type`}
 import org.http4s.{Charset, ContentCoding, Headers, Media}
 import org.ivovk.connect_rpc_scala.http.Headers.`Connect-Timeout-Ms`
+import org.ivovk.connect_rpc_scala.http.codec.MessageCodec
 import scalapb.{GeneratedMessage as Message, GeneratedMessageCompanion as Companion}
 
 object RequestEntity {

core/src/main/scala/org/ivovk/connect_rpc_scala/http/RequestEntity.scala

@@ -0,0 +1,20 @@
+package org.ivovk.connect_rpc_scala.http.codec
+
+import cats.effect.Sync
+import fs2.Stream
+import fs2.compression.Compression
+import org.http4s.ContentCoding
+
+class Compressor[F[_] : Sync] {
+
+  given Compression[F] = Compression.forSync[F]
+
+  def decompressed(encoding: Option[ContentCoding], body: Stream[F, Byte]): Stream[F, Byte] =
+    body.through(encoding match {
+      case Some(ContentCoding.gzip) =>
+        Compression[F].gunzip().andThen(_.flatMap(_.content))
+      case _ =>
+        identity
+    })
+
+}