fix(eio-client): abort the pending polling requests once the close packet is sent

packages/engine.io-client/lib/transports/polling-fetch.ts

@@ -10,6 +10,9 @@ import { Polling } from "./polling.js";
  * @see https://nodejs.org/api/globals.html#fetch
  */
 export class Fetch extends Polling {
+  private _abortController =
+    typeof AbortController === "function" ? new AbortController() : null;
+
   override doPoll() {
     this._fetch()
       .then((res) => {
@@ -24,20 +27,30 @@ export class Fetch extends Polling {
       });
   }
 
-  override doWrite(data: string, callback: () => void) {
+  override doWrite(
+    data: string,
+    callback: () => void,
+    errorCallback?: () => void,
+  ) {
     this._fetch(data)
       .then((res) => {
         if (!res.ok) {
-          return this.onError("fetch write error", res.status, res);
+          this.onError("fetch write error", res.status, res);
+          return errorCallback && errorCallback();
         }
 
         callback();
       })
       .catch((err) => {
         this.onError("fetch write error", err);
+        errorCallback && errorCallback();
       });
   }
 
+  protected override abortRequests() {
+    this._abortController?.abort();
+  }
+
   private _fetch(data?: string) {
     const isPost = data !== undefined;
     const headers = new Headers(this.opts.extraHeaders);
@@ -53,6 +66,7 @@ export class Fetch extends Polling {
       body: isPost ? data : null,
       headers,
       credentials: this.opts.withCredentials ? "include" : "omit",
+      signal: this._abortController?.signal,
     }).then((res) => {
       // @ts-ignore getSetCookie() was added in Node.js v19.7.0
       this.socket._cookieJar?.parseCookies(res.headers.getSetCookie());

packages/engine.io-client/lib/transports/polling-xhr.node.ts

@@ -23,4 +23,9 @@ export class XHR extends BaseXHR {
       opts as RequestOptions,
     );
   }
+
+  protected override abortRequests() {
+    // note: the pending requests are not aborted, because aborting a pending request with the "xmlhttprequest-ssl"
+    // package throws an uncaught exception (see https://github.com/mjwwit/node-XMLHttpRequest/issues/37)
+  }
 }

packages/engine.io-client/lib/transports/polling-xhr.ts

@@ -16,6 +16,7 @@ export abstract class BaseXHR extends Polling {
   protected readonly xd: boolean;
 
   private pollXhr: any;
+  private _requests = new Set<Request>();
 
   /**
    * XHR Polling constructor.
@@ -56,14 +57,20 @@ export abstract class BaseXHR extends Polling {
    * @param {Function} fn - called upon flush.
    * @private
    */
-  override doWrite(data: string, fn: () => void) {
+  override doWrite(data: string, fn: () => void, errorFn?: () => void) {
     const req = this.request({
       method: "POST",
       data: data,
     });
-    req.on("success", fn);
+    this._requests.add(req);
+    req.on("success", () => {
+      this._requests.delete(req);
+      fn();
+    });
     req.on("error", (xhrStatus, context) => {
+      this._requests.delete(req);
       this.onError("xhr post error", xhrStatus, context);
+      errorFn && errorFn();
     });
   }
 
@@ -75,12 +82,22 @@ export abstract class BaseXHR extends Polling {
   override doPoll() {
     debug("xhr poll");
     const req = this.request();
-    req.on("data", this.onData.bind(this));
+    this._requests.add(req);
+    req.on("data", (data) => {
+      this._requests.delete(req);
+      this.onData(data);
+    });
     req.on("error", (xhrStatus, context) => {
+      this._requests.delete(req);
       this.onError("xhr poll error", xhrStatus, context);
     });
     this.pollXhr = req;
   }
+
+  protected override abortRequests() {
+    this._requests.forEach((req) => req.abort(true));
+    this._requests.clear();
+  }
 }
 
 interface RequestReservedEvents {
@@ -282,10 +299,12 @@ export class Request extends Emitter<
   /**
    * Aborts the request.
    *
+   * @param {Boolean} force - whether the underlying XMLHttpRequest object should be aborted too (which closes the
+   * connection)
    * @package
    */
-  public abort() {
-    this._cleanup();
+  public abort(force?: boolean) {
+    this._cleanup(force);
   }
 }
 

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

@@ -122,7 +122,14 @@ export abstract class Polling extends Transport {
   override doClose() {
     const close = () => {
       debug("writing close packet");
-      this.write([{ type: "close" }]);
+      this.writable = false;
+
+      encodePayload([{ type: "close" }], (data) => {
+        // once the close packet is sent (or cannot be sent), the pending requests (for example the long-polling
+        // request, if the server did not answer it) are not needed anymore
+        const onSettled = () => this.abortRequests();
+        this.doWrite(data, onSettled, onSettled);
+      });
     };
 
     if ("open" === this.readyState) {
@@ -174,6 +181,17 @@ export abstract class Polling extends Transport {
     return this.createUri(schema, query);
   }
 
+  /**
+   * Aborts the pending requests, once the transport is closed.
+   *
+   * @protected
+   */
+  protected abortRequests() {}
+
   abstract doPoll();
-  abstract doWrite(data: string, callback: () => void);
+  abstract doWrite(
+    data: string,
+    callback: () => void,
+    errorCallback?: () => void,
+  );
 }

packages/engine.io-client/test/node.js

@@ -5,8 +5,46 @@ const { repeat } = require("./util");
 const expect = require("expect.js");
 const { parse } = require("../build/cjs/globals.node.js");
 const { CookieJar } = require("../build/cjs/globals.node.js");
+const { createServer } = require("node:http");
+const { useFetch } = require("./support/env");
 
 describe("node.js", () => {
+  it("should abort the long-polling request once the close packet is sent", function (done) {
+    if (!useFetch) {
+      // aborting a pending request with the "xmlhttprequest-ssl" package throws an uncaught exception (see
+      // https://github.com/mjwwit/node-XMLHttpRequest/issues/37)
+      return this.skip();
+    }
+
+    let socket;
+
+    const httpServer = createServer((req, res) => {
+      if (req.method === "POST") {
+        req.resume();
+        return req.on("end", () => res.end("ok"));
+      }
+      if (!new URL(req.url, "http://localhost").searchParams.has("sid")) {
+        return res.end(
+          '0{"sid":"abc","upgrades":[],"pingInterval":10000,"pingTimeout":5000,"maxPayload":1000000}',
+        );
+      }
+      // the long-polling request is never answered
+      res.on("close", () => {
+        httpServer.close();
+        done();
+      });
+
+      // the socket is closed while the long-polling request is pending
+      socket.close();
+    });
+
+    httpServer.listen(0, () => {
+      socket = new Socket(`http://localhost:${httpServer.address().port}`, {
+        transports: ["polling"],
+      });
+    });
+  });
+
   describe("autoRef option", () => {
     const runFixture = (filename, done) =>
       execFile(