fix(engine): properly pause the polling transport during upgrade

Before that change, the client was stuck waiting for the ongoing HTTP
long-polling request to be cleanly closed by the server, and the
upgrade process would fail after the 10s delay.

Note: the Node.js client implementation does not resume polling
("unpause") in that case, which looks like a bug.

Related:

- #4
- socketio/engine.io@00f9738
- socketio/engine.io@1c96ca4
- https://github.com/socketio/engine.io-client/blob/dfee8ded722a8c4f9f773505d0c77b4561569863/lib/transports/polling.ts#L84-L116

packages/engine.io/lib/socket.ts

@@ -28,6 +28,8 @@ interface SocketEvents {
   close: (reason: CloseReason) => void;
 }
 
+const FAST_UPGRADE_INTERVAL_MS = 100;
+
 export class Socket extends EventEmitter<
   never,
   never,
@@ -202,22 +204,44 @@ export class Socket extends EventEmitter<
     }, this.opts.upgradeTimeout);
 
     transport.on("close", () => {
+      clearInterval(fastUpgradeTimerId);
       transport.off();
     });
 
+    let fastUpgradeTimerId: number;
+
+    // we need to make sure that no packets gets lost during the upgrade, so the client does not cancel the HTTP
+    // long-polling request itself, instead the server sends a "noop" packet to cleanly end any ongoing polling request
+    const sendNoopPacket = () => {
+      if (this.transport.name === "polling" && this.transport.writable) {
+        getLogger("engine.io").debug(
+          "[socket] writing a noop packet to polling for fast upgrade",
+        );
+        this.transport.send([{ type: "noop" }]);
+      }
+    };
+
     transport.on("packet", (packet) => {
       if (packet.type === "ping" && packet.data === "probe") {
         getLogger("engine.io").debug(
           "[socket] got probe ping packet, sending pong",
         );
         transport.send([{ type: "pong", data: "probe" }]);
+
+        sendNoopPacket();
+        fastUpgradeTimerId = setInterval(
+          sendNoopPacket,
+          FAST_UPGRADE_INTERVAL_MS,
+        );
+
         this.emitReserved("upgrading", transport);
-      } else if (packet.type === "upgrade") {
+      } else if (packet.type === "upgrade" && this.readyState !== "closed") {
         getLogger("engine.io").debug("[socket] got upgrade packet - upgrading");
 
         this.upgradeState = "upgraded";
 
         clearTimeout(timeoutId);
+        clearInterval(fastUpgradeTimerId);
         transport.off();
         this.closeTransport();
         this.bindTransport(transport);

packages/engine.io/test/upgrade.test.ts

@@ -12,7 +12,7 @@ describe("upgrade", () => {
   it("should upgrade", () => {
     const engine = new Server();
 
-    return testServeWithAsyncResults(engine, 2, async (port, partialDone) => {
+    return testServeWithAsyncResults(engine, 3, async (port, partialDone) => {
       engine.on("connection", (socket) => {
         socket.on("message", (val) => {
           assertEquals(val, "upgraded!");
@@ -30,6 +30,18 @@ describe("upgrade", () => {
 
       const sid = await parseSessionID(response);
 
+      fetch(
+        `http://localhost:${port}/engine.io/?EIO=4&transport=polling&sid=${sid}`,
+        {
+          method: "get",
+        },
+      ).then(async (response) => {
+        assertEquals(response.status, 200);
+        assertEquals(await response.text(), "6"); // "noop" packet
+
+        partialDone();
+      });
+
       const socket = new WebSocket(
         `ws://localhost:${port}/engine.io/?EIO=4&transport=websocket&sid=${sid}`,
       );