Introduce separate type for Any's in error details. Split codec class…

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,6 @@ 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 java.io.InputStream
 import java.nio.ByteBuffer
@@ -37,15 +36,12 @@ object Main extends IOApp.Simple {
       )
 
       app <- ConnectRouteBuilder.forService[IO](service)
-        // 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]
-          )
+        .withJsonCodecConfigurator { 
+          // Registering message types in TypeRegistry is required to pass com.google.protobuf.any.Any
+          // JSON-serialization conformance tests
+          _
+            .registerType[connectrpc.conformance.v1.UnaryRequest]
+            .registerType[connectrpc.conformance.v1.IdempotentUnaryRequest]
         }
         .build
 

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,49 +9,75 @@ 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 {
 
   def forService[F[_] : Async](service: ServerServiceDefinition): ConnectRouteBuilder[F] =
-    ConnectRouteBuilder(Seq(service))
+    forServices(Seq(service))
 
   def forServices[F[_] : Async](service: ServerServiceDefinition, other: ServerServiceDefinition*): ConnectRouteBuilder[F] =
-    ConnectRouteBuilder(service +: other)
+    forServices(service +: other)
 
   def forServices[F[_] : Async](services: Seq[ServerServiceDefinition]): ConnectRouteBuilder[F] =
-    ConnectRouteBuilder(services)
+    new ConnectRouteBuilder(
+      services = services,
+      serverConfigurator = identity,
+      channelConfigurator = identity,
+      customJsonCodec = None,
+      pathPrefix = Uri.Path.Root,
+      executor = ExecutionContext.global,
+      waitForShutdown = 5.seconds,
+      treatTrailersAsHeaders = true,
+    )
 
 }
 
-case class ConnectRouteBuilder[F[_] : Async] private(
+final class ConnectRouteBuilder[F[_] : Async] private(
   services: Seq[ServerServiceDefinition],
-  jsonPrinterConfigurator: Endo[Printer] = identity,
-  serverConfigurator: Endo[ServerBuilder[_]] = identity,
-  channelConfigurator: Endo[ManagedChannelBuilder[_]] = identity,
-  pathPrefix: Uri.Path = Uri.Path.Root,
-  executor: Executor = ExecutionContext.global,
-  waitForShutdown: Duration = 5.seconds,
-  treatTrailersAsHeaders: Boolean = true,
+  serverConfigurator: Endo[ServerBuilder[_]],
+  channelConfigurator: Endo[ManagedChannelBuilder[_]],
+  customJsonCodec: Option[JsonMessageCodec[F]],
+  pathPrefix: Uri.Path,
+  executor: Executor,
+  waitForShutdown: Duration,
+  treatTrailersAsHeaders: Boolean,
 ) {
 
-  import Mappings.*
-
-  def withJsonPrinterConfigurator(method: Endo[Printer]): ConnectRouteBuilder[F] =
-    copy(jsonPrinterConfigurator = method)
+  private def copy(
+    services: Seq[ServerServiceDefinition] = services,
+    serverConfigurator: Endo[ServerBuilder[_]] = serverConfigurator,
+    channelConfigurator: Endo[ManagedChannelBuilder[_]] = channelConfigurator,
+    customJsonCodec: Option[JsonMessageCodec[F]] = customJsonCodec,
+    pathPrefix: Uri.Path = pathPrefix,
+    executor: Executor = executor,
+    waitForShutdown: Duration = waitForShutdown,
+    treatTrailersAsHeaders: Boolean = treatTrailersAsHeaders,
+  ): ConnectRouteBuilder[F] =
+    new ConnectRouteBuilder(
+      services,
+      serverConfigurator,
+      channelConfigurator,
+      customJsonCodec,
+      pathPrefix,
+      executor,
+      waitForShutdown,
+      treatTrailersAsHeaders,
+    )
 
   def withServerConfigurator(method: Endo[ServerBuilder[_]]): ConnectRouteBuilder[F] =
     copy(serverConfigurator = method)
 
   def withChannelConfigurator(method: Endo[ManagedChannelBuilder[_]]): ConnectRouteBuilder[F] =
     copy(channelConfigurator = method)
 
+  def withJsonCodecConfigurator(method: Endo[JsonMessageCodecBuilder[F]]): ConnectRouteBuilder[F] =
+    copy(customJsonCodec = Some(method(JsonMessageCodecBuilder[F]()).build))
+
   def withPathPrefix(path: Uri.Path): ConnectRouteBuilder[F] =
     copy(pathPrefix = path)
 
@@ -62,32 +88,28 @@ case class ConnectRouteBuilder[F[_] : Async] private(
     copy(waitForShutdown = duration)
 
   /**
-   * If enabled, trailers will be treated as headers (no "trailer-" prefix).
+   * When enabled, response trailers are treated as headers (no "trailer-" prefix added).
    *
    * Both `fs2-grpc` and `zio-grpc` support trailing headers only, so enabling this option is a single way to
-   * send headers from the server to the client.
+   * send headers from the server to a client.
    *
    * Enabled by default.
    */
   def withTreatTrailersAsHeaders(enabled: Boolean): ConnectRouteBuilder[F] =
     copy(treatTrailersAsHeaders = enabled)
 
   /**
-   * Method can be used if you want to add additional routes to the server.
-   * Otherwise, it is preferred to use the [[build]] method.
+   * Use this method only if you want to add additional routes to the server.
+   *
+   * Otherwise, [[build]] method should be preferred.
    */
   def buildRoutes: Resource[F, HttpRoutes[F]] = {
     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
+      )
 
   }