Implement multipart for netty-future, netty-cats & netty-zio

Author: ajozwik

.github/workflows/ci.yml

@@ -38,7 +38,7 @@ jobs:
     steps:
     - name: Checkout
       uses: actions/checkout@v6
-    - uses: sbt/setup-sbt@af116cce31c00823d3903ce687f9cda3a4f19f1b # v1, specifically v1.2.1
+    - uses: sbt/setup-sbt@18c5326132e2b0900fd6bd70f8b2b212364d11c8 # v1, specifically v1.3.0
     - name: Set up JDK 
       uses: actions/setup-java@v5
       with:
@@ -169,7 +169,7 @@ jobs:
     steps:
       - name: Checkout
         uses: actions/checkout@v6
-      - uses: sbt/setup-sbt@af116cce31c00823d3903ce687f9cda3a4f19f1b # v1, specifically v1.2.1
+      - uses: sbt/setup-sbt@18c5326132e2b0900fd6bd70f8b2b212364d11c8 # v1, specifically v1.3.0
       - name: Set up JDK
         uses: actions/setup-java@v5
         with:

.github/workflows/dependency-graph.yml

@@ -15,5 +15,5 @@ jobs:
     runs-on: ubuntu-24.04
     steps:
       - uses: actions/checkout@v6
-      - uses: sbt/setup-sbt@af116cce31c00823d3903ce687f9cda3a4f19f1b # v1, specifically v1.2.1
+      - uses: sbt/setup-sbt@18c5326132e2b0900fd6bd70f8b2b212364d11c8 # v1, specifically v1.3.0
       - uses: scalacenter/sbt-dependency-submission@d84eef4c09e633bcf5f113bcad7fd5e9af1baee9 # v3, specifically v3.1.1

build.sbt

@@ -546,7 +546,9 @@ lazy val tests: ProjectMatrix = (projectMatrix in file("tests"))
       "com.softwaremill.common" %%% "tagging" % "2.3.5",
       scalaTest.value,
       logback
