Add encodeBody, encode gzip, deflate and br, closes #72

Responses with `Content-Type`s including `gzip`, `deflate` or `br`
are now automatically encoded, unless `encodeBody` is set to
`manual`.

packages/core/src/standards/http.ts

@@ -327,12 +327,14 @@ export function withImmutableHeaders(req: Request): Request {
 }
 
 export interface ResponseInit extends BaseResponseInit {
+  readonly encodeBody?: "auto" | "manual";
   readonly webSocket?: WebSocket;
 }
 
 const kWaitUntil = Symbol("kWaitUntil");
 
 const enumerableResponseKeys: (keyof Response)[] = [
+  "encodeBody",
   "webSocket",
   "url",
   "redirected",
@@ -351,27 +353,31 @@ export class Response<
     return new Response(res.body, res);
   }
 
+  // https://developers.cloudflare.com/workers/runtime-apis/response#properties
+  // noinspection TypeScriptFieldCanBeMadeReadonly
+  #encodeBody: "auto" | "manual";
   // noinspection TypeScriptFieldCanBeMadeReadonly
   #status?: number;
   readonly #webSocket?: WebSocket;
   [kWaitUntil]?: Promise<WaitUntil>;
 
-  // TODO: add encodeBody: https://developers.cloudflare.com/workers/runtime-apis/response#properties
-
   constructor(body?: BodyInit, init?: ResponseInit | Response | BaseResponse) {
+    let encodeBody: string | undefined;
     let status: number | undefined;
     let webSocket: WebSocket | undefined;
     if (init instanceof BaseResponse && body === init.body) {
       // For cloning
       super(init);
     } else {
       if (init instanceof Response) {
+        encodeBody = init.#encodeBody;
         // Don't pass our strange hybrid Response to undici
         init = init[kInner];
-      } else if (!(init instanceof BaseResponse) /* ResponseInit */) {
+      } else if (!(init instanceof BaseResponse) /* ResponseInit */ && init) {
+        encodeBody = init.encodeBody;
         // Status 101 Switching Protocols would normally throw a RangeError, but we
         // need to allow it for WebSockets
-        if (init?.webSocket) {
+        if (init.webSocket) {
           if (init.status !== 101) {
             throw new RangeError(
               "Responses with a WebSocket must have status code 101."
@@ -384,6 +390,13 @@ export class Response<
       }
       super(new BaseResponse(body, init));
     }
+
+    encodeBody ??= "auto";
+    if (encodeBody !== "auto" && encodeBody !== "manual") {
+      throw new TypeError(`encodeBody: unexpected value: ${encodeBody}`);
+    }
+    this.#encodeBody = encodeBody;
+
     this.#status = status;
     this.#webSocket = webSocket;
 
@@ -399,13 +412,18 @@ export class Response<
     const clone = new Response(innerClone.body, innerClone);
     clone[kInputGated] = this[kInputGated];
     clone[kFormDataFiles] = this[kFormDataFiles];
+    clone.#encodeBody = this.#encodeBody;
     // Technically don't need to copy status, as it should only be set for
     // WebSocket handshake responses
     clone.#status = this.#status;
     clone[kWaitUntil] = this[kWaitUntil];
     return clone;
   }
 
+  get encodeBody(): "auto" | "manual" {
+    return this.#encodeBody;
+  }
+
   get webSocket(): WebSocket | undefined {
     return this.#webSocket;
   }

packages/core/test/standards/http.spec.ts

@@ -447,14 +447,20 @@ test("Response: can construct new Response from existing Response", async (t) =>
 test("Response: supports non-standard properties", (t) => {
   const pair = new WebSocketPair();
   const res = new Response(null, {
+    encodeBody: "manual",
     status: 101,
     webSocket: pair["0"],
     headers: { "X-Key": "value" },
   });
+  t.is(res.encodeBody, "manual");
   t.is(res.status, 101);
   t.is(res.webSocket, pair[0]);
   t.is(res.headers.get("X-Key"), "value");
 });
+test("Response: encodeBody defaults to auto", (t) => {
+  const res = new Response(null);
+  t.is(res.encodeBody, "auto");
+});
 test("Response: requires status 101 for WebSocket response", (t) => {
   const pair = new WebSocketPair();
   t.throws(() => new Response(null, { webSocket: pair["0"] }), {
@@ -470,16 +476,18 @@ test("Response: only allows status 101 for WebSocket response", (t) => {
   });
 });
 test("Response: clones non-standard properties", async (t) => {
-  const res = new Response("body");
+  const res = new Response("body", { encodeBody: "manual" });
   const waitUntil = [1, "2", true];
   withWaitUntil(res, Promise.resolve(waitUntil));
   t.is(await res.waitUntil(), waitUntil);
   const res2 = res.clone();
+  t.is(res2.encodeBody, "manual");
   t.is(await res2.waitUntil(), waitUntil);
 
   // Check prototype correct and clone still clones non-standard properties
   t.is(Object.getPrototypeOf(res2), Response.prototype);
   const res3 = res2.clone();
+  t.is(res3.encodeBody, "manual");
   t.is(await res3.waitUntil(), waitUntil);
   t.is(await res.text(), "body");
   t.is(await res2.text(), "body");
@@ -533,6 +541,7 @@ test("Response: Object.keys() returns getters", async (t) => {
     "body",
     "bodyUsed",
     "headers",
+    "encodeBody",
     "webSocket",
     "url",
     "redirected",

packages/http-server/src/index.ts

@@ -3,8 +3,11 @@
 import assert from "assert";
 import http, { OutgoingHttpHeaders } from "http";
 import https from "https";
+import { PassThrough, Transform } from "stream";
 import { arrayBuffer } from "stream/consumers";
+import { pipeline } from "stream/promises";
 import { URL } from "url";
+import zlib from "zlib";
 import {
   CorePluginSignatures,
   MiniflareCore,
@@ -76,7 +79,8 @@ export async function convertNodeRequest(
   // We're a bit naughty here mutating the incoming request, but this ensures
   // the headers are included in the pretty-error page. If we used the new
   // converted Request instance's headers, we wouldn't have connection, keep-
-  // alive, etc as we strip those
+  // alive, etc as we strip those. We need to take ownership of the request
+  // anyway though, since we're consuming its body.
   req.headers["x-forwarded-proto"] ??= proto;
   req.headers["x-real-ip"] ??= ip;
   req.headers["cf-connecting-ip"] ??= ip;
@@ -162,19 +166,53 @@ export function createRequestListener<Plugins extends HTTPPluginSignatures>(
         waitUntil = response.waitUntil();
         status = response.status;
         const headers: OutgoingHttpHeaders = {};
-        for (const [key, value] of response.headers) {
-          if (key.length === 10 && key.toLowerCase() === "set-cookie") {
+        // eslint-disable-next-line prefer-const
+        for (let [key, value] of response.headers) {
+          key = key.toLowerCase();
+          if (key === "set-cookie") {
             // Multiple Set-Cookie headers should be treated as separate headers
             headers["set-cookie"] = response.headers.getAll("set-cookie");
           } else {
             headers[key] = value;
           }
         }
+
+        // If a Content-Encoding is set, and the user hasn't encoded the body,
+        // we're responsible for doing so.
+        const encoders: Transform[] = [];
+        if (headers["content-encoding"] && response.encodeBody === "auto") {
+          // Content-Length will be wrong as it's for the decoded length
+          delete headers["content-length"];
+          // Reverse of https://github.com/nodejs/undici/blob/48d9578f431cbbd6e74f77455ba92184f57096cf/lib/fetch/index.js#L1660
+          const codings = headers["content-encoding"]
+            .toString()
+            .toLowerCase()
+            .split(",")
+            .map((x) => x.trim());
+          for (const coding of codings) {
+            if (/(x-)?gzip/.test(coding)) {
+              encoders.push(zlib.createGzip());
+            } else if (/(x-)?deflate/.test(coding)) {
+              encoders.push(zlib.createDeflate());
+            } else if (coding === "br") {
+              encoders.push(zlib.createBrotliCompress());
+            } else {
+              // Unknown encoding, don't do any encoding at all
+              mf.log.warn(
+                `Unknown encoding \"${coding}\", sending plain response...`
+              );
+              delete headers["content-encoding"];
+              encoders.length = 0;
+              break;
+            }
+          }
+        }
         res?.writeHead(status, headers);
 
         // Add live reload script if enabled and this is an HTML response
         if (
           HTTPPlugin.liveReload &&
+          response.encodeBody === "auto" &&
           response.headers
             .get("content-type")
             ?.toLowerCase()
@@ -196,12 +234,18 @@ export function createRequestListener<Plugins extends HTTPPluginSignatures>(
         }
 
         // Response body may be null if empty
-        if (response.body) {
-          for await (const chunk of response.body) {
-            if (chunk) res?.write(chunk);
+        if (res) {
+          const passThrough = new PassThrough();
+          // @ts-expect-error passThrough is definitely a PipelineSource
+          const pipelinePromise = pipeline(passThrough, ...encoders, res);
+          if (response.body) {
+            for await (const chunk of response.body) {
+              if (chunk) passThrough.write(chunk);
+            }
           }
+          passThrough.end();
+          await pipelinePromise;
         }
-        res?.end();
       } catch (e: any) {
         // MIME types aren't case sensitive
         const accept = req.headers.accept?.toLowerCase() ?? "";

packages/http-server/test/index.spec.ts

@@ -4,7 +4,9 @@ import http from "http";
 import https from "https";
 import { AddressInfo } from "net";
 import { Readable } from "stream";
+import { buffer, text } from "stream/consumers";
 import { setTimeout } from "timers/promises";
+import zlib from "zlib";
 import {
   BindingsPlugin,
   IncomingRequestCfProperties,
@@ -29,7 +31,7 @@ import {
   useTmp,
 } from "@miniflare/shared-test";
 import { MessageEvent, WebSocketPlugin } from "@miniflare/web-sockets";
-import test, { ExecutionContext } from "ava";
+import test, { ExecutionContext, Macro } from "ava";
 import StandardWebSocket from "ws";
 
 function listen(
@@ -86,10 +88,9 @@ function request(
         headers,
         rejectUnauthorized: false,
       },
-      (res) => {
-        let body = "";
-        res.on("data", (chunk) => (body += chunk));
-        res.on("end", () => resolve([body, res.headers, res.statusCode ?? 0]));
+      async (res) => {
+        const body = await text(res);
+        resolve([body, res.headers, res.statusCode ?? 0]);
       }
     );
   });
@@ -390,6 +391,74 @@ test("createRequestListener: includes CF-* headers in html error response", asyn
   t.regex(body, /CF-CONNECTING-IP/);
 });
 
+const longText = "".padStart(1024, "x");
+const autoEncodeMacro: Macro<
+  [encoding: string, decompress: (buffer: Buffer) => Buffer, encodes?: boolean]
+> = async (t, encoding, decompress, encodes = true) => {
+  const mf = useMiniflareWithHandler(
+    { HTTPPlugin, BindingsPlugin },
+    { bindings: { longText, encoding } },
+    (globals) => {
+      return new globals.Response(globals.longText, {
+        headers: { "Content-Encoding": globals.encoding },
+      });
+    }
+  );
+  const port = await listen(t, http.createServer(createRequestListener(mf)));
+  return new Promise<void>((resolve) => {
+    http.get({ port }, async (res) => {
+      t.is(res.headers["content-length"], undefined);
+      t.is(res.headers["transfer-encoding"], "chunked");
+      t.is(res.headers["content-encoding"], encodes ? encoding : undefined);
+      const compressed = await buffer(res);
+      const decompressed = decompress(compressed);
+      if (encodes) t.true(compressed.byteLength < decompressed.byteLength);
+      t.is(decompressed.toString("utf8"), longText);
+      resolve();
+    });
+  });
+};
+autoEncodeMacro.title = (providedTitle, encoding, decompress, encodes = true) =>
+  `createRequestListener: ${
+    encodes ? "auto-encodes" : "doesn't encode"
+  } response with Content-Encoding: ${encoding}`;
+test(autoEncodeMacro, "gzip", (buffer) => zlib.gunzipSync(buffer));
+test(autoEncodeMacro, "deFlaTe", (buffer) => zlib.inflateSync(buffer));
+test(autoEncodeMacro, "br", (buffer) => zlib.brotliDecompressSync(buffer));
+test(autoEncodeMacro, "deflate, gZip", (buffer) =>
+  zlib.inflateSync(zlib.gunzipSync(buffer))
+);
+// Should skip all encoding with single unknown encoding
+test(autoEncodeMacro, "deflate, unknown, gzip", (buffer) => buffer, false);
+test("createRequestListener: skips encoding already encoded data", async (t) => {
+  const encoded = new Uint8Array(zlib.gzipSync(Buffer.from(longText, "utf8")));
+  const mf = useMiniflareWithHandler(
+    { HTTPPlugin, BindingsPlugin },
+    { bindings: { encoded } },
+    (globals) => {
+      return new globals.Response(globals.encoded, {
+        encodeBody: "manual",
+        headers: {
+          "Content-Length": globals.encoded.byteLength.toString(),
+          "Content-Encoding": "gzip",
+        },
+      });
+    }
+  );
+  const port = await listen(t, http.createServer(createRequestListener(mf)));
+  return new Promise<void>((resolve) => {
+    http.get({ port }, async (res) => {
+      t.is(res.headers["content-length"], encoded.byteLength.toString());
+      t.is(res.headers["content-encoding"], "gzip");
+      const compressed = await buffer(res);
+      const decompressed = zlib.gunzipSync(compressed);
+      t.true(compressed.byteLength < decompressed.byteLength);
+      t.is(decompressed.toString("utf8"), longText);
+      resolve();
+    });
+  });
+});
+
 test("createServer: handles regular requests", async (t) => {
   const mf = useMiniflareWithHandler({ HTTPPlugin }, {}, (globals) => {
     return new globals.Response("body");