Flow debugger initial pass

red.js

@@ -37,7 +37,8 @@ var knownOpts = {
     "userDir":[path],
     "port": Number,
     "v": Boolean,
-    "help": Boolean
+    "help": Boolean,
+    "debug": Boolean
 };
 var shortHands = {
     "s":["--settings"],
@@ -120,6 +121,10 @@ if (parsedArgs.v) {
     settings.verbose = true;
 }
 
+if (parsedArgs.debug) {
+    settings.enableDebugger = true;
+}
+
 if (settings.https) {
     server = https.createServer(settings.https,function(req,res){app(req,res);});
 } else {

red/runtime/nodes/Node.js

@@ -29,6 +29,7 @@ function Node(n) {
     this.type = n.type;
     this.z = n.z;
     this._closeCallbacks = [];
+    this._config = n;
 
     if (n.name) {
         this.name = n.name;

red/runtime/nodes/debugger/index.js

@@ -0,0 +1,262 @@
+/**
+ * Copyright 2016 IBM Corp.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ **/
+
+var readline = require('readline');
+var rl;
+
+var runtime;
+var components;
+
+var breakpoints = {};
+
+var enabled = false;
+
+function enable() {
+    enabled = true;
+}
+function disable() {
+    enabled = false;
+}
+
+function formatNode(n) {
+    return n.id+" ["+n.type+"]"+(n.name?" "+n.name:"");
+}
+function listNodes() {
+    var flowIds = runtime.nodes.listFlows();
+    flowIds.forEach(function(id) {
+        var flow = runtime.nodes.getFlow(id);
+        console.log(flow.id+" [flow]"+(flow.label?" "+flow.label:""));
+        if (flow.subflows) {
+            flow.subflows.forEach(function(sf) {
+                console.log("  - "+formatNode(sf));
+                sf.configs.forEach(function(n) {
+                    console.log("     - "+formatNode(n));
+                })
+                sf.nodes.forEach(function(n) {
+                    console.log("     - "+formatNode(n));
+                })
+
+            })
+        }
+        if (flow.configs) {
+            flow.configs.forEach(function(n) {
+                console.log("  - "+formatNode(n));
+            })
+        }
+        if (flow.nodes) {
+            flow.nodes.forEach(function(n) {
+                console.log("  - "+formatNode(n));
+            })
+        }
+
+    })
+}
+function listNode(id) {
+    console.log(id);
+    var node = runtime.nodes.getNode(id);
+    if (node) {
+        console.log(node);
+    }
+}
+function listFlows() {
+}
+
+function handleAddBreakpoint(args) {
+    var valid = false;
+    var bp = {};
+    if (args.length > 0) {
+        bp.node = args[0];
+        valid = true;
+        if (args.length > 1) {
+            if (args[1]==='i' || args[1]==='o') {
+                bp.type = args[1];
+                if (args.length > 2) {
+                    bp.index = args[2];
+                }
+            } else {
+                valid = false;
+            }
+        }
+    }
+    if (valid) {
+        var id = addBreakpoint(bp);
+        console.log("Added breakpoint",id?id:"");
+    } else {
+        console.log("break <nodeid> [i|o] <index>");
+    }
+}
+
+function handleRemoveBreakpoint(args) {
+    if (args.length === 0) {
+        breakpoints = {};
+        console.log("Cleared all breakpoints");
+        return;
+    }
+    if (args.length === 1 && breakpoints.hasOwnProperty(args[0])) {
+        delete breakpoints[args[0]];
+        console.log("Cleared all breakpoints on node",args[0]);
+        return;
+    }
+    if (args.length === 2 && breakpoints.hasOwnProperty(args[0]) && (args[1] === 'i' || args[1] === 'o')) {
+        var c = Object.keys(breakpoints[args[0]][args[1]]).length;
+        breakpoints[args[0]][args[1]] = {};
+        breakpoints[args[0]].c -= c;
+        if (breakpoints[args[0]].c === 0) {
+            delete breakpoints[args[0]];
+        }
+        console.log("Cleared all",(args[1]==='i'?'input':'output'),"breakpoints on node",args[0]);
+        return;
+    }
+    if (args.length === 3 && breakpoints.hasOwnProperty(args[0]) && (args[1] === 'i' || args[1] === 'o')) {
+        var id = args.join(":");
+        if (removeBreakpoint(args.join(":"))) {
+            console.log("Cleared breakpoint",id);
+        }
+    }
+}
+function handleInfo(args) {
+    if (args[0] === 'breakpoints') {
+        console.log(breakpoints);
+    }
+}
+function init(_runtime,_components) {
+    runtime = _runtime;
+    components = _components;
+    breakpoints = {};
+
+
+    enabled = true;
+
+    rl = readline.createInterface({
+        input: process.stdin,
+        output: process.stdout,
+        prompt: 'nrdb> '
+    });
+    rl.on('line', function(input) {
+        input = input.trim();
+        var parts = input.split(/\s+/);
+        var cmd = parts[0];
+        var args = parts.slice(1)
+
+        switch(cmd) {
+            case 'pause':
+            case 'p':
+                console.log("pausing");
+                components.router.pause();
+                break;
+            case 'continue':
+            case 'c':
+                console.log("resuming");
+                components.router.resume();
+                break;
+            case 'nodes': listNodes(); break;
+            case 'flows': listFlows(); break;
+            case 'node': if (args.length === 1) { listNode(args[0]); } break;
+            case 'break': handleAddBreakpoint(args); break;
+            case 'delete': handleRemoveBreakpoint(args); break;
+            case 'info': handleInfo(args); break;
+        }
+        var m = /^node\s+([^\s]+)$/.exec(input);
+        if (m) {
+            listNode(m[1]);
+        }
+        rl.prompt();
+    }).on('close', function() {
+        console.log("exiting nrdb");
+        runtime.stop().then(function() {
+            process.exit(0);
+        });
+    })
+    rl.setPrompt('nrdb> ');
+    rl.prompt();
+
+}
+
+/*
+ * {
+ *    node: id,
+ *    type: 'i/o',
+ *    index: 0
+ * }
+ */
+function addBreakpoint(bp) {
+    if (!bp.hasOwnProperty('node')) {
+        return;
+    }
+    var node = runtime.nodes.getNode(bp.node);
+    if (!node || !node._config.hasOwnProperty('wires')) {
+        return;
+    }
+
+    if (!bp.hasOwnProperty('type')) {
+        addBreakpoint({node:bp.node,type:'i'});
+        addBreakpoint({node:bp.node,type:'o'});
+    } else if (!bp.hasOwnProperty('index')) {
+        if (bp.type === 'i') {
+            addBreakpoint({node:bp.node,type:'i',index:0});
+        } else {
+            for (var i=0;i<node._config.wires.length;i++) {
+                addBreakpoint({node:bp.node,type:'o',index:i});
+            }
+        }
+        return;
+    } else {
+        breakpoints[bp.node] = breakpoints[bp.node] || {c:0,i:{},o:{}};
+        if (!breakpoints[bp.node][bp.type][bp.index]) {
+            breakpoints[bp.node][bp.type][bp.index] = true;
+            breakpoints[bp.node].c++;
+        }
+        return bp.node+":"+bp.type+":"+bp.index;
+    }
+}
+function removeBreakpoint(id) {
+    var m = /^(.+):([io]):(\d+)$/.exec(id);
+    if (m) {
+        var node = m[1];
+        var portType = m[2];
+        var portIndex = m[3];
+        if (breakpoints[node]) {
+            delete breakpoints[node][portType][portIndex];
+            breakpoints[node].c--;
+            if (breakpoints[node].c === 0) {
+                delete breakpoints[node];
+            }
+            return true;
+        }
+    }
+    return false;
+}
+
+function getBreakpoints() {
+    return breakpoints;
+}
+
+function checkSendEvent(evt) {
+    return (enabled && (
+        (breakpoints.hasOwnProperty(evt.sourceNode.id) && breakpoints[evt.sourceNode.id].o[evt.sourcePort]) ||
+        (breakpoints.hasOwnProperty(evt.destinationNode.id) && breakpoints[evt.destinationNode.id].i[evt.destinationPort])
+    ));
+}
+
+module.exports = {
+    init:init,
+    enable: enable,
+    disable: disable,
+    addBreakpoint:addBreakpoint,
+    removeBreakpoint: removeBreakpoint,
+    getBreakpoints: getBreakpoints,
+    checkSendEvent: checkSendEvent
+}

red/runtime/nodes/flows/index.js

@@ -586,6 +586,10 @@ function removeFlow(id) {
     });
 }
 
+function listFlows() {
+    return Object.keys(activeFlows);
+}
+
 module.exports = {
     init: init,
 
@@ -633,6 +637,7 @@ module.exports = {
     getFlow: getFlow,
     updateFlow: updateFlow,
     removeFlow: removeFlow,
+    listFlows: listFlows,
     disableFlow:null,
     enableFlow:null
 

red/runtime/nodes/index.js

@@ -19,7 +19,9 @@ var credentials = require("./credentials");
 var flows = require("./flows");
 var context = require("./context");
 var Node = require("./Node");
+var router = require("./router");
 var log = require("../log");
+var nrdb = require("./debugger");
 
 var events = require("../events");
 
@@ -77,6 +79,10 @@ function init(runtime) {
     flows.init(runtime);
     registry.init(runtime);
     context.init(runtime.settings);
+    router.init(runtime);
+    if (settings.enableDebugger) {
+        nrdb.init(runtime,{router:router});
+    }
 }
 
 function disableNode(id) {
@@ -134,6 +140,7 @@ module.exports = {
     stopFlows:  flows.stopFlows,
     setFlows:   flows.setFlows,
     getFlows:   flows.getFlows,
+    listFlows:  flows.listFlows,
 
     addFlow:     flows.addFlow,
     getFlow:     flows.getFlow,
@@ -142,10 +149,12 @@ module.exports = {
     // disableFlow: flows.disableFlow,
     // enableFlow:  flows.enableFlow,
 
-
     // Credentials
     addCredentials: credentials.add,
     getCredentials: credentials.get,
     deleteCredentials: credentials.delete,
     getCredentialDefinition: credentials.getDefinition
+
+    // Router
+
 };

red/runtime/nodes/router/index.js

@@ -18,25 +18,55 @@ var util = require("util");
 
 var flows = require("../flows");
 var redUtil = require("../../util");
-
+var redDebugger = require("../debugger");
 var runtime;
 
 var routes = {};
 var sendQueue = [];
-
+var paused = false;
-
 
 function init(_runtime) {
     runtime = _runtime;
     wires = {};
 }
 
+
+function pause() {
+    paused = true;
+}
+function resume() {
+    paused = false;
+    setImmediate(processSendEvent);
+}
+
+
 function add(sourceNode, wires) {
     routes[sourceNode.id] = wires;
 }
 function remove(sourceNode) {
     delete routes[sourceNode.id];
 }
+
+function processSendEvent() {
+    if (!paused) {
+        if (sendQueue.length > 0) {
+            var sendEvent = sendQueue.shift();
+            //console.log(ev.sourceNode.id+"["+ev.sourcePort+"] -> "+ev.destinationNode.id+"["+ev.destinationPort+"] : "+redDebugger.checkSendEvent(ev));
+            if (!sendEvent.triggered && redDebugger.checkSendEvent(sendEvent)) {
+                sendEvent.triggered = true;
+                sendQueue.unshift(sendEvent);
+                pause();
+
+            } else {
+                sendEvent.destinationNode.receive(sendEvent.msg);
+            }
+        }
+        if (!paused && sendQueue.length > 0) {
+            setImmediate(processSendEvent);
+        }
+    }
+}
+
 function send(sourceNode, msg) {
     if (msg === null || typeof msg === "undefined") {
         return;
@@ -73,13 +103,20 @@ function send(sourceNode, msg) {
                                     if (!sentMessageId) {
                                         sentMessageId = m._msgid;
                                     }
+                                    var sendEvent = {
+                                        sourceNode: sourceNode,
+                                        sourcePort:i,
+                                        destinationNode:node,
+                                        destinationPort:0
+                                    }
+
                                     if (msgSent) {
-                                        var clonedmsg = redUtil.cloneMessage(m);
+                                        sendEvent.msg = redUtil.cloneMessage(m);
-                                        sendEvents.push({n:node,m:clonedmsg});
                                     } else {
-                                        sendEvents.push({n:node,m:m});
+                                        sendEvent.msg = m;
                                         msgSent = true;
                                     }
+                                    sendEvents.push(sendEvent);
                                 }
                             }
                         }
@@ -91,16 +128,16 @@ function send(sourceNode, msg) {
         if (!sentMessageId) {
             sentMessageId = redUtil.generateId();
         }
-        sourceNode.metric("send",{_msgid:sentMessageId});
-
         for (i=0;i<sendEvents.length;i++) {
             var ev = sendEvents[i];
             /* istanbul ignore else */
-            if (!ev.m._msgid) {
+            if (!ev.msg._msgid) {
-                ev.m._msgid = sentMessageId;
+                ev.msg._msgid = sentMessageId;
             }
-            ev.n.receive(ev.m);
+            sendQueue.push(ev);
         }
+        sourceNode.metric("send",{_msgid:sentMessageId});
+        processSendEvent();
     }
 }
 
@@ -108,5 +145,7 @@ module.exports = {
     init:init,
     add:add,
     remove: remove,
-    send:send
+    send:send,
+    pause:pause,
+    resume:resume
 }