Initial commit for OpenECOMP SDN-C OA&M

Change-Id: I7ab579fd0d206bf356f36d52dcdf4f71f1fa2680
Signed-off-by: Timoney, Daniel (dt5972) <dtimoney@att.com>

Former-commit-id: 2a9f0edd09581f907e62ec4689b5ac94dd5382ba

1 files changed, 220 insertions, 0 deletions

diff --git a/dgbuilder/red/nodes/flows.js b/dgbuilder/red/nodes/flows.js
new file mode 100644
index 00000000..b0b5d514
--- /dev/null
+++ b/dgbuilder/red/nodes/flows.js
@@ -0,0 +1,220 @@
+/**
+ * Copyright 2014 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 util = require("util");
+var when = require("when");
+
+var typeRegistry = require("./registry");
+var credentials = require("./credentials");
+var log = require("../log");
+var events = require("../events");
+
+var storage = null;
+
+var nodes = {};
+var activeConfig = [];
+var missingTypes = [];
+
+events.on('type-registered',function(type) {
+        if (missingTypes.length > 0) {
+            var i = missingTypes.indexOf(type);
+            if (i != -1) {
+                missingTypes.splice(i,1);
+                util.log("[red] Missing type registered: "+type);
+                if (missingTypes.length === 0) {
+                    parseConfig();
+                }
+            }
+        }
+});
+
+/**
+ * Parses the current activeConfig and creates the required node instances
+ */ 
+function parseConfig() {
+    var i;
+    var nt;
+    missingTypes = [];
+    
+    // Scan the configuration for any unknown node types
+    for (i=0;i<activeConfig.length;i++) {
+        var type = activeConfig[i].type;
+        // TODO: remove workspace in next release+1
+        if (type != "workspace" && type != "tab") {
+            nt = typeRegistry.get(type);
+            if (!nt && missingTypes.indexOf(type) == -1) {
+                missingTypes.push(type);
+            }
+        }
+    }
+    // Abort if there are any missing types
+    if (missingTypes.length > 0) {
+        util.log("[red] Waiting for missing types to be registered:");
+        for (i=0;i<missingTypes.length;i++) {
+            util.log("[red]  - "+missingTypes[i]);
+        }
+        return;
+    }
+
+    util.log("[red] Starting flows");
+    events.emit("nodes-starting");
+    
+    // Instantiate each node in the flow
+    for (i=0;i<activeConfig.length;i++) {
+        var nn = null;
+        // TODO: remove workspace in next release+1
+        if (activeConfig[i].type != "workspace" && activeConfig[i].type != "tab") {
+            nt = typeRegistry.get(activeConfig[i].type);
+            if (nt) {
+                try {
+                    nn = new nt(activeConfig[i]);
+                }
+                catch (err) {
+                    util.log("[red] "+activeConfig[i].type+" : "+err);
+                }
+            }
+            // console.log(nn);
+            if (nn === null) {
+                util.log("[red] unknown type: "+activeConfig[i].type);
+            }
+        }
+    }
+    // Clean up any orphaned credentials
+    credentials.clean(flowNodes.get);
+    events.emit("nodes-started");
+}
+
+/**
+ * Stops the current activeConfig
+ */
+function stopFlows() {
+    if (activeConfig&&activeConfig.length > 0) {
+        util.log("[red] Stopping flows");
+    }
+    return flowNodes.clear();
+}
+
+var flowNodes = module.exports = {
+    init: function(_storage) {
+        storage = _storage;
+    },
+    
+    /**
+     * Load the current activeConfig from storage and start it running
+     * @return a promise for the loading of the config
+     */
+    load: function() {
+        return storage.getFlows().then(function(flows) {
+            return credentials.load().then(function() {
+                activeConfig = flows;
+                if (activeConfig && activeConfig.length > 0) {
+                    parseConfig();
+                }
+            });
+        }).otherwise(function(err) {
+            util.log("[red] Error loading flows : "+err);
+        });
+    },
+    
+    /**
+     * Add a node to the current active set
+     * @param n the node to add
+     */
+    add: function(n) {
+        nodes[n.id] = n;
+        n.on("log",log.log);
+    },
+    
+    /**
+     * Get a node
+     * @param i the node id
+     * @return the node
+     */
+    get: function(i) {
+        return nodes[i];
+    },
+    
+    /**
+     * Stops all active nodes and clears the active set
+     * @return a promise for the stopping of all active nodes
+     */
+    clear: function() {
+        return when.promise(function(resolve) {
+            events.emit("nodes-stopping");
+            var promises = [];
+            for (var n in nodes) {
+                if (nodes.hasOwnProperty(n)) {
+                    try {
+                        var p = nodes[n].close();
+                        if (p) {
+                            promises.push(p);
+                        }
+                    } catch(err) {
+                        nodes[n].error(err);
+                    }
+                }
+            }
+            when.settle(promises).then(function() {
+                events.emit("nodes-stopped");
+                nodes = {};
+                resolve();
+            });
+        });
+    },
+    
+    /**
+     * Provides an iterator over the active set of nodes
+     * @param cb a function to be called for each node in the active set
+     */
+    each: function(cb) {
+        for (var n in nodes) {
+            if (nodes.hasOwnProperty(n)) {
+                cb(nodes[n]);
+            }
+        }
+    },
+
+    /**
+     * @return the active configuration
+     */
+    getFlows: function() {
+        return activeConfig;
+    },
+    
+    /**
+     * Sets the current active config.
+     * @param config the configuration to enable
+     * @return a promise for the starting of the new flow
+     */
+    setFlows: function (config) {
+        // Extract any credential updates
+        for (var i=0; i<config.length; i++) {
+            var node = config[i];
+            if (node.credentials) {
+                credentials.extract(node);
+                delete node.credentials;
+            }
+        }
+        return credentials.save()
+            .then(function() { return storage.saveFlows(config);})
+            .then(function() { return stopFlows();})
+            .then(function () {
+                activeConfig = config;
+                parseConfig();
+            });
+    },
+    stopFlows: stopFlows
+};