diff --git a/packages/engine.io-client/lib/transports/polling-fetch.ts b/packages/engine.io-client/lib/transports/polling-fetch.ts index 7d85b66c6..64fffa827 100644 --- a/packages/engine.io-client/lib/transports/polling-fetch.ts +++ b/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()); diff --git a/packages/engine.io-client/lib/transports/polling-xhr.node.ts b/packages/engine.io-client/lib/transports/polling-xhr.node.ts index 172463fcf..5730da547 100644 --- a/packages/engine.io-client/lib/transports/polling-xhr.node.ts +++ b/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) + } } diff --git a/packages/engine.io-client/lib/transports/polling-xhr.ts b/packages/engine.io-client/lib/transports/polling-xhr.ts index 2196d2d22..b2bca0172 100644 --- a/packages/engine.io-client/lib/transports/polling-xhr.ts +++ b/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(); /** * 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); } } diff --git a/packages/engine.io-client/lib/transports/polling.ts b/packages/engine.io-client/lib/transports/polling.ts index 2f09e8021..70d2b38e2 100644 --- a/packages/engine.io-client/lib/transports/polling.ts +++ b/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, + ); } diff --git a/packages/engine.io-client/test/node.js b/packages/engine.io-client/test/node.js index 1f756a13b..83276190d 100644 --- a/packages/engine.io-client/test/node.js +++ b/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(