Server : Apache/2.4.41 (Ubuntu) System : Linux absol.cf 5.4.0-198-generic #218-Ubuntu SMP Fri Sep 27 20:18:53 UTC 2024 x86_64 User : www-data ( 33) PHP Version : 7.4.33 Disable Function : pcntl_alarm,pcntl_fork,pcntl_waitpid,pcntl_wait,pcntl_wifexited,pcntl_wifstopped,pcntl_wifsignaled,pcntl_wifcontinued,pcntl_wexitstatus,pcntl_wtermsig,pcntl_wstopsig,pcntl_signal,pcntl_signal_get_handler,pcntl_signal_dispatch,pcntl_get_last_error,pcntl_strerror,pcntl_sigprocmask,pcntl_sigwaitinfo,pcntl_sigtimedwait,pcntl_exec,pcntl_getpriority,pcntl_setpriority,pcntl_async_signals,pcntl_unshare, Directory : /proc/self/root/usr/local/lib/node_modules/socket.io/node_modules/engine.io/lib/
Upload File :
Current File : //proc/self/root/usr/local/lib/node_modules/socket.io/node_modules/engine.io/lib/socket.js
const EventEmitter = require("events");
const debug = require("debug")("engine:socket");
class Socket extends EventEmitter {
/**
* Client class (abstract).
*
* @api private
*/
constructor(id, server, transport, req, protocol) {
super();
this.id = id;
this.server = server;
this.upgrading = false;
this.upgraded = false;
this.readyState = "opening";
this.writeBuffer = [];
this.packetsFn = [];
this.sentCallbackFn = [];
this.cleanupFn = [];
this.request = req;
this.protocol = protocol;
// Cache IP since it might not be in the req later
if (req.websocket && req.websocket._socket) {
this.remoteAddress = req.websocket._socket.remoteAddress;
} else {
this.remoteAddress = req.connection.remoteAddress;
}
this.checkIntervalTimer = null;
this.upgradeTimeoutTimer = null;
this.pingTimeoutTimer = null;
this.pingIntervalTimer = null;
this.setTransport(transport);
this.onOpen();
}
/**
* Called upon transport considered open.
*
* @api private
*/
onOpen() {
this.readyState = "open";
// sends an `open` packet
this.transport.sid = this.id;
this.sendPacket(
"open",
JSON.stringify({
sid: this.id,
upgrades: this.getAvailableUpgrades(),
pingInterval: this.server.opts.pingInterval,
pingTimeout: this.server.opts.pingTimeout
})
);
if (this.server.opts.initialPacket) {
this.sendPacket("message", this.server.opts.initialPacket);
}
this.emit("open");
if (this.protocol === 3) {
// in protocol v3, the client sends a ping, and the server answers with a pong
this.resetPingTimeout(
this.server.opts.pingInterval + this.server.opts.pingTimeout
);
} else {
// in protocol v4, the server sends a ping, and the client answers with a pong
this.schedulePing();
}
}
/**
* Called upon transport packet.
*
* @param {Object} packet
* @api private
*/
onPacket(packet) {
if ("open" === this.readyState) {
// export packet event
debug("packet");
this.emit("packet", packet);
// Reset ping timeout on any packet, incoming data is a good sign of
// other side's liveness
this.resetPingTimeout(
this.server.opts.pingInterval + this.server.opts.pingTimeout
);
switch (packet.type) {
case "ping":
if (this.transport.protocol !== 3) {
this.onError("invalid heartbeat direction");
return;
}
debug("got ping");
this.sendPacket("pong");
this.emit("heartbeat");
break;
case "pong":
if (this.transport.protocol === 3) {
this.onError("invalid heartbeat direction");
return;
}
debug("got pong");
this.schedulePing();
this.emit("heartbeat");
break;
case "error":
this.onClose("parse error");
break;
case "message":
this.emit("data", packet.data);
this.emit("message", packet.data);
break;
}
} else {
debug("packet received with closed socket");
}
}
/**
* Called upon transport error.
*
* @param {Error} error object
* @api private
*/
onError(err) {
debug("transport error");
this.onClose("transport error", err);
}
/**
* Pings client every `this.pingInterval` and expects response
* within `this.pingTimeout` or closes connection.
*
* @api private
*/
schedulePing() {
clearTimeout(this.pingIntervalTimer);
this.pingIntervalTimer = setTimeout(() => {
debug(
"writing ping packet - expecting pong within %sms",
this.server.opts.pingTimeout
);
this.sendPacket("ping");
this.resetPingTimeout(this.server.opts.pingTimeout);
}, this.server.opts.pingInterval);
}
/**
* Resets ping timeout.
*
* @api private
*/
resetPingTimeout(timeout) {
clearTimeout(this.pingTimeoutTimer);
this.pingTimeoutTimer = setTimeout(() => {
if (this.readyState === "closed") return;
this.onClose("ping timeout");
}, timeout);
}
/**
* Attaches handlers for the given transport.
*
* @param {Transport} transport
* @api private
*/
setTransport(transport) {
const onError = this.onError.bind(this);
const onPacket = this.onPacket.bind(this);
const flush = this.flush.bind(this);
const onClose = this.onClose.bind(this, "transport close");
this.transport = transport;
this.transport.once("error", onError);
this.transport.on("packet", onPacket);
this.transport.on("drain", flush);
this.transport.once("close", onClose);
// this function will manage packet events (also message callbacks)
this.setupSendCallback();
this.cleanupFn.push(function() {
transport.removeListener("error", onError);
transport.removeListener("packet", onPacket);
transport.removeListener("drain", flush);
transport.removeListener("close", onClose);
});
}
/**
* Upgrades socket to the given transport
*
* @param {Transport} transport
* @api private
*/
maybeUpgrade(transport) {
debug(
'might upgrade socket transport from "%s" to "%s"',
this.transport.name,
transport.name
);
this.upgrading = true;
const self = this;
// set transport upgrade timer
self.upgradeTimeoutTimer = setTimeout(function() {
debug("client did not complete upgrade - closing transport");
cleanup();
if ("open" === transport.readyState) {
transport.close();
}
}, this.server.opts.upgradeTimeout);
function onPacket(packet) {
if ("ping" === packet.type && "probe" === packet.data) {
transport.send([{ type: "pong", data: "probe" }]);
self.emit("upgrading", transport);
clearInterval(self.checkIntervalTimer);
self.checkIntervalTimer = setInterval(check, 100);
} else if ("upgrade" === packet.type && self.readyState !== "closed") {
debug("got upgrade packet - upgrading");
cleanup();
self.transport.discard();
self.upgraded = true;
self.clearTransport();
self.setTransport(transport);
self.emit("upgrade", transport);
self.schedulePing();
self.flush();
if (self.readyState === "closing") {
transport.close(function() {
self.onClose("forced close");
});
}
} else {
cleanup();
transport.close();
}
}
// we force a polling cycle to ensure a fast upgrade
function check() {
if ("polling" === self.transport.name && self.transport.writable) {
debug("writing a noop packet to polling for fast upgrade");
self.transport.send([{ type: "noop" }]);
}
}
function cleanup() {
self.upgrading = false;
clearInterval(self.checkIntervalTimer);
self.checkIntervalTimer = null;
clearTimeout(self.upgradeTimeoutTimer);
self.upgradeTimeoutTimer = null;
transport.removeListener("packet", onPacket);
transport.removeListener("close", onTransportClose);
transport.removeListener("error", onError);
self.removeListener("close", onClose);
}
function onError(err) {
debug("client did not complete upgrade - %s", err);
cleanup();
transport.close();
transport = null;
}
function onTransportClose() {
onError("transport closed");
}
function onClose() {
onError("socket closed");
}
transport.on("packet", onPacket);
transport.once("close", onTransportClose);
transport.once("error", onError);
self.once("close", onClose);
}
/**
* Clears listeners and timers associated with current transport.
*
* @api private
*/
clearTransport() {
let cleanup;
const toCleanUp = this.cleanupFn.length;
for (let i = 0; i < toCleanUp; i++) {
cleanup = this.cleanupFn.shift();
cleanup();
}
// silence further transport errors and prevent uncaught exceptions
this.transport.on("error", function() {
debug("error triggered by discarded transport");
});
// ensure transport won't stay open
this.transport.close();
clearTimeout(this.pingTimeoutTimer);
}
/**
* Called upon transport considered closed.
* Possible reasons: `ping timeout`, `client error`, `parse error`,
* `transport error`, `server close`, `transport close`
*/
onClose(reason, description) {
if ("closed" !== this.readyState) {
this.readyState = "closed";
// clear timers
clearTimeout(this.pingIntervalTimer);
clearTimeout(this.pingTimeoutTimer);
clearInterval(this.checkIntervalTimer);
this.checkIntervalTimer = null;
clearTimeout(this.upgradeTimeoutTimer);
const self = this;
// clean writeBuffer in next tick, so developers can still
// grab the writeBuffer on 'close' event
process.nextTick(function() {
self.writeBuffer = [];
});
this.packetsFn = [];
this.sentCallbackFn = [];
this.clearTransport();
this.emit("close", reason, description);
}
}
/**
* Setup and manage send callback
*
* @api private
*/
setupSendCallback() {
const self = this;
this.transport.on("drain", onDrain);
this.cleanupFn.push(function() {
self.transport.removeListener("drain", onDrain);
});
// the message was sent successfully, execute the callback
function onDrain() {
if (self.sentCallbackFn.length > 0) {
const seqFn = self.sentCallbackFn.splice(0, 1)[0];
if ("function" === typeof seqFn) {
debug("executing send callback");
seqFn(self.transport);
} else if (Array.isArray(seqFn)) {
debug("executing batch send callback");
const l = seqFn.length;
let i = 0;
for (; i < l; i++) {
if ("function" === typeof seqFn[i]) {
seqFn[i](self.transport);
}
}
}
}
}
}
/**
* Sends a message packet.
*
* @param {String} message
* @param {Object} options
* @param {Function} callback
* @return {Socket} for chaining
* @api public
*/
send(data, options, callback) {
this.sendPacket("message", data, options, callback);
return this;
}
write(data, options, callback) {
this.sendPacket("message", data, options, callback);
return this;
}
/**
* Sends a packet.
*
* @param {String} packet type
* @param {String} optional, data
* @param {Object} options
* @api private
*/
sendPacket(type, data, options, callback) {
if ("function" === typeof options) {
callback = options;
options = null;
}
options = options || {};
options.compress = false !== options.compress;
if ("closing" !== this.readyState && "closed" !== this.readyState) {
debug('sending packet "%s" (%s)', type, data);
const packet = {
type: type,
options: options
};
if (data) packet.data = data;
// exports packetCreate event
this.emit("packetCreate", packet);
this.writeBuffer.push(packet);
// add send callback to object, if defined
if (callback) this.packetsFn.push(callback);
this.flush();
}
}
/**
* Attempts to flush the packets buffer.
*
* @api private
*/
flush() {
if (
"closed" !== this.readyState &&
this.transport.writable &&
this.writeBuffer.length
) {
debug("flushing buffer to transport");
this.emit("flush", this.writeBuffer);
this.server.emit("flush", this, this.writeBuffer);
const wbuf = this.writeBuffer;
this.writeBuffer = [];
if (!this.transport.supportsFraming) {
this.sentCallbackFn.push(this.packetsFn);
} else {
this.sentCallbackFn.push.apply(this.sentCallbackFn, this.packetsFn);
}
this.packetsFn = [];
this.transport.send(wbuf);
this.emit("drain");
this.server.emit("drain", this);
}
}
/**
* Get available upgrades for this socket.
*
* @api private
*/
getAvailableUpgrades() {
const availableUpgrades = [];
const allUpgrades = this.server.upgrades(this.transport.name);
let i = 0;
const l = allUpgrades.length;
for (; i < l; ++i) {
const upg = allUpgrades[i];
if (this.server.opts.transports.indexOf(upg) !== -1) {
availableUpgrades.push(upg);
}
}
return availableUpgrades;
}
/**
* Closes the socket and underlying transport.
*
* @param {Boolean} optional, discard
* @return {Socket} for chaining
* @api public
*/
close(discard) {
if ("open" !== this.readyState) return;
this.readyState = "closing";
if (this.writeBuffer.length) {
this.once("drain", this.closeTransport.bind(this, discard));
return;
}
this.closeTransport(discard);
}
/**
* Closes the underlying transport.
*
* @param {Boolean} discard
* @api private
*/
closeTransport(discard) {
if (discard) this.transport.discard();
this.transport.close(this.onClose.bind(this, "forced close"));
}
}
module.exports = Socket;
Server : Apache/2.4.41 (Ubuntu)