diff --git a/packages/engine.io-client/lib/socket.ts b/packages/engine.io-client/lib/socket.ts index b548ced5ab..8b96c6cb67 100644 --- a/packages/engine.io-client/lib/socket.ts +++ b/packages/engine.io-client/lib/socket.ts @@ -689,6 +689,34 @@ export class SocketWithoutUpgrade extends Emitter< } } + /** + * Removes packets that have not been handed to the transport yet. + * + * Packets from one Socket.IO event share the same `options` object. If any of + * those packets is already part of the current flush batch, all related + * packets are left intact so a multi-packet payload cannot be partially sent. + * + * @param options - the options object shared by the packets to remove + * @private + */ + /* private */ _removeFromWriteBuffer(options: WriteOptions) { + if (options == null) { + return; + } + + for (let i = 0; i < this._prevBufferLen; i++) { + if (this.writeBuffer[i].options === options) { + return; + } + } + + for (let i = this.writeBuffer.length - 1; i >= this._prevBufferLen; i--) { + if (this.writeBuffer[i].options === options) { + this.writeBuffer.splice(i, 1); + } + } + } + /** * Flush write buffers. * diff --git a/packages/engine.io-client/test/socket.js b/packages/engine.io-client/test/socket.js index 621c895009..54c9a627cb 100644 --- a/packages/engine.io-client/test/socket.js +++ b/packages/engine.io-client/test/socket.js @@ -13,6 +13,56 @@ const { repeat } = require("./util"); describe("Socket", function () { this.timeout(10000); + describe("_removeFromWriteBuffer", () => { + it("should remove all matching packets that have not been flushed", () => { + const socket = Object.create(Socket.prototype); + const options = {}; + const otherPacket = { options: {} }; + socket.writeBuffer = [{ options }, otherPacket, { options }]; + socket._prevBufferLen = 0; + + socket._removeFromWriteBuffer(options); + + expect(socket.writeBuffer).to.eql([otherPacket]); + }); + + it("should preserve all matching packets when one has been flushed", () => { + const socket = Object.create(Socket.prototype); + const options = {}; + const packets = [{ options }, { options }]; + socket.writeBuffer = packets.slice(); + socket._prevBufferLen = 1; + + socket._removeFromWriteBuffer(options); + + expect(socket.writeBuffer).to.eql(packets); + }); + + it("should not remove already flushed packets or change the flush cursor", () => { + const socket = Object.create(Socket.prototype); + const options = {}; + const flushed = { options: {} }; + socket.writeBuffer = [flushed, { options }, { options }]; + socket._prevBufferLen = 1; + + socket._removeFromWriteBuffer(options); + + expect(socket.writeBuffer).to.eql([flushed]); + expect(socket._prevBufferLen).to.be(1); + }); + + it("should ignore a missing options object", () => { + const socket = Object.create(Socket.prototype); + const packet = { options: {} }; + socket.writeBuffer = [packet]; + socket._prevBufferLen = 0; + + socket._removeFromWriteBuffer(undefined); + + expect(socket.writeBuffer).to.eql([packet]); + }); + }); + describe("filterUpgrades", () => { it("should return only available transports", () => { const socket = new Socket({ transports: ["polling"] }); diff --git a/packages/socket.io-client/lib/socket.ts b/packages/socket.io-client/lib/socket.ts index cc8ff87df0..53c28aaa11 100644 --- a/packages/socket.io-client/lib/socket.ts +++ b/packages/socket.io-client/lib/socket.ts @@ -440,7 +440,7 @@ export class Socket< debug("emitting packet with ack id %d", id); const ack = args.pop() as (...args: any[]) => void; - this._registerAckCallback(id, ack); + this._registerAckCallback(id, ack, packet); packet.id = id; } @@ -465,7 +465,11 @@ export class Socket< /** * @private */ - private _registerAckCallback(id: number, ack: (...args: any[]) => void) { + private _registerAckCallback( + id: number, + ack: (...args: any[]) => void, + packet: Packet & { options?: { compress?: boolean } }, + ) { const timeout = this.flags.timeout ?? this._opts.ackTimeout; if (timeout === undefined) { this.acks[id] = ack; @@ -475,6 +479,9 @@ export class Socket< // @ts-ignore const timer = this.io.setTimeoutFn(() => { delete this.acks[id]; + // The packet may already have been written to the Engine.IO buffer + // (connection still "open" but transport temporarily unwritable). + this.io.engine?._removeFromWriteBuffer(packet.options); for (let i = 0; i < this.sendBuffer.length; i++) { if (this.sendBuffer[i].id === id) { debug("removing packet with ack id %d from the buffer", id); diff --git a/packages/socket.io-client/test/socket.ts b/packages/socket.io-client/test/socket.ts index 0149ad64b6..3a61f4b1f4 100644 --- a/packages/socket.io-client/test/socket.ts +++ b/packages/socket.io-client/test/socket.ts @@ -596,6 +596,95 @@ describe("socket", () => { }); }); + it("should not flush a timed out event when the transport becomes writable again", () => { + return wrap((done) => { + const socket = io(BASE_URL + "/", { + forceNew: true, + transports: ["websocket"], + }); + + socket.on("connect", () => { + const engine = socket.io.engine; + engine.transport.writable = false; + + const results: { one?: unknown; three?: unknown } = {}; + let timedOut = false; + + socket.emit("tracked", "one", (value) => { + results.one = value; + maybeDone(); + }); + + socket.timeout(50).emit("tracked", "two", (err) => { + expect(err).to.be.an(Error); + timedOut = true; + + engine.transport.writable = true; + // @ts-ignore flush remaining Engine.IO packets + engine.flush(); + }); + + socket.timeout(1000).emit("tracked", "three", (err, value) => { + results.three = err || value; + maybeDone(); + }); + + function maybeDone() { + if (results.one === undefined || results.three === undefined) { + return; + } + + expect(timedOut).to.be(true); + expect(results.one).to.be("one"); + expect(results.three).to.be("three"); + + socket.emit("getTracked", (events) => { + expect(events).to.eql(["one", "three"]); + success(done, socket); + }); + } + }); + }); + }); + + it("should discard a timed out packet buffered by Engine.IO", () => { + return wrap((done) => { + const socket = io(BASE_URL + "/", { + forceNew: true, + transports: ["websocket"], + }); + + socket.on("connect", () => { + const engine = socket.io.engine; + engine.transport.writable = false; + + const enginePackets = []; + const onPacketCreate = (packet) => { + enginePackets.push(packet); + }; + engine.on("packetCreate", onPacketCreate); + + socket.timeout(50).emit("tracked", Uint8Array.from([1]), (err) => { + expect(err).to.be.an(Error); + expect(enginePackets.length).to.be(2); + enginePackets.forEach((packet) => { + expect(engine.writeBuffer.indexOf(packet)).to.be(-1); + }); + + engine.transport.writable = true; + // @ts-ignore flush remaining Engine.IO packets + engine.flush(); + + socket.emit("getTracked", (events) => { + expect(events).to.eql([]); + success(done, socket); + }); + }); + engine.off("packetCreate", onPacketCreate); + }); + }); + }); + it("should timeout when the server does not acknowledge the event", () => { return wrap((done) => { const socket = io(BASE_URL + "/"); diff --git a/packages/socket.io-client/test/support/server.ts b/packages/socket.io-client/test/support/server.ts index 52a7d18f6a..9fe75391a9 100644 --- a/packages/socket.io-client/test/support/server.ts +++ b/packages/socket.io-client/test/support/server.ts @@ -60,6 +60,18 @@ export function createServer() { cb(arg); }); + socket.on("tracked", (arg, cb) => { + socket.data.tracked = socket.data.tracked || []; + socket.data.tracked.push(arg); + if (typeof cb === "function") { + cb(arg); + } + }); + + socket.on("getTracked", (cb) => { + cb(socket.data.tracked || []); + }); + // ack tests socket.on("ack", () => { socket.emit("ack", (a, b) => {