Multipart zio#1559

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

@@ -0,0 +1,30 @@
+package example
+
+import zio.http._
+import zio.http.model.Multipart.FileUpload
+import zio.stream.ZPipeline
+import zio.{ZIO, ZIOAppDefault, ZLayer}
+
+object MultipartServer extends ZIOAppDefault {
+
+  val app = Http.collectZIO[Request] { case req =>
+    ZIO.scoped {
+      req.body.multipart.flatMap(x =>
+        ZIO.foreachDiscard(x)(m =>
+          (ZIO.debug(s"${m.name} - ${m.getClass.getSimpleName} - ${m.length}")) *>
+            ZIO.whenCase(m) { case u: FileUpload =>
+              ZIO.debug(s"File name: ${u.filename}}") *>
+                u.content.via(ZPipeline.utf8Decode >>> ZPipeline.splitLines).foreach(ZIO.debug(_))
+            },
+        ),
+      )
+    } as Response.ok
+  }
+
+  val run =
+    ZIO.debug("To test upload big file:\ncurl -v -F key1=value1 -F upload=@bigfile.txt localhost:8080") *>
+      Server
+        .serve(app)
+        .provide(ZLayer.succeed(ServerConfig.default.objectAggregator(1024 * 3000)), Server.live)
+        .exitCode
+}

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

@@ -1,17 +1,20 @@
 package zio.http
 
 import io.netty.buffer.{ByteBuf, ByteBufUtil, Unpooled}
-import io.netty.channel.{Channel => JChannel, DefaultFileRegion}
-import io.netty.handler.codec.http.LastHttpContent
+import io.netty.channel.{DefaultFileRegion, Channel => JChannel}
+import io.netty.handler.codec.http.{FullHttpRequest, LastHttpContent}
+import io.netty.handler.codec.http.multipart.{HttpPostMultipartRequestDecoder, InterfaceHttpData}
 import io.netty.util.AsciiString
 import zio._
-import zio.http.model.HTTP_CHARSET
+import zio.http.model.{HTTP_CHARSET, Multipart}
 import zio.http.service.Ctx
 import zio.stream.ZStream
 
 import java.io.FileInputStream
 import java.nio.charset.Charset
-import zio.stacktracer.TracingImplicits.disableAutoTrace // scalafix:ok;
+import zio.stacktracer.TracingImplicits.disableAutoTrace
+
+import scala.jdk.CollectionConverters._ // scalafix:ok;
 
 /**
  * Holds Body that needs to be written on the HttpChannel
@@ -44,6 +47,10 @@ trait Body { self =>
   def isComplete: Boolean
 
   def write(ctx: Ctx)(implicit trace: Trace): Task[Boolean]
+
+  // TODO It looks that multipart/form-data can not be used together with chunked transfer.
+  // So I need to implement it only in fromFullHttpRequest. Is it correct?
+  def multipart(implicit trace: Trace): RIO[Scope, List[Multipart]] = ZIO.succeed(List.empty)
 }
 
 object Body {
@@ -97,6 +104,27 @@ object Body {
     override def toString(): String = s"Body.fromByteBuf($byteBuf)"
   }
 
+  def fromFullHttpRequest(request: FullHttpRequest): Body = new Body {
+    override def isComplete: Boolean = true
+
+    override def asChunk(implicit trace: Trace): Task[Chunk[Byte]] = ZIO.attempt {
+      Chunk.fromArray(ByteBufUtil.getBytes(request.content()))
+    }
+
+    override def asStream(implicit trace: Trace): ZStream[Any, Throwable, Byte] =
+      ZStream.unwrap(asChunk.map(ZStream.fromChunk(_)))
+
+    override def write(ctx: Ctx)(implicit trace: Trace): Task[Boolean] =
+      ZIO.attempt(ctx.write(request.content()): Unit).as(false)
+
+    override def multipart(implicit trace: Trace): RIO[Scope, List[Multipart]] =
+      ZIO
+        .succeed(new HttpPostMultipartRequestDecoder(request))
+        .withFinalizer(decoder => ZIO.attempt(decoder.destroy()).orDie)
+        .map(_.getBodyHttpDatas.asScala.toList)
+        .flatMap(ZIO.foreach(_)(MultipartConverter.convert))
+  }
+
   /**
    * Helper to create Body from CharSequence
    */

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

