mirror of https://github.com/node-red/node-red.git
Merge pull request #5438 from Dennis-SEG/fix/harden-nodes-against-uncaught-exceptions
fix: prevent uncaught exceptions in core node event handlerspull/5467/head^2
Nick O'Leary
GitHub
commit
620da24e01
|
|
@ -105,18 +105,26 @@ module.exports = function(RED) {
|
|||
}
|
||||
node.activeProcesses[child.pid] = child;
|
||||
child.stdout.on('data', function (data) {
|
||||
if (node.activeProcesses.hasOwnProperty(child.pid) && node.activeProcesses[child.pid] !== null) {
|
||||
// console.log('[exec] stdout: ' + data,child.pid);
|
||||
if (isUtf8(data)) { msg.payload = data.toString(); }
|
||||
else { msg.payload = data; }
|
||||
nodeSend([RED.util.cloneMessage(msg),null,null]);
|
||||
try {
|
||||
if (node.activeProcesses.hasOwnProperty(child.pid) && node.activeProcesses[child.pid] !== null) {
|
||||
// console.log('[exec] stdout: ' + data,child.pid);
|
||||
if (isUtf8(data)) { msg.payload = data.toString(); }
|
||||
else { msg.payload = data; }
|
||||
nodeSend([RED.util.cloneMessage(msg),null,null]);
|
||||
}
|
||||
} catch (err) {
|
||||
node.error(err.toString());
|
||||
}
|
||||
});
|
||||
child.stderr.on('data', function (data) {
|
||||
if (node.activeProcesses.hasOwnProperty(child.pid) && node.activeProcesses[child.pid] !== null) {
|
||||
if (isUtf8(data)) { msg.payload = data.toString(); }
|
||||
else { msg.payload = data; }
|
||||
nodeSend([null,RED.util.cloneMessage(msg),null]);
|
||||
try {
|
||||
if (node.activeProcesses.hasOwnProperty(child.pid) && node.activeProcesses[child.pid] !== null) {
|
||||
if (isUtf8(data)) { msg.payload = data.toString(); }
|
||||
else { msg.payload = data; }
|
||||
nodeSend([null,RED.util.cloneMessage(msg),null]);
|
||||
}
|
||||
} catch (err) {
|
||||
node.error(err.toString());
|
||||
}
|
||||
});
|
||||
child.on('close', function (code,signal) {
|
||||
|
|
|
|||
|
|
@ -227,6 +227,7 @@ module.exports = function(RED) {
|
|||
* Handle the payload / packet recieved in MQTT In and MQTT Sub nodes
|
||||
*/
|
||||
function subscriptionHandler(node, datatype ,topic, payload, packet) {
|
||||
if (!packet) { packet = {}; }
|
||||
const msg = {topic:topic, payload:null, qos:packet.qos, retain:packet.retain};
|
||||
const v5 = (node && node.brokerConn)
|
||||
? node.brokerConn.v5()
|
||||
|
|
@ -1074,12 +1075,16 @@ module.exports = function(RED) {
|
|||
|
||||
if (!subscription.handler) {
|
||||
subscription.handler = function (mtopic, mpayload, mpacket) {
|
||||
const sops = subscription.options ? subscription.options.properties : {}
|
||||
const pops = mpacket.properties || {}
|
||||
if (subIdsAvailable && pops.subscriptionIdentifier && sops.subscriptionIdentifier && (pops.subscriptionIdentifier !== sops.subscriptionIdentifier)) {
|
||||
//do nothing as subscriptionIdentifier does not match
|
||||
} else if (matchTopic(topic, mtopic)) {
|
||||
subscription.callback && subscription.callback(mtopic, mpayload, mpacket)
|
||||
try {
|
||||
const sops = subscription.options ? subscription.options.properties : {}
|
||||
const pops = (mpacket && mpacket.properties) || {}
|
||||
if (subIdsAvailable && pops.subscriptionIdentifier && sops.subscriptionIdentifier && (pops.subscriptionIdentifier !== sops.subscriptionIdentifier)) {
|
||||
//do nothing as subscriptionIdentifier does not match
|
||||
} else if (matchTopic(topic, mtopic)) {
|
||||
subscription.callback && subscription.callback(mtopic, mpayload, mpacket)
|
||||
}
|
||||
} catch (err) {
|
||||
node.error("MQTT subscription handler error: " + err.toString());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -297,7 +297,11 @@ module.exports = function(RED) {
|
|||
}
|
||||
msg._session = {type:"websocket",id:id};
|
||||
for (var i = 0; i < this._inputNodes.length; i++) {
|
||||
this._inputNodes[i].send(msg);
|
||||
try {
|
||||
this._inputNodes[i].send(msg);
|
||||
} catch (err) {
|
||||
this.error(RED._("websocket.errors.send-error") + " " + err.toString());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -127,32 +127,36 @@ module.exports = function(RED) {
|
|||
connectionPool[id] = client;
|
||||
|
||||
client.on('data', function (data) {
|
||||
if (node.datatype != 'buffer') {
|
||||
data = data.toString(node.datatype);
|
||||
}
|
||||
if (node.stream) {
|
||||
var msg;
|
||||
if ((node.datatype) === "utf8" && node.newline !== "") {
|
||||
buffer = buffer+data;
|
||||
var parts = buffer.split(node.newline);
|
||||
for (var i = 0; i<parts.length-1; i+=1) {
|
||||
msg = {topic:node.topic, payload:parts[i]};
|
||||
if (node.trim == true) { msg.payload += node.newline; }
|
||||
try {
|
||||
if (node.datatype != 'buffer') {
|
||||
data = data.toString(node.datatype);
|
||||
}
|
||||
if (node.stream) {
|
||||
var msg;
|
||||
if ((node.datatype) === "utf8" && node.newline !== "") {
|
||||
buffer = buffer+data;
|
||||
var parts = buffer.split(node.newline);
|
||||
for (var i = 0; i<parts.length-1; i+=1) {
|
||||
msg = {topic:node.topic, payload:parts[i]};
|
||||
if (node.trim == true) { msg.payload += node.newline; }
|
||||
msg._session = {type:"tcp",id:id};
|
||||
node.send(msg);
|
||||
}
|
||||
buffer = parts[parts.length-1];
|
||||
} else {
|
||||
msg = {topic:node.topic, payload:data};
|
||||
msg._session = {type:"tcp",id:id};
|
||||
node.send(msg);
|
||||
}
|
||||
buffer = parts[parts.length-1];
|
||||
} else {
|
||||
msg = {topic:node.topic, payload:data};
|
||||
msg._session = {type:"tcp",id:id};
|
||||
node.send(msg);
|
||||
}
|
||||
} else {
|
||||
if ((typeof data) === "string") {
|
||||
buffer = buffer+data;
|
||||
} else {
|
||||
buffer = Buffer.concat([buffer,data],buffer.length+data.length);
|
||||
if ((typeof data) === "string") {
|
||||
buffer = buffer+data;
|
||||
} else {
|
||||
buffer = Buffer.concat([buffer,data],buffer.length+data.length);
|
||||
}
|
||||
}
|
||||
} catch (err) {
|
||||
node.error(RED._("tcpin.errors.error",{error:err.toString()}));
|
||||
}
|
||||
});
|
||||
client.on('end', function() {
|
||||
|
|
@ -222,35 +226,39 @@ module.exports = function(RED) {
|
|||
|
||||
var buffer = (node.datatype == 'buffer') ? Buffer.alloc(0) : "";
|
||||
socket.on('data', function (data) {
|
||||
if (node.datatype != 'buffer') {
|
||||
data = data.toString(node.datatype);
|
||||
}
|
||||
if (node.stream) {
|
||||
var msg;
|
||||
if ((typeof data) === "string" && node.newline !== "") {
|
||||
buffer = buffer+data;
|
||||
var parts = buffer.split(node.newline);
|
||||
for (var i = 0; i<parts.length-1; i+=1) {
|
||||
msg = {topic:node.topic, payload:parts[i], ip:socket.remoteAddress, port:socket.remotePort};
|
||||
if (node.trim == true) { msg.payload += node.newline; }
|
||||
try {
|
||||
if (node.datatype != 'buffer') {
|
||||
data = data.toString(node.datatype);
|
||||
}
|
||||
if (node.stream) {
|
||||
var msg;
|
||||
if ((typeof data) === "string" && node.newline !== "") {
|
||||
buffer = buffer+data;
|
||||
var parts = buffer.split(node.newline);
|
||||
for (var i = 0; i<parts.length-1; i+=1) {
|
||||
msg = {topic:node.topic, payload:parts[i], ip:socket.remoteAddress, port:socket.remotePort};
|
||||
if (node.trim == true) { msg.payload += node.newline; }
|
||||
msg._session = {type:"tcp",id:id};
|
||||
node.send(msg);
|
||||
}
|
||||
buffer = parts[parts.length-1];
|
||||
} else {
|
||||
msg = {topic:node.topic, payload:data, ip:socket.remoteAddress, port:socket.remotePort};
|
||||
msg._session = {type:"tcp",id:id};
|
||||
node.send(msg);
|
||||
}
|
||||
buffer = parts[parts.length-1];
|
||||
} else {
|
||||
msg = {topic:node.topic, payload:data, ip:socket.remoteAddress, port:socket.remotePort};
|
||||
msg._session = {type:"tcp",id:id};
|
||||
node.send(msg);
|
||||
}
|
||||
}
|
||||
else {
|
||||
if ((typeof data) === "string") {
|
||||
buffer = buffer+data;
|
||||
} else {
|
||||
buffer = Buffer.concat([buffer,data],buffer.length+data.length);
|
||||
else {
|
||||
if ((typeof data) === "string") {
|
||||
buffer = buffer+data;
|
||||
} else {
|
||||
buffer = Buffer.concat([buffer,data],buffer.length+data.length);
|
||||
}
|
||||
fromi = socket.remoteAddress;
|
||||
fromp = socket.remotePort;
|
||||
}
|
||||
fromi = socket.remoteAddress;
|
||||
fromp = socket.remotePort;
|
||||
} catch (err) {
|
||||
node.error(RED._("tcpin.errors.error",{error:err.toString()}));
|
||||
}
|
||||
});
|
||||
socket.on('end', function() {
|
||||
|
|
@ -678,117 +686,121 @@ module.exports = function(RED) {
|
|||
}
|
||||
var chunk = "";
|
||||
clients[connection_id].client.on('data', function(data) {
|
||||
if (node.out === "sit") { // if we are staying connected just send the buffer
|
||||
if (clients[connection_id]) {
|
||||
const msg = clients[connection_id].lastMsg || {};
|
||||
msg.payload = RED.util.cloneMessage(data);
|
||||
if (node.ret === "string") {
|
||||
try {
|
||||
if (node.newline && node.newline !== "" ) {
|
||||
chunk += msg.payload.toString();
|
||||
let parts = chunk.split(node.newline);
|
||||
for (var p=0; p<parts.length-1; p+=1) {
|
||||
let m = RED.util.cloneMessage(msg);
|
||||
m.payload = parts[p];
|
||||
if (node.trim == true) { m.payload += node.newline; }
|
||||
nodeSend(m);
|
||||
}
|
||||
chunk = parts[parts.length-1];
|
||||
}
|
||||
else {
|
||||
msg.payload = msg.payload.toString();
|
||||
nodeSend(msg);
|
||||
}
|
||||
}
|
||||
catch(e) { node.error(RED._("tcpin.errors.bad-string"), msg); }
|
||||
}
|
||||
else { nodeSend(msg); }
|
||||
}
|
||||
}
|
||||
// else if (node.splitc === 0) {
|
||||
// clients[connection_id].msg.payload = data;
|
||||
// node.send(clients[connection_id].msg);
|
||||
// }
|
||||
else {
|
||||
for (var j = 0; j < data.length; j++ ) {
|
||||
if (node.out === "time") {
|
||||
if (clients[connection_id]) {
|
||||
// do the timer thing
|
||||
if (clients[connection_id].timeout) {
|
||||
i += 1;
|
||||
buf[i] = data[j];
|
||||
}
|
||||
else {
|
||||
clients[connection_id].timeout = setTimeout(function () {
|
||||
if (clients[connection_id]) {
|
||||
clients[connection_id].timeout = null;
|
||||
const msg = clients[connection_id].lastMsg || {};
|
||||
msg.payload = Buffer.alloc(i+1);
|
||||
buf.copy(msg.payload,0,0,i+1);
|
||||
if (node.ret === "string") {
|
||||
try { msg.payload = msg.payload.toString(); }
|
||||
catch(e) { node.error("Failed to create string", msg); }
|
||||
}
|
||||
nodeSend(msg);
|
||||
if (clients[connection_id].client) {
|
||||
node.status({});
|
||||
clients[connection_id].client.destroy();
|
||||
delete clients[connection_id];
|
||||
}
|
||||
try {
|
||||
if (node.out === "sit") { // if we are staying connected just send the buffer
|
||||
if (clients[connection_id]) {
|
||||
const msg = clients[connection_id].lastMsg || {};
|
||||
msg.payload = RED.util.cloneMessage(data);
|
||||
if (node.ret === "string") {
|
||||
try {
|
||||
if (node.newline && node.newline !== "" ) {
|
||||
chunk += msg.payload.toString();
|
||||
let parts = chunk.split(node.newline);
|
||||
for (var p=0; p<parts.length-1; p+=1) {
|
||||
let m = RED.util.cloneMessage(msg);
|
||||
m.payload = parts[p];
|
||||
if (node.trim == true) { m.payload += node.newline; }
|
||||
nodeSend(m);
|
||||
}
|
||||
}, node.splitc);
|
||||
i = 0;
|
||||
buf[0] = data[j];
|
||||
chunk = parts[parts.length-1];
|
||||
}
|
||||
else {
|
||||
msg.payload = msg.payload.toString();
|
||||
nodeSend(msg);
|
||||
}
|
||||
}
|
||||
catch(e) { node.error(RED._("tcpin.errors.bad-string"), msg); }
|
||||
}
|
||||
else { nodeSend(msg); }
|
||||
}
|
||||
}
|
||||
// else if (node.splitc === 0) {
|
||||
// clients[connection_id].msg.payload = data;
|
||||
// node.send(clients[connection_id].msg);
|
||||
// }
|
||||
else {
|
||||
for (var j = 0; j < data.length; j++ ) {
|
||||
if (node.out === "time") {
|
||||
if (clients[connection_id]) {
|
||||
// do the timer thing
|
||||
if (clients[connection_id].timeout) {
|
||||
i += 1;
|
||||
buf[i] = data[j];
|
||||
}
|
||||
else {
|
||||
clients[connection_id].timeout = setTimeout(function () {
|
||||
if (clients[connection_id]) {
|
||||
clients[connection_id].timeout = null;
|
||||
const msg = clients[connection_id].lastMsg || {};
|
||||
msg.payload = Buffer.alloc(i+1);
|
||||
buf.copy(msg.payload,0,0,i+1);
|
||||
if (node.ret === "string") {
|
||||
try { msg.payload = msg.payload.toString(); }
|
||||
catch(e) { node.error("Failed to create string", msg); }
|
||||
}
|
||||
nodeSend(msg);
|
||||
if (clients[connection_id].client) {
|
||||
node.status({});
|
||||
clients[connection_id].client.destroy();
|
||||
delete clients[connection_id];
|
||||
}
|
||||
}
|
||||
}, node.splitc);
|
||||
i = 0;
|
||||
buf[0] = data[j];
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
// count bytes into a buffer...
|
||||
else if (node.out == "count") {
|
||||
buf[i] = data[j];
|
||||
i += 1;
|
||||
if ( i >= node.splitc) {
|
||||
if (clients[connection_id]) {
|
||||
const msg = clients[connection_id].lastMsg || {};
|
||||
msg.payload = Buffer.alloc(i);
|
||||
buf.copy(msg.payload,0,0,i);
|
||||
if (node.ret === "string") {
|
||||
try { msg.payload = msg.payload.toString(); }
|
||||
catch(e) { node.error("Failed to create string", msg); }
|
||||
// count bytes into a buffer...
|
||||
else if (node.out == "count") {
|
||||
buf[i] = data[j];
|
||||
i += 1;
|
||||
if ( i >= node.splitc) {
|
||||
if (clients[connection_id]) {
|
||||
const msg = clients[connection_id].lastMsg || {};
|
||||
msg.payload = Buffer.alloc(i);
|
||||
buf.copy(msg.payload,0,0,i);
|
||||
if (node.ret === "string") {
|
||||
try { msg.payload = msg.payload.toString(); }
|
||||
catch(e) { node.error("Failed to create string", msg); }
|
||||
}
|
||||
nodeSend(msg);
|
||||
if (clients[connection_id].client) {
|
||||
node.status({});
|
||||
clients[connection_id].client.destroy();
|
||||
delete clients[connection_id];
|
||||
}
|
||||
i = 0;
|
||||
}
|
||||
nodeSend(msg);
|
||||
if (clients[connection_id].client) {
|
||||
node.status({});
|
||||
clients[connection_id].client.destroy();
|
||||
delete clients[connection_id];
|
||||
}
|
||||
i = 0;
|
||||
}
|
||||
}
|
||||
}
|
||||
// look for a char
|
||||
else {
|
||||
buf[i] = data[j];
|
||||
i += 1;
|
||||
if (data[j] == node.splitc) {
|
||||
if (clients[connection_id]) {
|
||||
const msg = clients[connection_id].lastMsg || {};
|
||||
msg.payload = Buffer.alloc(i);
|
||||
buf.copy(msg.payload,0,0,i);
|
||||
if (node.ret === "string") {
|
||||
try { msg.payload = msg.payload.toString(); }
|
||||
catch(e) { node.error("Failed to create string", msg); }
|
||||
// look for a char
|
||||
else {
|
||||
buf[i] = data[j];
|
||||
i += 1;
|
||||
if (data[j] == node.splitc) {
|
||||
if (clients[connection_id]) {
|
||||
const msg = clients[connection_id].lastMsg || {};
|
||||
msg.payload = Buffer.alloc(i);
|
||||
buf.copy(msg.payload,0,0,i);
|
||||
if (node.ret === "string") {
|
||||
try { msg.payload = msg.payload.toString(); }
|
||||
catch(e) { node.error("Failed to create string", msg); }
|
||||
}
|
||||
nodeSend(msg);
|
||||
if (clients[connection_id].client) {
|
||||
node.status({});
|
||||
clients[connection_id].client.destroy();
|
||||
delete clients[connection_id];
|
||||
}
|
||||
i = 0;
|
||||
}
|
||||
nodeSend(msg);
|
||||
if (clients[connection_id].client) {
|
||||
node.status({});
|
||||
clients[connection_id].client.destroy();
|
||||
delete clients[connection_id];
|
||||
}
|
||||
i = 0;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch (err) {
|
||||
node.error(RED._("tcpin.errors.error",{error:err.toString()}));
|
||||
}
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -98,15 +98,19 @@ module.exports = function(RED) {
|
|||
});
|
||||
|
||||
server.on('message', function (message, remote) {
|
||||
var msg;
|
||||
if (node.datatype =="base64") {
|
||||
msg = { payload:message.toString('base64'), fromip:remote.address+':'+remote.port, ip:remote.address, port:remote.port };
|
||||
} else if (node.datatype =="utf8") {
|
||||
msg = { payload:message.toString('utf8'), fromip:remote.address+':'+remote.port, ip:remote.address, port:remote.port };
|
||||
} else {
|
||||
msg = { payload:message, fromip:remote.address+':'+remote.port, ip:remote.address, port:remote.port };
|
||||
try {
|
||||
var msg;
|
||||
if (node.datatype =="base64") {
|
||||
msg = { payload:message.toString('base64'), fromip:remote.address+':'+remote.port, ip:remote.address, port:remote.port };
|
||||
} else if (node.datatype =="utf8") {
|
||||
msg = { payload:message.toString('utf8'), fromip:remote.address+':'+remote.port, ip:remote.address, port:remote.port };
|
||||
} else {
|
||||
msg = { payload:message, fromip:remote.address+':'+remote.port, ip:remote.address, port:remote.port };
|
||||
}
|
||||
node.send(msg);
|
||||
} catch (err) {
|
||||
node.error(RED._("udp.errors.error",{error:err.toString()}));
|
||||
}
|
||||
node.send(msg);
|
||||
});
|
||||
|
||||
server.on('listening', function () {
|
||||
|
|
|
|||
Loading…
Reference in New Issue