dgbuilder/core_nodes/io/22-websocket.js

   1 /**\r
   2  * Copyright 2013 IBM Corp.\r
   3  *\r
   4  * Licensed under the Apache License, Version 2.0 (the "License");\r
   5  * you may not use this file except in compliance with the License.\r
   6  * You may obtain a copy of the License at\r
   7  *\r
   8  * http://www.apache.org/licenses/LICENSE-2.0\r
   9  *\r
  10  * Unless required by applicable law or agreed to in writing, software\r
  11  * distributed under the License is distributed on an "AS IS" BASIS,\r
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r
  13  * See the License for the specific language governing permissions and\r
  14  * limitations under the License.\r
  15  **/\r
  16 \r
  17 module.exports = function(RED) {\r
  18     "use strict";\r
  19     var ws = require("ws"),\r
  20         inspect = require("sys").inspect;\r
  21 \r
  22     // A node red node that sets up a local websocket server\r
  23     function WebSocketListenerNode(n) {\r
  24         // Create a RED node\r
  25         RED.nodes.createNode(this,n);\r
  26 \r
  27         var node = this;\r
  28 \r
  29         // Store local copies of the node configuration (as defined in the .html)\r
  30         node.path = n.path;\r
  31         node.wholemsg = (n.wholemsg === "true");\r
  32 \r
  33         node._inputNodes = [];    // collection of nodes that want to receive events\r
  34 \r
  35         var path = RED.settings.httpNodeRoot || "/";\r
  36         path = path + (path.slice(-1) == "/" ? "":"/") + (node.path.charAt(0) == "/" ? node.path.substring(1) : node.path);\r
  37 \r
  38         // Workaround https://github.com/einaros/ws/pull/253\r
  39         // Listen for 'newListener' events from RED.server\r
  40         node._serverListeners = {};\r
  41 \r
  42         var storeListener = function(/*String*/event,/*function*/listener){\r
  43             if(event == "error" || event == "upgrade" || event == "listening"){\r
  44                 node._serverListeners[event] = listener;\r
  45             }\r
  46         }\r
  47 \r
  48         node._clients = {};\r
  49 \r
  50         RED.server.addListener('newListener',storeListener);\r
  51 \r
  52         // Create a WebSocket Server\r
  53         node.server = new ws.Server({server:RED.server,path:path});\r
  54 \r
  55         // Workaround https://github.com/einaros/ws/pull/253\r
  56         // Stop listening for new listener events\r
  57         RED.server.removeListener('newListener',storeListener);\r
  58 \r
  59         node.server.on('connection', function(socket){\r
  60             var id = (1+Math.random()*4294967295).toString(16);\r
  61             node._clients[id] = socket;\r
  62             socket.on('close',function() {\r
  63                 delete node._clients[id];\r
  64             });\r
  65             socket.on('message',function(data,flags){\r
  66                 node.handleEvent(id,socket,'message',data,flags);\r
  67             });\r
  68             socket.on('error', function(err) {\r
  69                 node.warn("An error occured on the ws connection: "+inspect(err));\r
  70             });\r
  71         });\r
  72 \r
  73         node.on("close", function() {\r
  74             // Workaround https://github.com/einaros/ws/pull/253\r
  75             // Remove listeners from RED.server\r
  76             var listener = null;\r
  77             for(var event in node._serverListeners) {\r
  78                 if (node._serverListeners.hasOwnProperty(event)) {\r
  79                     listener = node._serverListeners[event];\r
  80                     if(typeof listener === "function"){\r
  81                         RED.server.removeListener(event,listener);\r
  82                     }\r
  83                 }\r
  84             }\r
  85             node._serverListeners = {};\r
  86             node.server.close();\r
  87             node._inputNodes = [];\r
  88         });\r
  89     }\r
  90     RED.nodes.registerType("websocket-listener",WebSocketListenerNode);\r
  91 \r
  92     WebSocketListenerNode.prototype.registerInputNode = function(/*Node*/handler){\r
  93         this._inputNodes.push(handler);\r
  94     }\r
  95 \r
  96     WebSocketListenerNode.prototype.handleEvent = function(id,/*socket*/socket,/*String*/event,/*Object*/data,/*Object*/flags){\r
  97         var msg;\r
  98         if (this.wholemsg) {\r
  99             try {\r
 100                 msg = JSON.parse(data);\r
 101             }\r
 102             catch(err) {\r
 103                 msg = { payload:data };\r
 104             }\r
 105         } else {\r
 106             msg = {\r
 107                 payload:data\r
 108             };\r
 109         }\r
 110         msg._session = {type:"websocket",id:id};\r
 111 \r
 112         for (var i = 0; i < this._inputNodes.length; i++) {\r
 113             this._inputNodes[i].send(msg);\r
 114         }\r
 115     }\r
 116 \r
 117     WebSocketListenerNode.prototype.broadcast = function(data){\r
 118         try {\r
 119             for (var i = 0; i < this.server.clients.length; i++) {\r
 120                 this.server.clients[i].send(data);\r
 121             }\r
 122         }\r
 123         catch(e) { // swallow any errors\r
 124             this.warn("ws:"+i+" : "+e);\r
 125         }\r
 126     }\r
 127 \r
 128     WebSocketListenerNode.prototype.send = function(id,data) {\r
 129         var session = this._clients[id];\r
 130         if (session) {\r
 131             try {\r
 132                 session.send(data);\r
 133             }\r
 134             catch(e) { // swallow any errors\r
 135             }\r
 136         }\r
 137     }\r
 138 \r
 139     function WebSocketInNode(n) {\r
 140         RED.nodes.createNode(this,n);\r
 141         this.server = n.server;\r
 142         var node = this;\r
 143         this.serverConfig = RED.nodes.getNode(this.server);\r
 144         if (this.serverConfig) {\r
 145             this.serverConfig.registerInputNode(this);\r
 146         } else {\r
 147             this.error("Missing server configuration");\r
 148         }\r
 149     }\r
 150     RED.nodes.registerType("websocket in",WebSocketInNode);\r
 151 \r
 152     function WebSocketOutNode(n) {\r
 153         RED.nodes.createNode(this,n);\r
 154         var node = this;\r
 155         this.server = n.server;\r
 156         this.serverConfig = RED.nodes.getNode(this.server);\r
 157         if (!this.serverConfig) {\r
 158             this.error("Missing server configuration");\r
 159         }\r
 160         this.on("input", function(msg) {\r
 161             var payload;\r
 162             if (this.serverConfig.wholemsg) {\r
 163                 delete msg._session;\r
 164                 payload = JSON.stringify(msg);\r
 165             } else {\r
 166                 if (!Buffer.isBuffer(msg.payload)) { // if it's not a buffer make sure it's a string.\r
 167                     payload = RED.util.ensureString(msg.payload);\r
 168                 }\r
 169                 else {\r
 170                     payload = msg.payload;\r
 171                 }\r
 172             }\r
 173             if (msg._session && msg._session.type == "websocket") {\r
 174                 node.serverConfig.send(msg._session.id,payload);\r
 175             } else {\r
 176                 node.serverConfig.broadcast(payload,function(error){\r
 177                     if (!!error) {\r
 178                         node.warn("An error occurred while sending:" + inspect(error));\r
 179                     }\r
 180                 });\r
 181             }\r
 182         });\r
 183     }\r
 184     RED.nodes.registerType("websocket out",WebSocketOutNode);\r
 185 }\r