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 {
   /**
@@ -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);
         }
       }
     }
@@ -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);
     }
   }
 

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");
 

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;
@@ -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

@@ -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");
   }