Implement multipart for netty-future, netty-cats & netty-zio (#5315)

Proposal of multipart implementation for netty - see #4851 - only
multipart.

Implementation based on `nettyServerSync` implementation - but without
OxStreams.

According to [Note: Netty's multipart decoder does not expose other part
headers, nor other disposition
params.](https://github.com/softwaremill/tapir/blob/a4343ea90db20bec3cdad835c0c8e119ef788120/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/NettyRequestBody.scala#L135)
the test with netty should ignoring part headers -
`partOtherHeaderSupport = false`

Issue for support headers has been requested: [FileUpload or HttpData
should contain HTTP-Headers
](netty/netty#15445)

What is missing?
 - Part headers if netty api expose it
 - Configuration for multipart temp directory and minimum disk size
 - Streaming - but it is not part of multipart. 
 
For sync implementation multipart depends on OxStreams - what is not
needed and can be moved to other submodule.

---------

Co-authored-by: Adam Warski <adam.warski@softwaremill.com>
Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

Author: 3 people

build.sbt

@@ -87,8 +87,8 @@ val commonSettings = commonSmlBuildSettings ++ ossPublishSettings ++ Seq(
       case _            => Seq("-Xmax-inlines", "64")
     }
   },
-  Test / scalacOptions += "-Wconf:msg=unused value of type org.scalatest.Assertion:s",
-  Test / scalacOptions += "-Wconf:msg=unused value of type org.scalatest.compatible.Assertion:s",
+  scalacOptions += "-Wconf:msg=unused value of type org.scalatest.Assertion:s",
+  scalacOptions += "-Wconf:msg=unused value of type org.scalatest.compatible.Assertion:s",
   evictionErrorLevel := Level.Info
 )
 

observability/zio-opentelemetry/src/test/scala/sttp/tapir/server/ziopentelemetry/ZIOpenTelemetryTracingSpec.scala

@@ -131,7 +131,7 @@ object ZIOpenTelemetryTracingSpec extends ZIOSpecDefault {
               // draining mimics the backend sending the body; tolerate failures so the body-listener callback (which
               // marks the span as errored) runs and the finished span can be inspected
               ZIO.logDebug("streaming") *>
-              stream.runDrain.catchAllCause(_ => ZIO.unit)
+                stream.runDrain.catchAllCause(_ => ZIO.unit)
             case other =>
               ZIO.logDebug(s"WHat is it: $other ?")
           }
@@ -727,7 +727,9 @@ object ZIOpenTelemetryTracingSpec extends ZIOSpecDefault {
       val ep: ServerEndpoint[ZioStreams, Task] = endpoint
         .in("stream")
         .out(streamTextBody(ZioStreams)(CodecFormat.TextPlain(), Some(StandardCharsets.UTF_8)))
-        .zServerLogic(_ => ZIO.succeed(ZStream.fromIterable("abc".getBytes(StandardCharsets.UTF_8)).flatMap(_ => ZStream.fail(new RuntimeException("boom")))))
+        .zServerLogic(_ =>
+          ZIO.succeed(ZStream.fromIterable("abc".getBytes(StandardCharsets.UTF_8)).flatMap(_ => ZStream.fail(new RuntimeException("boom"))))
+        )
 
       val request = serverRequestFromUri(uri"http://example.com/stream")
       for {

server/netty-server/cats/src/main/scala/sttp/tapir/server/netty/cats/NettyCatsServerInterpreter.scala

@@ -33,7 +33,12 @@ trait NettyCatsServerInterpreter[F[_]] {
 
     val serverInterpreter = new ServerInterpreter[Fs2Streams[F] with WebSockets, F, NettyResponse, Fs2Streams[F]](
       FilterServerEndpoints(ses),
-      new NettyCatsRequestBody(createFile, Fs2StreamCompatible[F](nettyServerOptions.dispatcher)),
+      new NettyCatsRequestBody(
+        createFile,
+        Fs2StreamCompatible[F](nettyServerOptions.dispatcher),
+        nettyServerOptions.multipartTempDirectory,
+        nettyServerOptions.multipartMinSizeForDisk
+      ),
       new NettyToStreamsResponseBody(Fs2StreamCompatible[F](nettyServerOptions.dispatcher)),
       interceptors,
       deleteFile

server/netty-server/cats/src/main/scala/sttp/tapir/server/netty/cats/NettyCatsServerOptions.scala

@@ -18,7 +18,9 @@ case class NettyCatsServerOptions[F[_]](
     interceptors: List[Interceptor[F]],
     createFile: ServerRequest => F[TapirFile],
     deleteFile: TapirFile => F[Unit],
-    dispatcher: Dispatcher[F]
+    dispatcher: Dispatcher[F],
+    multipartTempDirectory: Option[TapirFile],
+    multipartMinSizeForDisk: Option[Long]
 ) {
   def prependInterceptor(i: Interceptor[F]): NettyCatsServerOptions[F] = copy(interceptors = i :: interceptors)
   def appendInterceptor(i: Interceptor[F]): NettyCatsServerOptions[F] = copy(interceptors = interceptors :+ i)
@@ -37,7 +39,9 @@ object NettyCatsServerOptions {
       interceptors,
       _ => Sync[F].delay(Defaults.createTempFile()),
       file => Sync[F].delay(Defaults.deleteFile()(file)),
-      dispatcher
+      dispatcher,
+      None,
+      None
     )
 
   def customiseInterceptors[F[_]: Async](

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

@@ -1,37 +1,37 @@
 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 scala.concurrent.Future
 
 private[cats] class NettyCatsRequestBody[F[_]: Async](
     val createFile: ServerRequest => F[TapirFile],
-    val streamCompatible: StreamCompatible[Fs2Streams[F]]
+    val streamCompatible: StreamCompatible[Fs2Streams[F]],
+    val multipartTempDirectory: Option[TapirFile],
+    val multipartMinSizeForDisk: Option[Long]
 ) extends NettyStreamingRequestBody[F, Fs2Streams[F]] {
 
   override implicit val monad: MonadError[F] = new CatsMonadError()
 
+  import cats.implicits._
+
+  override protected def listMonadToMonadOfList(l: List[F[RawPart]]): F[List[RawPart]] = l.sequence
+
+  override protected def fromFuture[T](f: Future[T]): F[T] = Async[F].fromFuture(Async[F].delay(f))
+
   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])
@@ -41,6 +41,4 @@ private[cats] class NettyCatsRequestBody[F[_]: Async](
       .compile
       .drain
 
-  override def writeBytesToFile(bytes: Array[Byte], file: TapirFile): F[Unit] =
-    monad.error(new UnsupportedOperationException("Multipart requests are not supported"))
 }

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

@@ -34,7 +34,7 @@ class NettyCatsServerTest extends TestSuite with EitherValues {
             createServerTest,
             interpreter,
             backend,
-            multipart = false
+            partOtherHeaderSupport = false
           )
             .tests() ++
             new ServerStreamingTests(createServerTest).tests(Fs2Streams[IO])(drainFs2) ++

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

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

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

@@ -17,7 +17,9 @@ import scala.concurrent.{Future, blocking}
 case class NettyFutureServerOptions(
     interceptors: List[Interceptor[Future]],
     createFile: ServerRequest => Future[TapirFile],
-    deleteFile: TapirFile => Future[Unit]
+    deleteFile: TapirFile => Future[Unit],
+    multipartTempDirectory: Option[TapirFile],
+    multipartMinSizeForDisk: Option[Long]
 ) {
   def prependInterceptor(i: Interceptor[Future]): NettyFutureServerOptions = copy(interceptors = i :: interceptors)
   def appendInterceptor(i: Interceptor[Future]): NettyFutureServerOptions = copy(interceptors = interceptors :+ i)
@@ -37,7 +39,9 @@ object NettyFutureServerOptions {
       file => {
         import scala.concurrent.ExecutionContext.Implicits.global
         Future(blocking(Defaults.deleteFile()(file)))
-      }
+      },
+      None,
+      None
     )
 
   /** Customise the interceptors that are being used when exposing endpoints as a server. */

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

@@ -7,41 +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],
+    val multipartTempDirectory: Option[TapirFile],
+    val multipartMinSizeForDisk: Option[Long]
+)(implicit ec: ExecutionContext)
+    extends NettyMonadRequestBody[Future, NoStreams] {
 
   override val streams: capabilities.Streams[NoStreams] = NoStreams
   override implicit val monad: MonadError[Future] = new FutureMonad()
 
+  protected def listMonadToMonadOfList(l: List[Future[RawPart]]): Future[List[RawPart]] = Future.sequence(l)
+
+  override protected def fromFuture[T](f: Future[T]): Future[T] = f
+
   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"))
-
   override def toStream(serverRequest: ServerRequest, maxBytes: Option[Long]): streams.BinaryStream =
     throw new UnsupportedOperationException()
 }

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

@@ -0,0 +1,144 @@
+package sttp.tapir.server.netty.internal
+
+import io.netty.handler.codec.http.HttpContent
+import io.netty.handler.codec.http.multipart.{HttpPostMultipartRequestDecoder, InterfaceHttpData}
+import org.playframework.netty.http.StreamedHttpRequest
+import org.reactivestreams.{Subscriber, Subscription}
+import sttp.capabilities.{StreamMaxLengthExceededException, Streams}
+import sttp.monad.syntax._
+import sttp.tapir.model.ServerRequest
+import sttp.tapir.server.interpreter.RawValue
+import sttp.tapir.{RawBodyType, RawPart, TapirFile}
+
+import java.nio.channels.{AsynchronousFileChannel, CompletionHandler}
+import java.nio.file.StandardOpenOption
+import java.nio.ByteBuffer
+import java.util.concurrent.atomic.{AtomicLong, AtomicReference}
+import scala.concurrent.{Future, Promise}
+
+/** Generic implementation of NettyRequestBody, for any effect F[_] and any stream type S. */
+private[netty] trait NettyMonadRequestBody[F[_], S <: Streams[S]] extends NettyRequestBody[F, S] {
+
+  protected def listMonadToMonadOfList(l: List[F[RawPart]]): F[List[RawPart]]
+  protected def fromFuture[T](f: Future[T]): F[T]
+
+  override def publisherToMultipart(
+      nettyRequest: StreamedHttpRequest,
+      serverRequest: ServerRequest,
+      m: RawBodyType.MultipartBody,
+      maxBytes: Option[Long]
+  ): F[RawValue[Seq[RawPart]]] = {
+
+    val promise = Promise[F[RawValue[Seq[RawPart]]]]()
+
+    val decoder = new HttpPostMultipartRequestDecoder(httpDataFactory, nettyRequest)
+    val subscriber = new MonadSubscriber(decoder, serverRequest, m, maxBytes, listMonadToMonadOfList)(promise)
+    nettyRequest.subscribe(subscriber)
+    val future = promise.future
+    val ff = fromFuture(future)
+    monad.flatten(ff)
+
+  }
+
+  private class MonadSubscriber(
+      decoder: HttpPostMultipartRequestDecoder,
+      serverRequest: ServerRequest,
+      m: RawBodyType.MultipartBody,
+      maxBytes: Option[Long],
+      seqMonadToMonadOfSeq: List[F[RawPart]] => F[List[RawPart]]
+  )(
+      promise: Promise[F[RawValue[Seq[RawPart]]]]
+  ) extends Subscriber[HttpContent] {
+    private val currentBytesRead: AtomicLong = new AtomicLong()
+    private val acc = new AtomicReference[List[F[RawPart]]](Nil)
+    private val subscription = new AtomicReference[Subscription]()
+
+    override def onSubscribe(s: Subscription): Unit = {
+      subscription.set(s)
+      subscription.get().request(1)
+    }
+
+    override def onNext(httpContent: HttpContent): Unit = {
+      val currentRead = currentBytesRead.accumulateAndGet(httpContent.content().readableBytes().toLong, (prev, toAdd) => prev + toAdd)
+      maxBytes match {
+        case Some(max) if currentRead > max =>
+          httpContent.release()
+          onError(StreamMaxLengthExceededException(max))
+        case _ =>
+          if (addContentSafe(httpContent)) {
+            val _ = actionOrOnError(
+              {
+                val parts = Iterator
+                  .continually(maybeNext())
+                  .takeWhile(_.nonEmpty)
+                  .flatten
+                  .flatMap(toPart)
+                  .toList
+                acc.getAndAccumulate(parts, (prev, toAdd) => prev ++ toAdd)
+                subscription.get().request(1)
+              },
+              ()
+            )
+          }
+      }
+    }
+
+    override def onError(t: Throwable): Unit = {
+      subscription.get().cancel()
+      val _ = promise.tryFailure(t)
+      decoder.destroy()
+    }
+
+    override def onComplete(): Unit = {
+      val f = seqMonadToMonadOfSeq(acc.get())
+      val r = f.map(p => RawValue.fromParts(p).copy(cleanup = Option(() => decoder.destroy())))
+      val _ = promise.trySuccess(r)
+    }
+
+    private def addContentSafe(httpContent: HttpContent): Boolean =
+      actionOrOnError(decoder.offer(httpContent), httpContent.release(): Unit)
+
+    private def actionOrOnError[T](action: => T, `finally`: => Unit): Boolean =
+      try {
+        val _ = action
+        true
+      } catch {
+        case r: RuntimeException =>
+          onError(r)
+          false
+      } finally {
+        `finally`
+      }
+
+    private def toPart(httpData: InterfaceHttpData): Option[F[RawPart]] =
+      m.partType(httpData.getName).map(partType => toRawPart(serverRequest, httpData, partType).map(identity))
+
+    private def maybeNext(): Option[InterfaceHttpData] = if (decoder.hasNext) Option(decoder.next()) else None
+  }
+
+  override def writeBytesToFile(bytes: Array[Byte], file: TapirFile): F[Unit] =
+    fromFuture(writeBytesToFileFuture(bytes, file))
+
+  private def writeBytesToFileFuture(bytes: Array[Byte], file: TapirFile): Future[Unit] = {
+    val promise = Promise[Unit]()
+    val channel = AsynchronousFileChannel.open(file.toPath, StandardOpenOption.WRITE, StandardOpenOption.CREATE)
+    channel.write(
+      ByteBuffer.wrap(bytes),
+      0,
+      channel,
+      new CompletionHandler[Integer, AutoCloseable]() {
+        override def completed(result: Integer, closeable: AutoCloseable): Unit = {
+          promise.success(())
+          closeable.close()
+        }
+
+        override def failed(exc: Throwable, closeable: AutoCloseable): Unit = {
+          promise.failure(exc)
+          closeable.close()
+        }
+      }
+    )
+    promise.future
+  }
+
+}