Add comms module

red/comms.js

@@ -0,0 +1,86 @@
+/**
+ * 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 ws = require("ws");
+var util = require("util");
+
+var server;
+var settings;
+
+var wsServer;
+var activeConnections = [];
+
+var heartbeatTimer;
+var lastSentTime;
+
+
+function init(_server,_settings) {
+    server = _server;
+    settings = _settings;
+}
+
+function start() {
+    var path = settings.httpAdminRoot || "/";
+    path = path + (path.slice(-1) == "/" ? "":"/") + "comms";
+    wsServer = new ws.Server({server:server,path:path});
+    
+    wsServer.on('connection',function(ws) {
+        activeConnections.push(ws);
+        ws.on('close',function() {
+            for (var i=0;i<activeConnections.length;i++) {
+                if (activeConnections[i] === ws) {
+                    activeConnections.splice(i,1);
+                    break;
+                }
+            }
+        });
+        ws.on('error', function(err) {
+            util.log("[red:comms] error : "+err.toString());
+        });
+    });
+    
+    wsServer.on('error', function(err) {
+        util.log("[red:comms] server error : "+err.toString());
+    });
+     
+    lastSentTime = Date.now();
+    
+    heartbeatTimer = setInterval(function() {
+        var now = Date.now();
+        if (now-lastSentTime > 15000) {
+            lastSentTime = now;
+            publish("hb",lastSentTime);
+        }
+    }, 15000);
+}
+
+function publish(topic,data) {
+    var msg = JSON.stringify({topic:topic,data:data});
+    activeConnections.forEach(function(conn) {
+        try {
+            conn.send(msg);
+        } catch(err) {
+            util.log("[red:comms] send error : "+err.toString());
+        }
+    });
+}
+
+
+module.exports = {
+    init:init,
+    start:start,
+    publish:publish,
+}

red/red.js

@@ -18,6 +18,7 @@ var events = require("./events");
 var server = require("./server");
 var nodes = require("./nodes");
 var library = require("./library");
+var comms = require("./comms");
 var log = require("./log");
 var fs = require("fs");
 var settings = null;
@@ -50,6 +51,7 @@ var RED = {
     stop: server.stop,
     nodes: nodes,
     library: library,
+    comms: comms,
     events: events,
     log: log
 };

red/server.js

@@ -20,6 +20,7 @@ var when = require('when');
 
 var createUI = require("./ui");
 var redNodes = require("./nodes");
+var comms = require("./comms");
 
 var app = null;
 var nodeApp = null;
@@ -30,6 +31,7 @@ var storage = null;
 function createServer(_server,_settings) {
     server = _server;
     settings = _settings;
+    comms.init(_server,_settings);
     storage = require("./storage");
     app = createUI(settings);
     nodeApp = express();
@@ -87,6 +89,7 @@ function start() {
             
             redNodes.loadFlows();
         });
+        comms.start();
     });
     
     return defer.promise;