Support websocket compression (permessage-deflate)

.Rbuildignore

@@ -22,3 +22,4 @@
 ^cran-comments\.md$
 ^CRAN-RELEASE$
 ^CRAN-SUBMISSION$
+^deflate-client$

.github/workflows/R-CMD-check.yaml

@@ -7,7 +7,7 @@ on:
     branches: [main, rc-**]
   pull_request:
   schedule:
-    - cron: '0 6 * * 1' # every monday
+    - cron: "0 6 * * 1" # every monday
 
 name: Package checks
 
@@ -20,3 +20,29 @@ jobs:
     uses: rstudio/shiny-workflows/.github/workflows/R-CMD-check.yaml@v1
     with:
       ubuntu: "ubuntu-20.04 ubuntu-latest"
+  permessage-deflate-test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+
+      - name: Setup R
+        uses: r-lib/actions/setup-r@v2
+
+      - name: Install R package dependencies
+        uses: r-lib/actions/setup-r-dependencies@v2
+
+      - name: Install R package
+        run: R CMD INSTALL .
+
+      - name: Setup Node.js
+        uses: actions/setup-node@v2
+        with:
+          node-version: "18"
+
+      - name: Install npm dependencies
+        run: npm install
+        working-directory: deflate-client
+
+      - name: Run npm tests
+        run: npm test
+        working-directory: deflate-client

deflate-client/.gitignore

@@ -0,0 +1 @@
+node_modules

deflate-client/README.md

@@ -0,0 +1,15 @@
+This directory contains Node.js based tests for exercising different WebSocket compression parameters in httpuv.
+
+To setup, you'll need Node.js installed, then:
+
+```sh
+npm install
+```
+
+Be sure to also install the version of httpuv you want to test.
+
+To run the test:
+
+```sh
+npm test
+```

deflate-client/index.mjs

@@ -0,0 +1,84 @@
+import assert from "assert";
+import child_process from "child_process";
+import { randomBytes } from "crypto";
+import { sleep, md5, WebSocketReceiver, withWS } from "./lib/util.mjs";
+
+const MAX_BITS = 21;
+
+async function runMD5Test(address, perMessageDeflate) {
+  await withWS(address + "md5", { perMessageDeflate }, async (ws, wsr) => {
+    process.stderr.write("Testing: ");
+    for (let i = 0; i <= MAX_BITS; i++) {
+      process.stderr.write(".");
+      const payload = randomBytes(2**i);
+      ws.send(payload);
+      const { data } = await wsr.nextEvent("message");
+      assert.strictEqual(data.toString("utf-8"), md5(payload), "md5 of payload must match server's response");    
+    }
+    process.stderr.write("\n");
+  });
+}
+
+async function runEchoTest(address, perMessageDeflate) {
+  await withWS(address + "echo", { perMessageDeflate }, async (ws, wsr) => {
+    process.stderr.write("Testing: ");
+    for (let i = 0; i <= MAX_BITS; i++) {
+      process.stderr.write(".");
+      const payload = randomBytes(2**i);
+      ws.send(payload);
+      const { data } = await wsr.nextEvent("message");
+      assert(data.equals(payload), "payload must match server's response");    
+    }
+    process.stderr.write("\n");
+  });
+}
+
+async function runTest(address, perMessageDeflate) {
+  await runMD5Test(address, perMessageDeflate);
+  await runEchoTest(address, perMessageDeflate);
+}
+
+async function main() {
+  await sleep(1000);
+  await runTest("ws://127.0.0.1:14252/", false);
+  await runTest("ws://127.0.0.1:14252/", true);
+  await runTest("ws://127.0.0.1:14252/", { threshold: 0 });
+  await runTest("ws://127.0.0.1:14252/", { threshold: 0, serverMaxWindowBits: 9 });
+  await runTest("ws://127.0.0.1:14252/", {
+    threshold: 0,
+    serverMaxWindowBits: 9,
+    clientMaxWindowBits: 9
+  });
+  await runTest("ws://127.0.0.1:14252/", {
+    threshold: 0,
+    serverMaxWindowBits: 9,
+    clientMaxWindowBits: 9,
+    serverNoContextTakeover: true
+  });
+  await runTest("ws://127.0.0.1:14252/", {
+    threshold: 0,
+    serverMaxWindowBits: 9,
+    clientMaxWindowBits: 9,
+    serverNoContextTakeover: true,
+    clientNoContextTakeover: true
+  });
+}
+
+console.error("Launching httpuv");
+const rprocess = child_process.spawn("Rscript", ["server.R"], {
+  stdio: ["inherit", "inherit", "inherit"]
+});
+process.on("exit", () => {
+  rprocess.kill();
+});
+
+main().then(
+  result => {
+    console.error("All tests passed!");
+    process.exit(0);
+  },
+  error => {
+    console.error(error ?? "An unknown error has occurred!");
+    process.exit(1);
+  }
+);

