Support for Streaming of Data Part of Multipart Form Fields

[khajavi]

Describe the bug
The current implementation allows streaming for form fields on the server side. However, there is a need to stream the body part of each form field individually without waiting for the entire form data to be received from the client.

To Reproduce

1. Run the provided server-side code.
2. Use the provided client-side code to send a multipart form with binary fields to the server.
3. Observe that the server waits for the entire form data to be received from the client, before processing them.

import zio._
import zio.http._
import zio.stream.{ZSink, ZStream}

object MultipartFormDataStreaming extends ZIOAppDefault {
  def logBytes = (b: Byte) => ZIO.log(s"received byte: $b")

  private val app: HttpApp[Any] =
    Routes(
      Method.POST / "upload-stream" / "simple"     -> handler { (req: Request) =>
        for {
          count <- req.body.asStream.tap(logBytes).run(ZSink.count)
        } yield Response.text(count.toString)
      },
      Method.POST / "upload-stream" / "form-field" -> handler { (req: Request) =>
        if (req.header(Header.ContentType).exists(_.mediaType == MediaType.multipart.`form-data`))
          for {
            form  <- req.body.asMultipartFormStream
            count <- form.fields
              .tap(f => ZIO.log(s"started reading new field: ${f.name}"))
              .flatMap {
                case sb: FormField.StreamingBinary => sb.data.tap(logBytes)
                case _                             => ZStream.empty
              }
              .run(ZSink.count)
          } yield Response.text(count.toString)
        else ZIO.succeed(Response(status = Status.NotFound))
      },
    ).sandbox.toHttpApp @@ Middleware.debug

  override def run: ZIO[Any with ZIOAppArgs with Scope, Any, Any] =
    Server.serve(app).provide(ZLayer.succeed(Server.Config.default.enableRequestStreaming), Server.live)
}

for {
  url    <- ZIO.fromEither(URL.decode("http://localhost:8080/upload-stream"))
  client <- ZIO.serviceWith[Client](_.url(url) @@ ZClientAspect.requestLogging())
  form = Form(
    Chunk(
      ("foo", "This is the first part of the foo form field."),
      ("foo", "This is the second part of the foo form field."),
      ("bar", "This is the body of the bar form field."),
    ).map { case (name, data) =>
      FormField.streamingBinaryField(
        name = name,
        data = ZStream.fromChunk(Chunk.fromArray(data.getBytes)).schedule(Schedule.fixed(200.milli)),
        mediaType = MediaType.application.`octet-stream`,
      )
    },
  )
  res <- client.request(
    Request
      .post(
        path = "form-field",
        body = Body.fromMultipartForm(form, Boundary("boundary123")),
      ),
  )
} yield ()

Expected behaviour
I expect to receive the data portion of each binary field as they are sent to the server, without waiting for the client to finish sending all bytes of the form field data. So when the server starts reading a new form field, the 'logBytes' logs incoming bytes as they are received.

[jdegoes]

/bounty $250

[algora-pbc]

💎 $250 bounty • ZIO

Steps to solve:

1. Start working: Comment /attempt #2734 with your implementation plan
2. Submit work: Create a pull request including /claim #2734 in the PR body to claim the bounty
3. Receive payment: 100% of the bounty is received 2-5 days post-reward. Make sure you are eligible for payouts

Additional opportunities:

• 🔴 Livestream on Algora TV while solving this bounty & earn $200 upon merge! Make sure to have your camera and microphone on. Comment /livestream once live

Thank you for contributing to zio/zio-http!

Add a bounty • Share on socials

Attempt	Started (GMT+0)	Solution
🟢 @kyri-petrou	Jun 9, 2024, 12:19:56 PM	#2899

[kyri-petrou]

/attempt #2734

Algora profile	Completed bounties	Tech	Active attempts	Options
@kyri-petrou	13 ZIO bounties	Scala, Python, Shell & more		Cancel attempt

[algora-pbc]

💡 @kyri-petrou submitted a pull request that claims the bounty. You can visit your bounty board to reward.

[algora-pbc]

🎉🎈 @kyri-petrou has been awarded $250! 🎈🎊

• Share on socials
• Give feedback