CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## 2023
 
+- [6.5.2](#652-2023-08-01) (Aug 2023)
+- [6.5.1](#651-2023-06-27) (Jun 2023)
 - [6.5.0](#650-2023-06-16) (Jun 2023)
 - [6.4.2](#642-2023-05-02) (May 2023)
 - [6.4.1](#641-2023-02-20) (Feb 2023)
@@ -48,6 +50,39 @@
 
 # Release notes
 
+## [6.5.2](https://github.com/socketio/engine.io/compare/6.5.1...6.5.2) (2023-08-01)
+
+
+### Bug Fixes
+
+* **webtransport:** add proper framing ([a306db0](https://github.com/socketio/engine.io/commit/a306db09e8ddb367c7d62f45fec920f979580b7c))
+
+
+### Dependencies
+
+- [`ws@~8.11.0`](https://github.com/websockets/ws/releases/tag/8.11.0) (no change)
+
+
+
+## [6.5.1](https://github.com/socketio/engine.io/compare/6.5.0...6.5.1) (2023-06-27)
+
+
+### Bug Fixes
+
+* prevent crash when accessing TextDecoder ([#684](https://github.com/socketio/engine.io/issues/684)) ([6dd2bc4](https://github.com/socketio/engine.io/commit/6dd2bc4f68edd7575c3844ae8ceadde649be95b2))
+
+
+### Credits
+
+Huge thanks to [@iowaguy](https://github.com/iowaguy) for helping!
+
+
+### Dependencies
+
+- [`ws@~8.11.0`](https://github.com/websockets/ws/releases/tag/8.11.0) (no change)
+
+
+
 ## [6.5.0](https://github.com/socketio/engine.io/compare/6.4.2...6.5.0) (2023-06-16)
 
 

examples/latency/package.json

@@ -3,7 +3,7 @@
   "version": "0.1.0",
   "dependencies": {
     "enchilada": "0.13.0",
-    "engine.io": "^6.2.1",
+    "engine.io": "^6.4.2",
     "engine.io-client": "^4.1.4",
     "express": "^4.18.2",
     "smoothie": "1.19.0"

lib/server.ts

@@ -16,11 +16,11 @@ import type { CookieSerializeOptions } from "cookie";
 import type { CorsOptions, CorsOptionsDelegate } from "cors";
 import type { Duplex } from "stream";
 import { WebTransport } from "./transports/webtransport";
+import { createPacketDecoderStream } from "engine.io-parser";
 
 const debug = debugModule("engine");
 
 const kResponseHeaders = Symbol("responseHeaders");
-const TEXT_DECODER = new TextDecoder();
 
 type Transport = "polling" | "websocket";
 
@@ -148,15 +148,13 @@ type Middleware = (
   next: (err?: any) => void
 ) => void;
 
-function parseSessionId(handshake: string) {
-  if (handshake.startsWith("0{")) {
-    try {
-      const parsed = JSON.parse(handshake.substring(1));
-      if (typeof parsed.sid === "string") {
-        return parsed.sid;
-      }
-    } catch (e) {}
-  }
+function parseSessionId(data: string) {
+  try {
+    const parsed = JSON.parse(data);
+    if (typeof parsed.sid === "string") {
+      return parsed.sid;
+    }
+  } catch (e) {}
 }
 
 export abstract class BaseServer extends EventEmitter {
@@ -535,7 +533,11 @@ export abstract class BaseServer extends EventEmitter {
     }
 
     const stream = result.value;
-    const reader = stream.readable.getReader();
+    const transformStream = createPacketDecoderStream(
+      this.opts.maxHttpBufferSize,
+      "nodebuffer"
+    );
+    const reader = stream.readable.pipeThrough(transformStream).getReader();
 
     // reading the first packet of the stream
     const { value, done } = await reader.read();
@@ -545,12 +547,13 @@ export abstract class BaseServer extends EventEmitter {
     }
 
     clearTimeout(timeout);
-    const handshake = TEXT_DECODER.decode(value);
 
-    // handshake is either
-    // "0" => new session
-    // '0{"sid":"xxxx"}' => upgrade
-    if (handshake === "0") {
+    if (value.type !== "open") {
+      debug("invalid WebTransport handshake");
+      return session.close();
+    }
+
+    if (value.data === undefined) {
       const transport = new WebTransport(session, stream, reader);
 
       // note: we cannot use "this.generateId()", because there is no "req" argument
@@ -571,7 +574,7 @@ export abstract class BaseServer extends EventEmitter {
       return;
     }
 
-    const sid = parseSessionId(handshake);
+    const sid = parseSessionId(value.data);
 
     if (!sid) {
       debug("invalid WebTransport handshake");

lib/transports/webtransport.ts

@@ -1,21 +1,9 @@
 import { Transport } from "../transport";
 import debugModule from "debug";
+import { createPacketEncoderStream } from "engine.io-parser";
 
 const debug = debugModule("engine:webtransport");
 
-const BINARY_HEADER = Buffer.of(54);
-
-function shouldIncludeBinaryHeader(packet, encoded) {
-  // 48 === "0".charCodeAt(0) (OPEN packet type)
-  // 54 === "6".charCodeAt(0) (NOOP packet type)
-  return (
-    packet.type === "message" &&
-    typeof packet.data !== "string" &&
-    encoded[0] >= 48 &&
-    encoded[0] <= 54
-  );
-}
-
 /**
  * Reference: https://developer.mozilla.org/en-US/docs/Web/API/WebTransport_API
  */
@@ -24,24 +12,24 @@ export class WebTransport extends Transport {
 
   constructor(private readonly session, stream, reader) {
     super({ _query: { EIO: "4" } });
-    this.writer = stream.writable.getWriter();
+
+    const transformStream = createPacketEncoderStream();
+    transformStream.readable.pipeTo(stream.writable);
+    this.writer = transformStream.writable.getWriter();
+
     (async () => {
-      let binaryFlag = false;
-      while (true) {
-        const { value, done } = await reader.read();
-        if (done) {
-          debug("session is closed");
-          break;
-        }
-        debug("received chunk: %o", value);
-        if (!binaryFlag && value.byteLength === 1 && value[0] === 54) {
-          binaryFlag = true;
-          continue;
+      try {
+        while (true) {
+          const { value, done } = await reader.read();
+          if (done) {
+            debug("session is closed");
+            break;
+          }
+          debug("received chunk: %o", value);
+          this.onPacket(value);
         }
-        this.onPacket(
-          this.parser.decodePacketFromBinary(value, binaryFlag, "nodebuffer")
-        );
-        binaryFlag = false;
+      } catch (e) {
+        debug("error while reading: %s", e.message);
       }
     })();
 
@@ -58,26 +46,20 @@ export class WebTransport extends Transport {
     return true;
   }
 
-  send(packets) {
+  async send(packets) {
     this.writable = false;
 
-    for (let i = 0; i < packets.length; i++) {
-      const packet = packets[i];
-      const isLast = i + 1 === packets.length;
-
-      this.parser.encodePacketToBinary(packet, (data) => {
-        if (shouldIncludeBinaryHeader(packet, data)) {
-          debug("writing binary header");
-          this.writer.write(BINARY_HEADER);
-        }
-        debug("writing chunk: %o", data);
-        this.writer.write(data);
-        if (isLast) {
-          this.writable = true;
-          this.emit("drain");
-        }
-      });
+    try {
+      for (let i = 0; i < packets.length; i++) {
+        const packet = packets[i];
+        await this.writer.write(packet);
+      }
+    } catch (e) {
+      debug("error while writing: %s", e.message);
     }
+
+    this.writable = true;
+    this.emit("drain");
   }
 
   doClose(fn) {