deflate-client/lib/util.mjs

@@ -0,0 +1,89 @@
+import { createHash } from "crypto";
+import { WebSocket } from "ws";
+
+export async function sleep(millis) {
+  await new Promise(resolve => {
+    setTimeout(() => {
+      resolve(undefined);
+    }, millis);
+  });
+} 
+
+export function md5(data) {
+  const hash = createHash("md5");
+  hash.update(data);
+  return hash.digest("hex");
+}
+
+export class WebSocketReceiver {
+  constructor(ws) {
+    this.messages = [];
+    this.pending = null;
+
+    ws.on("open", () => {
+      this.#log("open");
+      this.#push({type: "open"});
+    });
+    ws.on("message", (data, isBinary) => {
+      this.#log("message");
+      this.#push({type: "message", data, isBinary});
+    });
+    ws.on("close", ({code, reason}) => {
+      this.#log("close");
+      this.#push({type: "close", code, reason});
+    });
+    ws.on("error", error => {
+      this.#log("error");
+      this.#push({type: "error", error});
+    });
+  }
+
+  #log(...args) {
+    // console.log(...args);
+  }
+
+  #push(message) {
+    this.messages.push(message);
+    if (this.messages.length === 1 && this.pending) {
+      const prevPending = this.pending
+      this.pending = null;
+      prevPending.resolve(null);
+    }
+  }
+
+  async nextEvent(type = "message") {
+    while (this.messages.length === 0) {
+      if (!this.pending) {
+        let resolve, reject;
+        let promise = new Promise((resolve_, reject_) => {
+          resolve = resolve_;
+          reject = reject_;
+        });
+        this.pending = {promise, resolve, reject};
+      }
+      await this.pending.promise;
+    }
+    const msg = this.messages.shift();
+    if (type && msg.type !== type) {
+      if (msg.type === "error") {
+        throw msg.error;
+      } else {
+        throw new Error(`Unexpected WebSocket event (expected '${type}', got '${msg.type}')`);
+      }
+    }
+    return msg;
+  }
+}
+
+export async function withWS(address, options, callback) {
+  console.log("Connecting to", address, "with", options);
+  const ws = new WebSocket(address, options);
+  try {
+    const wsr = new WebSocketReceiver(ws);
+    await wsr.nextEvent("open");
+
+    return await callback(ws, wsr);
+  } finally {
+    ws.close();
+  }
+}

deflate-client/package.json

@@ -0,0 +1,12 @@
+{
+  "name": "deflate-client",
+  "version": "1.0.0",
+  "description": "This directory contains Node.js based tests for exercising different WebSocket compression parameters in httpuv.",
+  "main": "index.js",
+  "scripts": {
+    "test": "node index.mjs"
+  },
+  "dependencies": {
+    "ws": "^8.2.1"
+  }
+}

deflate-client/server.R

@@ -0,0 +1,28 @@
+library(httpuv)
+
+
+echo_server <- function(ws) {
+  ws$onMessage(function(isBinary, data) {
+    ws$send(data)
+  })
+}
+
+md5_server <- function(ws) {
+  ws$onMessage(function(isBinary, data) {
+    ws$send(digest::digest(data, serialize = FALSE))
+  })
+}
+
+server <- list(
+  onWSOpen = function(ws) {
+    if (identical(ws$request$PATH_INFO, "/echo")) {
+      echo_server(ws)
+    } else if (identical(ws$request$PATH_INFO, "/md5")) {
+      md5_server(ws)
+    } else {
+      ws$close()
+    }
+  }
+)
+
+httpuv::runServer("127.0.0.1", 14252, server)

src/constants.h

@@ -54,4 +54,23 @@ static inline std::string trim(const std::string &s) {
   return s.substr(start, end-start);
 }
 
+static inline std::vector<std::string> split(const std::string& s, const std::string& delim) {
+  std::vector<std::string> results;
+  size_t pos = 0;
+  while (true) {
+    size_t i = s.find(delim, pos);
+    if (i == std::string::npos) {
+      break;
+    }
+    if (i != pos) {
+      results.push_back(s.substr(pos, i - pos));
+    }
+    pos = i + 1;
+  }
+  if (pos != s.length()) {
+    results.push_back(s.substr(pos));
+  }
+  return results;
+}
+
 #endif // CONSTANTS_H