#1050: http4s server and client migrated to cats-effect 3 / http4s 1.x, some other modules partially as well

Author: adamw

build.sbt

@@ -34,7 +34,8 @@ val commonSettings = commonSmlBuildSettings ++ ossPublishSettings ++ Seq(
   // slow down for CI
   Test / parallelExecution := false,
   // remove false alarms about unused implicit definitions in macros
-  scalacOptions += "-Ywarn-macros:after"
+  scalacOptions += "-Ywarn-macros:after",
+  evictionErrorLevel := Level.Info
 )
 
 val commonJvmSettings: Seq[Def.Setting[_]] = commonSettings
@@ -137,12 +138,12 @@ lazy val rootProject = (project in file("."))
   .settings(
     publishArtifact := false,
     name := "tapir",
-    testJVM := (test in Test).all(filterProject(p => !p.contains("JS"))).value,
-    testJS := (test in Test).all(filterProject(_.contains("JS"))).value,
-    testDocs := (test in Test).all(filterProject(p => p.contains("Docs") || p.contains("openapi") || p.contains("asyncapi"))).value,
-    testServers := (test in Test).all(filterProject(p => p.contains("Server"))).value,
-    testClients := (test in Test).all(filterProject(p => p.contains("Client"))).value,
-    testOther := (test in Test)
+    testJVM := (Test / test).all(filterProject(p => !p.contains("JS"))).value,
+    testJS := (Test / test).all(filterProject(_.contains("JS"))).value,
+    testDocs := (Test / test).all(filterProject(p => p.contains("Docs") || p.contains("openapi") || p.contains("asyncapi"))).value,
+    testServers := (Test / test).all(filterProject(p => p.contains("Server"))).value,
+    testClients := (Test / test).all(filterProject(p => p.contains("Client"))).value,
+    testOther := (Test / test)
       .all(
         filterProject(p =>
           !p.contains("Server") && !p.contains("Client") && !p.contains("Docs") && !p.contains("openapi") && !p.contains("asyncapi")
@@ -155,14 +156,14 @@ lazy val rootProject = (project in file("."))
 // start a test server before running tests of a client interpreter; this is required both for JS tests run inside a
 // nodejs/browser environment, as well as for JVM tests where akka-http isn't available (e.g. dotty).
 val clientTestServerSettings = Seq(
-  test in Test := (test in Test)
-    .dependsOn(startClientTestServer in clientTestServer2_13)
+  Test / test := (Test / test)
+    .dependsOn(clientTestServer2_13 / startClientTestServer)
     .value,
-  testOnly in Test := (testOnly in Test)
-    .dependsOn(startClientTestServer in clientTestServer2_13)
+  Test / testOnly := (Test / testOnly)
+    .dependsOn(clientTestServer2_13 / startClientTestServer)
     .evaluated,
-  testOptions in Test += Tests.Setup(() => {
-    val port = (clientTestServerPort in clientTestServer2_13).value
+  Test / testOptions += Tests.Setup(() => {
+    val port = (clientTestServer2_13 / clientTestServerPort).value
     PollingUtils.waitUntilServerAvailable(new URL(s"http://localhost:$port"))
   })
 )
@@ -171,16 +172,16 @@ lazy val clientTestServer = (projectMatrix in file("client/testserver"))
   .settings(commonJvmSettings)
   .settings(
     name := "testing-server",
-    skip in publish := true,
+    publish / skip := true,
     libraryDependencies ++= loggerDependencies ++ Seq(
       "org.http4s" %% "http4s-dsl" % Versions.http4s,
       "org.http4s" %% "http4s-blaze-server" % Versions.http4s,
       "org.http4s" %% "http4s-circe" % Versions.http4s
     ),
     // the test server needs to be started before running any client tests
-    mainClass in reStart := Some("sttp.tapir.client.tests.HttpServer"),
-    reStartArgs in reStart := Seq(s"${(clientTestServerPort in Test).value}"),
-    fullClasspath in reStart := (fullClasspath in Test).value,
+    reStart / mainClass := Some("sttp.tapir.client.tests.HttpServer"),
+    reStart / reStartArgs := Seq(s"${(Test / clientTestServerPort).value}"),
+    reStart / fullClasspath := (reStart / fullClasspath).value,
     clientTestServerPort := 51823,
     startClientTestServer := reStart.toTask("").value
   )
@@ -205,17 +206,17 @@ lazy val core: ProjectMatrix = (projectMatrix in file("core"))
       scalaTestPlusScalaCheck.value % Test,
       "com.47deg" %%% "scalacheck-toolbox-datetime" % "0.5.0" % Test
     ),
-    unmanagedSourceDirectories in Compile += {
-      val sourceDir = (sourceDirectory in Compile).value
+    Compile / unmanagedSourceDirectories += {
+      val sourceDir = (Compile / sourceDirectory).value
       CrossVersion.partialVersion(scalaVersion.value) match {
         case Some((2, n)) if n >= 13 => sourceDir / "scala-2.13+"
         case _                       => sourceDir / "scala-2.13-"
       }
     },
     // Until https://youtrack.jetbrains.com/issue/SCL-18636 is fixed and IntelliJ properly imports projects with
     // generated sources, they are explicitly added to git. See also below: commented out plugin.
-    unmanagedSourceDirectories in Compile += {
-      (sourceDirectory in Compile).value / "boilerplate-gen"
+    Compile / unmanagedSourceDirectories += {
+      (Compile / sourceDirectory).value / "boilerplate-gen"
     }
   )
   .jvmPlatform(
@@ -748,11 +749,7 @@ lazy val finatraServerCats: ProjectMatrix =
     .settings(commonJvmSettings)
     .settings(
       name := "tapir-finatra-server-cats",
-      libraryDependencies ++= Seq(
-        "org.typelevel" %% "cats-effect" % Versions.catsEffect,
-        "io.catbird" %% "catbird-finagle" % Versions.catbird,
-        "io.catbird" %% "catbird-effect" % Versions.catbird
-      )
+      libraryDependencies ++= Seq("org.typelevel" %% "cats-effect" % Versions.catsEffect)
     )
     .jvmPlatform(scalaVersions = scala2_12Versions)
     .dependsOn(finatraServer % "compile->compile;test->test", serverTests % Test)

client/http4s-client/src/main/scala/sttp/tapir/client/http4s/EndpointToHttp4sClient.scala

@@ -1,11 +1,13 @@
 package sttp.tapir.client.http4s
 
 import cats.Applicative
-import cats.effect.{Blocker, ContextShift, Effect, Sync}
+import cats.effect.Async
 import cats.implicits._
 import fs2.Chunk
+import fs2.io.file.Files
 import org.http4s._
 import org.http4s.headers.`Content-Type`
+import org.typelevel.ci.CIString
 import sttp.capabilities.Streams
 import sttp.capabilities.fs2.Fs2Streams
 import sttp.model.ResponseMetadata
@@ -29,9 +31,9 @@ import sttp.tapir.{
 import java.io.{ByteArrayInputStream, File, InputStream}
 import java.nio.ByteBuffer
 
-private[http4s] class EndpointToHttp4sClient(blocker: Blocker, clientOptions: Http4sClientOptions) {
+private[http4s] class EndpointToHttp4sClient(clientOptions: Http4sClientOptions) {
 
-  def toHttp4sRequest[I, E, O, R, F[_]: ContextShift: Effect](
+  def toHttp4sRequest[I, E, O, R, F[_]: Async](
       e: Endpoint[I, E, O, R],
       baseUriStr: Option[String]
   ): I => (Request[F], Response[F] => F[DecodeResult[Either[E, O]]]) = { params =>
@@ -46,7 +48,7 @@ private[http4s] class EndpointToHttp4sClient(blocker: Blocker, clientOptions: Ht
     (request, responseParser)
   }
 
-  def toHttp4sRequestUnsafe[I, E, O, R, F[_]: ContextShift: Effect](
+  def toHttp4sRequestUnsafe[I, E, O, R, F[_]: Async](
       e: Endpoint[I, E, O, R],
       baseUriStr: Option[String]
   ): I => (Request[F], Response[F] => F[Either[E, O]]) = { params =>
@@ -63,7 +65,7 @@ private[http4s] class EndpointToHttp4sClient(blocker: Blocker, clientOptions: Ht
   }
 
   @scala.annotation.tailrec
-  private def setInputParams[I, F[_]: ContextShift: Effect](
+  private def setInputParams[I, F[_]: Async](
       input: EndpointInput[I],
       params: Params,
       req: Request[F]
@@ -95,12 +97,12 @@ private[http4s] class EndpointToHttp4sClient(blocker: Blocker, clientOptions: Ht
       case EndpointIO.StreamBodyWrapper(StreamBodyIO(streams, _, _, _)) =>
         setStreamingBody(streams)(value.asInstanceOf[streams.BinaryStream], req)
       case EndpointIO.Header(name, codec, _) =>
-        val headers = codec.encode(value).map(value => Header(name, value))
+        val headers = codec.encode(value).map(value => Header.Raw(CIString(name), value): Header.ToRaw)
         req.putHeaders(headers: _*)
       case EndpointIO.Headers(codec, _) =>
-        val headers = codec.encode(value).map(h => Header(h.name, h.value))
+        val headers = codec.encode(value).map(h => Header.Raw(CIString(h.name), h.value): Header.ToRaw)
         req.putHeaders(headers: _*)
-      case EndpointIO.FixedHeader(h, _, _)           => req.putHeaders(Header(h.name, h.value))
+      case EndpointIO.FixedHeader(h, _, _)           => req.putHeaders(Header.Raw(CIString(h.name), h.value))
       case EndpointInput.ExtractFromRequest(_, _)    => req // ignoring
       case a: EndpointInput.Auth[_]                  => setInputParams(a.input, params, req)
       case EndpointInput.Pair(left, right, _, split) => handleInputPair(left, right, params, split, req)
@@ -110,7 +112,7 @@ private[http4s] class EndpointToHttp4sClient(blocker: Blocker, clientOptions: Ht
     }
   }
 
-  private def setBody[R, T, CF <: CodecFormat, F[_]: ContextShift: Effect](
+  private def setBody[R, T, CF <: CodecFormat, F[_]: Async](
       value: T,
       bodyType: RawBodyType[R],
       codec: Codec[R, T, CF],
@@ -128,10 +130,10 @@ private[http4s] class EndpointToHttp4sClient(blocker: Blocker, clientOptions: Ht
         val entityEncoder = EntityEncoder.chunkEncoder[F].contramap(Chunk.byteBuffer)
         req.withEntity(encoded.asInstanceOf[ByteBuffer])(entityEncoder)
       case RawBodyType.InputStreamBody =>
-        val entityEncoder = EntityEncoder.inputStreamEncoder[F, InputStream](blocker)
+        val entityEncoder = EntityEncoder.inputStreamEncoder[F, InputStream]
         req.withEntity(Applicative[F].pure(encoded.asInstanceOf[InputStream]))(entityEncoder)
       case RawBodyType.FileBody =>
-        val entityEncoder = EntityEncoder.fileEncoder[F](blocker)
+        val entityEncoder = EntityEncoder.fileEncoder[F]
         req.withEntity(encoded.asInstanceOf[File])(entityEncoder)
       case _: RawBodyType.MultipartBody =>
         throw new IllegalArgumentException("Multipart body isn't supported yet")
@@ -151,7 +153,7 @@ private[http4s] class EndpointToHttp4sClient(blocker: Blocker, clientOptions: Ht
         throw new IllegalArgumentException("Only Fs2Streams streaming is supported")
     }
 
-  private def handleInputPair[I, F[_]: ContextShift: Effect](
+  private def handleInputPair[I, F[_]: Async](
       left: EndpointInput[_],
       right: EndpointInput[_],
       params: Params,
@@ -164,15 +166,15 @@ private[http4s] class EndpointToHttp4sClient(blocker: Blocker, clientOptions: Ht
     setInputParams(right.asInstanceOf[EndpointInput[Any]], rightParams, req2)
   }
 
-  private def handleMapped[II, T, F[_]: ContextShift: Effect](
+  private def handleMapped[II, T, F[_]: Async](
       tuple: EndpointInput[II],
       codec: Mapping[T, II],
       params: Params,
       req: Request[F]
   ): Request[F] =
     setInputParams(tuple.asInstanceOf[EndpointInput[Any]], ParamsAsAny(codec.encode(params.asAny.asInstanceOf[II])), req)
 
-  private def parseHttp4sResponse[I, E, O, R, F[_]: Sync: ContextShift](
+  private def parseHttp4sResponse[I, E, O, R, F[_]: Async](
       e: Endpoint[I, E, O, R]
   ): Response[F] => F[DecodeResult[Either[E, O]]] = { response =>
     val code = sttp.model.StatusCode(response.status.code)
@@ -181,15 +183,15 @@ private[http4s] class EndpointToHttp4sClient(blocker: Blocker, clientOptions: Ht
     val output = if (code.isSuccess) e.output else e.errorOutput
 
     // headers with cookies
-    val headers = response.headers.toList.map(h => sttp.model.Header(h.name.toString(), h.value)).toVector
+    val headers = response.headers.headers.map(h => sttp.model.Header(h.name.toString, h.value)).toVector
 
     parser(response).map { responseBody =>
       val params = clientOutputParams(output, responseBody, ResponseMetadata(code, response.status.reason, headers))
       params.map(_.asAny).map(p => if (code.isSuccess) Right(p.asInstanceOf[O]) else Left(p.asInstanceOf[E]))
     }
   }
 
-  private def responseFromOutput[F[_]: Sync: ContextShift](out: EndpointOutput[_]): Response[F] => F[Any] = { response =>
+  private def responseFromOutput[F[_]: Async](out: EndpointOutput[_]): Response[F] => F[Any] = { response =>
     bodyIsStream(out) match {
       case Some(streams) =>
         streams match {
@@ -211,7 +213,7 @@ private[http4s] class EndpointToHttp4sClient(blocker: Blocker, clientOptions: Ht
               response.body.compile.toVector.map(_.toArray).map(new ByteArrayInputStream(_)).map(_.asInstanceOf[Any])
             case RawBodyType.FileBody =>
               val file = clientOptions.createFile()
-              response.body.through(fs2.io.file.writeAll(file.toPath, blocker)).compile.drain.map(_ => file.asInstanceOf[Any])
+              response.body.through(Files[F].writeAll(file.toPath)).compile.drain.map(_ => file.asInstanceOf[Any])
             case RawBodyType.MultipartBody(_, _) => throw new IllegalArgumentException("Multipart bodies aren't supported in responses")
           }
           .getOrElse[F[Any]](((): Any).pure[F])

client/http4s-client/src/main/scala/sttp/tapir/client/http4s/Http4sClientInterpreter.scala

@@ -1,10 +1,10 @@
 package sttp.tapir.client.http4s
 
-import cats.effect.{Blocker, ContextShift, Effect}
+import cats.effect.Async
 import org.http4s.{Request, Response}
 import sttp.tapir.{DecodeResult, Endpoint}
 
-abstract class Http4sClientInterpreter[F[_]: ContextShift: Effect] {
+abstract class Http4sClientInterpreter[F[_]: Async] {
 
   /** Interprets the endpoint as a client call, using the given `baseUri` as the starting point to create the target
     * uri. If `baseUri` is not provided, the request will be a relative one.
@@ -15,10 +15,9 @@ abstract class Http4sClientInterpreter[F[_]: ContextShift: Effect] {
     *  - a response parser that extracts the expected entity from the received `org.http4s.Response[F]`.
     */
   def toRequest[I, E, O, R](e: Endpoint[I, E, O, R], baseUri: Option[String])(implicit
-      blocker: Blocker,
       clientOptions: Http4sClientOptions
   ): I => (Request[F], Response[F] => F[DecodeResult[Either[E, O]]]) =
-    new EndpointToHttp4sClient(blocker, clientOptions).toHttp4sRequest[I, E, O, R, F](e, baseUri)
+    new EndpointToHttp4sClient(clientOptions).toHttp4sRequest[I, E, O, R, F](e, baseUri)
 
   /** Interprets the endpoint as a client call, using the given `baseUri` as the starting point to create the target
     * uri. If `baseUri` is not provided, the request will be a relative one.
@@ -29,12 +28,11 @@ abstract class Http4sClientInterpreter[F[_]: ContextShift: Effect] {
     *  - a response parser that extracts the expected entity from the received `org.http4s.Response[F]`.
     */
   def toRequestUnsafe[I, E, O, R](e: Endpoint[I, E, O, R], baseUri: Option[String])(implicit
-      blocker: Blocker,
       clientOptions: Http4sClientOptions
   ): I => (Request[F], Response[F] => F[Either[E, O]]) =
-    new EndpointToHttp4sClient(blocker, clientOptions).toHttp4sRequestUnsafe[I, E, O, R, F](e, baseUri)
+    new EndpointToHttp4sClient(clientOptions).toHttp4sRequestUnsafe[I, E, O, R, F](e, baseUri)
 }
 
 object Http4sClientInterpreter {
-  def apply[F[_]: ContextShift: Effect]: Http4sClientInterpreter[F] = new Http4sClientInterpreter[F] {}
+  def apply[F[_]: Async]: Http4sClientInterpreter[F] = new Http4sClientInterpreter[F] {}
 }

client/http4s-client/src/test/scala/sttp/tapir/client/http4s/Http4ClientStreamingTests.scala

@@ -1,6 +1,7 @@
 package sttp.tapir.client.http4s
 
 import cats.effect.IO
+import cats.effect.unsafe.implicits.global
 import fs2.text
 import sttp.capabilities.fs2.Fs2Streams
 import sttp.tapir.client.tests.ClientStreamingTests

client/http4s-client/src/test/scala/sttp/tapir/client/http4s/Http4sClientTests.scala

@@ -1,6 +1,6 @@
 package sttp.tapir.client.http4s
 
-import cats.effect.{Blocker, ContextShift, IO, Timer}
+import cats.effect.IO
 import org.http4s.client.blaze.BlazeClientBuilder
 import org.http4s.{Request, Response}
 import sttp.tapir.client.tests.ClientTests
@@ -9,10 +9,6 @@ import sttp.tapir.{DecodeResult, Endpoint}
 import scala.concurrent.ExecutionContext.global
 
 abstract class Http4sClientTests[R] extends ClientTests[R] {
-  implicit val cs: ContextShift[IO] = IO.contextShift(global)
-  implicit val timer: Timer[IO] = IO.timer(global)
-  implicit val blocker: Blocker = Blocker.liftExecutionContext(global)
-
   override def send[I, E, O, FN[_]](e: Endpoint[I, E, O, R], port: Port, args: I, scheme: String = "http"): IO[Either[E, O]] = {
     val (request, parseResponse) = Http4sClientInterpreter[IO].toRequestUnsafe(e, Some(s"http://localhost:$port")).apply(args)
 

client/tests/src/main/scala/sttp/tapir/client/tests/ClientBasicTests.scala

@@ -1,5 +1,7 @@
 package sttp.tapir.client.tests
 
+import cats.effect.unsafe.implicits.global
+
 import sttp.model.{QueryParams, StatusCode}
 import sttp.tapir._
 import sttp.tapir.model.UsernamePassword

client/tests/src/main/scala/sttp/tapir/client/tests/ClientMultipartTests.scala

@@ -1,5 +1,6 @@
 package sttp.tapir.client.tests
 
+import cats.effect.unsafe.implicits.global
 import sttp.tapir.tests._
 
 trait ClientMultipartTests { this: ClientTests[Any] =>

client/tests/src/main/scala/sttp/tapir/client/tests/ClientStreamingTests.scala

@@ -1,5 +1,6 @@
 package sttp.tapir.client.tests
 
+import cats.effect.unsafe.implicits.global
 import sttp.capabilities.Streams
 import sttp.tapir.DecodeResult
 import sttp.tapir.tests.{in_stream_out_stream, not_existing_endpoint}

client/tests/src/main/scala/sttp/tapir/client/tests/ClientTests.scala

@@ -3,6 +3,7 @@ package sttp.tapir.client.tests
 import java.io.InputStream
 
 import cats.effect._
+import cats.effect.unsafe.implicits.global
 import cats.implicits._
 import org.scalatest.BeforeAndAfterAll
 import org.scalatest.funsuite.AsyncFunSuite

client/tests/src/main/scala/sttp/tapir/client/tests/ClientWebSocketTests.scala

@@ -1,6 +1,7 @@
 package sttp.tapir.client.tests
 
 import cats.effect.IO
+import cats.effect.unsafe.implicits.global
 import sttp.capabilities.{Streams, WebSockets}
 import sttp.tapir._
 import sttp.tapir.json.circe._
@@ -43,7 +44,9 @@ trait ClientWebSocketTests[S] { this: ClientTests[S with WebSockets] =>
 
     test("web sockets, client-terminated echo using fragmented frames") {
       send(
-        endpoint.get.in("ws" / "echo" / "fragmented").out(webSocketBody[String, CodecFormat.TextPlain, WebSocketFrame, CodecFormat.TextPlain].apply(streams)),
+        endpoint.get
+          .in("ws" / "echo" / "fragmented")
+          .out(webSocketBody[String, CodecFormat.TextPlain, WebSocketFrame, CodecFormat.TextPlain].apply(streams)),
         port,
         (),
         "ws"