You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

269 lines
7.7 KiB

1 month ago
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.Client = void 0;
const socket_io_parser_1 = require("socket.io-parser");
const debugModule = require("debug");
const url = require("url");
const debug = debugModule("socket.io:client");
class Client {
/**
* Client constructor.
*
* @param server instance
* @param conn
* @package
*/
constructor(server, conn) {
this.sockets = new Map();
this.nsps = new Map();
this.server = server;
this.conn = conn;
this.encoder = server.encoder;
this.decoder = new server._parser.Decoder();
this.id = conn.id;
this.setup();
}
/**
* @return the reference to the request that originated the Engine.IO connection
*
* @public
*/
get request() {
return this.conn.request;
}
/**
* Sets up event listeners.
*
* @private
*/
setup() {
this.onclose = this.onclose.bind(this);
this.ondata = this.ondata.bind(this);
this.onerror = this.onerror.bind(this);
this.ondecoded = this.ondecoded.bind(this);
// @ts-ignore
this.decoder.on("decoded", this.ondecoded);
this.conn.on("data", this.ondata);
this.conn.on("error", this.onerror);
this.conn.on("close", this.onclose);
this.connectTimeout = setTimeout(() => {
if (this.nsps.size === 0) {
debug("no namespace joined yet, close the client");
this.close();
}
else {
debug("the client has already joined a namespace, nothing to do");
}
}, this.server._connectTimeout);
}
/**
* Connects a client to a namespace.
*
* @param {String} name - the namespace
* @param {Object} auth - the auth parameters
* @private
*/
connect(name, auth = {}) {
if (this.server._nsps.has(name)) {
debug("connecting to namespace %s", name);
return this.doConnect(name, auth);
}
this.server._checkNamespace(name, auth, (dynamicNspName) => {
if (dynamicNspName) {
this.doConnect(name, auth);
}
else {
debug("creation of namespace %s was denied", name);
this._packet({
type: socket_io_parser_1.PacketType.CONNECT_ERROR,
nsp: name,
data: {
message: "Invalid namespace",
},
});
}
});
}
/**
* Connects a client to a namespace.
*
* @param name - the namespace
* @param {Object} auth - the auth parameters
*
* @private
*/
doConnect(name, auth) {
const nsp = this.server.of(name);
nsp._add(this, auth, (socket) => {
this.sockets.set(socket.id, socket);
this.nsps.set(nsp.name, socket);
if (this.connectTimeout) {
clearTimeout(this.connectTimeout);
this.connectTimeout = undefined;
}
});
}
/**
* Disconnects from all namespaces and closes transport.
*
* @private
*/
_disconnect() {
for (const socket of this.sockets.values()) {
socket.disconnect();
}
this.sockets.clear();
this.close();
}
/**
* Removes a socket. Called by each `Socket`.
*
* @private
*/
_remove(socket) {
if (this.sockets.has(socket.id)) {
const nsp = this.sockets.get(socket.id).nsp.name;
this.sockets.delete(socket.id);
this.nsps.delete(nsp);
}
else {
debug("ignoring remove for %s", socket.id);
}
}
/**
* Closes the underlying connection.
*
* @private
*/
close() {
if ("open" === this.conn.readyState) {
debug("forcing transport close");
this.conn.close();
this.onclose("forced server close");
}
}
/**
* Writes a packet to the transport.
*
* @param {Object} packet object
* @param {Object} opts
* @private
*/
_packet(packet, opts = {}) {
if (this.conn.readyState !== "open") {
debug("ignoring packet write %j", packet);
return;
}
const encodedPackets = opts.preEncoded
? packet // previous versions of the adapter incorrectly used socket.packet() instead of writeToEngine()
: this.encoder.encode(packet);
this.writeToEngine(encodedPackets, opts);
}
writeToEngine(encodedPackets, opts) {
if (opts.volatile && !this.conn.transport.writable) {
debug("volatile packet is discarded since the transport is not currently writable");
return;
}
const packets = Array.isArray(encodedPackets)
? encodedPackets
: [encodedPackets];
for (const encodedPacket of packets) {
this.conn.write(encodedPacket, opts);
}
}
/**
* Called with incoming transport data.
*
* @private
*/
ondata(data) {
// try/catch is needed for protocol violations (GH-1880)
try {
this.decoder.add(data);
}
catch (e) {
debug("invalid packet format");
this.onerror(e);
}
}
/**
* Called when parser fully decodes a packet.
*
* @private
*/
ondecoded(packet) {
let namespace;
let authPayload;
if (this.conn.protocol === 3) {
const parsed = url.parse(packet.nsp, true);
namespace = parsed.pathname;
authPayload = parsed.query;
}
else {
namespace = packet.nsp;
authPayload = packet.data;
}
const socket = this.nsps.get(namespace);
if (!socket && packet.type === socket_io_parser_1.PacketType.CONNECT) {
this.connect(namespace, authPayload);
}
else if (socket &&
packet.type !== socket_io_parser_1.PacketType.CONNECT &&
packet.type !== socket_io_parser_1.PacketType.CONNECT_ERROR) {
process.nextTick(function () {
socket._onpacket(packet);
});
}
else {
debug("invalid state (packet type: %s)", packet.type);
this.close();
}
}
/**
* Handles an error.
*
* @param {Object} err object
* @private
*/
onerror(err) {
for (const socket of this.sockets.values()) {
socket._onerror(err);
}
this.conn.close();
}
/**
* Called upon transport close.
*
* @param reason
* @param description
* @private
*/
onclose(reason, description) {
debug("client close with reason %s", reason);
// ignore a potential subsequent `close` event
this.destroy();
// `nsps` and `sockets` are cleaned up seamlessly
for (const socket of this.sockets.values()) {
socket._onclose(reason, description);
}
this.sockets.clear();
this.decoder.destroy(); // clean up decoder
}
/**
* Cleans up event listeners.
* @private
*/
destroy() {
this.conn.removeListener("data", this.ondata);
this.conn.removeListener("error", this.onerror);
this.conn.removeListener("close", this.onclose);
// @ts-ignore
this.decoder.removeListener("decoded", this.ondecoded);
if (this.connectTimeout) {
clearTimeout(this.connectTimeout);
this.connectTimeout = undefined;
}
}
}
exports.Client = Client;