feat: slice write buffer according to the maxPayload value

The server will now include a "maxPayload" field in the handshake details, allowing the clients to decide how many packets they have to send to stay under the maxHttpBufferSize value. Related: - socketio/socket.io-client#1531 - socketio/engine.io@088dcb4
socketio · Mar 12, 2022 · 46fdc2f · 46fdc2f
1 parent f4725f1
commit 46fdc2f
Show file tree

Hide file tree

Showing 11 changed files with 13,694 additions and 261 deletions.
diff --git a/lib/socket.ts b/lib/socket.ts
@@ -1,5 +1,5 @@
 import { transports } from "./transports/index.js";
-import { installTimerFunctions } from "./util.js";
+import { installTimerFunctions, byteLength } from "./util.js";
 import parseqs from "parseqs";
 import parseuri from "parseuri";
 import debugModule from "debug"; // debug()
@@ -249,6 +249,7 @@ export class Socket extends Emitter<{}, {}, SocketReservedEvents> {
   private clearTimeoutFn: typeof clearTimeout;
   private offlineEventListener;
   private upgrading: boolean;
+  private maxPayload?: number;
 
   private readonly opts: Partial<SocketOptions>;
   private readonly secure: boolean;
@@ -676,6 +677,7 @@ export class Socket extends Emitter<{}, {}, SocketReservedEvents> {
     this.upgrades = this.filterUpgrades(data.upgrades);
     this.pingInterval = data.pingInterval;
     this.pingTimeout = data.pingTimeout;
+    this.maxPayload = data.maxPayload;
     this.onOpen();
     // In case open handler closes socket
     if ("closed" === this.readyState) return;
@@ -729,15 +731,46 @@ export class Socket extends Emitter<{}, {}, SocketReservedEvents> {
       !this.upgrading &&
       this.writeBuffer.length
     ) {
-      debug("flushing %d packets in socket", this.writeBuffer.length);
-      this.transport.send(this.writeBuffer);
+      const packets = this.getWritablePackets();
+      debug("flushing %d packets in socket", packets.length);
+      this.transport.send(packets);
       // keep track of current length of writeBuffer
       // splice writeBuffer and callbackBuffer on `drain`
-      this.prevBufferLen = this.writeBuffer.length;
+      this.prevBufferLen = packets.length;
       this.emitReserved("flush");
     }
   }
 
+  /**
+   * Ensure the encoded size of the writeBuffer is below the maxPayload value sent by the server (only for HTTP
+   * long-polling)
+   *
+   * @private
+   */
+  private getWritablePackets() {
+    const shouldCheckPayloadSize =
+      this.maxPayload &&
+      this.transport.name === "polling" &&
+      this.writeBuffer.length > 1;
+    if (!shouldCheckPayloadSize) {
+      return this.writeBuffer;
+    }
+    let payloadSize = 1; // first packet type
+    for (let i = 0; i < this.writeBuffer.length; i++) {
+      const data = this.writeBuffer[i].data;
+      if (data) {
+        payloadSize += byteLength(data);
+      }
+      if (i > 0 && payloadSize > this.maxPayload) {
+        debug("only send %d out of %d packets", i, this.writeBuffer.length);
+        return this.writeBuffer.slice(0, i);
+      }
+      payloadSize += 2; // separator + packet type
+    }
+    debug("payload size is %d (max: %d)", payloadSize, this.maxPayload);
+    return this.writeBuffer;
+  }
+
   /**
    * Sends a message.
    *

diff --git a/lib/transports/websocket.ts b/lib/transports/websocket.ts
@@ -141,6 +141,7 @@ export class WS extends Transport {
 
           if (this.opts.perMessageDeflate) {
             const len =
+              // @ts-ignore
               "string" === typeof data ? Buffer.byteLength(data) : data.length;
             if (len < this.opts.perMessageDeflate.threshold) {
               opts.compress = false;

diff --git a/lib/util.ts b/lib/util.ts
@@ -22,3 +22,34 @@ export function installTimerFunctions(obj, opts) {
     obj.clearTimeoutFn = clearTimeout.bind(globalThis);
   }
 }
+
+// base64 encoded buffers are about 33% bigger (https://en.wikipedia.org/wiki/Base64)
+const BASE64_OVERHEAD = 1.33;
+
+// we could also have used `new Blob([obj]).size`, but it isn't supported in IE9
+export function byteLength(obj) {
+  if (typeof obj === "string") {
+    return utf8Length(obj);
+  }
+  // arraybuffer or blob
+  return Math.ceil((obj.byteLength || obj.size) * BASE64_OVERHEAD);
+}
+
+function utf8Length(str) {
+  let c = 0,
+    length = 0;
+  for (let i = 0, l = str.length; i < l; i++) {
+    c = str.charCodeAt(i);
+    if (c < 0x80) {
+      length += 1;
+    } else if (c < 0x800) {
+      length += 2;
+    } else if (c < 0xd800 || c >= 0xe000) {
+      length += 3;
+    } else {
+      i++;
+      length += 4;
+    }
+  }
+  return length;
+}