added argument to drain that provides the list of packets that are drained

Author: bytenik

packages/engine.io/lib/socket.ts

@@ -14,7 +14,10 @@ export interface SendOptions {
 
 type ReadyState = "opening" | "open" | "closing" | "closed";
 
-type SendCallback = (transport: Transport) => void;
+type SendCallback = (
+  transport: Transport,
+  packets: Packet[] | undefined
+) => void;
 
 export class Socket extends EventEmitter {
   /**
@@ -80,7 +83,7 @@ export class Socket extends EventEmitter {
     server: BaseServer,
     transport: Transport,
     req: EngineRequest,
-    protocol: number,
+    protocol: number
   ) {
     super();
     this.id = id;
@@ -125,7 +128,7 @@ export class Socket extends EventEmitter {
         pingInterval: this.server.opts.pingInterval,
         pingTimeout: this.server.opts.pingTimeout,
         maxPayload: this.server.opts.maxHttpBufferSize,
-      }),
+      })
     );
 
     if (this.server.opts.initialPacket) {
@@ -212,7 +215,7 @@ export class Socket extends EventEmitter {
     this.pingIntervalTimer = setTimeout(() => {
       debug(
         "writing ping packet - expecting pong within %sms",
-        this.server.opts.pingTimeout,
+        this.server.opts.pingTimeout
       );
       this.sendPacket("ping");
       this.resetPingTimeout();
@@ -233,7 +236,7 @@ export class Socket extends EventEmitter {
       },
       this.protocol === 3
         ? this.server.opts.pingInterval + this.server.opts.pingTimeout
-        : this.server.opts.pingTimeout,
+        : this.server.opts.pingTimeout
     );
   }
 
@@ -271,13 +274,13 @@ export class Socket extends EventEmitter {
    *
    * @private
    */
-  private onDrain() {
+  private onDrain(packets: Packet[] | undefined) {
     if (this.sentCallbackFn.length > 0) {
       debug("executing batch send callback");
       const seqFn = this.sentCallbackFn.shift();
       if (seqFn) {
         for (let i = 0; i < seqFn.length; i++) {
-          seqFn[i](this.transport);
+          seqFn[i](this.transport, packets);
         }
       }
     }
@@ -293,7 +296,7 @@ export class Socket extends EventEmitter {
     debug(
       'might upgrade socket transport from "%s" to "%s"',
       this.transport.name,
-      transport.name,
+      transport.name
     );
 
     this.upgrading = true;
@@ -468,7 +471,7 @@ export class Socket extends EventEmitter {
     type: PacketType,
     data?: RawData,
     options: SendOptions = {},
-    callback?: SendCallback,
+    callback?: SendCallback
   ) {
     if ("function" === typeof options) {
       callback = options;
@@ -525,8 +528,8 @@ export class Socket extends EventEmitter {
       }
 
       this.transport.send(wbuf);
-      this.emit("drain");
-      this.server.emit("drain", this);
+      this.emit("drain", wbuf);
+      this.server.emit("drain", this, wbuf);
     }
   }
 
@@ -568,7 +571,7 @@ export class Socket extends EventEmitter {
     if (this.writeBuffer.length) {
       debug(
         "there are %d remaining packets in the buffer, waiting for the 'drain' event",
-        this.writeBuffer.length,
+        this.writeBuffer.length
       );
       this.once("drain", () => {
         debug("all packets have been sent, closing the transport");

packages/engine.io/lib/transport.ts

@@ -4,6 +4,7 @@ import * as parser_v3 from "./parser-v3/index";
 import debugModule from "debug";
 import type { IncomingMessage, ServerResponse } from "http";
 import { Packet, RawData } from "engine.io-parser";
+import type * as Parser from "engine.io-parser";
 
 const debug = debugModule("engine:transport");
 
@@ -69,7 +70,7 @@ export abstract class Transport extends EventEmitter {
       "readyState updated from %s to %s (%s)",
       this._readyState,
       state,
-      this.name,
+      this.name
     );
     this._readyState = state;
   }

packages/engine.io/lib/transports-uws/polling.ts

@@ -3,6 +3,7 @@ import { createGzip, createDeflate } from "zlib";
 import * as accepts from "accepts";
 import debugModule from "debug";
 import { HttpRequest, HttpResponse } from "uWebSockets.js";
+import { Packet } from "engine.io-parser";
 
 const debug = debugModule("engine:polling");
 
@@ -253,7 +254,7 @@ export class Polling extends Transport {
    * @param {Object} packet
    * @private
    */
-  send(packets) {
+  send(packets: Packet[]) {
     this.writable = false;
 
     if (this.shouldClose) {
@@ -263,11 +264,11 @@ export class Polling extends Transport {
       this.shouldClose = null;
     }
 
-    const doWrite = (data) => {
+    const doWrite = (data: string) => {
       const compress = packets.some((packet) => {
         return packet.options && packet.options.compress;
       });
-      this.write(data, { compress });
+      this.write(data, { compress, source: packets });
     };
 
     if (this.protocol === 3) {
@@ -288,7 +289,7 @@ export class Polling extends Transport {
     debug('writing "%s"', data);
     this.doWrite(data, options, () => {
       this.req.cleanup();
-      this.emit("drain");
+      this.emit("drain", options.source);
     });
   }
 

packages/engine.io/lib/transports-uws/websocket.ts

@@ -55,7 +55,7 @@ export class WebSocket extends Transport {
         this.socket.send(data, isBinary, compress);
 
         if (isLast) {
-          this.emit("drain");
+          this.emit("drain", packets);
           this.writable = true;
           this.emit("ready");
         }

packages/engine.io/lib/transports/polling.ts

@@ -229,7 +229,7 @@ export class Polling extends Transport {
       const compress = packets.some((packet) => {
         return packet.options && packet.options.compress;
       });
-      this.write(data, { compress });
+      this.write(data, { compress, source: packets });
     };
 
     if (this.protocol === 3) {
@@ -250,7 +250,7 @@ export class Polling extends Transport {
     debug('writing "%s"', data);
     this.doWrite(data, options, () => {
       this.req.cleanup();
-      this.emit("drain");
+      this.emit("drain", options.source);
     });
   }
 

packages/engine.io/lib/transports/websocket.ts

@@ -7,6 +7,7 @@ const debug = debugModule("engine:ws");
 export class WebSocket extends Transport {
   protected perMessageDeflate: any;
   private socket: any;
+  private currentPackets: Packet[] | undefined;
 
   /**
    * WebSocket transport
@@ -44,6 +45,8 @@ export class WebSocket extends Transport {
   send(packets: Packet[]) {
     this.writable = false;
 
+    this.currentPackets = packets;
+
     for (let i = 0; i < packets.length; i++) {
       const packet = packets[i];
       const isLast = i + 1 === packets.length;
@@ -54,13 +57,13 @@ export class WebSocket extends Transport {
         this.socket._sender.sendFrame(
           // @ts-ignore
           packet.options.wsPreEncodedFrame,
-          isLast ? this._onSentLast : this._onSent,
+          isLast ? this._onSentLast : this._onSent
         );
       } else {
         this.parser.encodePacket(
           packet,
           this.supportsBinary,
-          isLast ? this._doSendLast : this._doSend,
+          isLast ? this._doSendLast : this._doSend
         );
       }
     }
@@ -98,8 +101,9 @@ export class WebSocket extends Transport {
     if (err) {
       this.onError("write error", err.stack);
     } else {
-      this.emit("drain");
+      this.emit("drain", this.currentPackets);
       this.writable = true;
+      this.currentPackets = undefined;
       this.emit("ready");
     }
   };

packages/engine.io/lib/transports/webtransport.ts

@@ -13,7 +13,7 @@ export class WebTransport extends Transport {
   constructor(
     private readonly session,
     stream,
-    reader,
+    reader
   ) {
     super({ _query: { EIO: "4" } });
 
@@ -60,7 +60,7 @@ export class WebTransport extends Transport {
       debug("error while writing: %s", e.message);
     }
 
-    this.emit("drain");
+    this.emit("drain", packets);
     this.writable = true;
     this.emit("ready");
   }