frodovdr/node-red
1
0
Fork 0
mirror of https://github.com/node-red/node-red.git synced 2023-10-10 13:36:53 +02:00
Code Issues Releases Wiki Activity

correlate joined/split messages

Browse Source
This commit is contained in:
Kunihiko Toumura 2021-09-02 14:33:14 +09:00
parent 40f816c311
commit e994b9d5e1
1 changed files with 88 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

114
packages/node_modules/@node-red/nodes/core/sequence/17-split.js vendored
View File

@ -18,20 +18,26 @@ module.exports = function(RED) {
"use strict"; "use strict";
function sendArray(node,msg,array,send) { function sendArray(node,msg,array,send) {
const corrIds = [];
for (var i = 0; i < array.length-1; i++) { for (var i = 0; i < array.length-1; i++) {
msg.payload = array[i]; msg.payload = array[i];
msg.parts.index = node.c++; msg.parts.index = node.c++;
if (node.stream !== true) { msg.parts.count = array.length; } if (node.stream !== true) { msg.parts.count = array.length; }
msg._msgid = RED.util.generateId();
corrIds.push(msg._msgid);
send(RED.util.cloneMessage(msg)); send(RED.util.cloneMessage(msg));
} }
if (node.stream !== true) { if (node.stream !== true) {
msg.payload = array[i]; msg.payload = array[i];
msg.parts.index = node.c++; msg.parts.index = node.c++;
msg.parts.count = array.length; msg.parts.count = array.length;
msg._msgid = RED.util.generateId();
corrIds.push(msg._msgid);
send(RED.util.cloneMessage(msg)); send(RED.util.cloneMessage(msg));
node.c = 0; node.c = 0;
} }
else { node.remainder = array[i]; } else { node.remainder = array[i]; }
return corrIds;
} }
function SplitNode(n) { function SplitNode(n) {
@ -69,15 +75,28 @@ module.exports = function(RED) {
node.buffer = Buffer.from([]); node.buffer = Buffer.from([]);
node.pendingDones = []; node.pendingDones = [];
this.on("input", function(msg, send, done) { this.on("input", function(msg, send, done) {
function swapIdThenDone(msgInfo, sentMsgIds) {
const currentMsgid = msgInfo.msg._msgid;
msgInfo.msg._msgid = msgInfo.msgid;
msgInfo.done();
let c = [];
if (msgInfo.corrMsgids) { Array.prototype.push.apply(c, msgInfo.corrMsgids); }
if (sentMsgIds) { Array.prototype.push.apply(c, sentMsgIds); }
if (c.length > 0) { node.metric("correlate",msgInfo.msg,{split: c}); }
msgInfo.msg._msgid = currentMsgid;
}
if (msg.hasOwnProperty("payload")) { if (msg.hasOwnProperty("payload")) {
if (msg.hasOwnProperty("parts")) { msg.parts = { parts:msg.parts }; } // push existing parts to a stack if (msg.hasOwnProperty("parts")) { msg.parts = { parts:msg.parts }; } // push existing parts to a stack
else { msg.parts = {}; } else { msg.parts = {}; }
msg.parts.id = RED.util.generateId(); // generate a random id msg.parts.id = RED.util.generateId(); // generate a random id
const origMsgid = msg._msgid;
delete msg._msgid; delete msg._msgid;
if (typeof msg.payload === "string") { // Split String into array if (typeof msg.payload === "string") { // Split String into array
msg.payload = (node.remainder || "") + msg.payload; msg.payload = (node.remainder || "") + msg.payload;
msg.parts.type = "string"; msg.parts.type = "string";
if (node.spltType === "len") { if (node.spltType === "len") {
const corrIds = [];
msg.parts.ch = ""; msg.parts.ch = "";
msg.parts.len = node.splt; msg.parts.len = node.splt;
var count = msg.payload.length/node.splt; var count = msg.payload.length/node.splt;
@ -94,23 +113,28 @@ module.exports = function(RED) {
msg.payload = data.substring(pos,pos+node.splt); msg.payload = data.substring(pos,pos+node.splt);
msg.parts.index = node.c++; msg.parts.index = node.c++;
pos += node.splt; pos += node.splt;
msg._msgid = RED.util.generateId();
corrIds.push(msg._msgid);
send(RED.util.cloneMessage(msg)); send(RED.util.cloneMessage(msg));
} }
if (count > 1) { if (count > 1) {
node.pendingDones.forEach(d => d()); node.pendingDones.forEach(msgInfo => swapIdThenDone(msgInfo, [corrIds[0]]));
node.pendingDones = []; node.pendingDones = [];
} }
node.remainder = data.substring(pos); node.remainder = data.substring(pos);
if ((node.stream !== true) || (node.remainder.length === node.splt)) { if ((node.stream !== true) || (node.remainder.length === node.splt)) {
msg.payload = node.remainder; msg.payload = node.remainder;
msg.parts.index = node.c++; msg.parts.index = node.c++;
msg._msgid = RED.util.generateId();
corrIds.push(msg._msgid);
send(RED.util.cloneMessage(msg)); send(RED.util.cloneMessage(msg));
node.pendingDones.forEach(d => d()); node.pendingDones.push({done,msg,msgid:origMsgid});
node.pendingDones.forEach(msgInfo => swapIdThenDone(msgInfo, corrIds));
node.pendingDones = []; node.pendingDones = [];
done();
node.remainder = ""; node.remainder = "";
} else { } else {
node.pendingDones.push(done); node.pendingDones.push({done,msg,msgid:origMsgid,
corrMsgids: corrIds});
} }
} }
else { else {
@ -125,8 +149,22 @@ module.exports = function(RED) {
a = msg.payload.split(node.splt); a = msg.payload.split(node.splt);
msg.parts.ch = node.splt; // pass the split char to other end for rejoin msg.parts.ch = node.splt; // pass the split char to other end for rejoin
} }
sendArray(node,msg,a,send); const corrIds = sendArray(node,msg,a,send);
done(); if (node.stream) {
if (a.length > 1) {
node.pendingDones.forEach(msgInfo => swapIdThenDone(msgInfo, [corrIds[0]]));
node.pendingDones = [];
if (a[a.length-1] == '') {
swapIdThenDone({done,msg,msgid:origMsgid}, corrIds);
} else {
node.pendingDones.push({done,msg,msgid:origMsgid,corrMsgId:corrIds});
}
} else {
node.pendingDones.push({done,msg,msgid:origMsgid,corrMsgId:corrIds});
}
} else {
swapIdThenDone({done,msg,msgid:origMsgid}, corrIds);
}
} }
} }
else if (Array.isArray(msg.payload)) { // then split array into messages else if (Array.isArray(msg.payload)) { // then split array into messages
@ -139,6 +177,7 @@ module.exports = function(RED) {
var pos = 0; var pos = 0;
var data = msg.payload; var data = msg.payload;
msg.parts.len = node.arraySplt; msg.parts.len = node.arraySplt;
const corrIds = []
for (var i=0; i<count; i++) { for (var i=0; i<count; i++) {
msg.payload = data.slice(pos,pos+node.arraySplt); msg.payload = data.slice(pos,pos+node.arraySplt);
if (node.arraySplt === 1) { if (node.arraySplt === 1) {
@ -146,15 +185,18 @@ module.exports = function(RED) {
} }
msg.parts.index = i; msg.parts.index = i;
pos += node.arraySplt; pos += node.arraySplt;
msg._msgid = RED.util.generateId();
corrIds.push(msg._msgid);
send(RED.util.cloneMessage(msg)); send(RED.util.cloneMessage(msg));
} }
done(); swapIdThenDone({done,msg,msgid:origMsgid}, corrIds);
} }
else if ((typeof msg.payload === "object") && !Buffer.isBuffer(msg.payload)) { else if ((typeof msg.payload === "object") && !Buffer.isBuffer(msg.payload)) {
var j = 0; var j = 0;
var l = Object.keys(msg.payload).length; var l = Object.keys(msg.payload).length;
var pay = msg.payload; var pay = msg.payload;
msg.parts.type = "object"; msg.parts.type = "object";
const corrIds = [];
for (var p in pay) { for (var p in pay) {
if (pay.hasOwnProperty(p)) { if (pay.hasOwnProperty(p)) {
msg.payload = pay[p]; msg.payload = pay[p];
@ -164,17 +206,20 @@ module.exports = function(RED) {
msg.parts.key = p; msg.parts.key = p;
msg.parts.index = j; msg.parts.index = j;
msg.parts.count = l; msg.parts.count = l;
msg._msgid = RED.util.generateId();
corrIds.push(msg._msgid);
send(RED.util.cloneMessage(msg)); send(RED.util.cloneMessage(msg));
j += 1; j += 1;
} }
} }
done(); swapIdThenDone({done,msg,msgid:origMsgid}, corrIds);
} }
else if (Buffer.isBuffer(msg.payload)) { else if (Buffer.isBuffer(msg.payload)) {
var len = node.buffer.length + msg.payload.length; var len = node.buffer.length + msg.payload.length;
var buff = Buffer.concat([node.buffer, msg.payload], len); var buff = Buffer.concat([node.buffer, msg.payload], len);
msg.parts.type = "buffer"; msg.parts.type = "buffer";
if (node.spltType === "len") { if (node.spltType === "len") {
const corrIds = [];
var count = buff.length/node.splt; var count = buff.length/node.splt;
if (Math.floor(count) !== count) { if (Math.floor(count) !== count) {
count = Math.ceil(count); count = Math.ceil(count);
@ -189,26 +234,31 @@ module.exports = function(RED) {
msg.payload = buff.slice(pos,pos+node.splt); msg.payload = buff.slice(pos,pos+node.splt);
msg.parts.index = node.c++; msg.parts.index = node.c++;
pos += node.splt; pos += node.splt;
msg._msgid = RED.util.generateId();
corrIds.push(msg._msgid);
send(RED.util.cloneMessage(msg)); send(RED.util.cloneMessage(msg));
} }
if (count > 1) { if (count > 1) {
node.pendingDones.forEach(d => d()); node.pendingDones.forEach(msgInfo => swapIdThenDone(msgInfo, [corrIds[0]]));
node.pendingDones = []; node.pendingDones = [];
} }
node.buffer = buff.slice(pos); node.buffer = buff.slice(pos);
if ((node.stream !== true) || (node.buffer.length === node.splt)) { if ((node.stream !== true) || (node.buffer.length === node.splt)) {
msg.payload = node.buffer; msg.payload = node.buffer;
msg.parts.index = node.c++; msg.parts.index = node.c++;
msg._msgid = RED.util.generateId();
corrIds.push(msg._msgid);
send(RED.util.cloneMessage(msg)); send(RED.util.cloneMessage(msg));
node.pendingDones.forEach(d => d()); node.pendingDones.push({done,msg,msgid:origMsgid});
node.pendingDones.forEach(msgInfo => swapIdThenDone(msgInfo, corrIds));
node.pendingDones = []; node.pendingDones = [];
done();
node.buffer = Buffer.from([]); node.buffer = Buffer.from([]);
} else { } else {
node.pendingDones.push(done); node.pendingDones.push({done,msg,msgid:origMsgid,corrMsgids:corrIds});
} }
} }
else { else {
const corrIds = [];
var count = 0; var count = 0;
if (node.spltType === "bin") { if (node.spltType === "bin") {
msg.parts.ch = node.spltBuffer; msg.parts.ch = node.spltBuffer;
@ -232,33 +282,37 @@ module.exports = function(RED) {
while (pos > -1) { while (pos > -1) {
msg.payload = buff.slice(p,pos); msg.payload = buff.slice(p,pos);
msg.parts.index = node.c++; msg.parts.index = node.c++;
msg._msgid = RED.util.generateId();
corrIds.push(msg._msgid);
send(RED.util.cloneMessage(msg)); send(RED.util.cloneMessage(msg));
i++; i++;
p = pos+node.splt.length; p = pos+node.splt.length;
pos = buff.indexOf(node.splt,p); pos = buff.indexOf(node.splt,p);
} }
if (count > 1) { if (count > 1) {
node.pendingDones.forEach(d => d()); node.pendingDones.forEach(msgInfo => swapIdThenDone(msgInfo, [corrIds[0]]));
node.pendingDones = []; node.pendingDones = [];
} }
if ((node.stream !== true) && (p < buff.length)) { if ((node.stream !== true) && (p < buff.length)) {
msg.payload = buff.slice(p,buff.length); msg.payload = buff.slice(p,buff.length);
msg.parts.index = node.c++; msg.parts.index = node.c++;
msg.parts.count = node.c++; msg.parts.count = node.c++;
msg._msgid = RED.util.generateId();
corrIds.push(msg._msgid);
send(RED.util.cloneMessage(msg)); send(RED.util.cloneMessage(msg));
node.pendingDones.forEach(d => d()); node.pendingDones.forEach(msgInfo => swapIdThenDone(msgInfo, corrIds));
node.pendingDones = []; node.pendingDones = [];
} }
else { else {
node.buffer = buff.slice(p,buff.length); node.buffer = buff.slice(p,buff.length);
node.pendingDones.push(done); node.pendingDones.push({done,msg,msgid:origMsgid,corrMsgids:corrIds});
} }
if (node.buffer.length == 0) { if (node.buffer.length == 0) {
done(); swapIdThenDone({done,msg,msgid:origMsgid},corrIds);
} }
} }
} else { // otherwise drop the message. } else { // otherwise drop the message.
done(); swapIdThenDone({done,msg,msgid:origMsgid});
} }
} }
}); });
@ -304,7 +358,7 @@ module.exports = function(RED) {
exp.assign("A", accumulator); exp.assign("A", accumulator);
RED.util.evaluateJSONataExpression(exp, msgInfo.msg, (err,result) => { RED.util.evaluateJSONataExpression(exp, msgInfo.msg, (err,result) => {
if (err) { if (err) {
return done(err); return done(err, null);
} }
if (msgInfos.length === 0) { if (msgInfos.length === 0) {
if (fixup) { if (fixup) {
@ -312,14 +366,16 @@ module.exports = function(RED) {
fixup.assign("A", result); fixup.assign("A", result);
RED.util.evaluateJSONataExpression(fixup, {}, (err, result) => { RED.util.evaluateJSONataExpression(fixup, {}, (err, result) => {
if (err) { if (err) {
return done(err); return done(err, null);
} }
msgInfo.send({payload: result}); const _msgid = RED.util.generateId()
done(); msgInfo.send({payload: result, _msgid});
done(null, _msgid);
}); });
} else { } else {
msgInfo.send({payload: result}); const _msgid = RED.util.generateId()
done(); msgInfo.send({payload: result, _msgid});
done(null, _msgid);
} }
} else { } else {
reduceMessageGroup(node,msgInfos,exp,fixup,count,result,done); reduceMessageGroup(node,msgInfos,exp,fixup,count,result,done);
@ -350,7 +406,9 @@ module.exports = function(RED) {
done(err); done(err);
return; return;
} else { } else {
preservedMsgInfos.forEach(mInfo => mInfo.done()); const corrIds = [];
preservedMsgInfos.forEach(mInfo => { corrIds.push(mInfo.msg._msgid); mInfo.done()});
if (corrIds.length > 0) { node.metric("correlate", {_msgid:result}, {join: corrIds}); }
done(); done();
} }
}) })
@ -529,6 +587,7 @@ module.exports = function(RED) {
group.send(RED.util.cloneMessage(group.msg)); group.send(RED.util.cloneMessage(group.msg));
group.dones.forEach(f => f()); group.dones.forEach(f => f());
group.dones = []; group.dones = [];
node.metric("correlate",group.msg,{join: group.corrIds});
} }
var pendingMessages = []; var pendingMessages = [];
@ -675,7 +734,8 @@ module.exports = function(RED) {
type:"object", type:"object",
msg:RED.util.cloneMessage(msg), msg:RED.util.cloneMessage(msg),
send: send, send: send,
dones: [] dones: [],
corrIds: []
}; };
} }
else { else {
@ -686,7 +746,8 @@ module.exports = function(RED) {
type:payloadType, type:payloadType,
msg:RED.util.cloneMessage(msg), msg:RED.util.cloneMessage(msg),
send: send, send: send,
dones: [] dones: [],
corrIds: []
}; };
if (payloadType === 'string') { if (payloadType === 'string') {
inflight[partId].joinChar = joinChar; inflight[partId].joinChar = joinChar;
@ -704,6 +765,7 @@ module.exports = function(RED) {
} }
} }
inflight[partId].dones.push(done); inflight[partId].dones.push(done);
inflight[partId].corrIds.push(msg._msgid);
var group = inflight[partId]; var group = inflight[partId];
if (payloadType === 'buffer') { if (payloadType === 'buffer') {