Init. Map does not function and only working on backend at the moment.
This commit is contained in:
eetnaviation
674
node_modules/socket.io-adapter/dist/cluster-adapter.js
generated
vendored
Normal file
674
node_modules/socket.io-adapter/dist/cluster-adapter.js
generated
vendored
Normal file
@ -0,0 +1,674 @@
|
||||
"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;
|
||||
Reference in New Issue
Block a user