fix(ext/http): flush gzip streaming response (#23991)

This commit changes `gzip` compression in `Deno.serve` API to flush data
after each write. There's a slight performance regression, but provided
test shows a scenario that was not possible before.

---------

Co-authored-by: Divy Srivastava <dj.srivastava23@gmail.com>

1 files changed, 57 insertions, 0 deletions

diff --git a/tests/unit/serve_test.ts b/tests/unit/serve_test.ts
index f7f01076b..ff77578e6 100644
--- a/tests/unit/serve_test.ts
+++ b/tests/unit/serve_test.ts
@@ -11,6 +11,7 @@ import {
   curlRequest,
   curlRequestWithStdErr,
   execCode,
+  execCode3,
   fail,
   tmpUnixSocketPath,
 } from "./test_util.ts";
@@ -3985,3 +3986,59 @@ Deno.test(
     assert(respText === "Internal Server Error");
   },
 );
+
+Deno.test(
+  {
+    permissions: { net: true, run: true, read: true },
+    ignore: Deno.build.os !== "linux",
+  },
+  async function gzipFlushResponseStream() {
+    const { promise, resolve } = Promise.withResolvers<void>();
+    const ac = new AbortController();
+
+    console.log("Starting server", servePort);
+    let timer: number | undefined = undefined;
+    let _controller;
+
+    const server = Deno.serve(
+      {
+        port: servePort,
+        onListen: onListen(resolve),
+        signal: ac.signal,
+      },
+      () => {
+        const body = new ReadableStream({
+          start(controller) {
+            timer = setInterval(() => {
+              const message = `It is ${new Date().toISOString()}\n`;
+              controller.enqueue(new TextEncoder().encode(message));
+            }, 1000);
+            _controller = controller;
+          },
+          cancel() {
+            if (timer !== undefined) {
+              clearInterval(timer);
+            }
+          },
+        });
+        return new Response(body, {
+          headers: {
+            "content-type": "text/plain",
+            "x-content-type-options": "nosniff",
+          },
+        });
+      },
+    );
+    await promise;
+    const e = await execCode3("/usr/bin/sh", [
+      "-c",
+      `curl --stderr - -N --compressed --no-progress-meter http://localhost:${servePort}`,
+    ]);
+    await e.waitStdoutText("It is ");
+    clearTimeout(timer);
+    _controller!.close();
+    await e.finished();
+    ac.abort();
+    await server.finished;
+  },
+);