fix(uws): properly handle chunked content (#642)

With the engine based on µWebSockets.js (introduced in version 6.1.0),
a huge request body split in multiple chunks would throw the following
error:

> node:buffer:254
>   TypedArrayPrototypeSet(target, source, targetStart);
>   ^
>
> TypeError: Cannot perform %TypedArray%.prototype.set on a detached ArrayBuffer
>     at Buffer.set (<anonymous>)
>     at _copyActual (node:buffer:254:3)
> node:buffer:254
>   TypedArrayPrototypeSet(target, source, targetStart);
>   ^
>
> TypeError: Cannot perform %TypedArray%.prototype.set on a detached ArrayBuffer
>     at Buffer.set (<anonymous>)
>     at _copyActual (node:buffer:254:3)
>     at Function.concat (node:buffer:562:12)
>     at onEnd (.../node_modules/engine.io/build/transports-uws/polling.js:126:32)
>     at .../node_modules/engine.io/build/transports-uws/polling.js:143:17

Note: µWebSockets.js does not currently support chunked transfer
encoding.

lib/transports-uws/polling.ts

@@ -122,6 +122,20 @@ export class Polling extends Transport {
       return;
     }
 
+    const expectedContentLength = Number(req.headers["content-length"]);
+
+    if (!expectedContentLength) {
+      this.onError("content-length header required");
+      res.writeStatus("411 Length Required").end();
+      return;
+    }
+
+    if (expectedContentLength > this.maxHttpBufferSize) {
+      this.onError("payload too large");
+      res.writeStatus("413 Payload Too Large").end();
+      return;
+    }
+
     const isBinary = "application/octet-stream" === req.headers["content-type"];
 
     if (isBinary && this.protocol === 4) {
@@ -131,11 +145,11 @@ export class Polling extends Transport {
     this.dataReq = req;
     this.dataRes = res;
 
-    let chunks = [];
-    let contentLength = 0;
+    let buffer;
+    let offset = 0;
 
     const cleanup = () => {
-      this.dataReq = this.dataRes = chunks = null;
+      this.dataReq = this.dataRes = null;
     };
 
     const onClose = () => {
@@ -154,8 +168,8 @@ export class Polling extends Transport {
       res.writeHeader(key, String(headers[key]));
     });
 
-    const onEnd = () => {
-      this.onData(Buffer.concat(chunks).toString());
+    const onEnd = buffer => {
+      this.onData(buffer.toString());
 
       if (this.readyState !== "closing") {
         res.end("ok");
@@ -165,18 +179,36 @@ export class Polling extends Transport {
 
     res.onAborted(onClose);
 
-    res.onData((chunk, isLast) => {
-      chunks.push(Buffer.from(chunk));
-      contentLength += Buffer.byteLength(chunk);
-      if (contentLength > this.maxHttpBufferSize) {
-        this.onError("payload too large");
-        res.writeStatus("413 Payload Too Large");
-        res.end();
+    res.onData((arrayBuffer, isLast) => {
+      const totalLength = offset + arrayBuffer.byteLength;
+      if (totalLength > expectedContentLength) {
+        this.onError("content-length mismatch");
+        res.close(); // calls onAborted
         return;
       }
+
+      if (!buffer) {
+        if (isLast) {
+          onEnd(Buffer.from(arrayBuffer));
+          return;
+        }
+        buffer = Buffer.allocUnsafe(expectedContentLength);
+      }
+
+      Buffer.from(arrayBuffer).copy(buffer, offset);
+
       if (isLast) {
-        onEnd();
+        if (totalLength != expectedContentLength) {
+          this.onError("content-length mismatch");
+          res.writeStatus("400 Content-Length Mismatch").end();
+          cleanup();
+          return;
+        }
+        onEnd(buffer);
+        return;
       }
+
+      offset = totalLength;
     });
   }
 

test/server.js

@@ -1955,6 +1955,65 @@ describe("server", () => {
       });
     });
 
+    it("should arrive when content is split in multiple chunks (polling)", done => {
+      const engine = listen(
+        {
+          maxHttpBufferSize: 1e10
+        },
+        port => {
+          const client = new ClientSocket(`ws://localhost:${port}`, {
+            transports: ["polling"]
+          });
+
+          engine.on("connection", socket => {
+            socket.on("message", data => {
+              client.close();
+              done();
+            });
+          });
+
+          client.on("open", () => {
+            client.send("a".repeat(1e6));
+          });
+        }
+      );
+    });
+
+    it("should arrive when content is sent with chunked transfer-encoding (polling)", function(done) {
+      if (process.env.EIO_WS_ENGINE === "uws") {
+        // µWebSockets.js does not currently support chunked encoding: https://github.com/uNetworking/uWebSockets.js/issues/669
+        return this.skip();
+      }
+      const engine = listen(port => {
+        const client = new ClientSocket(`ws://localhost:${port}`, {
+          transports: ["polling"]
+        });
+
+        engine.on("connection", socket => {
+          socket.on("message", data => {
+            expect(data).to.eql("123");
+
+            client.close();
+            done();
+          });
+        });
+
+        client.on("open", () => {
+          const req = http.request({
+            host: "localhost",
+            port,
+            path: `/engine.io/?EIO=4&transport=polling&sid=${client.id}`,
+            method: "POST"
+          });
+
+          req.write(process.env.EIO_CLIENT === "3" ? "4:41" : "41");
+          req.write("2");
+          req.write("3");
+          req.end();
+        });
+      });
+    });
+
     it("should arrive as ArrayBuffer if requested when binary data sent as Buffer (polling)", done => {
       const binaryData = Buffer.allocUnsafe(5);
       for (let i = 0; i < binaryData.length; i++) {