-    )
+    ),
+    scalacOptions += "-Wconf:msg=unused value of type org.scalatest.Assertion:s",
+    scalacOptions += "-Wconf:msg=unused value of type org.scalatest.compatible.Assertion:s"
   )
   .jvmPlatform(
     scalaVersions = scala2And3Versions,
@@ -1646,6 +1648,7 @@ lazy val nettyServer: ProjectMatrix = (projectMatrix in file("server/netty-serve
       "io.netty" % "netty-all" % Versions.nettyAll,
       "org.playframework.netty" % "netty-reactive-streams-http" % Versions.nettyReactiveStreams,
       "org.apache.httpcomponents" % "httpmime" % "4.5.14",
+      "org.scala-lang.modules" %% "scala-java8-compat" % Versions.scalaJava8Compat,
       slf4j
     )
   )

openapi-codegen/sbt-plugin/src/sbt-test/sbt-openapi-codegen/caching/project/build.properties

@@ -1 +1 @@
-sbt.version=1.12.11
+sbt.version=1.12.12

perf-tests/perf-tests-e2e/src/main/scala/sttp/tapir/perf/netty/cats/NettyCats.scala

@@ -12,8 +12,6 @@ import sttp.tapir.server.netty.cats.NettyCatsServer
 import sttp.tapir.server.netty.cats.NettyCatsServerOptions
 import sttp.capabilities.fs2.Fs2Streams
 
-import scala.concurrent.duration._
-
 object Tapir extends Endpoints {
   val wsResponseStream = Stream.fixedRate[IO](WebSocketSingleResponseLag, dampen = false)
   val wsEndpoint = wsBaseEndpoint

project/Versions.scala

@@ -49,8 +49,8 @@ object Versions {
   val play29Server = "2.9.11"
   val tethys = "0.29.8"
   val vertx = "5.1.2"
-  val jsScalaJavaTime = "2.6.0"
-  val nativeScalaJavaTime = "2.6.0"
+  val jsScalaJavaTime = "2.7.0"
+  val nativeScalaJavaTime = "2.7.0"
   val jwtScala = "10.0.4"
   val derevo = "0.14.0"
   val newtype = "0.4.4"

project/build.properties

@@ -1 +1 @@
-sbt.version=1.12.11
+sbt.version=1.12.12

server/netty-server/cats/src/main/scala/sttp/tapir/server/netty/cats/internal/NettyCatsRequestBody.scala

@@ -1,37 +1,39 @@
 package sttp.tapir.server.netty.cats.internal
 
 import cats.effect.Async
-import cats.syntax.all._
 import fs2.Chunk
 import fs2.io.file.{Files, Path}
 import io.netty.handler.codec.http.HttpContent
-import org.playframework.netty.http.StreamedHttpRequest
 import org.reactivestreams.Publisher
 import sttp.capabilities.fs2.Fs2Streams
 import sttp.monad.MonadError
 import sttp.tapir.integ.cats.effect.CatsMonadError
 import sttp.tapir.model.ServerRequest
-import sttp.tapir.server.interpreter.RawValue
 import sttp.tapir.server.netty.internal.{NettyStreamingRequestBody, StreamCompatible}
-import sttp.tapir.{RawBodyType, RawPart, TapirFile}
+import sttp.tapir.{RawPart, TapirFile}
+
+import java.nio.file.{Files => JFiles}
+import scala.util.Try
 
 private[cats] class NettyCatsRequestBody[F[_]: Async](
     val createFile: ServerRequest => F[TapirFile],
-    val streamCompatible: StreamCompatible[Fs2Streams[F]]
-) extends NettyStreamingRequestBody[F, Fs2Streams[F]] {
+    val streamCompatible: StreamCompatible[Fs2Streams[F]],
+    multipartTempDirectory: Option[TapirFile] = None,
+    multipartMinSizeForDisk: Option[Long] = None
+) extends NettyStreamingRequestBody[F, Fs2Streams[F]](
+      multipartTempDirectory,
+      multipartMinSizeForDisk
+    ) {
 
   override implicit val monad: MonadError[F] = new CatsMonadError()
 
+  import cats.implicits._
+
+  override protected val listMonadToMonadOfList: List[F[RawPart]] => F[List[RawPart]] = _.sequence
+
   override def publisherToBytes(publisher: Publisher[HttpContent], contentLength: Option[Long], maxBytes: Option[Long]): F[Array[Byte]] =
     streamCompatible.fromPublisher(publisher, maxBytes).compile.to(Chunk).map(_.toArray[Byte])
 
-  def publisherToMultipart(
-      nettyRequest: StreamedHttpRequest,
-      serverRequest: ServerRequest,
-      m: RawBodyType.MultipartBody,
-      maxBytes: Option[Long]
-  ): F[RawValue[Seq[RawPart]]] = monad.error(new UnsupportedOperationException("Multipart requests are not supported"))
-
   override def writeToFile(serverRequest: ServerRequest, file: TapirFile, maxBytes: Option[Long]): F[Unit] =
     (toStream(serverRequest, maxBytes)
       .asInstanceOf[streamCompatible.streams.BinaryStream])
@@ -42,5 +44,5 @@ private[cats] class NettyCatsRequestBody[F[_]: Async](
       .drain
 
   override def writeBytesToFile(bytes: Array[Byte], file: TapirFile): F[Unit] =
-    monad.error(new UnsupportedOperationException("Multipart requests are not supported"))
+    monad.fromTry(for { _ <- Try(JFiles.write(file.toPath, bytes)) } yield ())
 }

server/netty-server/cats/src/test/scala/sttp/tapir/server/netty/cats/NettyCatsServerTest.scala

@@ -24,17 +24,15 @@ class NettyCatsServerTest extends TestSuite with EitherValues {
 
           val interpreter = new NettyCatsTestServerInterpreter(eventLoopGroup, dispatcher)
           val createServerTest = new DefaultCreateServerTest(backend, interpreter)
-          val ioSleeper: Sleeper[IO] = new Sleeper[IO] {
-            override def sleep(duration: FiniteDuration): IO[Unit] = IO.sleep(duration)
-          }
+          val ioSleeper: Sleeper[IO] = (duration: FiniteDuration) => IO.sleep(duration)
           def drainFs2(stream: Fs2Streams[IO]#BinaryStream): IO[Unit] =
             stream.compile.drain.void
 
           val tests = new AllServerTests(
             createServerTest,
             interpreter,
             backend,
-            multipart = false
+            partOtherHeaderSupport = false
           )
             .tests() ++
             new ServerStreamingTests(createServerTest).tests(Fs2Streams[IO])(drainFs2) ++

server/netty-server/cats/src/test/scala/sttp/tapir/server/netty/cats/NettyCatsTestServerInterpreter.scala

@@ -2,16 +2,17 @@ package sttp.tapir.server.netty.cats
 
 import cats.effect.std.Dispatcher
 import cats.effect.{IO, Resource}
-import io.netty.channel.nio.NioEventLoopGroup
+import io.netty.channel.EventLoopGroup
 import sttp.capabilities.WebSockets
 import sttp.tapir.server.ServerEndpoint
 import sttp.tapir.server.netty.{NettyConfig, Route}
 import sttp.tapir.server.tests.TestServerInterpreter
 import sttp.tapir.tests._
 import sttp.capabilities.fs2.Fs2Streams
+
 import scala.concurrent.duration.FiniteDuration
 
-class NettyCatsTestServerInterpreter(eventLoopGroup: NioEventLoopGroup, dispatcher: Dispatcher[IO])
+class NettyCatsTestServerInterpreter(eventLoopGroup: EventLoopGroup, dispatcher: Dispatcher[IO])
     extends TestServerInterpreter[IO, Fs2Streams[IO] with WebSockets, NettyCatsServerOptions[IO], Route[IO]] {
   override def route(es: List[ServerEndpoint[Fs2Streams[IO] with WebSockets, IO]], interceptors: Interceptors): Route[IO] = {
     val serverOptions: NettyCatsServerOptions[IO] = interceptors(