Fix response compression for static files (#2856)

zio-http/jvm/src/main/scala/zio/http/netty/NettyBodyWriter.scala

@@ -16,21 +16,32 @@
 
 package zio.http.netty
 
+import scala.annotation.tailrec
+
 import zio.Chunk.ByteArray
 import zio._
 import zio.stacktracer.TracingImplicits.disableAutoTrace
 
+import zio.stream.ZStream
+
 import zio.http.Body
 import zio.http.Body._
 import zio.http.netty.NettyBody.{AsciiStringBody, AsyncBody, ByteBufBody, UnsafeAsync}
 
 import io.netty.buffer.Unpooled
 import io.netty.channel._
 import io.netty.handler.codec.http.{DefaultHttpContent, LastHttpContent}
+import io.netty.handler.stream.ChunkedNioFile
 
 object NettyBodyWriter {
 
-  def writeAndFlush(body: Body, contentLength: Option[Long], ctx: ChannelHandlerContext)(implicit
+  @tailrec
+  def writeAndFlush(
+    body: Body,
+    contentLength: Option[Long],
+    ctx: ChannelHandlerContext,
+    compressionEnabled: Boolean,
+  )(implicit
     trace: Trace,
   ): Option[Task[Unit]] = {
 
@@ -46,19 +57,21 @@ object NettyBodyWriter {
     }
 
     body match {
-      case body: ByteBufBody                  =>
+      case body: ByteBufBody                    =>
         ctx.write(new DefaultHttpContent(body.byteBuf))
         ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT)
         None
-      case body: FileBody                     =>
-        val file = body.file
-        // Write the content.
-        ctx.write(new DefaultFileRegion(file, 0, body.fileSize))
-
-        // Write the end marker.
+      case body: FileBody if compressionEnabled =>
+        // We need to stream the file when compression is enabled otherwise the response encoding fails
+        val stream = ZStream.fromFile(body.file)
+        val size   = Some(body.fileSize)
+        val s      = StreamBody(stream, knownContentLength = size, mediaType = body.mediaType)
+        NettyBodyWriter.writeAndFlush(s, size, ctx, compressionEnabled)
+      case body: FileBody                       =>
+        ctx.write(new DefaultFileRegion(body.file, 0, body.fileSize))
         ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT)
         None
-      case AsyncBody(async, _, _, _)          =>
+      case AsyncBody(async, _, _, _)            =>
         async(
           new UnsafeAsync {
             override def apply(message: Chunk[Byte], isLast: Boolean): Unit = {
@@ -74,10 +87,10 @@ object NettyBodyWriter {
           },
         )
         None
-      case AsciiStringBody(asciiString, _, _) =>
+      case AsciiStringBody(asciiString, _, _)   =>
         writeArray(asciiString.array(), isLast = true)
         None
-      case StreamBody(stream, _, _, _)        =>
+      case StreamBody(stream, _, _, _)          =>
         Some(
           contentLength.orElse(body.knownContentLength) match {
             case Some(length) =>
@@ -118,13 +131,13 @@ object NettyBodyWriter {
               }
           },
         )
-      case ArrayBody(data, _, _)              =>
+      case ArrayBody(data, _, _)                =>
         writeArray(data, isLast = true)
         None
-      case ChunkBody(data, _, _)              =>
+      case ChunkBody(data, _, _)                =>
         writeArray(data.toArray, isLast = true)
         None
-      case EmptyBody                          =>
+      case EmptyBody                            =>
         ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT)
         None
     }

zio-http/jvm/src/main/scala/zio/http/netty/client/ClientInboundHandler.scala

@@ -56,7 +56,7 @@ final class ClientInboundHandler(
         ctx.writeAndFlush(fullRequest)
       case _: HttpRequest               =>
         ctx.write(jReq)
-        NettyBodyWriter.writeAndFlush(req.body, None, ctx).foreach { effect =>
+        NettyBodyWriter.writeAndFlush(req.body, None, ctx, compressionEnabled = false).foreach { effect =>
           rtm.run(ctx, NettyRuntime.noopEnsuring)(effect)(Unsafe.unsafe, trace)
         }
     }

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

@@ -187,7 +187,7 @@ private[zio] final case class ServerInboundHandler(
             }
 
           ctx.writeAndFlush(jResponse)
-          NettyBodyWriter.writeAndFlush(response.body, contentLength, ctx)
+          NettyBodyWriter.writeAndFlush(response.body, contentLength, ctx, isResponseCompressible(jRequest))
         } else {
           ctx.writeAndFlush(jResponse)
           None
@@ -206,6 +206,16 @@ private[zio] final case class ServerInboundHandler(
     }
   }
 
+  private def isResponseCompressible(req: HttpRequest): Boolean = {
+    config.responseCompression match {
+      case None      => false
+      case Some(cfg) =>
+        val headers    = req.headers()
+        val headerName = Header.AcceptEncoding.name
+        cfg.options.exists(opt => headers.containsValue(headerName, opt.kind.name, true))
+    }
+  }
+
   private def makeZioRequest(ctx: ChannelHandlerContext, nettyReq: HttpRequest): Request = {
     val nettyHttpVersion = nettyReq.protocolVersion()
     val protocolVersion  = nettyHttpVersion match {

zio-http/jvm/src/test/scala/zio/http/ResponseCompressionSpec.scala

@@ -73,7 +73,12 @@ object ResponseCompressionSpec extends ZIOHttpSpec {
         ),
     )
 
-  private val app                              = text ++ stream
+  private val file: Routes[Any, Response] =
+    Routes(
+      Method.GET / "file" -> Handler.fromResource("TestStatic/TestFile1.txt"),
+    ).sandbox
+
+  private val app                              = text ++ stream ++ file
   private lazy val serverConfig: Server.Config = Server.Config.default.port(0).responseCompression()
 
   override def spec =
@@ -100,6 +105,22 @@ object ResponseCompressionSpec extends ZIOHttpSpec {
       test("with Response.stream (chunked)") {
         streamTest("stream-chunked")
       },
+      test("with files") {
+        for {
+          server       <- ZIO.service[Server]
+          client       <- ZIO.service[Client]
+          _            <- server.install(app)
+          response     <- client.request(
+            Request(
+              method = Method.GET,
+              url = URL(Path.root / "file", kind = URL.Location.Absolute(Scheme.HTTP, "localhost", Some(server.port))),
+            )
+              .addHeader(Header.AcceptEncoding(Header.AcceptEncoding.GZip(), Header.AcceptEncoding.Deflate())),
+          )
+          res          <- response.body.asChunk
+          decompressed <- decompressed(res)
+        } yield assertTrue(decompressed == "This file is added for testing Static File Server.")
+      },
     ).provide(
       ZLayer.succeed(serverConfig),
       Server.customized,

zio-http/shared/src/main/scala/zio/http/Server.scala

@@ -289,11 +289,13 @@ object Server extends ServerPlatformSpecific {
       def deflate(level: Int = DefaultLevel, bits: Int = DefaultBits, mem: Int = DefaultMem): CompressionOptions =
         CompressionOptions(level, bits, mem, CompressionType.Deflate)
 
-      sealed trait CompressionType
+      sealed trait CompressionType {
+        val name: String
+      }
 
       private[http] object CompressionType {
-        case object GZip    extends CompressionType
-        case object Deflate extends CompressionType
+        case object GZip    extends CompressionType { val name = "gzip"    }
+        case object Deflate extends CompressionType { val name = "deflate" }
 
         lazy val config: zio.Config[CompressionType] =
           zio.Config.string.mapOrFail {