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

Merge remote-tracking branch 'upstream/0.19' into json-schema

Browse Source
This commit is contained in:
Nathanaël Lécaudé 2018-07-10 11:11:15 -04:00
commit eea85485e6
31 changed files with 1607 additions and 766 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
Gruntfile.js
View File

@ -24,6 +24,10 @@ module.exports = function(grunt) {
nodemonArgs.push(flowFile);
}
var nonHeadless = grunt.option('non-headless');
if (nonHeadless) {
process.env.NODE_RED_NON_HEADLESS = 'true';
}
grunt.initConfig({
pkg: grunt.file.readJSON('package.json'),
paths: {

7
editor/js/ui/common/popover.js
View File

@ -48,12 +48,15 @@ RED.popover = (function() {
var openPopup = function(instant) {
if (active) {
div = $('<div class="red-ui-popover red-ui-popover-'+direction+'"></div>').appendTo("body");
div = $('<div class="red-ui-popover red-ui-popover-'+direction+'"></div>');
if (size !== "default") {
div.addClass("red-ui-popover-size-"+size);
}
if (typeof content === 'function') {
var result = content.call(res);
if (result === null) {
return;
}
if (typeof result === 'string') {
div.text(result);
} else {
@ -65,7 +68,7 @@ RED.popover = (function() {
if (width !== "auto") {
div.width(width);
}
div.appendTo("body");
var targetPos = target.offset();
var targetWidth = target.outerWidth();

70
editor/js/ui/common/typedInput.js
View File

@ -14,10 +14,38 @@
* limitations under the License.
**/
(function($) {
var contextParse = function(v) {
var parts = RED.utils.parseContextKey(v);
return {
option: parts.store,
value: parts.key
}
}
var contextExport = function(v,opt) {
if (!opt) {
return v;
}
var store = ((typeof opt === "string")?opt:opt.value)
if (store !== RED.settings.context.default) {
return "#:("+store+")::"+v;
} else {
return v;
}
}
var allOptions = {
msg: {value:"msg",label:"msg.",validate:RED.utils.validatePropertyExpression},
flow: {value:"flow",label:"flow.",validate:RED.utils.validatePropertyExpression},
global: {value:"global",label:"global.",validate:RED.utils.validatePropertyExpression},
flow: {value:"flow",label:"flow.",hasValue:true,
options:[],
validate:RED.utils.validatePropertyExpression,
parse: contextParse,
export: contextExport
},
global: {value:"global",label:"global.",hasValue:true,
options:[],
validate:RED.utils.validatePropertyExpression,
parse: contextParse,
export: contextExport
},
str: {value:"str",label:"string",icon:"red/images/typedInput/az.png"},
num: {value:"num",label:"number",icon:"red/images/typedInput/09.png",validate:/^[+-]?[0-9]*\.?[0-9]*([eE][-+]?[0-9]+)?$/},
bool: {value:"bool",label:"boolean",icon:"red/images/typedInput/bool.png",options:["true","false"]},
@ -87,12 +115,24 @@
$.widget( "nodered.typedInput", {
_create: function() {
try {
if (!nlsd && RED && RED._) {
for (var i in allOptions) {
if (allOptions.hasOwnProperty(i)) {
allOptions[i].label = RED._("typedInput.type."+i,{defaultValue:allOptions[i].label});
}
}
var contextStores = RED.settings.context.stores;
var contextOptions = contextStores.map(function(store) {
return {value:store,label: store, icon:'<i class="red-ui-typedInput-icon fa fa-database" style="color: #'+(store==='memory'?'ddd':'777')+'"></i>'}
})
if (contextOptions.length < 2) {
allOptions.flow.options = [];
allOptions.global.options = [];
} else {
allOptions.flow.options = contextOptions;
allOptions.global.options = contextOptions;
}
}
nlsd = true;
var that = this;
@ -170,6 +210,9 @@
// explicitly set optionSelectTrigger display to inline-block otherwise jQ sets it to 'inline'
this.optionSelectTrigger = $('<button tabindex="0" class="red-ui-typedInput-option-trigger" style="display:inline-block"><span class="red-ui-typedInput-option-caret"><i class="red-ui-typedInput-icon fa fa-sort-desc"></i></span></button>').appendTo(this.uiSelect);
this.optionSelectLabel = $('<span class="red-ui-typedInput-option-label"></span>').prependTo(this.optionSelectTrigger);
RED.popover.tooltip(this.optionSelectLabel,function() {
return that.optionValue;
});
this.optionSelectTrigger.click(function(event) {
event.preventDefault();
that._showOptionSelectMenu();
@ -186,6 +229,9 @@
this.optionExpandButton = $('<button tabindex="0" class="red-ui-typedInput-option-expand" style="display:inline-block"><i class="red-ui-typedInput-icon fa fa-ellipsis-h"></i></button>').appendTo(this.uiSelect);
this.type(this.options.default||this.typeList[0].value);
}catch(err) {
console.log(err.stack);
}
},
_showTypeMenu: function() {
if (this.typeList.length > 1) {
@ -467,7 +513,9 @@
var opt = this.typeMap[type];
if (opt && this.propertyType !== type) {
this.propertyType = type;
if (this.typeField) {
this.typeField.val(type);
}
this.selectLabel.empty();
var image;
if (opt.icon) {
@ -539,19 +587,31 @@
var parts = opt.parse(this.input.val());
if (parts.option) {
selectedOption = parts.option;
if (!this.activeOptions.hasOwnProperty(selectedOption)) {
parts.option = Object.keys(this.activeOptions)[0];
selectedOption = parts.option
}
}
this.input.val(parts.value);
if (opt.export) {
this.element.val(opt.export(parts.value,parts.option||selectedOption));
}
}
if (typeof selectedOption === "string") {
this.optionValue = selectedOption;
this._updateOptionSelectLabel(this.activeOptions[selectedOption]);
if (!this.activeOptions.hasOwnProperty(selectedOption)) {
selectedOption = Object.keys(this.activeOptions)[0];
}
if (!selectedOption) {
this.optionSelectTrigger.hide();
} else {
this._updateOptionSelectLabel(this.activeOptions[selectedOption]);
}
} else if (selectedOption) {
this.optionValue = selectedOption.value;
this._updateOptionSelectLabel(selectedOption);
} else {
this.optionSelectTrigger.hide();
}
}
}
@ -574,6 +634,7 @@
}
this.elementDiv.show();
}
if (this.optionExpandButton) {
if (opt.expand && typeof opt.expand === 'function') {
this.optionExpandButton.show();
this.optionExpandButton.off('click');
@ -584,6 +645,7 @@
} else {
this.optionExpandButton.hide();
}
}
this.input.trigger('change',this.propertyType,this.value());
}
if (image) {

17
editor/js/ui/utils.js
View File

@ -828,6 +828,20 @@ RED.utils = (function() {
return payload;
}
function parseContextKey(key) {
var parts = {};
var m = /^#:\((\S+?)\)::(.*)$/.exec(key);
if (m) {
parts.store = m[1];
parts.key = m[2];
} else {
parts.key = key;
if (RED.settings.context) {
parts.store = RED.settings.context.default;
}
}
return parts;
}
return {
createObjectElement: buildMessageElement,
@ -839,6 +853,7 @@ RED.utils = (function() {
getNodeIcon: getNodeIcon,
getNodeLabel: getNodeLabel,
addSpinnerOverlay: addSpinnerOverlay,
decodeObject: decodeObject
decodeObject: decodeObject,
parseContextKey: parseContextKey
}
})();

2
editor/js/ui/view.js
View File

@ -491,7 +491,7 @@ RED.view = (function() {
var midX = Math.floor(destX-dx/2);
var midY = Math.floor(destY-dy/2);
//
if (dy == 0) {
if (dy === 0) {
midY = destY + node_height;
}
var cp_height = node_height/2;

11
nodes/core/core/20-inject.html
View File

@ -237,10 +237,13 @@ If you want every 20 minutes from now - use the <i>"interval"</i> option.</p>
} else {
return this._("inject.timestamp")+suffix;
}
} else if (this.payloadType === 'flow' && this.payload.length < 19) {
return 'flow.'+this.payload+suffix;
} else if (this.payloadType === 'global' && this.payload.length < 17) {
return 'global.'+this.payload+suffix;
} else if (this.payloadType === 'flow' || this.payloadType === 'global') {
var key = this.payload;
var m = /^#:\((\S+?)\)::(.*)$/.exec(key);
if (m) {
key = m[2];
}
return 'flow.'+key+suffix;
} else {
return this._("inject.inject")+suffix;
}

14
nodes/core/core/20-inject.js
View File

@ -63,8 +63,9 @@ module.exports = function(RED) {
}
this.on("input",function(msg) {
try {
msg.topic = this.topic;
if (this.payloadType !== 'flow' && this.payloadType !== 'global') {
try {
if ( (this.payloadType == null && this.payload === "") || this.payloadType === "date") {
msg.payload = Date.now();
} else if (this.payloadType == null) {
@ -79,6 +80,17 @@ module.exports = function(RED) {
} catch(err) {
this.error(err,msg);
}
} else {
RED.util.evaluateNodeProperty(this.payload,this.payloadType,this,msg, function(err,res) {
if (err) {
node.error(err,msg);
} else {
msg.payload = res;
node.send(msg);
}
});
}
});
}

111
nodes/core/core/89-trigger.js
View File

@ -76,8 +76,43 @@ module.exports = function(RED) {
var node = this;
node.topics = {};
this.on("input", function(msg) {
var pendingMessages = [];
var activeMessagePromise = null;
var processMessageQueue = function(msg) {
if (msg) {
// A new message has arrived - add it to the message queue
pendingMessages.push(msg);
if (activeMessagePromise !== null) {
// The node is currently processing a message, so do nothing
// more with this message
return;
}
}
if (pendingMessages.length === 0) {
// There are no more messages to process, clear the active flag
// and return
activeMessagePromise = null;
return;
}
// There are more messages to process. Get the next message and
// start processing it. Recurse back in to check for any more
var nextMsg = pendingMessages.shift();
activeMessagePromise = processMessage(nextMsg)
.then(processMessageQueue)
.catch((err) => {
node.error(err,nextMsg);
return processMessageQueue();
});
}
this.on('input', function(msg) {
processMessageQueue(msg);
});
var processMessage = function(msg) {
var topic = msg.topic || "_none";
var promise;
if (node.bytopic === "all") { topic = "_none"; }
node.topics[topic] = node.topics[topic] || {};
if (msg.hasOwnProperty("reset") || ((node.reset !== '') && msg.hasOwnProperty("payload") && (msg.payload !== null) && msg.payload.toString && (msg.payload.toString() == node.reset)) ) {
@ -88,18 +123,39 @@ module.exports = function(RED) {
}
else {
if (((!node.topics[topic].tout) && (node.topics[topic].tout !== 0)) || (node.loop === true)) {
promise = Promise.resolve();
if (node.op2type === "pay" || node.op2type === "payl") { node.topics[topic].m2 = RED.util.cloneMessage(msg.payload); }
else if (node.op2Templated) { node.topics[topic].m2 = mustache.render(node.op2,msg); }
else if (node.op2type !== "nul") {
node.topics[topic].m2 = RED.util.evaluateNodeProperty(node.op2,node.op2type,node,msg);
promise = new Promise((resolve,reject) => {
RED.util.evaluateNodeProperty(node.op2,node.op2type,node,msg,(err,value) => {
if (err) {
reject(err);
} else {
node.topics[topic].m2 = value;
resolve();
}
});
});
}
return promise.then(() => {
promise = Promise.resolve();
if (node.op1type === "pay") { }
else if (node.op1Templated) { msg.payload = mustache.render(node.op1,msg); }
else if (node.op1type !== "nul") {
msg.payload = RED.util.evaluateNodeProperty(node.op1,node.op1type,node,msg);
promise = new Promise((resolve,reject) => {
RED.util.evaluateNodeProperty(node.op1,node.op1type,node,msg,(err,value) => {
if (err) {
reject(err);
} else {
msg.payload = value;
resolve();
}
});
});
}
return promise.then(() => {
if (node.duration === 0) { node.topics[topic].tout = 0; }
else if (node.loop === true) {
/* istanbul ignore else */
@ -115,21 +171,40 @@ module.exports = function(RED) {
node.topics[topic].tout = setTimeout(function() {
var msg2 = null;
if (node.op2type !== "nul") {
var promise = Promise.resolve();
msg2 = RED.util.cloneMessage(msg);
if (node.op2type === "flow" || node.op2type === "global") {
node.topics[topic].m2 = RED.util.evaluateNodeProperty(node.op2,node.op2type,node,msg);
promise = new Promise((resolve,reject) => {
RED.util.evaluateNodeProperty(node.op2,node.op2type,node,msg,(err,value) => {
if (err) {
reject(err);
} else {
node.topics[topic].m2 = value;
resolve();
}
});
});
}
promise.then(() => {
msg2.payload = node.topics[topic].m2;
delete node.topics[topic];
node.send(msg2);
}
else { delete node.topics[topic]; }
node.status({});
}).catch(err => {
node.error(err);
});
} else {
delete node.topics[topic];
node.status({});
}
}, node.duration);
}
}
node.status({fill:"blue",shape:"dot",text:" "});
if (node.op1type !== "nul") { node.send(RED.util.cloneMessage(msg)); }
});
});
}
else if ((node.extend === "true" || node.extend === true) && (node.duration > 0)) {
/* istanbul ignore else */
@ -138,10 +213,24 @@ module.exports = function(RED) {
if (node.topics[topic].tout) { clearTimeout(node.topics[topic].tout); }
node.topics[topic].tout = setTimeout(function() {
var msg2 = null;
var promise = Promise.resolve();
if (node.op2type !== "nul") {
if (node.op2type === "flow" || node.op2type === "global") {
node.topics[topic].m2 = RED.util.evaluateNodeProperty(node.op2,node.op2type,node,msg);
promise = new Promise((resolve,reject) => {
RED.util.evaluateNodeProperty(node.op2,node.op2type,node,msg,(err,value) => {
if (err) {
reject(err);
} else {
node.topics[topic].m2 = value;
resolve();
}
});
});
}
}
promise.then(() => {
if (node.op2type !== "nul") {
if (node.topics[topic] !== undefined) {
msg2 = RED.util.cloneMessage(msg);
msg2.payload = node.topics[topic].m2;
@ -150,13 +239,17 @@ module.exports = function(RED) {
delete node.topics[topic];
node.status({});
node.send(msg2);
}).catch(err => {
node.error(err);
});
}, node.duration);
}
else {
if (node.op2type === "payl") { node.topics[topic].m2 = RED.util.cloneMessage(msg.payload); }
}
}
});
return Promise.resolve();
}
this.on("close", function() {
for (var t in node.topics) {
/* istanbul ignore else */

40
nodes/core/hardware/nrgpio.py
View File

@ -23,10 +23,6 @@ from time import sleep
bounce = 25;
if sys.version_info >= (3,0):
print("Sorry - currently only configured to work with python 2.x")
sys.exit(1)
if len(sys.argv) > 2:
cmd = sys.argv[1].lower()
pin = int(sys.argv[2])
@ -34,7 +30,7 @@ if len(sys.argv) > 2:
GPIO.setwarnings(False)
if cmd == "pwm":
#print "Initialised pin "+str(pin)+" to PWM"
#print("Initialised pin "+str(pin)+" to PWM")
try:
freq = int(sys.argv[3])
except:
@ -54,10 +50,10 @@ if len(sys.argv) > 2:
GPIO.cleanup(pin)
sys.exit(0)
except Exception as ex:
print "bad data: "+data
print("bad data: "+data)
elif cmd == "buzz":
#print "Initialised pin "+str(pin)+" to Buzz"
#print("Initialised pin "+str(pin)+" to Buzz")
GPIO.setup(pin,GPIO.OUT)
p = GPIO.PWM(pin, 100)
p.stop()
@ -76,10 +72,10 @@ if len(sys.argv) > 2:
GPIO.cleanup(pin)
sys.exit(0)
except Exception as ex:
print "bad data: "+data
print("bad data: "+data)
elif cmd == "out":
#print "Initialised pin "+str(pin)+" to OUT"
#print("Initialised pin "+str(pin)+" to OUT")
GPIO.setup(pin,GPIO.OUT)
if len(sys.argv) == 4:
GPIO.output(pin,int(sys.argv[3]))
@ -103,11 +99,11 @@ if len(sys.argv) > 2:
GPIO.output(pin,data)
elif cmd == "in":
#print "Initialised pin "+str(pin)+" to IN"
#print("Initialised pin "+str(pin)+" to IN")
bounce = float(sys.argv[4])
def handle_callback(chan):
sleep(bounce/1000.0)
print GPIO.input(chan)
print(GPIO.input(chan))
if sys.argv[3].lower() == "up":
GPIO.setup(pin,GPIO.IN,GPIO.PUD_UP)
@ -116,7 +112,7 @@ if len(sys.argv) > 2:
else:
GPIO.setup(pin,GPIO.IN)
print GPIO.input(pin)
print(GPIO.input(pin))
GPIO.add_event_detect(pin, GPIO.BOTH, callback=handle_callback, bouncetime=int(bounce))
while True:
@ -129,7 +125,7 @@ if len(sys.argv) > 2:
sys.exit(0)
elif cmd == "byte":
#print "Initialised BYTE mode - "+str(pin)+
#print("Initialised BYTE mode - "+str(pin)+)
list = [7,11,13,12,15,16,18,22]
GPIO.setup(list,GPIO.OUT)
@ -152,7 +148,7 @@ if len(sys.argv) > 2:
GPIO.output(list[bit], data & mask)
elif cmd == "borg":
#print "Initialised BORG mode - "+str(pin)+
#print("Initialised BORG mode - "+str(pin)+)
GPIO.setup(11,GPIO.OUT)
GPIO.setup(13,GPIO.OUT)
GPIO.setup(15,GPIO.OUT)
@ -190,7 +186,7 @@ if len(sys.argv) > 2:
button = ord( buf[0] ) & pin # mask out just the required button(s)
if button != oldbutt: # only send if changed
oldbutt = button
print button
print(button)
while True:
try:
@ -215,7 +211,7 @@ if len(sys.argv) > 2:
# type,code,value
print("%u,%u" % (code, value))
event = file.read(EVENT_SIZE)
print "0,0"
print("0,0")
file.close()
sys.exit(0)
except:
@ -225,14 +221,14 @@ if len(sys.argv) > 2:
elif len(sys.argv) > 1:
cmd = sys.argv[1].lower()
if cmd == "rev":
print GPIO.RPI_REVISION
print(GPIO.RPI_REVISION)
elif cmd == "ver":
print GPIO.VERSION
print(GPIO.VERSION)
elif cmd == "info":
print GPIO.RPI_INFO
print(GPIO.RPI_INFO)
else:
print "Bad parameters - in|out|pwm|buzz|byte|borg|mouse|kbd|ver|info {pin} {value|up|down}"
print " only ver (gpio version) and info (board information) accept no pin parameter."
print("Bad parameters - in|out|pwm|buzz|byte|borg|mouse|kbd|ver|info {pin} {value|up|down}")
print(" only ver (gpio version) and info (board information) accept no pin parameter.")
else:
print "Bad parameters - in|out|pwm|buzz|byte|borg|mouse|kbd|ver|info {pin} {value|up|down}"
print("Bad parameters - in|out|pwm|buzz|byte|borg|mouse|kbd|ver|info {pin} {value|up|down}")

8
nodes/core/logic/10-switch.html
View File

@ -99,11 +99,17 @@
}
return v;
}
function prop2name(key) {
var result = RED.utils.parseContextKey(key);
return result.key;
}
function getValueLabel(t,v) {
if (t === 'str') {
return '"'+clipValueLength(v)+'"';
} else if (t === 'msg' || t==='flow' || t==='global') {
} else if (t === 'msg') {
return t+"."+clipValueLength(v);
} else if (t === 'flow' || t === 'global') {
return t+"."+clipValueLength(prop2name(v));
}
return clipValueLength(v);
}

376
nodes/core/logic/10-switch.js
View File

@ -59,21 +59,157 @@ module.exports = function(RED) {
'else': function(a) { return a === true; }
};
var _max_kept_msgs_count = undefined;
var _maxKeptCount;
function max_kept_msgs_count(node) {
if (_max_kept_msgs_count === undefined) {
function getMaxKeptCount() {
if (_maxKeptCount === undefined) {
var name = "nodeMessageBufferMaxLength";
if (RED.settings.hasOwnProperty(name)) {
_max_kept_msgs_count = RED.settings[name];
_maxKeptCount = RED.settings[name];
}
else {
_max_kept_msgs_count = 0;
_maxKeptCount = 0;
}
}
return _max_kept_msgs_count;
return _maxKeptCount;
}
function getProperty(node,msg) {
return new Promise((resolve,reject) => {
if (node.propertyType === 'jsonata') {
RED.util.evaluateJSONataExpression(node.property,msg,(err,value) => {
if (err) {
reject(RED._("switch.errors.invalid-expr",{error:err.message}));
} else {
resolve(value);
}
});
} else {
RED.util.evaluateNodeProperty(node.property,node.propertyType,node,msg,(err,value) => {
if (err) {
resolve(undefined);
} else {
resolve(value);
}
});
}
});
}
function getV1(node,msg,rule,hasParts) {
return new Promise( (resolve,reject) => {
if (rule.vt === 'prev') {
resolve(node.previousValue);
} else if (rule.vt === 'jsonata') {
var exp = rule.v;
if (rule.t === 'jsonata_exp') {
if (hasParts) {
exp.assign("I", msg.parts.index);
exp.assign("N", msg.parts.count);
}
}
RED.util.evaluateJSONataExpression(exp,msg,(err,value) => {
if (err) {
reject(RED._("switch.errors.invalid-expr",{error:err.message}));
} else {
resolve(value);
}
});
} else if (rule.vt === 'json') {
resolve("json");
} else if (rule.vt === 'null') {
resolve("null");
} else {
RED.util.evaluateNodeProperty(rule.v,rule.vt,node,msg, function(err,value) {
if (err) {
resolve(undefined);
} else {
resolve(value);
}
});
}
});
}
function getV2(node,msg,rule) {
return new Promise((resolve,reject) => {
var v2 = rule.v2;
if (rule.v2t === 'prev') {
resolve(node.previousValue);
} else if (rule.v2t === 'jsonata') {
RED.util.evaluateJSONataExpression(rule.v2,msg,(err,value) => {
if (err) {
reject(RED._("switch.errors.invalid-expr",{error:err.message}));
} else {
resolve(value);
}
});
} else if (typeof v2 !== 'undefined') {
RED.util.evaluateNodeProperty(rule.v2,rule.v2t,node,msg, function(err,value) {
if (err) {
resolve(undefined);
} else {
resolve(value);
}
});
} else {
resolve(v2);
}
})
}
function applyRule(node, msg, property, state) {
return new Promise((resolve,reject) => {
var rule = node.rules[state.currentRule];
var v1,v2;
getV1(node,msg,rule,state.hasParts).then(value => {
v1 = value;
}).then(()=>getV2(node,msg,rule)).then(value => {
v2 = value;
}).then(() => {
if (rule.t == "else") {
property = state.elseflag;
state.elseflag = true;
}
if (operators[rule.t](property,v1,v2,rule.case,msg.parts)) {
state.onward.push(msg);
state.elseflag = false;
if (node.checkall == "false") {
return resolve(false);
}
} else {
state.onward.push(null);
}
resolve(state.currentRule < node.rules.length - 1);
});
})
}
function applyRules(node, msg, property,state) {
if (!state) {
state = {
currentRule: 0,
elseflag: true,
onward: [],
hasParts: msg.hasOwnProperty("parts") &&
msg.parts.hasOwnProperty("id") &&
msg.parts.hasOwnProperty("index")
}
}
return applyRule(node,msg,property,state).then(hasMore => {
if (hasMore) {
state.currentRule++;
return applyRules(node,msg,property,state);
} else {
node.previousValue = property;
return state.onward;
}
});
}
function SwitchNode(n) {
RED.nodes.createNode(this, n);
this.rules = n.rules || [];
@ -94,10 +230,10 @@ module.exports = function(RED) {
var node = this;
var valid = true;
var repair = n.repair;
var needs_count = repair;
var needsCount = repair;
for (var i=0; i<this.rules.length; i+=1) {
var rule = this.rules[i];
needs_count = needs_count || ((rule.t === "tail") || (rule.t === "jsonata_exp"));
needsCount = needsCount || ((rule.t === "tail") || (rule.t === "jsonata_exp"));
if (!rule.vt) {
if (!isNaN(Number(rule.v))) {
rule.vt = 'num';
@ -142,26 +278,26 @@ module.exports = function(RED) {
return;
}
var pending_count = 0;
var pending_id = 0;
var pending_in = {};
var pending_out = {};
var pendingCount = 0;
var pendingId = 0;
var pendingIn = {};
var pendingOut = {};
var received = {};
function add2group_in(id, msg, parts) {
if (!(id in pending_in)) {
pending_in[id] = {
function addMessageToGroup(id, msg, parts) {
if (!(id in pendingIn)) {
pendingIn[id] = {
count: undefined,
msgs: [],
seq_no: pending_id++
seq_no: pendingId++
};
}
var group = pending_in[id];
var group = pendingIn[id];
group.msgs.push(msg);
pending_count++;
var max_msgs = max_kept_msgs_count(node);
if ((max_msgs > 0) && (pending_count > max_msgs)) {
clear_pending();
pendingCount++;
var max_msgs = getMaxKeptCount();
if ((max_msgs > 0) && (pendingCount > max_msgs)) {
clearPending();
node.error(RED._("switch.errors.too-many"), msg);
}
if (parts.hasOwnProperty("count")) {
@ -170,32 +306,29 @@ module.exports = function(RED) {
return group;
}
function del_group_in(id, group) {
pending_count -= group.msgs.length;
delete pending_in[id];
}
function add2pending_in(msg) {
function addMessageToPending(msg) {
var parts = msg.parts;
if (parts.hasOwnProperty("id") &&
parts.hasOwnProperty("index")) {
var group = add2group_in(parts.id, msg, parts);
// We've already checked the msg.parts has the require bits
var group = addMessageToGroup(parts.id, msg, parts);
var msgs = group.msgs;
var count = group.count;
if (count === msgs.length) {
for (var i = 0; i < msgs.length; i++) {
var msg = msgs[i];
// We have a complete group - send the individual parts
return msgs.reduce((promise, msg) => {
return promise.then((result) => {
msg.parts.count = count;
process_msg(msg, false);
return processMessage(msg, false);
})
}, Promise.resolve()).then( () => {
pendingCount -= group.msgs.length;
delete pendingIn[parts.id];
});
}
del_group_in(parts.id, group);
}
return true;
}
return false;
return Promise.resolve();
}
function send_group(onwards, port_count) {
function sendGroup(onwards, port_count) {
var counts = new Array(port_count).fill(0);
for (var i = 0; i < onwards.length; i++) {
var onward = onwards[i];
@ -230,141 +363,104 @@ module.exports = function(RED) {
}
}
function send2ports(onward, msg) {
function sendGroupMessages(onward, msg) {
var parts = msg.parts;
var gid = parts.id;
received[gid] = ((gid in received) ? received[gid] : 0) +1;
var send_ok = (received[gid] === parts.count);
if (!(gid in pending_out)) {
pending_out[gid] = {
if (!(gid in pendingOut)) {
pendingOut[gid] = {
onwards: []
};
}
var group = pending_out[gid];
var group = pendingOut[gid];
var onwards = group.onwards;
onwards.push(onward);
pending_count++;
pendingCount++;
if (send_ok) {
send_group(onwards, onward.length, msg);
pending_count -= onward.length;
delete pending_out[gid];
sendGroup(onwards, onward.length, msg);
pendingCount -= onward.length;
delete pendingOut[gid];
delete received[gid];
}
var max_msgs = max_kept_msgs_count(node);
if ((max_msgs > 0) && (pending_count > max_msgs)) {
clear_pending();
var max_msgs = getMaxKeptCount();
if ((max_msgs > 0) && (pendingCount > max_msgs)) {
clearPending();
node.error(RED._("switch.errors.too-many"), msg);
}
}
function msg_has_parts(msg) {
if (msg.hasOwnProperty("parts")) {
var parts = msg.parts;
return (parts.hasOwnProperty("id") &&
parts.hasOwnProperty("index"));
}
return false;
}
function process_msg(msg, check_parts) {
var has_parts = msg_has_parts(msg);
if (needs_count && check_parts && has_parts &&
add2pending_in(msg)) {
return;
function processMessage(msg, checkParts) {
var hasParts = msg.hasOwnProperty("parts") &&
msg.parts.hasOwnProperty("id") &&
msg.parts.hasOwnProperty("index");
if (needsCount && checkParts && hasParts) {
return addMessageToPending(msg);
}
var onward = [];
try {
var prop;
if (node.propertyType === 'jsonata') {
prop = RED.util.evaluateJSONataExpression(node.property,msg);
} else {
prop = RED.util.evaluateNodeProperty(node.property,node.propertyType,node,msg);
}
var elseflag = true;
for (var i=0; i<node.rules.length; i+=1) {
var rule = node.rules[i];
var test = prop;
var v1,v2;
if (rule.vt === 'prev') {
v1 = node.previousValue;
} else if (rule.vt === 'jsonata') {
try {
var exp = rule.v;
if (rule.t === 'jsonata_exp') {
if (has_parts) {
exp.assign("I", msg.parts.index);
exp.assign("N", msg.parts.count);
}
}
v1 = RED.util.evaluateJSONataExpression(exp,msg);
} catch(err) {
node.error(RED._("switch.errors.invalid-expr",{error:err.message}));
return;
}
} else if (rule.vt === 'json') {
v1 = "json";
} else if (rule.vt === 'null') {
v1 = "null";
} else {
try {
v1 = RED.util.evaluateNodeProperty(rule.v,rule.vt,node,msg);
} catch(err) {
v1 = undefined;
}
}
v2 = rule.v2;
if (rule.v2t === 'prev') {
v2 = node.previousValue;
} else if (rule.v2t === 'jsonata') {
try {
v2 = RED.util.evaluateJSONataExpression(rule.v2,msg);
} catch(err) {
node.error(RED._("switch.errors.invalid-expr",{error:err.message}));
return;
}
} else if (typeof v2 !== 'undefined') {
try {
v2 = RED.util.evaluateNodeProperty(rule.v2,rule.v2t,node,msg);
} catch(err) {
v2 = undefined;
}
}
if (rule.t == "else") { test = elseflag; elseflag = true; }
if (operators[rule.t](test,v1,v2,rule.case,msg.parts)) {
onward.push(msg);
elseflag = false;
if (node.checkall == "false") { break; }
} else {
onward.push(null);
}
}
node.previousValue = prop;
if (!repair || !has_parts) {
return getProperty(node,msg)
.then(property => applyRules(node,msg,property))
.then(onward => {
if (!repair || !hasParts) {
node.send(onward);
}
else {
send2ports(onward, msg);
sendGroupMessages(onward, msg);
}
} catch(err) {
}).catch(err => {
node.warn(err);
}
});
}
function clear_pending() {
pending_count = 0;
pending_id = 0;
pending_in = {};
pending_out = {};
function clearPending() {
pendingCount = 0;
pendingId = 0;
pendingIn = {};
pendingOut = {};
received = {};
}
var pendingMessages = [];
var activeMessagePromise = null;
var processMessageQueue = function(msg) {
if (msg) {
// A new message has arrived - add it to the message queue
pendingMessages.push(msg);
if (activeMessagePromise !== null) {
// The node is currently processing a message, so do nothing
// more with this message
return;
}
}
if (pendingMessages.length === 0) {
// There are no more messages to process, clear the active flag
// and return
activeMessagePromise = null;
return;
}
// There are more messages to process. Get the next message and
// start processing it. Recurse back in to check for any more
var nextMsg = pendingMessages.shift();
activeMessagePromise = processMessage(nextMsg,true)
.then(processMessageQueue)
.catch((err) => {
node.error(err,nextMsg);
return processMessageQueue();
});
}
this.on('input', function(msg) {
process_msg(msg, true);
processMessageQueue(msg);
});
this.on('close', function() {
clear_pending();
clearPending();
});
}

12
nodes/core/logic/15-change.html
View File

@ -54,6 +54,10 @@
outputs: 1,
icon: "swap.png",
label: function() {
function prop2name(type, key) {
var result = RED.utils.parseContextKey(key);
return type +"." +result.key;
}
if (this.name) {
return this.name;
}
@ -70,13 +74,13 @@
} else {
if (this.rules.length == 1) {
if (this.rules[0].t === "set") {
return this._("change.label.set",{property:(this.rules[0].pt||"msg")+"."+this.rules[0].p});
return this._("change.label.set",{property:prop2name((this.rules[0].pt||"msg"), this.rules[0].p)});
} else if (this.rules[0].t === "change") {
return this._("change.label.change",{property:(this.rules[0].pt||"msg")+"."+this.rules[0].p});
return this._("change.label.change",{property:prop2name((this.rules[0].pt||"msg"), this.rules[0].p)});
} else if (this.rules[0].t === "move") {
return this._("change.label.move",{property:(this.rules[0].pt||"msg")+"."+this.rules[0].p});
return this._("change.label.move",{property:prop2name((this.rules[0].pt||"msg"), this.rules[0].p)});
} else {
return this._("change.label.delete",{property:(this.rules[0].pt||"msg")+"."+this.rules[0].p});
return this._("change.label.delete",{property:prop2name((this.rules[0].pt||"msg"), this.rules[0].p)});
}
} else {
return this._("change.label.changeCount",{count:this.rules.length});

176
nodes/core/logic/15-change.js
View File

@ -98,19 +98,13 @@ module.exports = function(RED) {
}
}
function applyRule(msg,rule) {
try {
var property = rule.p;
function getToValue(msg,rule) {
var value = rule.to;
if (rule.tot === 'json') {
value = JSON.parse(rule.to);
} else if (rule.tot === 'bin') {
value = Buffer.from(JSON.parse(rule.to))
}
var current;
var fromValue;
var fromType;
var fromRE;
if (rule.tot === "msg") {
value = RED.util.getMessageProperty(msg,rule.to);
} else if (rule.tot === 'flow') {
@ -120,22 +114,37 @@ module.exports = function(RED) {
} else if (rule.tot === 'date') {
value = Date.now();
} else if (rule.tot === 'jsonata') {
try{
value = RED.util.evaluateJSONataExpression(rule.to,msg);
} catch(err) {
node.error(RED._("change.errors.invalid-expr",{error:err.message}),msg);
return;
return new Promise((resolve,reject) => {
RED.util.evaluateJSONataExpression(rule.to,msg, (err, value) => {
if (err) {
reject(RED._("change.errors.invalid-expr",{error:err.message}))
} else {
resolve(value);
}
});
});
}
return Promise.resolve(value);
}
function getFromValue(msg,rule) {
var fromValue;
var fromType;
var fromRE;
if (rule.t === 'change') {
if (rule.fromt === 'msg' || rule.fromt === 'flow' || rule.fromt === 'global') {
return new Promise((resolve,reject) => {
if (rule.fromt === "msg") {
fromValue = RED.util.getMessageProperty(msg,rule.from);
} else if (rule.fromt === 'flow') {
fromValue = node.context().flow.get(rule.from);
} else if (rule.fromt === 'global') {
fromValue = node.context().global.get(rule.from);
resolve(RED.util.getMessageProperty(msg,rule.from));
} else if (rule.fromt === 'flow' || rule.fromt === 'global') {
node.context()[rule.fromt].get(rule.from,(err,fromValue) => {
if (err) {
reject(err);
} else {
resolve(fromValue);
}
});
}
}).then(fromValue => {
if (typeof fromValue === 'number' || fromValue instanceof Number) {
fromType = 'num';
} else if (typeof fromValue === 'boolean') {
@ -149,21 +158,45 @@ module.exports = function(RED) {
try {
fromRE = new RegExp(fromRE, "g");
} catch (e) {
valid = false;
node.error(RED._("change.errors.invalid-from",{error:e.message}),msg);
reject(new Error(RED._("change.errors.invalid-from",{error:e.message})));
return;
}
} else {
node.error(RED._("change.errors.invalid-from",{error:"unsupported type: "+(typeof fromValue)}),msg);
return
reject(new Error(RED._("change.errors.invalid-from",{error:"unsupported type: "+(typeof fromValue)})));
return;
}
return {
fromType,
fromValue,
fromRE
}
});
} else {
fromType = rule.fromt;
fromValue = rule.from;
fromRE = rule.fromRE;
}
}
return Promise.resolve({
fromType,
fromValue,
fromRE
});
}
function applyRule(msg,rule) {
var property = rule.p;
var current;
var fromValue;
var fromType;
var fromRE;
try {
return getToValue(msg,rule).then(value => {
return getFromValue(msg,rule).then(fromParts => {
fromValue = fromParts.fromValue;
fromType = fromParts.fromType;
fromRE = fromParts.fromRE;
if (rule.pt === 'msg') {
try {
if (rule.t === 'delete') {
RED.util.setMessageProperty(msg,property,undefined);
} else if (rule.t === 'set') {
@ -189,68 +222,97 @@ module.exports = function(RED) {
}
}
}
} catch(err) {}
return msg;
} else if (rule.pt === 'flow' || rule.pt === 'global') {
return new Promise((resolve,reject) => {
var target = node.context()[rule.pt];
var callback = err => {
if (err) {
reject(err);
} else {
resolve(msg);
}
else {
var target;
if (rule.pt === 'flow') {
target = node.context().flow;
} else if (rule.pt === 'global') {
target = node.context().global;
}
if (target) {
if (rule.t === 'delete') {
target.set(property,undefined);
target.set(property,undefined,callback);
} else if (rule.t === 'set') {
target.set(property,value);
target.set(property,value,callback);
} else if (rule.t === 'change') {
current = target.get(property);
target.get(property,(err,current) => {
if (err) {
reject(err);
return;
}
if (typeof current === 'string') {
if ((fromType === 'num' || fromType === 'bool' || fromType === 'str') && current === fromValue) {
// str representation of exact from number/boolean
// only replace if they match exactly
target.set(property,value);
target.set(property,value,callback);
} else {
current = current.replace(fromRE,value);
target.set(property,current);
target.set(property,current,callback);
}
} else if ((typeof current === 'number' || current instanceof Number) && fromType === 'num') {
if (current == Number(fromValue)) {
target.set(property,value);
target.set(property,value,callback);
}
} else if (typeof current === 'boolean' && fromType === 'bool') {
if (current.toString() === fromValue) {
target.set(property,value);
target.set(property,value,callback);
}
}
});
}
});
}
});
}).catch(err => {
node.error(err, msg);
return null;
});
} catch(err) {
return Promise.resolve(msg);
}
}
function applyRules(msg, currentRule) {
var r = node.rules[currentRule];
var rulePromise;
if (r.t === "move") {
if ((r.tot !== r.pt) || (r.p.indexOf(r.to) !== -1)) {
rulePromise = applyRule(msg,{t:"set", p:r.to, pt:r.tot, to:r.p, tot:r.pt}).then(
msg => applyRule(msg,{t:"delete", p:r.p, pt:r.pt})
);
}
} catch(err) {/*console.log(err.stack)*/}
else { // 2 step move if we are moving from a child
rulePromise = applyRule(msg,{t:"set", p:"_temp_move", pt:r.tot, to:r.p, tot:r.pt}).then(
msg => applyRule(msg,{t:"delete", p:r.p, pt:r.pt})
).then(
msg => applyRule(msg,{t:"set", p:r.to, pt:r.tot, to:"_temp_move", tot:r.pt})
).then(
msg => applyRule(msg,{t:"delete", p:"_temp_move", pt:r.pt})
)
}
} else {
rulePromise = applyRule(msg,r);
}
return rulePromise.then(
msg => {
if (!msg) {
return
} else if (currentRule === node.rules.length - 1) {
return msg;
} else {
return applyRules(msg, currentRule+1);
}
}
);
}
if (valid) {
this.on('input', function(msg) {
for (var i=0; i<this.rules.length; i++) {
if (this.rules[i].t === "move") {
var r = this.rules[i];
if ((r.tot !== r.pt) || (r.p.indexOf(r.to) !== -1)) {
msg = applyRule(msg,{t:"set", p:r.to, pt:r.tot, to:r.p, tot:r.pt});
applyRule(msg,{t:"delete", p:r.p, pt:r.pt});
}
else { // 2 step move if we are moving from a child
msg = applyRule(msg,{t:"set", p:"_temp_move", pt:r.tot, to:r.p, tot:r.pt});
applyRule(msg,{t:"delete", p:r.p, pt:r.pt});
msg = applyRule(msg,{t:"set", p:r.to, pt:r.tot, to:"_temp_move", tot:r.pt});
applyRule(msg,{t:"delete", p:"_temp_move", pt:r.pt});
}
} else {
msg = applyRule(msg,this.rules[i]);
}
if (msg === null) {
return;
}
}
node.send(msg);
applyRules(msg, 0)
.then( msg => { if (msg) { node.send(msg) }} )
.catch( err => node.error(err, msg))
});
}
}

193
nodes/core/logic/17-split.js
View File

@ -233,7 +233,7 @@ module.exports = function(RED) {
RED.nodes.registerType("split",SplitNode);
var _max_kept_msgs_count = undefined;
var _max_kept_msgs_count;
function max_kept_msgs_count(node) {
if (_max_kept_msgs_count === undefined) {
@ -252,7 +252,15 @@ module.exports = function(RED) {
exp.assign("I", index);
exp.assign("N", count);
exp.assign("A", accum);
return RED.util.evaluateJSONataExpression(exp, msg);
return new Promise((resolve,reject) => {
RED.util.evaluateJSONataExpression(exp, msg, (err, result) => {
if (err) {
reject(err);
} else {
resolve(result);
}
});
});
}
function apply_f(exp, accum, count) {
@ -269,32 +277,37 @@ module.exports = function(RED) {
return exp
}
function reduce_and_send_group(node, group) {
function reduceAndSendGroup(node, group) {
var is_right = node.reduce_right;
var flag = is_right ? -1 : 1;
var msgs = group.msgs;
var accum = eval_exp(node, node.exp_init, node.exp_init_type);
return getInitialReduceValue(node, node.exp_init, node.exp_init_type).then(accum => {
var reduce_exp = node.reduce_exp;
var reduce_fixup = node.reduce_fixup;
var count = group.count;
msgs.sort(function(x,y) {
var ix = x.parts.index;
var iy = y.parts.index;
if (ix < iy) return -flag;
if (ix > iy) return flag;
if (ix < iy) {return -flag;}
if (ix > iy) {return flag;}
return 0;
});
for(var msg of msgs) {
accum = apply_r(reduce_exp, accum, msg, msg.parts.index, count);
}
return msgs.reduce((promise, msg) => promise.then(accum => apply_r(reduce_exp, accum, msg, msg.parts.index, count)), Promise.resolve(accum))
.then(accum => {
if(reduce_fixup !== undefined) {
accum = apply_f(reduce_fixup, accum, count);
}
node.send({payload: accum});
});
}).catch(err => {
throw new Error(RED._("join.errors.invalid-expr",{error:e.message}));
});
}
function reduce_msg(node, msg) {
if(msg.hasOwnProperty('parts')) {
var promise;
if (msg.hasOwnProperty('parts')) {
var parts = msg.parts;
var pending = node.pending;
var pending_count = node.pending_count;
@ -317,60 +330,77 @@ module.exports = function(RED) {
}
msgs.push(msg);
pending_count++;
if(msgs.length === group.count) {
delete pending[gid];
try {
pending_count -= msgs.length;
reduce_and_send_group(node, group);
} catch(e) {
node.error(RED._("join.errors.invalid-expr",{error:e.message})); }
}
var completeProcess = function() {
node.pending_count = pending_count;
var max_msgs = max_kept_msgs_count(node);
if ((max_msgs > 0) && (pending_count > max_msgs)) {
node.pending = {};
node.pending_count = 0;
node.error(RED._("join.too-many"), msg);
var promise = Promise.reject(RED._("join.too-many"));
promise.catch(()=>{});
return promise;
}
return Promise.resolve();
}
else {
if(msgs.length === group.count) {
delete pending[gid];
pending_count -= msgs.length;
promise = reduceAndSendGroup(node, group).then(completeProcess);
} else {
promise = completeProcess();
}
} else {
node.send(msg);
}
if (!promise) {
promise = Promise.resolve();
}
return promise;
}
function eval_exp(node, exp, exp_type) {
if(exp_type === "flow") {
return node.context().flow.get(exp);
function getInitialReduceValue(node, exp, exp_type) {
return new Promise((resolve,reject) => {
if(exp_type === "flow" || exp_type === "global") {
node.context()[exp_type].get(exp,(err,value) => {
if (err) {
reject(err);
} else {
resolve(value);
}
else if(exp_type === "global") {
return node.context().global.get(exp);
}
else if(exp_type === "str") {
return exp;
}
else if(exp_type === "num") {
return Number(exp);
}
else if(exp_type === "bool") {
if (exp === 'true') {
return true;
}
else if (exp === 'false') {
return false;
}
}
else if ((exp_type === "bin") ||
(exp_type === "json")) {
return JSON.parse(exp);
}
else if(exp_type === "date") {
return Date.now();
}
else if(exp_type === "jsonata") {
});
return;
} else if(exp_type === "jsonata") {
var jexp = RED.util.prepareJSONataExpression(exp, node);
return RED.util.evaluateJSONataExpression(jexp, {});
RED.util.evaluateJSONataExpression(jexp, {},(err,value) => {
if (err) {
reject(err);
} else {
resolve(value);
}
throw new Error("unexpected initial value type");
});
return;
}
var result;
if(exp_type === "str") {
result = exp;
} else if(exp_type === "num") {
result = Number(exp);
} else if(exp_type === "bool") {
if (exp === 'true') {
result = true;
} else if (exp === 'false') {
result = false;
}
} else if ((exp_type === "bin") || (exp_type === "json")) {
result = JSON.parse(exp);
} else if(exp_type === "date") {
result = Date.now();
} else {
reject(new Error("unexpected initial value type"));
return;
}
resolve(result);
});
}
function JoinNode(n) {
@ -437,7 +467,8 @@ module.exports = function(RED) {
newArray = newArray.concat(n);
})
group.payload = newArray;
} else if (group.type === 'buffer') {
}
else if (group.type === 'buffer') {
var buffers = [];
var bufferLen = 0;
if (group.joinChar !== undefined) {
@ -450,7 +481,8 @@ module.exports = function(RED) {
buffers.push(group.payload[i]);
bufferLen += group.payload[i].length;
}
} else {
}
else {
bufferLen = group.bufferLen;
buffers = group.payload;
}
@ -463,7 +495,8 @@ module.exports = function(RED) {
groupJoinChar = group.joinChar.toString();
}
RED.util.setMessageProperty(group.msg,node.property,group.payload.join(groupJoinChar));
} else {
}
else {
if (node.propertyType === 'full') {
group.msg = RED.util.cloneMessage(group.msg);
}
@ -471,13 +504,48 @@ module.exports = function(RED) {
}
if (group.msg.hasOwnProperty('parts') && group.msg.parts.hasOwnProperty('parts')) {
group.msg.parts = group.msg.parts.parts;
} else {
}
else {
delete group.msg.parts;
}
delete group.msg.complete;
node.send(group.msg);
}
var pendingMessages = [];
var activeMessagePromise = null;
// In reduce mode, we must process messages fully in order otherwise
// groups may overlap and cause unexpected results. The use of JSONata
// means some async processing *might* occur if flow/global context is
// accessed.
var processReduceMessageQueue = function(msg) {
if (msg) {
// A new message has arrived - add it to the message queue
pendingMessages.push(msg);
if (activeMessagePromise !== null) {
// The node is currently processing a message, so do nothing
// more with this message
return;
}
}
if (pendingMessages.length === 0) {
// There are no more messages to process, clear the active flag
// and return
activeMessagePromise = null;
return;
}
// There are more messages to process. Get the next message and
// start processing it. Recurse back in to check for any more
var nextMsg = pendingMessages.shift();
activeMessagePromise = reduce_msg(node, nextMsg)
.then(processReduceMessageQueue)
.catch((err) => {
node.error(err,nextMsg);
return processReduceMessageQueue();
});
}
this.on("input", function(msg) {
try {
var property;
@ -516,8 +584,7 @@ module.exports = function(RED) {
propertyIndex = msg.parts.index;
}
else if (node.mode === 'reduce') {
reduce_msg(node, msg);
return;
return processReduceMessageQueue(msg);
}
else {
// Use the node configuration to identify all of the group information
@ -525,7 +592,7 @@ module.exports = function(RED) {
payloadType = node.build;
targetCount = node.count;
joinChar = node.joiner;
if (targetCount === 0 && msg.hasOwnProperty('parts')) {
if (n.count === "" && msg.hasOwnProperty('parts')) {
targetCount = msg.parts.count || 0;
}
if (node.build === 'object') {
@ -554,7 +621,7 @@ module.exports = function(RED) {
payload:{},
targetCount:targetCount,
type:"object",
msg:msg
msg:RED.util.cloneMessage(msg)
};
}
else if (node.accumulate === true) {
@ -564,7 +631,7 @@ module.exports = function(RED) {
payload:{},
targetCount:targetCount,
type:payloadType,
msg:msg
msg:RED.util.cloneMessage(msg)
}
if (payloadType === 'string' || payloadType === 'array' || payloadType === 'buffer') {
inflight[partId].payload = [];
@ -576,7 +643,7 @@ module.exports = function(RED) {
payload:[],
targetCount:targetCount,
type:payloadType,
msg:msg
msg:RED.util.cloneMessage(msg)
};
if (payloadType === 'string') {
inflight[partId].joinChar = joinChar;
@ -627,14 +694,14 @@ module.exports = function(RED) {
}
}
}
// TODO: currently reuse the last received - add option to pick first received
group.msg = msg;
group.msg = Object.assign(group.msg, msg);
var tcnt = group.targetCount;
if (msg.hasOwnProperty("parts")) { tcnt = group.targetCount || msg.parts.count; }
if ((tcnt > 0 && group.currentCount >= tcnt) || msg.hasOwnProperty('complete')) {
completeSend(partId);
}
} catch(err) {
}
catch(err) {
console.log(err.stack);
}
});

166
nodes/core/logic/18-sort.js
View File

@ -17,7 +17,7 @@
module.exports = function(RED) {
"use strict";
var _max_kept_msgs_count = undefined;
var _max_kept_msgs_count;
function max_kept_msgs_count(node) {
if (_max_kept_msgs_count === undefined) {
@ -32,30 +32,20 @@ module.exports = function(RED) {
return _max_kept_msgs_count;
}
function eval_jsonata(node, code, val) {
try {
return RED.util.evaluateJSONataExpression(code, val);
}
catch (e) {
node.error(RED._("sort.invalid-exp"));
throw e;
}
}
function get_context_val(node, name, dval) {
var context = node.context();
var val = context.get(name);
if (val === undefined) {
context.set(name, dval);
return dval;
}
return val;
}
// function get_context_val(node, name, dval) {
// var context = node.context();
// var val = context.get(name);
// if (val === undefined) {
// context.set(name, dval);
// return dval;
// }
// return val;
// }
function SortNode(n) {
RED.nodes.createNode(this, n);
var node = this;
var pending = get_context_val(node, 'pending', {})
var pending = {};//get_context_val(node, 'pending', {})
var pending_count = 0;
var pending_id = 0;
var order = n.order || "ascending";
@ -76,11 +66,10 @@ module.exports = function(RED) {
}
}
var dir = (order === "ascending") ? 1 : -1;
var conv = as_num
? function(x) { return Number(x); }
var conv = as_num ? function(x) { return Number(x); }
: function(x) { return x; };
function gen_comp(key) {
function generateComparisonFunction(key) {
return function(x, y) {
var xp = conv(key(x));
var yp = conv(key(y));
@ -90,69 +79,91 @@ module.exports = function(RED) {
};
}
function send_group(group) {
var key = key_is_exp
? function(msg) {
return eval_jsonata(node, key_exp, msg);
}
: function(msg) {
return RED.util.getMessageProperty(msg, key_prop);
};
var comp = gen_comp(key);
function sortMessageGroup(group) {
var promise;
var msgs = group.msgs;
if (key_is_exp) {
var evaluatedDataPromises = msgs.map(msg => {
return new Promise((resolve,reject) => {
RED.util.evaluateJSONataExpression(key_exp, msg, (err, result) => {
resolve({
item: msg,
sortValue: result
})
});
})
});
promise = Promise.all(evaluatedDataPromises).then(evaluatedElements => {
// Once all of the sort keys are evaluated, sort by them
var comp = generateComparisonFunction(elem=>elem.sortValue);
return evaluatedElements.sort(comp).map(elem=>elem.item);
});
} else {
var key = function(msg) {
return ;
}
var comp = generateComparisonFunction(msg => RED.util.getMessageProperty(msg, key_prop));
try {
msgs.sort(comp);
}
catch (e) {
return; // not send when error
}
promise = Promise.resolve(msgs);
}
return promise.then(msgs => {
for (var i = 0; i < msgs.length; i++) {
var msg = msgs[i];
msg.parts.index = i;
node.send(msg);
}
});
}
function sort_payload(msg) {
function sortMessageProperty(msg) {
var data = RED.util.getMessageProperty(msg, target_prop);
if (Array.isArray(data)) {
var key = key_is_exp
? function(elem) {
return eval_jsonata(node, key_exp, elem);
}
: function(elem) { return elem; };
var comp = gen_comp(key);
if (key_is_exp) {
// key is an expression. Evaluated the expression for each item
// to get its sort value. As this could be async, need to do
// it first.
var evaluatedDataPromises = data.map(elem => {
return new Promise((resolve,reject) => {
RED.util.evaluateJSONataExpression(key_exp, elem, (err, result) => {
resolve({
item: elem,
sortValue: result
})
});
})
})
return Promise.all(evaluatedDataPromises).then(evaluatedElements => {
// Once all of the sort keys are evaluated, sort by them
// and reconstruct the original message item with the newly
// sorted values.
var comp = generateComparisonFunction(elem=>elem.sortValue);
data = evaluatedElements.sort(comp).map(elem=>elem.item);
RED.util.setMessageProperty(msg, target_prop,data);
return true;
})
} else {
var comp = generateComparisonFunction(elem=>elem);
try {
data.sort(comp);
} catch (e) {
return Promise.resolve(false);
}
catch (e) {
return false;
return Promise.resolve(true);
}
return true;
}
return false;
return Promise.resolve(false);
}
function check_parts(parts) {
if (parts.hasOwnProperty("id") &&
parts.hasOwnProperty("index")) {
return true;
}
return false;
}
function clear_pending() {
for(var key in pending) {
node.log(RED._("sort.clear"), pending[key].msgs[0]);
delete pending[key];
}
pending_count = 0;
}
function remove_oldest_pending() {
var oldest = undefined;
var oldest_key = undefined;
function removeOldestPending() {
var oldest;
var oldest_key;
for(var key in pending) {
if (pending.hasOwnProperty(key)) {
var item = pending[key];
if((oldest === undefined) ||
(oldest.seq_no > item.seq_no)) {
@ -160,6 +171,7 @@ module.exports = function(RED) {
oldest_key = key;
}
}
}
if(oldest !== undefined) {
delete pending[oldest_key];
return oldest.msgs.length;
@ -167,15 +179,17 @@ module.exports = function(RED) {
return 0;
}
function process_msg(msg) {
function processMessage(msg) {
if (target_is_prop) {
if (sort_payload(msg)) {
sortMessageProperty(msg).then(send => {
if (send) {
node.send(msg);
}
return;
}).catch(err => {
});
}
var parts = msg.parts;
if (!check_parts(parts)) {
if (!parts.hasOwnProperty("id") || !parts.hasOwnProperty("index")) {
return;
}
var gid = parts.id;
@ -195,23 +209,29 @@ module.exports = function(RED) {
pending_count++;
if (group.count === msgs.length) {
delete pending[gid]
send_group(group);
sortMessageGroup(group);
pending_count -= msgs.length;
}
} else {
var max_msgs = max_kept_msgs_count(node);
if ((max_msgs > 0) && (pending_count > max_msgs)) {
pending_count -= remove_oldest_pending();
pending_count -= removeOldestPending();
node.error(RED._("sort.too-many"), msg);
}
}
}
this.on("input", function(msg) {
process_msg(msg);
processMessage(msg);
});
this.on("close", function() {
clear_pending();
})
for(var key in pending) {
if (pending.hasOwnProperty(key)) {
node.log(RED._("sort.clear"), pending[key].msgs[0]);
delete pending[key];
}
}
pending_count = 0; })
}
RED.nodes.registerType("sort", SortNode);

37
red/api/admin/context.js
View File

@ -29,7 +29,6 @@ module.exports = {
var scope = req.params.scope;
var id = req.params.id;
var key = req.params[0];
var result = {};
var ctx;
if (scope === 'global') {
ctx = redNodes.getContext('global');
@ -43,19 +42,31 @@ module.exports = {
}
if (ctx) {
if (key) {
result = util.encodeObject({msg:ctx.get(key)});
ctx.get(key,function(err, v) {
res.json(util.encodeObject({msg:v}));
});
return;
} else {
var keys = ctx.keys();
var i = 0;
var l = keys.length;
while(i < l) {
var k = keys[i];
result[k] = util.encodeObject({msg:ctx.get(k)});
i++;
}
}
}
ctx.keys(function(err, keys) {
var result = {};
var c = keys.length;
if (c === 0) {
res.json(result);
} else {
keys.forEach(function(key) {
ctx.get(key,function(err, v) {
result[key] = util.encodeObject({msg:v});
c--;
if (c === 0) {
res.json(result);
}
});
});
}
});
}
} else {
res.json({});
}
}
}

2
red/api/editor/settings.js
View File

@ -40,6 +40,8 @@ module.exports = {
})
}
safeSettings.context = runtime.nodes.listContextStores();
var themeSettings = theme.settings();
if (themeSettings) {
safeSettings.editorTheme = themeSettings;

164
red/runtime/nodes/context/index.js
View File

@ -19,30 +19,37 @@ var log = require("../../log");
var memory = require("./memory");
var settings;
// A map of scope id to context instance
var contexts = {};
var globalContext = null;
var externalContexts = {};
var hasExternalContext = false;
// A map of store name to instance
var stores = {};
var storeList = [];
var defaultStore;
// Whether there context storage has been configured or left as default
var hasConfiguredStore = false;
function init(_settings) {
settings = _settings;
externalContexts = {};
// init memory plugin
stores = {};
var seed = settings.functionGlobalContext || {};
externalContexts["_"] = memory();
externalContexts["_"].setGlobalContext(seed);
globalContext = createContext("global",seed);
contexts['global'] = globalContext;
contexts['global'] = createContext("global",seed);
stores["_"] = new memory();
defaultStore = "memory";
}
function load() {
return new Promise(function(resolve,reject) {
// load & init plugins in settings.contextStorage
var plugins = settings.contextStorage;
var plugins = settings.contextStorage || {};
var defaultIsAlias = false;
var promises = [];
if (plugins) {
if (plugins && Object.keys(plugins).length > 0) {
var hasDefault = plugins.hasOwnProperty('default');
var defaultName;
for (var pluginName in plugins) {
if (plugins.hasOwnProperty(pluginName)) {
// "_" is a reserved name - do not allow it to be overridden
@ -59,6 +66,9 @@ function load() {
defaultIsAlias = true;
continue;
}
if (!hasDefault && !defaultName) {
defaultName = pluginName;
}
var plugin;
if (plugins[pluginName].hasOwnProperty("module")) {
// Get the provided config and copy in the 'approved' top-level settings (eg userDir)
@ -79,7 +89,7 @@ function load() {
}
try {
// Create a new instance of the plugin by calling its module function
externalContexts[pluginName] = plugin(config);
stores[pluginName] = plugin(config);
} catch(err) {
return reject(new Error(log._("context.error-loading-module",{module:pluginName,message:err.toString()})));
}
@ -91,23 +101,41 @@ function load() {
}
// Open all of the configured contexts
for (var plugin in externalContexts) {
if (externalContexts.hasOwnProperty(plugin)) {
promises.push(externalContexts[plugin].open());
for (var plugin in stores) {
if (stores.hasOwnProperty(plugin)) {
promises.push(stores[plugin].open());
}
}
// There is a 'default' listed in the configuration
if (hasDefault) {
// If 'default' is an alias, point it at the right module - we have already
// checked that it exists
// checked that it exists. If it isn't an alias, then it will
// already be set to a configured store
if (defaultIsAlias) {
externalContexts["default"] = externalContexts[plugins["default"]];
}
}
if (promises.length === 0) {
promises.push(externalContexts["_"].open())
stores["_"] = stores[plugins["default"]];
defaultStore = plugins["default"];
} else {
hasExternalContext = true;
stores["_"] = stores["default"];
defaultStore = "default";
}
} else if (defaultName) {
// No 'default' listed, so pick first in list as the default
stores["_"] = stores[defaultName];
defaultStore = defaultName;
defaultIsAlias = true;
} else {
// else there were no stores list the config object - fall through
// to below where we default to a memory store
storeList = ["memory"];
defaultStore = "memory";
}
hasConfiguredStore = true;
storeList = Object.keys(stores).filter(n=>!(defaultIsAlias && n==="default") && n!== "_");
} else {
// No configured plugins
promises.push(stores["_"].open())
storeList = ["memory"];
defaultStore = "memory";
}
return resolve(Promise.all(promises));
});
@ -122,12 +150,12 @@ function copySettings(config, settings){
}
function getContextStorage(storage) {
if (externalContexts.hasOwnProperty(storage)) {
if (stores.hasOwnProperty(storage)) {
// A known context
return externalContexts[storage];
} else if (externalContexts.hasOwnProperty("default")) {
return stores[storage];
} else if (stores.hasOwnProperty("_")) {
// Not known, but we have a default to fall back to
return externalContexts["default"];
return stores["_"];
} else {
// Not known and no default configured
var contextError = new Error(log._("context.error-use-undefined-storage", {storage:storage}));
@ -136,34 +164,61 @@ function getContextStorage(storage) {
}
}
function createContext(id,seed) {
// Seed is only set for global context - sourced from functionGlobalContext
var scope = id;
var obj = seed || {};
var seedKeys;
if (seed) {
seedKeys = Object.keys(seed);
}
obj.get = function(key, storage, callback) {
var context;
if (!storage && !callback) {
context = externalContexts["_"];
context = stores["_"];
} else {
if (typeof storage === 'function') {
callback = storage;
storage = "default";
storage = "_";
}
if (typeof callback !== 'function'){
throw new Error("Callback must be a function");
}
context = getContextStorage(storage);
}
if (seed) {
// Get the value from the underlying store. If it is undefined,
// check the seed for a default value.
if (callback) {
context.get(scope,key,function(err, v) {
if (v === undefined) {
callback(err, seed[key]);
} else {
callback(err, v);
}
})
} else {
// No callback, attempt to do this synchronously
var storeValue = context.get(scope,key);
if (storeValue === undefined) {
return seed[key];
} else {
return storeValue;
}
}
} else {
return context.get(scope, key, callback);
}
};
obj.set = function(key, value, storage, callback) {
var context;
if (!storage && !callback) {
context = externalContexts["_"];
context = stores["_"];
} else {
if (typeof storage === 'function') {
callback = storage;
storage = "default";
storage = "_";
}
if (callback && typeof callback !== 'function') {
throw new Error("Callback must be a function");
@ -175,18 +230,29 @@ function createContext(id,seed) {
obj.keys = function(storage, callback) {
var context;
if (!storage && !callback) {
context = externalContexts["_"];
context = stores["_"];
} else {
if (typeof storage === 'function') {
callback = storage;
storage = "default";
storage = "_";
}
if (typeof callback !== 'function') {
throw new Error("Callback must be a function");
}
context = getContextStorage(storage);
}
if (seed) {
if (callback) {
context.keys(scope, function(err,keys) {
callback(err,Array.from(new Set(seedKeys.concat(keys)).keys()));
});
} else {
var keys = context.keys(scope);
return Array.from(new Set(seedKeys.concat(keys)).keys())
}
} else {
return context.keys(scope, callback);
}
};
return obj;
}
@ -203,21 +269,20 @@ function getContext(localId,flowId) {
if (flowId) {
newContext.flow = getContext(flowId);
}
if (globalContext) {
newContext.global = globalContext;
}
newContext.global = contexts['global'];
contexts[contextId] = newContext;
return newContext;
}
function deleteContext(id,flowId) {
if(!hasExternalContext){
if(!hasConfiguredStore){
// only delete context if there's no configured storage.
var contextId = id;
if (flowId) {
contextId = id+":"+flowId;
}
delete contexts[contextId];
return externalContexts["_"].delete(contextId);
return stores["_"].delete(contextId);
}else{
return Promise.resolve();
}
@ -225,9 +290,9 @@ function deleteContext(id,flowId) {
function clean(flowConfig) {
var promises = [];
for(var plugin in externalContexts){
if(externalContexts.hasOwnProperty(plugin)){
promises.push(externalContexts[plugin].clean(Object.keys(flowConfig.allNodes)));
for(var plugin in stores){
if(stores.hasOwnProperty(plugin)){
promises.push(stores[plugin].clean(Object.keys(flowConfig.allNodes)));
}
}
for (var id in contexts) {
@ -243,17 +308,22 @@ function clean(flowConfig) {
function close() {
var promises = [];
for(var plugin in externalContexts){
if(externalContexts.hasOwnProperty(plugin)){
promises.push(externalContexts[plugin].close());
for(var plugin in stores){
if(stores.hasOwnProperty(plugin)){
promises.push(stores[plugin].close());
}
}
return Promise.all(promises);
}
function listStores() {
return {default:defaultStore,stores:storeList};
}
module.exports = {
init: init,
load: load,
listStores: listStores,
get: getContext,
delete: deleteContext,
clean: clean,

114
red/runtime/nodes/context/localfilesystem.js
View File

@ -14,10 +14,39 @@
* limitations under the License.
**/
/**
* Local file-system based context storage
*
* Configuration options:
* {
* base: "contexts", // the base directory to use
* // default: "contexts"
* dir: "/path/to/storage", // the directory to create the base directory in
* // default: settings.userDir
* cache: true // whether to cache contents in memory
* // default: true
* }
*
*
* $HOME/.node-red/contexts
* ├── global
* │ └── global_context.json
* ├── <id of Flow 1>
* │ ├── flow_context.json
* │ ├── <id of Node a>.json
* │ └── <id of Node b>.json
* └── <id of Flow 2>
* ├── flow_context.json
* ├── <id of Node x>.json
* └── <id of Node y>.json
*/
var fs = require('fs-extra');
var path = require("path");
var util = require("../../util");
var MemoryStore = require("./memory");
function getStoragePath(storageBaseDir, scope) {
if(scope.indexOf(":") === -1){
if(scope === "global"){
@ -76,10 +105,54 @@ function loadFile(storagePath){
function LocalFileSystem(config){
this.config = config;
this.storageBaseDir = getBasePath(this.config);
if (config.hasOwnProperty('cache')?config.cache:true) {
this.cache = MemoryStore({});
}
}
LocalFileSystem.prototype.open = function(){
var self = this;
if (this.cache) {
var scopes = [];
var promises = [];
var subdirs = [];
var subdirPromises = [];
return fs.readdir(self.storageBaseDir).then(function(dirs){
dirs.forEach(function(fn) {
var p = getStoragePath(self.storageBaseDir ,fn)+".json";
scopes.push(fn);
promises.push(loadFile(p));
subdirs.push(path.join(self.storageBaseDir,fn));
subdirPromises.push(fs.readdir(path.join(self.storageBaseDir,fn)));
})
return Promise.all(subdirPromises);
}).then(function(dirs) {
dirs.forEach(function(files,i) {
files.forEach(function(fn) {
if (fn !== 'flow.json' && fn !== 'global.json') {
scopes.push(fn.substring(0,fn.length-5)+":"+scopes[i]);
promises.push(loadFile(path.join(subdirs[i],fn)))
}
});
})
return Promise.all(promises);
}).then(function(res) {
scopes.forEach(function(scope,i) {
var data = res[i]?JSON.parse(res[i]):{};
Object.keys(data).forEach(function(key) {
self.cache.set(scope,key,data[key]);
})
});
}).catch(function(err){
if(err.code == 'ENOENT') {
return fs.mkdir(self.storageBaseDir);
}else{
return Promise.reject(err);
}
});
} else {
return Promise.resolve();
}
}
LocalFileSystem.prototype.close = function(){
@ -87,6 +160,9 @@ LocalFileSystem.prototype.close = function(){
}
LocalFileSystem.prototype.get = function(scope, key, callback) {
if (this.cache) {
return this.cache.get(scope,key,callback);
}
if(typeof callback !== "function"){
throw new Error("Callback must be a function");
}
@ -102,7 +178,7 @@ LocalFileSystem.prototype.get = function(scope, key, callback) {
});
};
LocalFileSystem.prototype.set =function(scope, key, value, callback) {
LocalFileSystem.prototype._set = function(scope, key, value, callback) {
var storagePath = getStoragePath(this.storageBaseDir ,scope);
loadFile(storagePath + ".json").then(function(data){
var obj = data ? JSON.parse(data) : {}
@ -117,9 +193,25 @@ LocalFileSystem.prototype.set =function(scope, key, value, callback) {
callback(err);
}
});
}
LocalFileSystem.prototype.set = function(scope, key, value, callback) {
if (this.cache) {
this.cache.set(scope,key,value,callback);
// With cache enabled, no need to re-read the file prior to writing.
var newContext = this.cache._export()[scope];
var storagePath = getStoragePath(this.storageBaseDir ,scope);
fs.outputFile(storagePath + ".json", JSON.stringify(newContext, undefined, 4), "utf8").catch(function(err) {
});
} else {
this._set(scope,key,value,callback);
}
};
LocalFileSystem.prototype.keys = function(scope, callback){
if (this.cache) {
return this.cache.keys(scope,callback);
}
if(typeof callback !== "function"){
throw new Error("Callback must be a function");
}
@ -136,12 +228,28 @@ LocalFileSystem.prototype.keys = function(scope, callback){
};
LocalFileSystem.prototype.delete = function(scope){
var storagePath = getStoragePath(this.storageBaseDir ,scope);
var cachePromise;
if (this.cache) {
cachePromise = this.cache.delete(scope);
} else {
cachePromise = Promise.resolve();
}
var that = this;
return cachePromise.then(function() {
var storagePath = getStoragePath(that.storageBaseDir,scope);
return fs.remove(storagePath + ".json");
});
}
LocalFileSystem.prototype.clean = function(activeNodes){
var self = this;
var cachePromise;
if (this.cache) {
cachePromise = this.cache.clean(activeNodes);
} else {
cachePromise = Promise.resolve();
}
return cachePromise.then(function() {
return fs.readdir(self.storageBaseDir).then(function(dirs){
return Promise.all(dirs.reduce(function(result, item){
if(item !== "global" && activeNodes.indexOf(item) === -1){
@ -156,9 +264,9 @@ LocalFileSystem.prototype.clean = function(activeNodes){
return Promise.reject(err);
}
});
});
}
module.exports = function(config){
return new LocalFileSystem(config);
};

7
red/runtime/nodes/context/memory.js
View File

@ -109,9 +109,10 @@ Memory.prototype.clean = function(activeNodes){
return Promise.resolve();
}
Memory.prototype.setGlobalContext= function(seed){
this.data["global"] = seed;
};
Memory.prototype._export = function() {
return this.data;
}
module.exports = function(config){
return new Memory(config);

3
red/runtime/nodes/index.js
View File

@ -222,5 +222,6 @@ module.exports = {
// Contexts
loadContextsPlugin: context.load,
closeContextsPlugin: context.close
closeContextsPlugin: context.close,
listContextStores: context.listStores
};

93
red/runtime/util.js
View File

@ -322,35 +322,63 @@ function evaluteEnvProperty(value) {
return value;
}
function evaluateNodeProperty(value, type, node, msg) {
var parseContextStore = function(key) {
var parts = {};
var m = /^#:\((\S+?)\)::(.*)$/.exec(key);
if (m) {
parts.store = m[1];
parts.key = m[2];
} else {
parts.key = key;
}
return parts;
}
function evaluateNodeProperty(value, type, node, msg, callback) {
var result = value;
if (type === 'str') {
return ""+value;
result = ""+value;
} else if (type === 'num') {
return Number(value);
result = Number(value);
} else if (type === 'json') {
return JSON.parse(value);
result = JSON.parse(value);
} else if (type === 're') {
return new RegExp(value);
result = new RegExp(value);
} else if (type === 'date') {
return Date.now();
result = Date.now();
} else if (type === 'bin') {
var data = JSON.parse(value);
return Buffer.from(data);
result = Buffer.from(data);
} else if (type === 'msg' && msg) {
return getMessageProperty(msg,value);
} else if (type === 'flow' && node) {
return node.context().flow.get(value);
} else if (type === 'global' && node) {
return node.context().global.get(value);
try {
result = getMessageProperty(msg,value);
} catch(err) {
if (callback) {
callback(err);
} else {
throw err;
}
return;
}
} else if ((type === 'flow' || type === 'global') && node) {
var contextKey = parseContextStore(value);
result = node.context()[type].get(contextKey.key,contextKey.store,callback);
if (callback) {
return;
}
} else if (type === 'bool') {
return /^true$/i.test(value);
result = /^true$/i.test(value);
} else if (type === 'jsonata') {
var expr = prepareJSONataExpression(value,node);
return evaluateJSONataExpression(expr,msg);
result = evaluateJSONataExpression(expr,msg);
} else if (type === 'env') {
return evaluteEnvProperty(value);
result = evaluteEnvProperty(value);
}
if (callback) {
callback(null,result);
} else {
return result;
}
return value;
}
function prepareJSONataExpression(value,node) {
@ -366,15 +394,44 @@ function prepareJSONataExpression(value,node) {
})
expr.registerFunction('clone', cloneMessage, '<(oa)-:o>');
expr._legacyMode = /(^|[^a-zA-Z0-9_'"])msg([^a-zA-Z0-9_'"]|$)/.test(value);
expr._node = node;
return expr;
}
function evaluateJSONataExpression(expr,msg) {
function evaluateJSONataExpression(expr,msg,callback) {
var context = msg;
if (expr._legacyMode) {
context = {msg:msg};
}
return expr.evaluate(context);
var bindings = {};
if (callback) {
// If callback provided, need to override the pre-assigned sync
// context functions to be their async variants
bindings.flowContext = function(val) {
return new Promise((resolve,reject) => {
expr._node.context().flow.get(val, function(err,value) {
if (err) {
reject(err);
} else {
resolve(value);
}
})
});
}
bindings.globalContext = function(val) {
return new Promise((resolve,reject) => {
expr._node.context().global.get(val, function(err,value) {
if (err) {
reject(err);
} else {
resolve(value);
}
})
});
}
}
return expr.evaluate(context, bindings, callback);
}

9
test/editor/wdio.conf.js
View File

@ -62,10 +62,11 @@ exports.config = {
//
browserName: 'chrome',
chromeOptions: {
// Runs tests without opening a broser.
args: ['--headless', '--disable-gpu', 'window-size=1920,1080'],
// Runs tests with opening a broser.
// args: ['--disable-gpu'],
args: process.env.NODE_RED_NON_HEADLESS
// Runs tests with opening a browser.
? ['--disable-gpu']
// Runs tests without opening a browser.
: ['--headless', '--disable-gpu', 'window-size=1920,1080']
},
}],
//

6
test/nodes/core/core/89-trigger_spec.js
View File

@ -288,7 +288,7 @@ describe('trigger node', function() {
it('should be able to return things from flow and global context variables', function(done) {
var spy = sinon.stub(RED.util, 'evaluateNodeProperty',
function(arg1, arg2, arg3, arg4) { return arg1; }
function(arg1, arg2, arg3, arg4, arg5) { if (arg5) { arg5(null, arg1) } else { return arg1; } }
);
var flow = [{"id":"n1", "type":"trigger", "name":"triggerNode", op1:"foo", op1type:"flow", op2:"bar", op2type:"global", duration:"20", wires:[["n2"]] },
{id:"n2", type:"helper"} ];
@ -386,7 +386,7 @@ describe('trigger node', function() {
it('should be able to extend the delay', function(done) {
this.timeout(5000); // add extra time for flake
var spy = sinon.stub(RED.util, 'evaluateNodeProperty',
function(arg1, arg2, arg3, arg4) { return arg1; }
function(arg1, arg2, arg3, arg4, arg5) { if (arg5) { arg5(null, arg1) } else { return arg1; } }
);
var flow = [{"id":"n1", "type":"trigger", "name":"triggerNode", extend:"true", op1type:"flow", op1:"foo", op2:"bar", op2type:"global", duration:"100", wires:[["n2"]] },
{id:"n2", type:"helper"} ];
@ -428,12 +428,10 @@ describe('trigger node', function() {
n2.on("input", function(msg) {
try {
if (c === 0) {
console.log(c,Date.now() - ss,msg);
msg.should.have.a.property("payload", "Hello");
c += 1;
}
else {
console.log(c,Date.now() - ss,msg);
msg.should.have.a.property("payload", "World");
(Date.now() - ss).should.be.greaterThan(150);
done();

22
test/nodes/core/logic/10-switch_spec.js
View File

@ -460,7 +460,7 @@ describe('switch Node', function() {
} catch(err) {
done(err);
}
},100)
},500)
});
});
@ -821,4 +821,24 @@ describe('switch Node', function() {
n1.receive({payload:1, parts:{index:0, count:4, id:222}});
});
});
it('should handle invalid jsonata expression', function(done) {
var flow = [{id:"switchNode1",type:"switch",name:"switchNode",property:"$invalidExpression(payload)",propertyType:"jsonata",rules:[{"t":"btwn","v":"$sqrt(16)","vt":"jsonata","v2":"$sqrt(36)","v2t":"jsonata"}],checkall:true,outputs:1,wires:[["helperNode1"]]},
{id:"helperNode1", type:"helper", wires:[]}];
helper.load(switchNode, flow, function() {
var n1 = helper.getNode("switchNode1");
setTimeout(function() {
var logEvents = helper.log().args.filter(function (evt) {
return evt[0].type == "switch";
});
var evt = logEvents[0][0];
evt.should.have.property('id', "switchNode1");
evt.should.have.property('type', "switch");
done();
}, 150);
n1.receive({payload:1});
});
});
});

23
test/nodes/core/logic/15-change_spec.js
View File

@ -15,6 +15,7 @@
**/
var should = require("should");
var sinon = require("sinon");
var changeNode = require("../../../../nodes/core/logic/15-change.js");
var helper = require("node-red-node-test-helper");
@ -454,6 +455,28 @@ describe('change Node', function() {
});
});
it('reports invalid jsonata expression', function(done) {
var flow = [{"id":"changeNode1","type":"change",rules:[{"t":"set","p":"payload","to":"$invalid(payload)","tot":"jsonata"}],"name":"changeNode","wires":[["helperNode1"]]},
{id:"helperNode1", type:"helper", wires:[]}];
helper.load(changeNode, flow, function() {
var changeNode1 = helper.getNode("changeNode1");
var helperNode1 = helper.getNode("helperNode1");
sinon.spy(changeNode1,"error");
helperNode1.on("input", function(msg) {
done("Invalid jsonata expression passed message through");
});
changeNode1.receive({payload:"Hello World!"});
setTimeout(function() {
try {
changeNode1.error.called.should.be.true();
done();
} catch(err) {
done(err);
}
},50);
});
});
});
describe('#change', function() {
it('changes the value of the message property', function(done) {

20
test/red/api/editor/settings_spec.js
View File

@ -59,7 +59,8 @@ describe("api/editor/settings", function() {
},
nodes: {
paletteEditorEnabled: function() { return true; },
getCredentialKeyType: function() { return "test-key-type"}
getCredentialKeyType: function() { return "test-key-type"},
listContextStores: function() { return {default: "foo", stores: ["foo","bar"]}}
},
log: { error: console.error },
storage: {}
@ -73,6 +74,7 @@ describe("api/editor/settings", function() {
}
res.body.should.have.property("httpNodeRoot","testHttpNodeRoot");
res.body.should.have.property("version","testVersion");
res.body.should.have.property("context",{default: "foo", stores: ["foo","bar"]});
res.body.should.have.property("paletteCategories",["red","blue","green"]);
res.body.should.have.property("editorTheme",{test:456});
res.body.should.have.property("testNodeSetting","helloWorld");
@ -95,7 +97,8 @@ describe("api/editor/settings", function() {
},
nodes: {
paletteEditorEnabled: function() { return true; },
getCredentialKeyType: function() { return "test-key-type"}
getCredentialKeyType: function() { return "test-key-type"},
listContextStores: function() { return {default: "foo", stores: ["foo","bar"]}}
},
log: { error: console.error },
storage: {}
@ -130,7 +133,8 @@ describe("api/editor/settings", function() {
},
nodes: {
paletteEditorEnabled: function() { return true; },
getCredentialKeyType: function() { return "test-key-type"}
getCredentialKeyType: function() { return "test-key-type"},
listContextStores: function() { return {default: "foo", stores: ["foo","bar"]}}
},
log: { error: console.error },
storage: {
@ -169,7 +173,8 @@ describe("api/editor/settings", function() {
},
nodes: {
paletteEditorEnabled: function() { return true; },
getCredentialKeyType: function() { return "test-key-type"}
getCredentialKeyType: function() { return "test-key-type"},
listContextStores: function() { return {default: "foo", stores: ["foo","bar"]}}
},
log: { error: console.error },
storage: {
@ -211,7 +216,8 @@ describe("api/editor/settings", function() {
},
nodes: {
paletteEditorEnabled: function() { return true; },
getCredentialKeyType: function() { return "test-key-type"}
getCredentialKeyType: function() { return "test-key-type"},
listContextStores: function() { return {default: "foo", stores: ["foo","bar"]}}
},
log: { error: console.error },
storage: {
@ -248,8 +254,8 @@ describe("api/editor/settings", function() {
},
nodes: {
paletteEditorEnabled: function() { return false; },
getCredentialKeyType: function() { return "test-key-type"}
getCredentialKeyType: function() { return "test-key-type"},
listContextStores: function() { return {default: "foo", stores: ["foo","bar"]}}
},
log: { error: console.error },
storage: {}

182
test/red/runtime/nodes/context/index_spec.js
View File

@ -23,7 +23,7 @@ describe('context', function() {
describe('local memory',function() {
beforeEach(function() {
Context.init({});
return Context.load();
Context.load();
});
afterEach(function() {
Context.clean({allNodes:{}});
@ -124,7 +124,7 @@ describe('context', function() {
});
});
it('enumerates context keys', function() {
it('enumerates context keys - sync', function() {
var context = Context.get("1","flowA");
var keys = context.keys();
@ -142,16 +142,64 @@ describe('context', function() {
keys[1].should.equal("abc");
});
it('should enumerate only context keys when GlobalContext was given', function() {
Context.init({functionGlobalContext: {foo:"bar"}});
return Context.load().then(function(){
it('enumerates context keys - async', function(done) {
var context = Context.get("1","flowA");
var keys = context.global.keys();
var keys = context.keys(function(err,keys) {
keys.should.be.an.Array();
keys.should.be.empty();
context.set("foo","bar");
keys = context.keys(function(err,keys) {
keys.should.have.length(1);
keys[0].should.equal("foo");
context.set("abc.def","bar");
keys = context.keys(function(err,keys) {
keys.should.have.length(2);
keys[1].should.equal("abc");
done();
});
});
});
});
it('should enumerate only context keys when GlobalContext was given - sync', function() {
Context.init({functionGlobalContext: {foo:"bar"}});
Context.load().then(function(){
var context = Context.get("1","flowA");
context.global.set("foo2","bar2");
var keys = context.global.keys();
keys.should.have.length(2);
keys[0].should.equal("foo");
keys[1].should.equal("foo2");
});
});
it('should enumerate only context keys when GlobalContext was given - async', function(done) {
Context.init({functionGlobalContext: {foo:"bar"}});
Context.load().then(function(){
var context = Context.get("1","flowA");
context.global.set("foo2","bar2");
context.global.keys(function(err,keys) {
keys.should.have.length(2);
keys[0].should.equal("foo");
keys[1].should.equal("foo2");
done();
});
}).catch(done);
});
it('returns functionGlobalContext value if store value undefined', function() {
Context.init({functionGlobalContext: {foo:"bar"}});
Context.load().then(function(){
var context = Context.get("1","flowA");
var v = context.global.get('foo');
v.should.equal('bar');
});
})
});
describe('external context storage',function() {
var resourcesDir = path.resolve(path.join(__dirname,"..","resources","context"));
@ -215,6 +263,11 @@ describe('context', function() {
config:{}
}
};
var memoryStorage ={
memory:{
module: "memory"
}
};
afterEach(function() {
sandbox.reset();
@ -226,18 +279,18 @@ describe('context', function() {
describe('load modules',function(){
it('should call open()', function() {
Context.init({contextStorage:contextDefaultStorage});
return Context.load().then(function(){
Context.load().then(function(){
stubOpen.called.should.be.true();
stubOpen2.called.should.be.true();
});
});
it('should load memory module', function() {
Context.init({contextStorage:{memory:{module:"memory"}}});
return Context.load();
Context.load();
});
it('should load localfilesystem module', function() {
Context.init({contextStorage:{file:{module:"localfilesystem",config:{dir:resourcesDir}}}});
return Context.load();
Context.load();
});
it('should accept special storage name', function(done) {
Context.init({
@ -257,7 +310,7 @@ describe('context', function() {
stubSet.calledWithExactly("1:flow","file","file2",cb).should.be.true();
stubSet.calledWithExactly("1:flow","num","num3",cb).should.be.true();
done();
});
}).catch(done);
});
it('should ignore reserved storage name `_`', function(done) {
Context.init({contextStorage:{_:{module:testPlugin}}});
@ -271,7 +324,7 @@ describe('context', function() {
stubGet.called.should.be.false();
stubKeys.called.should.be.false();
done();
});
}).catch(done);
});
it('should fail when using invalid default context', function(done) {
Context.init({contextStorage:{default:"noexist"}});
@ -300,14 +353,15 @@ describe('context', function() {
});
describe('close modules',function(){
it('should call close()', function() {
it('should call close()', function(done) {
Context.init({contextStorage:contextDefaultStorage});
return Context.load().then(function(){
Context.load().then(function(){
return Context.close().then(function(){
stubClose.called.should.be.true();
stubClose2.called.should.be.true();
done();
});
});
}).catch(done);
});
});
@ -324,7 +378,7 @@ describe('context', function() {
stubGet.calledWithExactly("1:flow","foo",cb).should.be.true();
stubKeys.calledWithExactly("1:flow",cb).should.be.true();
done();
});
}).catch(done);
});
it('should store flow property to external context storage',function(done) {
Context.init({contextStorage:contextStorage});
@ -338,7 +392,7 @@ describe('context', function() {
stubGet.calledWithExactly("flow","foo",cb).should.be.true();
stubKeys.calledWithExactly("flow",cb).should.be.true();
done();
});
}).catch(done);
});
it('should store global property to external context storage',function(done) {
Context.init({contextStorage:contextStorage});
@ -349,10 +403,10 @@ describe('context', function() {
context.global.get("foo","test",cb);
context.global.keys("test",cb);
stubSet.calledWithExactly("global","foo","bar",cb).should.be.true();
stubGet.calledWithExactly("global","foo",cb).should.be.true();
stubKeys.calledWithExactly("global",cb).should.be.true();
stubGet.calledWith("global","foo").should.be.true();
stubKeys.calledWith("global").should.be.true();
done();
});
}).catch(done);
});
it('should store data to the default context when non-existent context storage was specified', function(done) {
Context.init({contextStorage:contextDefaultStorage});
@ -369,7 +423,7 @@ describe('context', function() {
stubGet2.calledWithExactly("1:flow","foo",cb).should.be.true();
stubKeys2.calledWithExactly("1:flow",cb).should.be.true();
done();
});
}).catch(done);
});
it('should use the default context', function(done) {
Context.init({contextStorage:contextDefaultStorage});
@ -386,7 +440,7 @@ describe('context', function() {
stubGet2.calledWithExactly("1:flow","foo",cb).should.be.true();
stubKeys2.calledWithExactly("1:flow",cb).should.be.true();
done();
});
}).catch(done);
});
it('should use the alias of default context', function(done) {
Context.init({contextStorage:contextDefaultStorage});
@ -403,7 +457,7 @@ describe('context', function() {
stubGet2.calledWithExactly("1:flow","foo",cb).should.be.true();
stubKeys2.calledWithExactly("1:flow",cb).should.be.true();
done();
});
}).catch(done);
});
it('should use default as the alias of other context', function(done) {
Context.init({contextStorage:contextAlias});
@ -417,22 +471,16 @@ describe('context', function() {
stubGet.calledWithExactly("1:flow","foo",cb).should.be.true();
stubKeys.calledWithExactly("1:flow",cb).should.be.true();
done();
}).catch(done);
});
});
it('should throw an error using undefined storage for local context', function(done) {
it('should not throw an error using undefined storage for local context', function(done) {
Context.init({contextStorage:contextStorage});
Context.load().then(function(){
var context = Context.get("1","flow");
var cb = function(){done("An error occurred")}
context.get("local","nonexist",cb);
should.fail(null, null, "An error was not thrown using undefined storage for local context");
}).catch(function(err) {
if (err.name === "ContextError") {
done();
} else {
done(err);
}
});
done()
}).catch(done);
});
it('should throw an error using undefined storage for flow context', function(done) {
Context.init({contextStorage:contextStorage});
@ -440,39 +488,89 @@ describe('context', function() {
var context = Context.get("1","flow");
var cb = function(){done("An error occurred")}
context.flow.get("flow","nonexist",cb);
should.fail(null, null, "An error was not thrown using undefined storage for flow context");
}).catch(function(err) {
if (err.name === "ContextError") {
done();
}).catch(done);
});
it('should return functionGlobalContext value as a default - synchronous', function(done) {
var fGC = { "foo": 456 };
Context.init({contextStorage:memoryStorage, functionGlobalContext:fGC });
Context.load().then(function() {
var context = Context.get("1","flow");
// Get foo - should be value from fGC
var v = context.global.get("foo");
v.should.equal(456);
// Update foo - should not touch fGC object
context.global.set("foo","new value");
fGC.foo.should.equal(456);
// Get foo - should be the updated value
v = context.global.get("foo");
v.should.equal("new value");
done();
}).catch(done);
})
it('should return functionGlobalContext value as a default - async', function(done) {
var fGC = { "foo": 456 };
Context.init({contextStorage:memoryStorage, functionGlobalContext:fGC });
Context.load().then(function() {
var context = Context.get("1","flow");
// Get foo - should be value from fGC
context.global.get("foo", function(err, v) {
if (err) {
done(err)
} else {
done(err);
v.should.equal(456);
// Update foo - should not touch fGC object
context.global.set("foo","new value", function(err) {
if (err) {
done(err)
} else {
fGC.foo.should.equal(456);
// Get foo - should be the updated value
context.global.get("foo", function(err, v) {
if (err) {
done(err)
} else {
v.should.equal("new value");
done();
}
});
}
});
}
});
}).catch(done);
})
});
describe('delete context',function(){
it('should not call delete() when external context storage is used', function() {
it('should not call delete() when external context storage is used', function(done) {
Context.init({contextStorage:contextDefaultStorage});
return Context.load().then(function(){
Context.load().then(function(){
Context.get("flowA");
return Context.delete("flowA").then(function(){
stubDelete.called.should.be.false();
stubDelete2.called.should.be.false();
done();
});
});
}).catch(done);
});
});
describe('clean context',function(){
it('should call clean()', function() {
it('should call clean()', function(done) {
Context.init({contextStorage:contextDefaultStorage});
return Context.load().then(function(){
Context.load().then(function(){
return Context.clean({allNodes:{}}).then(function(){
stubClean.calledWithExactly([]).should.be.true();
stubClean2.calledWithExactly([]).should.be.true();
done();
});
});
}).catch(done);
});
});
});

4
test/red/runtime/nodes/context/localfilesystem_spec.js
View File

@ -29,7 +29,7 @@ describe('localfilesystem',function() {
});
beforeEach(function() {
context = LocalFileSystem({dir: resourcesDir});
context = LocalFileSystem({dir: resourcesDir, cache: false});
return context.open();
});
@ -308,7 +308,7 @@ describe('localfilesystem',function() {
done();
});
});
});
}).catch(done);
});
});
});

26
test/red/runtime/nodes/context/memory_spec.js
View File

@ -103,19 +103,6 @@ describe('memory',function() {
keysY[0].should.equal("hoge");
});
it('should enumerate only context keys when GlobalContext was given', function() {
var keys = context.keys("global");
keys.should.be.an.Array();
keys.should.be.empty();
var data = {
foo: "bar"
}
context.setGlobalContext(data);
keys = context.keys("global");
keys.should.have.length(1);
keys[0].should.equal("foo");
});
});
describe('#delete',function() {
@ -163,17 +150,4 @@ describe('memory',function() {
});
});
describe('#setGlobalContext',function() {
it('should initialize global context with argument', function() {
var keys = context.keys("global");
keys.should.be.an.Array();
keys.should.be.empty();
var data = {
foo: "bar"
}
context.setGlobalContext(data);
context.get("global","foo").should.equal("bar");
});
});
});

28
test/red/runtime/util_spec.js
View File

@ -307,6 +307,10 @@ describe("red/util", function() {
},{});
result.should.eql("123");
});
it('returns null', function() {
var result = util.evaluateNodeProperty(null,'null');
(result === null).should.be.true();
})
describe('environment variable', function() {
before(function() {
process.env.NR_TEST_A = "foo";
@ -454,6 +458,30 @@ describe("red/util", function() {
var result = util.evaluateJSONataExpression(expr,{payload:"hello"});
should.not.exist(result);
});
it('handles async flow context access', function(done) {
var expr = util.prepareJSONataExpression('$flowContext("foo")',{context:function() { return {flow:{get: function(key,callback) { setTimeout(()=>{callback(null,{'foo':'bar'}[key])},10)}}}}});
util.evaluateJSONataExpression(expr,{payload:"hello"},function(err,value) {
try {
should.not.exist(err);
value.should.eql("bar");
done();
} catch(err2) {
done(err2);
}
});
})
it('handles async global context access', function(done) {
var expr = util.prepareJSONataExpression('$globalContext("foo")',{context:function() { return {global:{get: function(key,callback) { setTimeout(()=>{callback(null,{'foo':'bar'}[key])},10)}}}}});
util.evaluateJSONataExpression(expr,{payload:"hello"},function(err,value) {
try {
should.not.exist(err);
value.should.eql("bar");
done();
} catch(err2) {
done(err2);
}
});
})
});