| "use strict"; |
| var __rest = (this && this.__rest) || function (s, e) { |
| var t = {}; |
| for (var p in s) if (Object.prototype.hasOwnProperty.call(s, p) && e.indexOf(p) < 0) |
| t[p] = s[p]; |
| if (s != null && typeof Object.getOwnPropertySymbols === "function") |
| for (var i = 0, p = Object.getOwnPropertySymbols(s); i < p.length; i++) { |
| if (e.indexOf(p[i]) < 0 && Object.prototype.propertyIsEnumerable.call(s, p[i])) |
| t[p[i]] = s[p[i]]; |
| } |
| return t; |
| }; |
| Object.defineProperty(exports, "__esModule", { value: true }); |
| exports.ClusterAdapterWithHeartbeat = exports.ClusterAdapter = exports.MessageType = void 0; |
| const in_memory_adapter_1 = require("./in-memory-adapter"); |
| const debug_1 = require("debug"); |
| const crypto_1 = require("crypto"); |
| const debug = (0, debug_1.debug)("socket.io-adapter"); |
| const EMITTER_UID = "emitter"; |
| const DEFAULT_TIMEOUT = 5000; |
| function randomId() { |
| return (0, crypto_1.randomBytes)(8).toString("hex"); |
| } |
| var MessageType; |
| (function (MessageType) { |
| MessageType[MessageType["INITIAL_HEARTBEAT"] = 1] = "INITIAL_HEARTBEAT"; |
| MessageType[MessageType["HEARTBEAT"] = 2] = "HEARTBEAT"; |
| MessageType[MessageType["BROADCAST"] = 3] = "BROADCAST"; |
| MessageType[MessageType["SOCKETS_JOIN"] = 4] = "SOCKETS_JOIN"; |
| MessageType[MessageType["SOCKETS_LEAVE"] = 5] = "SOCKETS_LEAVE"; |
| MessageType[MessageType["DISCONNECT_SOCKETS"] = 6] = "DISCONNECT_SOCKETS"; |
| MessageType[MessageType["FETCH_SOCKETS"] = 7] = "FETCH_SOCKETS"; |
| MessageType[MessageType["FETCH_SOCKETS_RESPONSE"] = 8] = "FETCH_SOCKETS_RESPONSE"; |
| MessageType[MessageType["SERVER_SIDE_EMIT"] = 9] = "SERVER_SIDE_EMIT"; |
| MessageType[MessageType["SERVER_SIDE_EMIT_RESPONSE"] = 10] = "SERVER_SIDE_EMIT_RESPONSE"; |
| MessageType[MessageType["BROADCAST_CLIENT_COUNT"] = 11] = "BROADCAST_CLIENT_COUNT"; |
| MessageType[MessageType["BROADCAST_ACK"] = 12] = "BROADCAST_ACK"; |
| MessageType[MessageType["ADAPTER_CLOSE"] = 13] = "ADAPTER_CLOSE"; |
| })(MessageType = exports.MessageType || (exports.MessageType = {})); |
| function encodeOptions(opts) { |
| return { |
| rooms: [...opts.rooms], |
| except: [...opts.except], |
| flags: opts.flags, |
| }; |
| } |
| function decodeOptions(opts) { |
| return { |
| rooms: new Set(opts.rooms), |
| except: new Set(opts.except), |
| flags: opts.flags, |
| }; |
| } |
| /** |
| * A cluster-ready adapter. Any extending class must: |
| * |
| * - implement {@link ClusterAdapter#doPublish} and {@link ClusterAdapter#doPublishResponse} |
| * - call {@link ClusterAdapter#onMessage} and {@link ClusterAdapter#onResponse} |
| */ |
| class ClusterAdapter extends in_memory_adapter_1.Adapter { |
| constructor(nsp) { |
| super(nsp); |
| this.requests = new Map(); |
| this.ackRequests = new Map(); |
| this.uid = randomId(); |
| } |
| /** |
| * Called when receiving a message from another member of the cluster. |
| * |
| * @param message |
| * @param offset |
| * @protected |
| */ |
| onMessage(message, offset) { |
| if (message.uid === this.uid) { |
| return debug("[%s] ignore message from self", this.uid); |
| } |
| debug("[%s] new event of type %d from %s", this.uid, message.type, message.uid); |
| switch (message.type) { |
| case MessageType.BROADCAST: { |
| const withAck = message.data.requestId !== undefined; |
| if (withAck) { |
| super.broadcastWithAck(message.data.packet, decodeOptions(message.data.opts), (clientCount) => { |
| debug("[%s] waiting for %d client acknowledgements", this.uid, clientCount); |
| this.publishResponse(message.uid, { |
| type: MessageType.BROADCAST_CLIENT_COUNT, |
| data: { |
| requestId: message.data.requestId, |
| clientCount, |
| }, |
| }); |
| }, (arg) => { |
| debug("[%s] received acknowledgement with value %j", this.uid, arg); |
| this.publishResponse(message.uid, { |
| type: MessageType.BROADCAST_ACK, |
| data: { |
| requestId: message.data.requestId, |
| packet: arg, |
| }, |
| }); |
| }); |
| } |
| else { |
| const packet = message.data.packet; |
| const opts = decodeOptions(message.data.opts); |
| this.addOffsetIfNecessary(packet, opts, offset); |
| super.broadcast(packet, opts); |
| } |
| break; |
| } |
| case MessageType.SOCKETS_JOIN: |
| super.addSockets(decodeOptions(message.data.opts), message.data.rooms); |
| break; |
| case MessageType.SOCKETS_LEAVE: |
| super.delSockets(decodeOptions(message.data.opts), message.data.rooms); |
| break; |
| case MessageType.DISCONNECT_SOCKETS: |
| super.disconnectSockets(decodeOptions(message.data.opts), message.data.close); |
| break; |
| case MessageType.FETCH_SOCKETS: { |
| debug("[%s] calling fetchSockets with opts %j", this.uid, message.data.opts); |
| super |
| .fetchSockets(decodeOptions(message.data.opts)) |
| .then((localSockets) => { |
| this.publishResponse(message.uid, { |
| type: MessageType.FETCH_SOCKETS_RESPONSE, |
| data: { |
| requestId: message.data.requestId, |
| sockets: localSockets.map((socket) => { |
| // remove sessionStore from handshake, as it may contain circular references |
| const _a = socket.handshake, { sessionStore } = _a, handshake = __rest(_a, ["sessionStore"]); |
| return { |
| id: socket.id, |
| handshake, |
| rooms: [...socket.rooms], |
| data: socket.data, |
| }; |
| }), |
| }, |
| }); |
| }); |
| break; |
| } |
| case MessageType.SERVER_SIDE_EMIT: { |
| const packet = message.data.packet; |
| const withAck = message.data.requestId !== undefined; |
| if (!withAck) { |
| this.nsp._onServerSideEmit(packet); |
| return; |
| } |
| let called = false; |
| const callback = (arg) => { |
| // only one argument is expected |
| if (called) { |
| return; |
| } |
| called = true; |
| debug("[%s] calling acknowledgement with %j", this.uid, arg); |
| this.publishResponse(message.uid, { |
| type: MessageType.SERVER_SIDE_EMIT_RESPONSE, |
| data: { |
| requestId: message.data.requestId, |
| packet: arg, |
| }, |
| }); |
| }; |
| this.nsp._onServerSideEmit([...packet, callback]); |
| break; |
| } |
| // @ts-ignore |
| case MessageType.BROADCAST_CLIENT_COUNT: |
| // @ts-ignore |
| case MessageType.BROADCAST_ACK: |
| // @ts-ignore |
| case MessageType.FETCH_SOCKETS_RESPONSE: |
| // @ts-ignore |
| case MessageType.SERVER_SIDE_EMIT_RESPONSE: |
| // extending classes may not make a distinction between a ClusterMessage and a ClusterResponse payload and may |
| // always call the onMessage() method |
| this.onResponse(message); |
| break; |
| default: |
| debug("[%s] unknown message type: %s", this.uid, message.type); |
| } |
| } |
| /** |
| * Called when receiving a response from another member of the cluster. |
| * |
| * @param response |
| * @protected |
| */ |
| onResponse(response) { |
| var _a, _b; |
| const requestId = response.data.requestId; |
| debug("[%s] received response %s to request %s", this.uid, response.type, requestId); |
| switch (response.type) { |
| case MessageType.BROADCAST_CLIENT_COUNT: { |
| (_a = this.ackRequests |
| .get(requestId)) === null || _a === void 0 ? void 0 : _a.clientCountCallback(response.data.clientCount); |
| break; |
| } |
| case MessageType.BROADCAST_ACK: { |
| (_b = this.ackRequests.get(requestId)) === null || _b === void 0 ? void 0 : _b.ack(response.data.packet); |
| break; |
| } |
| case MessageType.FETCH_SOCKETS_RESPONSE: { |
| const request = this.requests.get(requestId); |
| if (!request) { |
| return; |
| } |
| request.current++; |
| response.data.sockets.forEach((socket) => request.responses.push(socket)); |
| if (request.current === request.expected) { |
| clearTimeout(request.timeout); |
| request.resolve(request.responses); |
| this.requests.delete(requestId); |
| } |
| break; |
| } |
| case MessageType.SERVER_SIDE_EMIT_RESPONSE: { |
| const request = this.requests.get(requestId); |
| if (!request) { |
| return; |
| } |
| request.current++; |
| request.responses.push(response.data.packet); |
| if (request.current === request.expected) { |
| clearTimeout(request.timeout); |
| request.resolve(null, request.responses); |
| this.requests.delete(requestId); |
| } |
| break; |
| } |
| default: |
| // @ts-ignore |
| debug("[%s] unknown response type: %s", this.uid, response.type); |
| } |
| } |
| async broadcast(packet, opts) { |
| var _a; |
| const onlyLocal = (_a = opts.flags) === null || _a === void 0 ? void 0 : _a.local; |
| if (!onlyLocal) { |
| try { |
| const offset = await this.publishAndReturnOffset({ |
| type: MessageType.BROADCAST, |
| data: { |
| packet, |
| opts: encodeOptions(opts), |
| }, |
| }); |
| this.addOffsetIfNecessary(packet, opts, offset); |
| } |
| catch (e) { |
| return debug("[%s] error while broadcasting message: %s", this.uid, e.message); |
| } |
| } |
| super.broadcast(packet, opts); |
| } |
| /** |
| * Adds an offset at the end of the data array in order to allow the client to receive any missed packets when it |
| * reconnects after a temporary disconnection. |
| * |
| * @param packet |
| * @param opts |
| * @param offset |
| * @private |
| */ |
| addOffsetIfNecessary(packet, opts, offset) { |
| var _a; |
| if (!this.nsp.server.opts.connectionStateRecovery) { |
| return; |
| } |
| const isEventPacket = packet.type === 2; |
| // packets with acknowledgement are not stored because the acknowledgement function cannot be serialized and |
| // restored on another server upon reconnection |
| const withoutAcknowledgement = packet.id === undefined; |
| const notVolatile = ((_a = opts.flags) === null || _a === void 0 ? void 0 : _a.volatile) === undefined; |
| if (isEventPacket && withoutAcknowledgement && notVolatile) { |
| packet.data.push(offset); |
| } |
| } |
| broadcastWithAck(packet, opts, clientCountCallback, ack) { |
| var _a; |
| const onlyLocal = (_a = opts === null || opts === void 0 ? void 0 : opts.flags) === null || _a === void 0 ? void 0 : _a.local; |
| if (!onlyLocal) { |
| const requestId = randomId(); |
| this.ackRequests.set(requestId, { |
| clientCountCallback, |
| ack, |
| }); |
| this.publish({ |
| type: MessageType.BROADCAST, |
| data: { |
| packet, |
| requestId, |
| opts: encodeOptions(opts), |
| }, |
| }); |
| // we have no way to know at this level whether the server has received an acknowledgement from each client, so we |
| // will simply clean up the ackRequests map after the given delay |
| setTimeout(() => { |
| this.ackRequests.delete(requestId); |
| }, opts.flags.timeout); |
| } |
| super.broadcastWithAck(packet, opts, clientCountCallback, ack); |
| } |
| async addSockets(opts, rooms) { |
| var _a; |
| const onlyLocal = (_a = opts.flags) === null || _a === void 0 ? void 0 : _a.local; |
| if (!onlyLocal) { |
| try { |
| await this.publishAndReturnOffset({ |
| type: MessageType.SOCKETS_JOIN, |
| data: { |
| opts: encodeOptions(opts), |
| rooms, |
| }, |
| }); |
| } |
| catch (e) { |
| debug("[%s] error while publishing message: %s", this.uid, e.message); |
| } |
| } |
| super.addSockets(opts, rooms); |
| } |
| async delSockets(opts, rooms) { |
| var _a; |
| const onlyLocal = (_a = opts.flags) === null || _a === void 0 ? void 0 : _a.local; |
| if (!onlyLocal) { |
| try { |
| await this.publishAndReturnOffset({ |
| type: MessageType.SOCKETS_LEAVE, |
| data: { |
| opts: encodeOptions(opts), |
| rooms, |
| }, |
| }); |
| } |
| catch (e) { |
| debug("[%s] error while publishing message: %s", this.uid, e.message); |
| } |
| } |
| super.delSockets(opts, rooms); |
| } |
| async disconnectSockets(opts, close) { |
| var _a; |
| const onlyLocal = (_a = opts.flags) === null || _a === void 0 ? void 0 : _a.local; |
| if (!onlyLocal) { |
| try { |
| await this.publishAndReturnOffset({ |
| type: MessageType.DISCONNECT_SOCKETS, |
| data: { |
| opts: encodeOptions(opts), |
| close, |
| }, |
| }); |
| } |
| catch (e) { |
| debug("[%s] error while publishing message: %s", this.uid, e.message); |
| } |
| } |
| super.disconnectSockets(opts, close); |
| } |
| async fetchSockets(opts) { |
| var _a; |
| const [localSockets, serverCount] = await Promise.all([ |
| super.fetchSockets(opts), |
| this.serverCount(), |
| ]); |
| const expectedResponseCount = serverCount - 1; |
| if (((_a = opts.flags) === null || _a === void 0 ? void 0 : _a.local) || expectedResponseCount <= 0) { |
| return localSockets; |
| } |
| const requestId = randomId(); |
| return new Promise((resolve, reject) => { |
| const timeout = setTimeout(() => { |
| const storedRequest = this.requests.get(requestId); |
| if (storedRequest) { |
| reject(new Error(`timeout reached: only ${storedRequest.current} responses received out of ${storedRequest.expected}`)); |
| this.requests.delete(requestId); |
| } |
| }, opts.flags.timeout || DEFAULT_TIMEOUT); |
| const storedRequest = { |
| type: MessageType.FETCH_SOCKETS, |
| resolve, |
| timeout, |
| current: 0, |
| expected: expectedResponseCount, |
| responses: localSockets, |
| }; |
| this.requests.set(requestId, storedRequest); |
| this.publish({ |
| type: MessageType.FETCH_SOCKETS, |
| data: { |
| opts: encodeOptions(opts), |
| requestId, |
| }, |
| }); |
| }); |
| } |
| async serverSideEmit(packet) { |
| const withAck = typeof packet[packet.length - 1] === "function"; |
| if (!withAck) { |
| return this.publish({ |
| type: MessageType.SERVER_SIDE_EMIT, |
| data: { |
| packet, |
| }, |
| }); |
| } |
| const ack = packet.pop(); |
| const expectedResponseCount = (await this.serverCount()) - 1; |
| debug('[%s] waiting for %d responses to "serverSideEmit" request', this.uid, expectedResponseCount); |
| if (expectedResponseCount <= 0) { |
| return ack(null, []); |
| } |
| const requestId = randomId(); |
| const timeout = setTimeout(() => { |
| const storedRequest = this.requests.get(requestId); |
| if (storedRequest) { |
| ack(new Error(`timeout reached: only ${storedRequest.current} responses received out of ${storedRequest.expected}`), storedRequest.responses); |
| this.requests.delete(requestId); |
| } |
| }, DEFAULT_TIMEOUT); |
| const storedRequest = { |
| type: MessageType.SERVER_SIDE_EMIT, |
| resolve: ack, |
| timeout, |
| current: 0, |
| expected: expectedResponseCount, |
| responses: [], |
| }; |
| this.requests.set(requestId, storedRequest); |
| this.publish({ |
| type: MessageType.SERVER_SIDE_EMIT, |
| data: { |
| requestId, |
| packet, |
| }, |
| }); |
| } |
| publish(message) { |
| this.publishAndReturnOffset(message).catch((err) => { |
| debug("[%s] error while publishing message: %s", this.uid, err); |
| }); |
| } |
| publishAndReturnOffset(message) { |
| message.uid = this.uid; |
| message.nsp = this.nsp.name; |
| return this.doPublish(message); |
| } |
| publishResponse(requesterUid, response) { |
| response.uid = this.uid; |
| response.nsp = this.nsp.name; |
| this.doPublishResponse(requesterUid, response).catch((err) => { |
| debug("[%s] error while publishing response: %s", this.uid, err); |
| }); |
| } |
| } |
| exports.ClusterAdapter = ClusterAdapter; |
| class ClusterAdapterWithHeartbeat extends ClusterAdapter { |
| constructor(nsp, opts) { |
| super(nsp); |
| this.nodesMap = new Map(); // uid => timestamp of last message |
| this.customRequests = new Map(); |
| this._opts = Object.assign({ |
| heartbeatInterval: 5000, |
| heartbeatTimeout: 10000, |
| }, opts); |
| this.cleanupTimer = setInterval(() => { |
| const now = Date.now(); |
| this.nodesMap.forEach((lastSeen, uid) => { |
| const nodeSeemsDown = now - lastSeen > this._opts.heartbeatTimeout; |
| if (nodeSeemsDown) { |
| debug("[%s] node %s seems down", this.uid, uid); |
| this.removeNode(uid); |
| } |
| }); |
| }, 1000); |
| } |
| init() { |
| this.publish({ |
| type: MessageType.INITIAL_HEARTBEAT, |
| }); |
| } |
| scheduleHeartbeat() { |
| if (this.heartbeatTimer) { |
| this.heartbeatTimer.refresh(); |
| } |
| else { |
| this.heartbeatTimer = setTimeout(() => { |
| this.publish({ |
| type: MessageType.HEARTBEAT, |
| }); |
| }, this._opts.heartbeatInterval); |
| } |
| } |
| close() { |
| this.publish({ |
| type: MessageType.ADAPTER_CLOSE, |
| }); |
| clearTimeout(this.heartbeatTimer); |
| if (this.cleanupTimer) { |
| clearInterval(this.cleanupTimer); |
| } |
| } |
| onMessage(message, offset) { |
| if (message.uid === this.uid) { |
| return debug("[%s] ignore message from self", this.uid); |
| } |
| if (message.uid && message.uid !== EMITTER_UID) { |
| // we track the UID of each sender, in order to know how many servers there are in the cluster |
| this.nodesMap.set(message.uid, Date.now()); |
| } |
| debug("[%s] new event of type %d from %s", this.uid, message.type, message.uid); |
| switch (message.type) { |
| case MessageType.INITIAL_HEARTBEAT: |
| this.publish({ |
| type: MessageType.HEARTBEAT, |
| }); |
| break; |
| case MessageType.HEARTBEAT: |
| // nothing to do |
| break; |
| case MessageType.ADAPTER_CLOSE: |
| this.removeNode(message.uid); |
| break; |
| default: |
| super.onMessage(message, offset); |
| } |
| } |
| serverCount() { |
| return Promise.resolve(1 + this.nodesMap.size); |
| } |
| publish(message) { |
| this.scheduleHeartbeat(); |
| return super.publish(message); |
| } |
| async serverSideEmit(packet) { |
| const withAck = typeof packet[packet.length - 1] === "function"; |
| if (!withAck) { |
| return this.publish({ |
| type: MessageType.SERVER_SIDE_EMIT, |
| data: { |
| packet, |
| }, |
| }); |
| } |
| const ack = packet.pop(); |
| const expectedResponseCount = this.nodesMap.size; |
| debug('[%s] waiting for %d responses to "serverSideEmit" request', this.uid, expectedResponseCount); |
| if (expectedResponseCount <= 0) { |
| return ack(null, []); |
| } |
| const requestId = randomId(); |
| const timeout = setTimeout(() => { |
| const storedRequest = this.customRequests.get(requestId); |
| if (storedRequest) { |
| ack(new Error(`timeout reached: missing ${storedRequest.missingUids.size} responses`), storedRequest.responses); |
| this.customRequests.delete(requestId); |
| } |
| }, DEFAULT_TIMEOUT); |
| const storedRequest = { |
| type: MessageType.SERVER_SIDE_EMIT, |
| resolve: ack, |
| timeout, |
| missingUids: new Set([...this.nodesMap.keys()]), |
| responses: [], |
| }; |
| this.customRequests.set(requestId, storedRequest); |
| this.publish({ |
| type: MessageType.SERVER_SIDE_EMIT, |
| data: { |
| requestId, |
| packet, |
| }, |
| }); |
| } |
| async fetchSockets(opts) { |
| var _a; |
| const [localSockets, serverCount] = await Promise.all([ |
| super.fetchSockets({ |
| rooms: opts.rooms, |
| except: opts.except, |
| flags: { |
| local: true, |
| }, |
| }), |
| this.serverCount(), |
| ]); |
| const expectedResponseCount = serverCount - 1; |
| if (((_a = opts.flags) === null || _a === void 0 ? void 0 : _a.local) || expectedResponseCount <= 0) { |
| return localSockets; |
| } |
| const requestId = randomId(); |
| return new Promise((resolve, reject) => { |
| const timeout = setTimeout(() => { |
| const storedRequest = this.customRequests.get(requestId); |
| if (storedRequest) { |
| reject(new Error(`timeout reached: missing ${storedRequest.missingUids.size} responses`)); |
| this.customRequests.delete(requestId); |
| } |
| }, opts.flags.timeout || DEFAULT_TIMEOUT); |
| const storedRequest = { |
| type: MessageType.FETCH_SOCKETS, |
| resolve, |
| timeout, |
| missingUids: new Set([...this.nodesMap.keys()]), |
| responses: localSockets, |
| }; |
| this.customRequests.set(requestId, storedRequest); |
| this.publish({ |
| type: MessageType.FETCH_SOCKETS, |
| data: { |
| opts: encodeOptions(opts), |
| requestId, |
| }, |
| }); |
| }); |
| } |
| onResponse(response) { |
| const requestId = response.data.requestId; |
| debug("[%s] received response %s to request %s", this.uid, response.type, requestId); |
| switch (response.type) { |
| case MessageType.FETCH_SOCKETS_RESPONSE: { |
| const request = this.customRequests.get(requestId); |
| if (!request) { |
| return; |
| } |
| response.data.sockets.forEach((socket) => request.responses.push(socket)); |
| request.missingUids.delete(response.uid); |
| if (request.missingUids.size === 0) { |
| clearTimeout(request.timeout); |
| request.resolve(request.responses); |
| this.customRequests.delete(requestId); |
| } |
| break; |
| } |
| case MessageType.SERVER_SIDE_EMIT_RESPONSE: { |
| const request = this.customRequests.get(requestId); |
| if (!request) { |
| return; |
| } |
| request.responses.push(response.data.packet); |
| request.missingUids.delete(response.uid); |
| if (request.missingUids.size === 0) { |
| clearTimeout(request.timeout); |
| request.resolve(null, request.responses); |
| this.customRequests.delete(requestId); |
| } |
| break; |
| } |
| default: |
| super.onResponse(response); |
| } |
| } |
| removeNode(uid) { |
| this.customRequests.forEach((request, requestId) => { |
| request.missingUids.delete(uid); |
| if (request.missingUids.size === 0) { |
| clearTimeout(request.timeout); |
| if (request.type === MessageType.FETCH_SOCKETS) { |
| request.resolve(request.responses); |
| } |
| else if (request.type === MessageType.SERVER_SIDE_EMIT) { |
| request.resolve(null, request.responses); |
| } |
| this.customRequests.delete(requestId); |
| } |
| }); |
| this.nodesMap.delete(uid); |
| } |
| } |
| exports.ClusterAdapterWithHeartbeat = ClusterAdapterWithHeartbeat; |
