Endpoint api en-/decodes based on accept header (#2366)

* `Endpoint` api en-/decodes based on accept header

The default codec is still json. Only added support for protobuf

* Fix scala 2.12 compilation

* Fix wildcard import

* Extract aspect creation to fix Scala 2.12 test

* Use accept/content-type header also for multipart encoding/decoding

* Formatting

* Make codec creation lazy and cache it

* Merge fix and formatting

* Fix Scala 2.12 build: concurrent map converter

* Use plain java map as cache; add missing schema types to TextCodec

* Fix type check

* Formatting

* Redesign EncoderDecoder creation for specific media type for efficiency

* Fix multipart encoding for endpoint API

* migrate into main

* Formatting

build.sbt

@@ -130,6 +130,7 @@ lazy val zioHttp = (project in file("zio-http"))
       `zio-streams`,
       `zio-schema`,
       `zio-schema-json`,
+      `zio-schema-protobuf`,
       `zio-test`,
       `zio-test-sbt`,
       `netty-incubator`,

project/Dependencies.scala

@@ -1,5 +1,5 @@
-import sbt.Keys.scalaVersion
 import sbt._
+import sbt.Keys.scalaVersion
 
 object Dependencies {
   val JwtCoreVersion                = "9.1.1"
@@ -29,12 +29,13 @@ object Dependencies {
   val `netty-incubator` =
     "io.netty.incubator" % "netty-incubator-transport-native-io_uring" % NettyIncubatorVersion classifier "linux-x86_64"
 
-  val zio               = "dev.zio" %% "zio"             % ZioVersion
-  val `zio-cli`         = "dev.zio" %% "zio-cli"         % ZioCliVersion
-  val `zio-streams`     = "dev.zio" %% "zio-streams"     % ZioVersion
-  val `zio-schema`      = "dev.zio" %% "zio-schema"      % ZioSchemaVersion
-  val `zio-schema-json` = "dev.zio" %% "zio-schema-json" % ZioSchemaVersion
-  val `zio-test`        = "dev.zio" %% "zio-test"        % ZioVersion % "test"
-  val `zio-test-sbt`    = "dev.zio" %% "zio-test-sbt"    % ZioVersion % "test"
+  val zio                   = "dev.zio" %% "zio"                 % ZioVersion
+  val `zio-cli`             = "dev.zio" %% "zio-cli"             % ZioCliVersion
+  val `zio-streams`         = "dev.zio" %% "zio-streams"         % ZioVersion
+  val `zio-schema`          = "dev.zio" %% "zio-schema"          % ZioSchemaVersion
+  val `zio-schema-json`     = "dev.zio" %% "zio-schema-json"     % ZioSchemaVersion
+  val `zio-schema-protobuf` = "dev.zio" %% "zio-schema-protobuf" % ZioSchemaVersion
+  val `zio-test`            = "dev.zio" %% "zio-test"            % ZioVersion % "test"
+  val `zio-test-sbt`        = "dev.zio" %% "zio-test-sbt"        % ZioVersion % "test"
 
 }

zio-http-example/src/main/scala/example/CombinerTypesExample.scala

@@ -1,7 +1,6 @@
 package example
 
 import zio.http.codec.HttpCodec._
-import zio.http.codec.HttpCodecType.PathQuery
 import zio.http.codec._
 
 object CombinerTypesExample extends App {

zio-http/src/main/scala/zio/http/MediaType.scala

@@ -27,7 +27,7 @@ final case class MediaType(
   extensions: Map[String, String] = Map.empty,
   parameters: Map[String, String] = Map.empty,
 ) {
-  def fullType: String = s"$mainType/$subType"
+  lazy val fullType: String = s"$mainType/$subType"
 }
 
 object MediaType extends MediaTypes {
@@ -37,11 +37,12 @@ object MediaType extends MediaTypes {
   def forContentType(contentType: String): Option[MediaType] = {
     val index = contentType.indexOf(";")
     if (index == -1)
-      contentTypeMap.get(contentType)
+      contentTypeMap.get(contentType).orElse(parseCustomMediaType(contentType))
     else {
       val (contentType1, parameter) = contentType.splitAt(index)
       contentTypeMap
         .get(contentType1)
+        .orElse(parseCustomMediaType(contentType1))
         .map(_.copy(parameters = parseOptionalParameters(parameter.split(";"))))
     }
   }

zio-http/src/main/scala/zio/http/Request.scala

@@ -152,4 +152,8 @@ object Request {
     } else {
       URL(Path(path))
     }
+
+  object Patch {
+    val empty: Patch = Patch(Headers.empty, QueryParams.empty)
+  }
 }

zio-http/src/main/scala/zio/http/ZClientAspect.scala

@@ -109,6 +109,14 @@ object ZClientAspect {
    * Client aspect that logs a debug message to the console after each request
    */
   final def debug(implicit trace: Trace): ZClientAspect[Nothing, Any, Nothing, Body, Nothing, Any, Nothing, Response] =
+    debug(PartialFunction.empty)
+
+  /**
+   * Client aspect that logs a debug message to the console after each request
+   */
+  final def debug(
+    extraMessage: PartialFunction[Response, String],
+  )(implicit trace: Trace): ZClientAspect[Nothing, Any, Nothing, Body, Nothing, Any, Nothing, Response] =
     new ZClientAspect[Nothing, Any, Nothing, Body, Nothing, Any, Nothing, Response] {
 
       /**
@@ -140,11 +148,10 @@ object ZClientAspect {
               .timed
               .tap {
                 case (duration, Exit.Success(response)) =>
-                  Console
-                    .printLine(
-                      s"${response.status.code} $method ${url.encode} ${duration.toMillis}ms",
-                    )
-                    .orDie
+                  {
+                    Console.printLine(s"${response.status.code} $method ${url.encode} ${duration.toMillis}ms") *>
+                      Console.printLine(extraMessage(response)).when(extraMessage.isDefinedAt(response))
+                  }.orDie
                 case (duration, Exit.Failure(cause))    =>
                   Console
                     .printLine(

zio-http/src/main/scala/zio/http/codec/HeaderCodecs.scala

@@ -20,8 +20,8 @@ import scala.util.Try
 
 import zio.stacktracer.TracingImplicits.disableAutoTrace
 
-import zio.http.Header
 import zio.http.Header.HeaderType
+import zio.http.{Header, MediaType}
 
 private[codec] trait HeaderCodecs {
   private[http] def headerCodec[A](name: String, value: TextCodec[A]): HeaderCodec[A] =

zio-http/src/main/scala/zio/http/codec/HttpCodec.scala

@@ -16,6 +16,8 @@
 
 package zio.http.codec
 
+import java.util.concurrent.ConcurrentHashMap
+
 import scala.language.implicitConversions
 import scala.reflect.ClassTag
 
@@ -26,8 +28,10 @@ import zio.stream.ZStream
 
 import zio.schema.Schema
 
+import zio.http.Header.Accept.MediaTypeWithQFactor
 import zio.http._
 import zio.http.codec.HttpCodec.{Annotated, Metadata}
+import zio.http.codec.internal.EncoderDecoder
 
 /**
  * A [[zio.http.codec.HttpCodec]] represents a codec for a part of an HTTP
@@ -44,7 +48,36 @@ import zio.http.codec.HttpCodec.{Annotated, Metadata}
 sealed trait HttpCodec[-AtomTypes, Value] {
   self =>
 
-  private lazy val encoderDecoder = zio.http.codec.internal.EncoderDecoder(self)
+  private lazy val encoderDecoders: ConcurrentHashMap[String, EncoderDecoder[_, _]] =
+    new ConcurrentHashMap[String, EncoderDecoder[_, _]]()
+
+  private lazy val defaultEncoderDecoder: EncoderDecoder[AtomTypes, Value]                              =
+    EncoderDecoder(self, None)
+  private def encoderDecoder(mediaTypes: Chunk[MediaTypeWithQFactor]): EncoderDecoder[AtomTypes, Value] =
+    if (mediaTypes.isEmpty) defaultEncoderDecoder
+    else {
+      var mediaType: Option[MediaTypeWithQFactor] = None
+      var i                                       = 0
+      while (i < mediaTypes.length) {
+        if (mediaTypes(i).qFactor.getOrElse(1.0) > mediaType.map(_.qFactor.getOrElse(1.0)).getOrElse(0.0)) {
+          mediaType = Some(mediaTypes(i))
+        }
+        i += 1
+      }
+      mediaType match {
+        case Some(mediaType) =>
+          encoderDecoders
+            .computeIfAbsent(
+              mediaType.mediaType.fullType,
+              mediaType => {
+                EncoderDecoder(self, Some(mediaType))
+              },
+            )
+            .asInstanceOf[EncoderDecoder[AtomTypes, Value]]
+        case None            =>
+          throw new IllegalArgumentException("No supported media type provided") // TODO: Better error handling
+      }
+    }
 
   /**
    * Returns a new codec that is the same as this one, but has attached docs,
@@ -158,13 +191,13 @@ sealed trait HttpCodec[-AtomTypes, Value] {
   private final def decode(url: URL, status: Status, method: Method, headers: Headers, body: Body)(implicit
     trace: Trace,
   ): Task[Value] =
-    encoderDecoder.decode(url, status, method, headers, body)
+    encoderDecoder(Chunk.empty).decode(url, status, method, headers, body)
 
   /**
    * Uses this codec to encode the Scala value into a request.
    */
   final def encodeRequest(value: Value): Request =
-    encodeWith(value)((url, _, method, headers, body) =>
+    encodeWith(value, Chunk.empty)((url, _, method, headers, body) =>
       Request(
         url = url,
         method = method.getOrElse(Method.GET),
@@ -179,7 +212,7 @@ sealed trait HttpCodec[-AtomTypes, Value] {
    * Uses this codec to encode the Scala value as a patch to a request.
    */
   final def encodeRequestPatch(value: Value): Request.Patch =
-    encodeWith(value)((url, _, _, headers, _) =>
+    encodeWith(value, Chunk.empty)((url, _, _, headers, _) =>
       Request.Patch(
         addQueryParams = url.queryParams,
         addHeaders = headers,
@@ -189,23 +222,23 @@ sealed trait HttpCodec[-AtomTypes, Value] {
   /**
    * Uses this codec to encode the Scala value as a response.
    */
-  final def encodeResponse[Z](value: Value): Response =
-    encodeWith(value)((_, status, _, headers, body) =>
+  final def encodeResponse[Z](value: Value, outputTypes: Chunk[MediaTypeWithQFactor]): Response =
+    encodeWith(value, outputTypes)((_, status, _, headers, body) =>
       Response(headers = headers, body = body, status = status.getOrElse(Status.Ok)),
     )
 
   /**
    * Uses this codec to encode the Scala value as a response patch.
    */
-  final def encodeResponsePatch[Z](value: Value): Response.Patch =
-    encodeWith(value)((_, status, _, headers, _) =>
+  final def encodeResponsePatch[Z](value: Value, outputTypes: Chunk[MediaTypeWithQFactor]): Response.Patch =
+    encodeWith(value, outputTypes)((_, status, _, headers, _) =>
       Response.Patch.addHeaders(headers) ++ status.map(Response.Patch.status(_)).getOrElse(Response.Patch.empty),
     )
 
-  private final def encodeWith[Z](value: Value)(
+  private final def encodeWith[Z](value: Value, outputTypes: Chunk[MediaTypeWithQFactor])(
     f: (URL, Option[Status], Option[Method], Headers, Body) => Z,
   ): Z =
-    encoderDecoder.encodeWith(value)(f)
+    encoderDecoder(outputTypes).encodeWith(value)(f)
 
   def examples(examples: Iterable[(String, Value)]): HttpCodec[AtomTypes, Value] =
     HttpCodec.Annotated(self, Metadata.Examples(Chunk.fromIterable(examples).toMap))

zio-http/src/main/scala/zio/http/codec/HttpCodecError.scala

@@ -69,6 +69,10 @@ object HttpCodecError {
   }
   final case class CustomError(message: String)                                                extends HttpCodecError
 
+  final case class UnsupportedContentType(contentType: String) extends HttpCodecError {
+    def message = s"Unsupported content type $contentType"
+  }
+
   def isHttpCodecError(cause: Cause[Any]): Boolean = {
     !cause.isFailure && cause.defects.forall(e => e.isInstanceOf[HttpCodecError])
   }

zio-http/src/main/scala/zio/http/codec/internal/BodyCodec.scala

@@ -16,16 +16,18 @@
 
 package zio.http.codec.internal
 
+import java.nio.charset.Charset
+
 import zio._
 import zio.stacktracer.TracingImplicits.disableAutoTrace
 
 import zio.stream.{ZPipeline, ZStream}
 
 import zio.schema._
-import zio.schema.codec.BinaryCodec
+import zio.schema.codec.{BinaryCodec, Codec}
 
 import zio.http.codec.HttpCodecError
-import zio.http.{Body, FormField, MediaType}
+import zio.http.{Body, MediaType}
 
 /**
  * A BodyCodec encapsulates the logic necessary to both encode and decode bodies
@@ -45,11 +47,21 @@ private[internal] sealed trait BodyCodec[A] { self =>
    */
   def decodeFromBody(body: Body, codec: BinaryCodec[Element])(implicit trace: Trace): IO[Throwable, A]
 
+  /**
+   * Attempts to decode the `A` from a body using the given codec.
+   */
+  def decodeFromBody(body: Body, codec: Codec[String, Char, Element])(implicit trace: Trace): IO[Throwable, A]
+
   /**
    * Encodes the `A` to a body in the given codec.
    */
   def encodeToBody(value: A, codec: BinaryCodec[Element])(implicit trace: Trace): Body
 
+  /**
+   * Encodes the `A` to a body in the given codec.
+   */
+  def encodeToBody(value: A, codec: Codec[String, Char, Element])(implicit trace: Trace): Body
+
   /**
    * Erases the type for easier use in the internal implementation.
    */
@@ -82,8 +94,13 @@ private[internal] object BodyCodec {
 
     def decodeFromBody(body: Body, codec: BinaryCodec[Unit])(implicit trace: Trace): IO[Nothing, Unit] = ZIO.unit
 
+    def decodeFromBody(body: Body, codec: Codec[String, Char, Unit])(implicit trace: Trace): IO[Nothing, Unit] =
+      ZIO.unit
+
     def encodeToBody(value: Unit, codec: BinaryCodec[Unit])(implicit trace: Trace): Body = Body.empty
 
+    def encodeToBody(value: Unit, codec: Codec[String, Char, Unit])(implicit trace: Trace): Body = Body.empty
+
     def schema: Schema[Unit] = Schema[Unit]
 
     def mediaType: Option[MediaType] = None
@@ -100,9 +117,16 @@ private[internal] object BodyCodec {
           ZIO.fromEither(codec.decode(chunk))
         }.flatMap(validateZIO(schema))
 
+    def decodeFromBody(body: Body, codec: Codec[String, Char, A])(implicit trace: Trace): IO[Throwable, A] =
+      if (schema == Schema[Unit]) ZIO.unit.asInstanceOf[IO[Throwable, A]]
+      else body.asString.flatMap(chunk => ZIO.fromEither(codec.decode(chunk)))
+
     def encodeToBody(value: A, codec: BinaryCodec[A])(implicit trace: Trace): Body =
       Body.fromChunk(codec.encode(value))
 
+    def encodeToBody(value: A, codec: Codec[String, Char, A])(implicit trace: Trace): Body =
+      Body.fromString(codec.encode(value))
+
     type Element = A
   }
 
@@ -113,9 +137,17 @@ private[internal] object BodyCodec {
     ): IO[Throwable, ZStream[Any, Nothing, E]] =
       ZIO.succeed((body.asStream >>> codec.streamDecoder >>> validateStream(schema)).orDie)
 
+    def decodeFromBody(body: Body, codec: Codec[String, Char, E])(implicit
+      trace: Trace,
+    ): IO[Throwable, ZStream[Any, Nothing, E]] =
+      ZIO.succeed((body.asStream >>> ZPipeline.decodeCharsWith(Charset.defaultCharset()) >>> codec.streamDecoder).orDie)
+
     def encodeToBody(value: ZStream[Any, Nothing, E], codec: BinaryCodec[E])(implicit trace: Trace): Body =
       Body.fromStream(value >>> codec.streamEncoder)
 
+    def encodeToBody(value: ZStream[Any, Nothing, E], codec: Codec[String, Char, E])(implicit trace: Trace): Body =
+      Body.fromStream(value >>> codec.streamEncoder.map(_.toByte))
+
     type Element = E
   }