@@ -0,0 +1,66 @@
+package zio.http
+
+import io.netty.handler.codec.http.multipart._
+import zio.{RIO, Scope, ZIO}
+import zio.http.model.Multipart
+import zio.http.model.Multipart.{Attribute, FileUpload}
+import zio.stream.ZStream
+
+import java.nio.file.Paths
+
+object MultipartConverter {
+
+  def convert(data: InterfaceHttpData): RIO[Scope, Multipart] = data match {
+
+    case attribute: io.netty.handler.codec.http.multipart.Attribute =>
+      val value: RIO[Scope, String] = attribute match {
+        case a: DiskAttribute   => ZIO.attempt(a.getValue)
+        case a: MemoryAttribute => ZIO.succeed(a.getValue)
+        case a: MixedAttribute  =>
+          // TODO should I just use this instead of pattern matching?
+          if (a.isInMemory)
+            ZIO.succeed(a.getValue)
+          else
+            ZIO.attempt(a.getValue)
+      }
+
+      value.map(v =>
+        Attribute(
+          attribute.getName,
+          attribute.length,
+          attribute.definedLength,
+          attribute.getCharset.name,
+          v,
+        ),
+      )
+
+    case fileUpload: io.netty.handler.codec.http.multipart.FileUpload =>
+      val value: ZStream[Any, Throwable, Byte] = fileUpload match {
+        case u: DiskFileUpload   =>
+          ZStream.fromPath(u.getFile.toPath)
+        case u: MemoryFileUpload =>
+          ZStream.fromIterable(u.get)
+        case u: MixedFileUpload  =>
+          // TODO should I just use this instead of pattern matching?
+          if (u.isInMemory)
+            ZStream.fromIterable(u.get)
+          else {
+            println(s"FFF ${u.getFile.toPath}")
+            ZStream.fromPath(u.getFile.toPath)
+          }
+      }
+      ZIO.succeed(
+        FileUpload(
+          fileUpload.getName,
+          fileUpload.length,
+          fileUpload.definedLength,
+          fileUpload.getCharset.name,
+          fileUpload.getFilename,
+          fileUpload.getContentType,
+          fileUpload.getContentTransferEncoding,
+          value,
+        ),
+      )
+  }
+
+}

zio-http/src/main/scala/zio/http/model/Multipart.scala

@@ -0,0 +1,51 @@
+package zio.http.model
+
+import zio.http.model.Multipart.{Attribute, FileUpload}
+import zio.stream.ZStream
+
+sealed trait Multipart {
+  def name: String
+  def length: Long
+
+  /**
+   * Returns the defined length of the HttpData.
+   *
+   * If no Content-Length is provided in the request, the defined length is
+   * always 0 (whatever during decoding or in final state).
+   *
+   * If Content-Length is provided in the request, this is this given defined
+   * length. This value does not change, whatever during decoding or in the
+   * final state.
+   *
+   * This method could be used for instance to know the amount of bytes
+   * transmitted for one particular HttpData, for example one {@link FileUpload}
+   * or any known big {@link Attribute}.
+   *
+   * @return
+   *   the defined length of the HttpData
+   */
+  def definedLength: Long
+  def charset: String
+}
+
+object Multipart {
+
+  case class Attribute(
+    name: String,
+    length: Long,
+    definedLength: Long,
+    charset: String,
+    value: String,
+  ) extends Multipart
+
+  case class FileUpload(
+    name: String,
+    length: Long,
+    definedLength: Long,
+    charset: String,
+    filename: String,
+    contentType: String,
+    contentTransferEncoding: String,
+    content: ZStream[Any, Throwable, Byte],
+  ) extends Multipart
+}

zio-http/src/main/scala/zio/http/netty/server/ServerInboundHandler.scala

@@ -113,7 +113,7 @@ private[zio] final case class ServerInboundHandler(
       nettyReq match {
         case nettyReq: FullHttpRequest =>
           Request(
-            Body.fromByteBuf(nettyReq.content()),
+            Body.fromFullHttpRequest(nettyReq),
             Headers.make(nettyReq.headers()),
             Method.fromHttpMethod(nettyReq.method()),
             URL.fromString(nettyReq.uri()).getOrElse(URL.empty),