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

Author: ajozwik

build.sbt

@@ -92,6 +92,11 @@ val commonSettings = commonSmlBuildSettings ++ ossPublishSettings ++ Seq(
   evictionErrorLevel := Level.Info
 )
 
+val testSettings = Seq(
+  scalacOptions += "-Wconf:msg=unused value of type org.scalatest.Assertion:s",
+  scalacOptions += "-Wconf:msg=unused value of type org.scalatest.compatible.Assertion:s"
+)
+
 val versioningSchemeSettings = Seq(versionScheme := Some("early-semver"))
 
 val enableMimaSettings = Seq(
@@ -539,6 +544,7 @@ lazy val testing: ProjectMatrix = (projectMatrix in file("testing"))
 
 lazy val tests: ProjectMatrix = (projectMatrix in file("tests"))
   .settings(commonSettings)
+  .settings(testSettings)
   .settings(
     name := "tapir-tests",
     libraryDependencies ++= Seq(
@@ -1375,6 +1381,7 @@ lazy val serverCore: ProjectMatrix = (projectMatrix in file("server/core"))
 
 lazy val serverTests: ProjectMatrix = (projectMatrix in file("server/tests"))
   .settings(commonSettings)
+  .settings(testSettings)
   .settings(
     name := "tapir-server-tests",
     libraryDependencies ++= Seq(
@@ -1646,6 +1653,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
     )
   )
@@ -2059,6 +2067,7 @@ lazy val awsExamples2_13 = awsExamples.jvm(scala2_13).dependsOn(awsSam.jvm(scala
 
 lazy val clientTests: ProjectMatrix = (projectMatrix in file("client/tests"))
   .settings(commonSettings)
+  .settings(testSettings)
   .settings(
     name := "tapir-client-tests"
   )
@@ -2326,6 +2335,7 @@ lazy val openapiCodegenCli: ProjectMatrix = (projectMatrix in file("openapi-code
 
 lazy val examples: ProjectMatrix = (projectMatrix in file("examples"))
   .settings(commonSettings)
+  .settings(testSettings)
   .settings(
     name := "tapir-examples",
     libraryDependencies ++= Seq(

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

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

@@ -1,37 +1,36 @@
 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}
 
 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 +41,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"))
+    Async[F].fromFuture(Async[F].delay(writeBytesToFileFuture(bytes, file)))
 }

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(

server/netty-server/src/main/scala/sttp/tapir/server/netty/NettyFutureServerInterpreter.scala

@@ -20,7 +20,7 @@ trait NettyFutureServerInterpreter {
     NettyServerInterpreter.toRoute(
       ses,
       nettyServerOptions.interceptors,
-      new NettyFutureRequestBody(nettyServerOptions.createFile),
+      new NettyFutureRequestBody(nettyServerOptions.createFile, None, None),
       new NettyToResponseBody[Future](RunAsync.Future),
       nettyServerOptions.deleteFile,
       RunAsync.Future

server/netty-server/src/main/scala/sttp/tapir/server/netty/NettyServerRequest.scala

@@ -4,7 +4,6 @@ import scala.collection.JavaConverters._
 import scala.collection.immutable.Seq
 import io.netty.channel.ChannelHandlerContext
 import io.netty.handler.codec.http.{HttpRequest, QueryStringDecoder}
-import io.netty.handler.ssl.SslHandler
 import sttp.model.{Header, Method, QueryParams, Uri}
 import sttp.tapir.{AttributeKey, AttributeMap}
 import sttp.tapir.model.{ConnectionInfo, ServerRequest}

server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/NettyBootstrap.scala

@@ -1,7 +1,7 @@
 package sttp.tapir.server.netty.internal
 
 import io.netty.bootstrap.ServerBootstrap
-import io.netty.channel.{Channel, ChannelFuture, ChannelHandler, ChannelInitializer, ChannelOption, EventLoopGroup}
+import io.netty.channel.{Channel, ChannelFuture, ChannelInitializer, ChannelOption, EventLoopGroup}
 import sttp.tapir.server.netty.NettyConfig
 
 import java.net.{InetSocketAddress, SocketAddress}

server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/NettyFutureRequestBody.scala

@@ -7,40 +7,38 @@ import sttp.capabilities
 import sttp.monad.{FutureMonad, MonadError}
 import sttp.tapir.capabilities.NoStreams
 import sttp.tapir.model.ServerRequest
-import sttp.tapir.server.interpreter.RawValue
 import sttp.tapir.server.netty.internal.reactivestreams._
-import sttp.tapir.{RawBodyType, RawPart, TapirFile}
+import sttp.tapir.{RawPart, TapirFile}
 
 import scala.concurrent.{ExecutionContext, Future}
 
-private[netty] class NettyFutureRequestBody(val createFile: ServerRequest => Future[TapirFile])(implicit ec: ExecutionContext)
-    extends NettyRequestBody[Future, NoStreams] {
+private[netty] class NettyFutureRequestBody(
+    val createFile: ServerRequest => Future[TapirFile],
+    multipartTempDirectory: Option[TapirFile],
+    multipartMinSizeForDisk: Option[Long]
+)(implicit ec: ExecutionContext)
+    extends NettyRequestBodyWithMultipartF[Future, NoStreams](multipartTempDirectory, multipartMinSizeForDisk) {
 
   override val streams: capabilities.Streams[NoStreams] = NoStreams
   override implicit val monad: MonadError[Future] = new FutureMonad()
 
+  protected val listMonadToMonadOfList: List[Future[RawPart]] => Future[List[RawPart]] = l => Future.sequence(l)
+
   override def publisherToBytes(
       publisher: Publisher[HttpContent],
       contentLength: Option[Long],
       maxBytes: Option[Long]
   ): Future[Array[Byte]] =
     SimpleSubscriber.processAll(publisher, contentLength, maxBytes)
 
-  override def publisherToMultipart(
-      nettyRequest: StreamedHttpRequest,
-      serverRequest: ServerRequest,
-      m: RawBodyType.MultipartBody,
-      maxBytes: Option[Long]
-  ): Future[RawValue[Seq[RawPart]]] = Future.failed(new UnsupportedOperationException("Multipart requests are not supported"))
-
   override def writeToFile(serverRequest: ServerRequest, file: TapirFile, maxBytes: Option[Long]): Future[Unit] =
     serverRequest.underlying match {
       case r: StreamedHttpRequest => FileWriterSubscriber.processAll(r, file.toPath, maxBytes)
       case _                      => monad.unit(()) // Empty request
     }
 
   override def writeBytesToFile(bytes: Array[Byte], file: TapirFile): Future[Unit] =
-    Future.failed(new UnsupportedOperationException("Multipart requests are not supported"))
+    writeBytesToFileFuture(bytes, file)
 
   override def toStream(serverRequest: ServerRequest, maxBytes: Option[Long]): streams.BinaryStream =
     throw new UnsupportedOperationException()

server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/NettyRequestBodyWithMultipart.scala

@@ -0,0 +1,20 @@
+package sttp.tapir.server.netty.internal
+
+import io.netty.handler.codec.http.multipart.{DefaultHttpDataFactory, HttpDataFactory}
+import sttp.capabilities.Streams
+import sttp.tapir.TapirFile
+
+private[netty] abstract class NettyRequestBodyWithMultipart[F[_], S <: Streams[S]](
+    multipartTempDirectory: Option[TapirFile],
+    multipartMinSizeForDisk: Option[Long]
+) extends NettyRequestBody[F, S] {
+  protected val httpDataFactory: HttpDataFactory = {
+    val factory = multipartMinSizeForDisk match {
+      case Some(minSize) => new DefaultHttpDataFactory(minSize)
+      case None          => new DefaultHttpDataFactory()
+    }
+    multipartTempDirectory.foreach(dir => factory.setBaseDir(dir.getPath))
+    factory
+  }
+
+}