feat: uws handling backpressure

src/response.js

@@ -72,8 +72,7 @@ module.exports = class Response extends Writable {
     #socket = null;
     #ended = false;
     #pendingChunks = [];
-    #lastWriteChunkTime = 0;
-    #writeTimeout = null;
+    #flushScheduled = false;
     req;
     constructor(res, req, app) {
         super();
@@ -158,32 +157,24 @@ module.exports = class Response extends Writable {
 
             if (this.chunkedTransfer) {
                 this.#pendingChunks.push(chunk);
-                const size = this.#pendingChunks.reduce((acc, chunk) => acc + chunk.byteLength, 0);
-                const now = performance.now();
-                // the first chunk is sent immediately (!this.#lastWriteChunkTime)
-                // the other chunks are sent when watermark is reached (size >= HIGH_WATERMARK) 
-                // or if elapsed 50ms of last send (now - this.#lastWriteChunkTime > 50)
-                if (!this.#lastWriteChunkTime || size >= HIGH_WATERMARK || now - this.#lastWriteChunkTime > 50) {
+                const size = this.#pendingChunks.reduce((acc, c) => acc + c.byteLength, 0);
+                if (size >= HIGH_WATERMARK) {
                     this._res.write(Buffer.concat(this.#pendingChunks, size));
                     this.#pendingChunks = [];
-                    this.#lastWriteChunkTime = now;
-                    if(this.#writeTimeout) {
-                        clearTimeout(this.#writeTimeout);
-                        this.#writeTimeout = null;
-                    }
-                } else if(!this.#writeTimeout) {
-                    this.#writeTimeout = setTimeout(() => {
-                        this.#writeTimeout = null;
-                        if(!this.finished && !this.aborted) this._res.cork(() => {
-                            if(this.#pendingChunks.length) {
-                                const size = this.#pendingChunks.reduce((acc, chunk) => acc + chunk.byteLength, 0);
-                                this._res.write(Buffer.concat(this.#pendingChunks, size));
+                }
+                if (!this.#flushScheduled) {
+                    this.#flushScheduled = true;
+                    setTimeout(() => {
+                        this.#flushScheduled = false;
+                        if (this.finished || this.aborted || !this.#pendingChunks.length) return;
+                        this._res.cork(() => {
+                            if (this.#pendingChunks.length) {
+                                const s = this.#pendingChunks.reduce((acc, c) => acc + c.byteLength, 0);
+                                this._res.write(Buffer.concat(this.#pendingChunks, s));
                                 this.#pendingChunks = [];
-                                this.#lastWriteChunkTime = performance.now();
                             }
                         });
-                    }, 50);
-                    this.#writeTimeout.unref();
+                    }, 0);
                 }
                 this.writingChunk = false;
                 callback(null);
@@ -313,9 +304,11 @@ module.exports = class Response extends Writable {
                 this._res.endWithoutBody(contentLength.toString());
             } else {
                 if(this.#pendingChunks.length) {
-                    this._res.write(Buffer.concat(this.#pendingChunks));
+                    const buf = this.#pendingChunks.length === 1
+                        ? this.#pendingChunks[0]
+                        : Buffer.concat(this.#pendingChunks);
                     this.#pendingChunks = [];
-                    this.lastWriteChunkTime = 0;
+                    this._res.write(buf);
                 }
                 if(data instanceof Buffer) {
                     data = data.buffer.slice(data.byteOffset, data.byteOffset + data.byteLength);

tests/test-compression-server.js

@@ -0,0 +1,34 @@
+const express = require('../src/index.js');
+const compression = require('compression');
+const app = express();
+
+// force encoding by query param ?compress=
+app.use((req, res, next) => {
+  const enc = req.query.compress;
+  if (enc) {
+    req.headers['accept-encoding'] = enc;
+  }
+  next();
+});
+
+app.use(compression({ threshold: 1 }));
+app.use(express.static('tests/parts'));
+
+const PORT = 13335;
+app.listen(PORT, () => {
+  const base = `http://localhost:${PORT}`;
+  const files = ['small-file.json', 'medium-file.json', 'large-file.json'];
+  const encodings = ['gzip', 'br', 'deflate'];
+
+  console.log(`\n  Server on ${base}\n`);
+  console.log('  === Links ===\n');
+
+  for (const file of files) {
+    console.log(`  ${file}`);
+    console.log(`     ${'identity'.padEnd(8)} : ${base}/${file}?compress=identity`);
+    for (const enc of encodings) {
+      console.log(`     ${enc.padEnd(8)} : ${base}/${file}?compress=${enc}`);
+    }
+    console.log();
+  }
+});