github/socket.io
1
1
Fork 0
mirror of https://github.com/socketio/socket.io.git synced 2026-04-30 03:00:39 -04:00
Code Issues Packages Projects Releases Wiki Activity

refactor: remove "self" references

Browse Source
This commit is contained in:
Damien Arrachequesne
2021-04-30 14:30:47 +02:00
parent 252754353a
commit 4c0aa73e06
4 changed files with 121 additions and 143 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
lib/server.js
View File

@@ -345,7 +345,6 @@ class Server extends EventEmitter {
return;
}
const socket = new Socket(id, this, transport, req, protocol);
const self = this;
transport.on("headers", (headers, req) => {
const isInitialRequest = !req._query.sid;
@@ -366,9 +365,9 @@ class Server extends EventEmitter {
this.clients[id] = socket;
this.clientsCount++;
socket.once("close", function() {
delete self.clients[id];
self.clientsCount--;
socket.once("close", () => {
delete this.clients[id];
this.clientsCount--;
});
this.emit("connection", socket);
@@ -382,7 +381,6 @@ class Server extends EventEmitter {
handleUpgrade(req, socket, upgradeHead) {
this.prepare(req);
const self = this;
this.verify(req, true, (errorCode, errorContext) => {
if (errorCode) {
this.emit("connection_error", {
@@ -399,8 +397,8 @@ class Server extends EventEmitter {
upgradeHead = null;
// delegate to ws
self.ws.handleUpgrade(req, socket, head, function(conn) {
self.onWebSocket(req, conn);
this.ws.handleUpgrade(req, socket, head, conn => {
this.onWebSocket(req, conn);
});
});
}
@@ -475,9 +473,7 @@ class Server extends EventEmitter {
* @param {Object} options
* @api public
*/
attach(server, options) {
const self = this;
options = options || {};
attach(server, options = {}) {
let path = (options.path || "/engine.io").replace(/\/$/, "");
const destroyUpgradeTimeout = options.destroyUpgradeTimeout || 1000;
@@ -492,14 +488,14 @@ class Server extends EventEmitter {
// cache and clean up listeners
const listeners = server.listeners("request").slice(0);
server.removeAllListeners("request");
server.on("close", self.close.bind(self));
server.on("listening", self.init.bind(self));
server.on("close", this.close.bind(this));
server.on("listening", this.init.bind(this));
// add request handler
server.on("request", function(req, res) {
server.on("request", (req, res) => {
if (check(req)) {
debug('intercepting request for path "%s"', path);
self.handleRequest(req, res);
this.handleRequest(req, res);
} else {
let i = 0;
const l = listeners.length;
@@ -509,10 +505,10 @@ class Server extends EventEmitter {
}
});
if (~self.opts.transports.indexOf("websocket")) {
server.on("upgrade", function(req, socket, head) {
if (~this.opts.transports.indexOf("websocket")) {
server.on("upgrade", (req, socket, head) => {
if (check(req)) {
self.handleUpgrade(req, socket, head);
this.handleUpgrade(req, socket, head);
} else if (false !== options.destroyUpgrade) {
// default node behavior is to disconnect when no handlers
// but by adding a handler, we prevent that

102
lib/socket.js
View File

@@ -211,10 +211,8 @@ class Socket extends EventEmitter {
this.upgrading = true;
const self = this;
// set transport upgrade timer
self.upgradeTimeoutTimer = setTimeout(function() {
this.upgradeTimeoutTimer = setTimeout(() => {
debug("client did not complete upgrade - closing transport");
cleanup();
if ("open" === transport.readyState) {
@@ -222,75 +220,75 @@ class Socket extends EventEmitter {
}
}, this.server.opts.upgradeTimeout);
function onPacket(packet) {
const 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") {
this.emit("upgrading", transport);
clearInterval(this.checkIntervalTimer);
this.checkIntervalTimer = setInterval(check, 100);
} else if ("upgrade" === packet.type && this.readyState !== "closed") {
debug("got upgrade packet - upgrading");
cleanup();
self.transport.discard();
self.upgraded = true;
self.clearTransport();
self.setTransport(transport);
self.emit("upgrade", transport);
self.flush();
if (self.readyState === "closing") {
transport.close(function() {
self.onClose("forced close");
this.transport.discard();
this.upgraded = true;
this.clearTransport();
this.setTransport(transport);
this.emit("upgrade", transport);
this.flush();
if (this.readyState === "closing") {
transport.close(() => {
this.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) {
const check = () => {
if ("polling" === this.transport.name && this.transport.writable) {
debug("writing a noop packet to polling for fast upgrade");
self.transport.send([{ type: "noop" }]);
this.transport.send([{ type: "noop" }]);
}
}
};
function cleanup() {
self.upgrading = false;
const cleanup = () => {
this.upgrading = false;
clearInterval(self.checkIntervalTimer);
self.checkIntervalTimer = null;
clearInterval(this.checkIntervalTimer);
this.checkIntervalTimer = null;
clearTimeout(self.upgradeTimeoutTimer);
self.upgradeTimeoutTimer = null;
clearTimeout(this.upgradeTimeoutTimer);
this.upgradeTimeoutTimer = null;
transport.removeListener("packet", onPacket);
transport.removeListener("close", onTransportClose);
transport.removeListener("error", onError);
self.removeListener("close", onClose);
}
this.removeListener("close", onClose);
};
function onError(err) {
const onError = err => {
debug("client did not complete upgrade - %s", err);
cleanup();
transport.close();
transport = null;
}
};
function onTransportClose() {
const onTransportClose = () => {
onError("transport closed");
}
};
function onClose() {
const onClose = () => {
onError("socket closed");
}
};
transport.on("packet", onPacket);
transport.once("close", onTransportClose);
transport.once("error", onError);
self.once("close", onClose);
this.once("close", onClose);
}
/**
@@ -335,11 +333,10 @@ class Socket extends EventEmitter {
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 = [];
process.nextTick(() => {
this.writeBuffer = [];
});
this.packetsFn = [];
this.sentCallbackFn = [];
@@ -354,32 +351,31 @@ class Socket extends EventEmitter {
* @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];
const onDrain = () => {
if (this.sentCallbackFn.length > 0) {
const seqFn = this.sentCallbackFn.splice(0, 1)[0];
if ("function" === typeof seqFn) {
debug("executing send callback");
seqFn(self.transport);
seqFn(this.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);
seqFn[i](this.transport);
}
}
}
}
}
};
this.transport.on("drain", onDrain);
this.cleanupFn.push(() => {
this.transport.removeListener("drain", onDrain);
});
}
/**

90
lib/transports/polling.js
View File

@@ -70,16 +70,14 @@ class Polling extends Transport {
this.req = req;
this.res = res;
const self = this;
const onClose = () => {
this.onError("poll connection closed prematurely");
};
function onClose() {
self.onError("poll connection closed prematurely");
}
function cleanup() {
const cleanup = () => {
req.removeListener("close", onClose);
self.req = self.res = null;
}
this.req = this.res = null;
};
req.cleanup = cleanup;
req.on("close", onClose);
@@ -118,21 +116,20 @@ class Polling extends Transport {
this.dataRes = res;
let chunks = isBinary ? Buffer.concat([]) : "";
const self = this;
function cleanup() {
const cleanup = () => {
req.removeListener("data", onData);
req.removeListener("end", onEnd);
req.removeListener("close", onClose);
self.dataReq = self.dataRes = chunks = null;
}
this.dataReq = this.dataRes = chunks = null;
};
function onClose() {
const onClose = () => {
cleanup();
self.onError("data request connection closed prematurely");
}
this.onError("data request connection closed prematurely");
};
function onData(data) {
const onData = data => {
let contentLength;
if (isBinary) {
chunks = Buffer.concat([chunks, data]);
@@ -142,14 +139,14 @@ class Polling extends Transport {
contentLength = Buffer.byteLength(chunks);
}
if (contentLength > self.maxHttpBufferSize) {
if (contentLength > this.maxHttpBufferSize) {
chunks = isBinary ? Buffer.concat([]) : "";
req.connection.destroy();
}
}
};
function onEnd() {
self.onData(chunks);
const onEnd = () => {
this.onData(chunks);
const headers = {
// text/html is required instead of text/plain to avoid an
@@ -158,10 +155,10 @@ class Polling extends Transport {
"Content-Length": 2
};
res.writeHead(200, self.headers(req, headers));
res.writeHead(200, this.headers(req, headers));
res.end("ok");
cleanup();
}
};
req.on("close", onClose);
if (!isBinary) req.setEncoding("utf8");
@@ -177,15 +174,14 @@ class Polling extends Transport {
*/
onData(data) {
debug('received "%s"', data);
const self = this;
const callback = function(packet) {
const callback = packet => {
if ("close" === packet.type) {
debug("got xhr close packet");
self.onClose();
this.onClose();
return false;
}
self.onPacket(packet);
this.onPacket(packet);
};
if (this.protocol === 3) {
@@ -247,9 +243,8 @@ class Polling extends Transport {
*/
write(data, options) {
debug('writing "%s"', data);
const self = this;
this.doWrite(data, options, function() {
self.req.cleanup();
this.doWrite(data, options, () => {
this.req.cleanup();
});
}
@@ -259,8 +254,6 @@ class Polling extends Transport {
* @api private
*/
doWrite(data, options, callback) {
const self = this;
// explicit UTF-8 is required for pages not served under utf
const isString = typeof data === "string";
const contentType = isString
@@ -271,6 +264,14 @@ class Polling extends Transport {
"Content-Type": contentType
};
const respond = data => {
headers["Content-Length"] =
"string" === typeof data ? Buffer.byteLength(data) : data.length;
this.res.writeHead(200, this.headers(this.req, headers));
this.res.end(data);
callback();
};
if (!this.httpCompression || !options.compress) {
respond(data);
return;
@@ -288,10 +289,10 @@ class Polling extends Transport {
return;
}
this.compress(data, encoding, function(err, data) {
this.compress(data, encoding, (err, data) => {
if (err) {
self.res.writeHead(500);
self.res.end();
this.res.writeHead(500);
this.res.end();
callback(err);
return;
}
@@ -299,14 +300,6 @@ class Polling extends Transport {
headers["Content-Encoding"] = encoding;
respond(data);
});
function respond(data) {
headers["Content-Length"] =
"string" === typeof data ? Buffer.byteLength(data) : data.length;
self.res.writeHead(200, self.headers(self.req, headers));
self.res.end(data);
callback();
}
}
/**
@@ -340,7 +333,6 @@ class Polling extends Transport {
doClose(fn) {
debug("closing");
const self = this;
let closeTimeoutTimer;
if (this.dataReq) {
@@ -348,6 +340,12 @@ class Polling extends Transport {
this.dataReq.destroy();
}
const onClose = () => {
clearTimeout(closeTimeoutTimer);
fn();
this.onClose();
};
if (this.writable) {
debug("transport writable - closing right away");
this.send([{ type: "close" }]);
@@ -360,12 +358,6 @@ class Polling extends Transport {
this.shouldClose = onClose;
closeTimeoutTimer = setTimeout(onClose, this.closeTimeout);
}
function onClose() {
clearTimeout(closeTimeoutTimer);
fn();
self.onClose();
}
}
/**

42
lib/transports/websocket.js
View File

@@ -63,38 +63,32 @@ class WebSocket extends Transport {
* @api private
*/
send(packets) {
var self = this;
for (var i = 0; i < packets.length; i++) {
var packet = packets[i];
this.parser.encodePacket(packet, self.supportsBinary, send);
}
function send(data) {
debug('writing "%s"', data);
for (let i = 0; i < packets.length; i++) {
const packet = packets[i];
// always creates a new object since ws modifies it
var opts = {};
const opts = {};
if (packet.options) {
opts.compress = packet.options.compress;
}
if (self.perMessageDeflate) {
var len =
"string" === typeof data ? Buffer.byteLength(data) : data.length;
if (len < self.perMessageDeflate.threshold) {
opts.compress = false;
this.parser.encodePacket(packet, this.supportsBinary, data => {
if (this.perMessageDeflate) {
const len =
"string" === typeof data ? Buffer.byteLength(data) : data.length;
if (len < this.perMessageDeflate.threshold) {
opts.compress = false;
}
}
}
debug('writing "%s"', data);
this.writable = false;
self.writable = false;
self.socket.send(data, opts, onEnd);
}
function onEnd(err) {
if (err) return self.onError("write error", err.stack);
self.writable = true;
self.emit("drain");
this.socket.send(data, opts, err => {
if (err) return this.onError("write error", err.stack);
this.writable = true;
this.emit("drain");
});
});
}
}