diff --git a/src/v1/diagnostics.test.ts b/src/v1/diagnostics.test.ts index ace0146..0d749a3 100644 --- a/src/v1/diagnostics.test.ts +++ b/src/v1/diagnostics.test.ts @@ -89,4 +89,13 @@ describe("DiagnosticsBatcher", () => { expect(logger.removeListener).toHaveBeenCalledWith("log", expect.any(Function)); clearIntervalSpy.mockRestore(); }); + + test("should remove the same logger listener it registered on shutdown", () => { + const diagnostics = new DiagnosticsBatcher(); + const registered = (logger.on as jest.Mock).mock.calls[0][1]; + + diagnostics.shutdown(); + + expect(logger.removeListener).toHaveBeenCalledWith("log", registered); + }); }); diff --git a/src/v1/diagnostics.ts b/src/v1/diagnostics.ts index 04cf4bd..5bcf128 100644 --- a/src/v1/diagnostics.ts +++ b/src/v1/diagnostics.ts @@ -9,12 +9,14 @@ export type Diagnostics = { export class DiagnosticsBatcher extends EventEmitter { private logEvents: any[]; private flushInterval: Timeout; + // Kept so shutdown() can remove the exact listener that was registered. + private logListener = this.handleLogEvent.bind(this); constructor(flushIntervalMillis = 300000) { super(); this.logEvents = new Array(); - logger.on("log", this.handleLogEvent.bind(this)); + logger.on("log", this.logListener); this.flushInterval = setInterval(this.flushDiagnostics.bind(this), flushIntervalMillis); } @@ -30,7 +32,7 @@ export class DiagnosticsBatcher extends EventEmitter { shutdown() { this.flushDiagnostics(); clearInterval(this.flushInterval); - logger.removeListener("log", this.handleLogEvent.bind(this)); + logger.removeListener("log", this.logListener); } private handleLogEvent(level: LogLevel, timestamp: any, ...args: any) { diff --git a/src/v1/signaling.serverClose.test.ts b/src/v1/signaling.serverClose.test.ts new file mode 100644 index 0000000..94cc7ea --- /dev/null +++ b/src/v1/signaling.serverClose.test.ts @@ -0,0 +1,67 @@ +/** + * @jest-environment node + */ +// Exercises the real rpc-websockets client (no mock) against a local server, because +// the bug only shows up with the library's real ordering: it clears its socket and +// "ready" flag synchronously on close, then emits "close" on the next tick. +import { AddressInfo } from "net"; +import { Server } from "rpc-websockets"; +import Signaling from "./signaling"; +import { DiagnosticsBatcher } from "./diagnostics"; +import logger from "../logging"; + +describe("Signaling server-initiated close", () => { + let server: Server; + + beforeEach(async () => { + server = new Server({ port: 0, host: "127.0.0.1" }); + await new Promise((resolve) => server.on("listening", resolve)); + server.register("setMediaPreferences", () => ({})); + }); + + afterEach(async () => { + await server.close(); + }); + + test("tears down without logging errors when the server closes the socket", async () => { + const errorSpy = jest.spyOn(logger, "error"); + const port = (server.wss.address() as AddressInfo).port; + server.on("connection", (socket: any) => { + socket.send(JSON.stringify({ notification: "ready", params: { endpointId: "e-test" } })); + // Close the way the gateway does when the endpoint is deleted. + setTimeout(() => socket.close(1000), 50); + }); + + const batcher = new DiagnosticsBatcher(); + const signaling = new Signaling(batcher); + await signaling.connect({ endpointToken: "t" }, { websocketUrl: `ws://127.0.0.1:${port}` }); + + await new Promise((resolve) => setTimeout(resolve, 200)); + + expect((signaling as any).ws).toBeNull(); + expect((signaling as any).isReady).toBe(false); + expect(errorSpy).not.toHaveBeenCalled(); + errorSpy.mockRestore(); + }); + + test("disconnect() cancels a reconnect already scheduled after a 1001 close", async () => { + const port = (server.wss.address() as AddressInfo).port; + let connections = 0; + server.on("connection", (socket: any) => { + connections++; + socket.send(JSON.stringify({ notification: "ready", params: { endpointId: "e-test" } })); + if (connections === 1) { + setTimeout(() => socket.close(1001), 50); + } + }); + + const signaling = new Signaling(new DiagnosticsBatcher()); + await signaling.connect({ endpointToken: "t" }, { websocketUrl: `ws://127.0.0.1:${port}` }); + await new Promise((resolve) => setTimeout(resolve, 100)); + signaling.disconnect(); + + // Past rpc-websockets' default 1s reconnect interval. + await new Promise((resolve) => setTimeout(resolve, 1500)); + expect(connections).toBe(1); + }); +}); diff --git a/src/v1/signaling.test.ts b/src/v1/signaling.test.ts index 3f6caea..eddb00d 100644 --- a/src/v1/signaling.test.ts +++ b/src/v1/signaling.test.ts @@ -7,6 +7,9 @@ jest.mock("rpc-websockets", () => { return { Client: jest.fn().mockImplementation(() => { const mockClient = { + // An open socket, as rpc-websockets reports it while connected. + ready: true, + socket: {}, on: jest.fn((event: string, callback: Function) => { // Automatically trigger 'ready' event for successful connections if (event === "ready") { diff --git a/src/v1/signaling.ts b/src/v1/signaling.ts index 9c9f1d8..0ba59df 100644 --- a/src/v1/signaling.ts +++ b/src/v1/signaling.ts @@ -185,7 +185,7 @@ class Signaling extends EventEmitter { private _disconnect(notifyLeave: boolean) { logger.debug("Disconnecting websocket"); if (this.ws) { - if (notifyLeave) { + if (notifyLeave && this.socketOpen) { try { this.ws.notify("leave"); } catch (err) { @@ -220,11 +220,18 @@ class Signaling extends EventEmitter { } catch (err) { logger.error("Error disabling auto-reconnect", err); } + // A reconnect may already be scheduled (e.g. after a 1001), which + // setAutoReconnect(false) does not cancel. + clearTimeout((this.ws as any).reconnect_timer_id); this.ws.removeAllListeners(); - try { - this.ws.close(); - } catch (err) { - logger.error(err); + // rpc-websockets drops its socket as soon as the server closes it, so there is + // nothing left to close on a server-initiated disconnect. + if ((this.ws as any).socket) { + try { + this.ws.close(); + } catch (err) { + logger.error(err); + } } this.ws = null; } @@ -277,7 +284,16 @@ class Signaling extends EventEmitter { }) as Promise; } + // rpc-websockets clears its ready flag synchronously when the socket closes, but only + // emits "close" on the next tick, so anything sent from the close handler would fail. + private get socketOpen(): boolean { + return Boolean((this.ws as any)?.ready); + } + private sendDiagnostics(diagnostics: Diagnostics): Promise { + if (!this.socketOpen) { + return Promise.resolve(); + } logger.debug(`Calling "deviceDiagnostics"`); return this.ws?.notify("deviceDiagnostics", diagnostics).catch((err: any) => { logger.error("Error sending diagnostics", err);