Update to WS 6.x and fix all it broke

Significant update to the ws module to get it completely up to date.

The jump from 1.x to 6.x has required a rewrite of our WS handling. Most
specifically the means by which you can have multiple ws servers on a
single http server has completely changed; we now have to handle the
'upgrade' event on the server ourselves.
This commit is contained in:
Nick O'Leary
2019-01-08 16:21:36 +00:00
parent 201d1926bc
commit ed31a0cf15
4 changed files with 79 additions and 55 deletions

View File

@@ -18,6 +18,23 @@ module.exports = function(RED) {
"use strict";
var ws = require("ws");
var inspect = require("util").inspect;
var url = require("url");
var serverUpgradeAdded = false;
function handleServerUpgrade(request, socket, head) {
const pathname = url.parse(request.url).pathname;
if (listenerNodes.hasOwnProperty(pathname)) {
listenerNodes[pathname].server.handleUpgrade(request, socket, head, function done(ws) {
listenerNodes[pathname].server.emit('connection', ws, request);
});
} else {
// Don't destroy the socket as other listeners may want to handle the
// event.
}
}
var listenerNodes = {};
var activeListenerNodes = 0;
// A node red node that sets up a local websocket server
function WebSocketListenerNode(n) {
@@ -53,13 +70,22 @@ module.exports = function(RED) {
function handleConnection(/*socket*/socket) {
var id = (1+Math.random()*4294967295).toString(16);
if (node.isServer) { node._clients[id] = socket; node.emit('opened',Object.keys(node._clients).length); }
if (node.isServer) {
node._clients[id] = socket;
node.emit('opened',Object.keys(node._clients).length);
}
socket.on('open',function() {
if (!node.isServer) { node.emit('opened',''); }
if (!node.isServer) {
node.emit('opened','');
}
});
socket.on('close',function() {
if (node.isServer) { delete node._clients[id]; node.emit('closed',Object.keys(node._clients).length); }
else { node.emit('closed'); }
if (node.isServer) {
delete node._clients[id];
node.emit('closed',Object.keys(node._clients).length);
} else {
node.emit('closed');
}
if (!node.closing && !node.isServer) {
clearTimeout(node.tout);
node.tout = setTimeout(function() { startconn(); }, 3000); // try to reconnect every 3 secs... bit fast ?
@@ -78,34 +104,29 @@ module.exports = function(RED) {
}
if (node.isServer) {
var path = RED.settings.httpNodeRoot || "/";
path = path + (path.slice(-1) == "/" ? "":"/") + (node.path.charAt(0) == "/" ? node.path.substring(1) : node.path);
// Workaround https://github.com/einaros/ws/pull/253
// Listen for 'newListener' events from RED.server
node._serverListeners = {};
var storeListener = function(/*String*/event,/*function*/listener) {
if (event == "error" || event == "upgrade" || event == "listening") {
node._serverListeners[event] = listener;
}
activeListenerNodes++;
if (!serverUpgradeAdded) {
RED.server.on('upgrade', handleServerUpgrade);
serverUpgradeAdded = true
}
RED.server.addListener('newListener',storeListener);
var path = RED.settings.httpNodeRoot || "/";
path = path + (path.slice(-1) == "/" ? "":"/") + (node.path.charAt(0) == "/" ? node.path.substring(1) : node.path);
node.fullPath = path;
if (listenerNodes.hasOwnProperty(path)) {
node.error(RED._("websocket.errors.duplicate-path",{path: node.path}));
return;
}
listenerNodes[node.fullPath] = node;
var serverOptions = {
server:RED.server,
path:path
noServer: true
}
if (RED.settings.webSocketNodeVerifyClient) {
serverOptions.verifyClient = RED.settings.webSocketNodeVerifyClient;
}
// Create a WebSocket Server
node.server = new ws.Server(serverOptions);
// Workaround https://github.com/einaros/ws/pull/253
// Stop listening for new listener events
RED.server.removeListener('newListener',storeListener);
node.server.setMaxListeners(0);
node.server.on('connection', handleConnection);
}
@@ -115,21 +136,17 @@ module.exports = function(RED) {
}
node.on("close", function() {
// Workaround https://github.com/einaros/ws/pull/253
// Remove listeners from RED.server
if (node.isServer) {
var listener = null;
for (var event in node._serverListeners) {
if (node._serverListeners.hasOwnProperty(event)) {
listener = node._serverListeners[event];
if (typeof listener === "function") {
RED.server.removeListener(event,listener);
}
}
}
node._serverListeners = {};
delete listenerNodes[node.fullPath];
node.server.close();
node._inputNodes = [];
activeListenerNodes--;
if (activeListenerNodes === 0 && serverUpgradeAdded) {
RED.server.removeListener('upgrade', handleServerUpgrade);
serverUpgradeAdded = false;
}
}
else {
node.closing = true;
@@ -177,11 +194,12 @@ module.exports = function(RED) {
}
WebSocketListenerNode.prototype.broadcast = function(data) {
var i;
try {
if (this.isServer) {
for (i = 0; i < this.server.clients.length; i++) {
this.server.clients[i].send(data);
for (let client in this._clients) {
if (this._clients.hasOwnProperty(client)) {
this._clients[client].send(data);
}
}
}
else {
@@ -215,8 +233,11 @@ module.exports = function(RED) {
this.serverConfig.on('opened', function(n) { node.status({fill:"green",shape:"dot",text:RED._("websocket.status.connected",{count:n})}); });
this.serverConfig.on('erro', function() { node.status({fill:"red",shape:"ring",text:"common.status.error"}); });
this.serverConfig.on('closed', function(n) {
if (n > 0) { node.status({fill:"green",shape:"dot",text:RED._("websocket.status.connected",{count:n})}); }
else { node.status({fill:"red",shape:"ring",text:"common.status.disconnected"}); }
if (n > 0) {
node.status({fill:"green",shape:"dot",text:RED._("websocket.status.connected",{count:n})});
} else {
node.status({fill:"red",shape:"ring",text:"common.status.disconnected"});
}
});
} else {
this.error(RED._("websocket.errors.missing-conf"));

View File

@@ -432,7 +432,8 @@
"errors": {
"connect-error": "An error occured on the ws connection: ",
"send-error": "An error occurred while sending: ",
"missing-conf": "Missing server configuration"
"missing-conf": "Missing server configuration",
"duplicate-path": "Cannot have two WebSocket listeners on the same path: __path__"
}
},
"watch": {