Add session awareness to TCP nodes

Closes #63,#65

nodes/core/io/31-tcpin.html

@@ -118,11 +118,12 @@
 <script type="text/x-red" data-template-name="tcp out">
     <div class="form-row">
         <label for="node-input-beserver"><i class="icon-resize-small"></i> Type</label>
-        <select id="node-input-beserver" style="width:120px; margin-right:5px;">
+        <select id="node-input-beserver" style="width:150px; margin-right:5px;">
             <option value="server">Listen on</option>
             <option value="client">Connect to</option>
+            <option value="reply">Reply to TCP</option>
         </select>
-        port <input type="text" id="node-input-port" style="width: 50px">
+        <span id="node-input-port-row">port <input type="text" id="node-input-port" style="width: 50px"></span>
     </div>
     
     <div class="form-row hidden" id="node-input-host-row" style="padding-left: 110px;"> 
@@ -143,7 +144,7 @@
 
 <script type="text/x-red" data-help-name="tcp out">
 	<p>Provides a choice of tcp outputs. Can either connect to a remote tcp port,
-	or accept incoming connections.</p>
+    accept incoming connections, or reply to messages received from a TCP In node.</p>
 	<p>Only <b>msg.payload</b> is sent.</p>
 	<p>If <b>msg.payload</b> is a string containing a base64 encoding of binary
 	data, the Base64 decoding option will cause it to be converted back to binary
@@ -155,8 +156,8 @@
         category: 'output',
         color:"Silver",
         defaults: {
-            host: {value:"",validate:function(v) { return (this.beserver == "server")||v.length > 0;} },
+            host: {value:"",validate:function(v) { return (this.beserver != "client")||v.length > 0;} },
-            port: {value:"",required:true},
+            port: {value:"",validate:function(v) { return (this.beserver == "reply")||RED.validators.number()(v) } },
             beserver: {value:"client",required:true},
             base64: {value:false,required:true},
             name: {value:""}
@@ -174,6 +175,13 @@
         oneditprepare: function() {
             var updateOptions = function() {
                 var sockettype = $("#node-input-beserver option:selected").val();
+                if (sockettype == "reply") {
+                    $("#node-input-port-row").hide();
+                    $("#node-input-host-row").hide();
+                } else {
+                    $("#node-input-port-row").show();
+                }
+                    
                 if (sockettype == "client") {
                     $("#node-input-host-row").show();
                 } else {

nodes/core/io/31-tcpin.js

@@ -18,6 +18,10 @@ var RED = require(process.env.NODE_RED_HOME+"/red/red");
 var reconnectTime = RED.settings.socketReconnectTime||10000;
 var net = require('net');
 
+var connectionPool = {};
+
+
+
 function TcpIn(n) {
     RED.nodes.createNode(this,n);
     this.host = n.host;
@@ -37,10 +41,12 @@ function TcpIn(n) {
         var reconnectTimeout;
         function setupTcpClient() {
             node.log("connecting to "+node.host+":"+node.port);
+            var id = (1+Math.random()*4294967295).toString(16);
             client = net.connect(node.port, node.host, function() {
                     buffer = (node.datatype == 'buffer')? new Buffer(0):"";
                     node.log("connected to "+node.host+":"+node.port);
             });
+            connectionPool[id] = client;
             
             client.on('data', function (data) {
                     if (node.datatype != 'buffer') {
@@ -52,11 +58,13 @@ function TcpIn(n) {
                             var parts = buffer.split(node.newline);
                             for (var i = 0;i<parts.length-1;i+=1) {
                                 var msg = {topic:node.topic, payload:parts[i]};
+                                msg._session = {type:"tcp",id:id};
                                 node.send(msg);
                             }
                             buffer = parts[parts.length-1];
                         } else {
                             var msg = {topic:node.topic, payload:data};
+                            msg._session = {type:"tcp",id:id};
                             node.send(msg);
                         }
                     } else {
@@ -70,12 +78,14 @@ function TcpIn(n) {
             client.on('end', function() {
                     if (!node.stream || (node.datatype == "utf8" && node.newline != "" && buffer.length > 0)) {
                         var msg = {topic:node.topic,payload:buffer};
+                        msg._session = {type:"tcp",id:id};
                         node.send(msg);
                         buffer = null;
                     }
             });
 
             client.on('close', function() {
+                    delete connectionPool[id];
                     node.log("connection lost to "+node.host+":"+node.port);
                     if (!node.closing) {
                         reconnectTimeout = setTimeout(setupTcpClient, reconnectTime);
@@ -95,6 +105,9 @@ function TcpIn(n) {
         });
     } else {
         var server = net.createServer(function (socket) {
+                var id = (1+Math.random()*4294967295).toString(16);
+                connectionPool[id] = socket;
+                
                 var buffer = (node.datatype == 'buffer')? new Buffer(0):"";
                 socket.on('data', function (data) {
                         if (node.datatype != 'buffer') {
@@ -107,11 +120,13 @@ function TcpIn(n) {
                                 var parts = buffer.split(node.newline);
                                 for (var i = 0;i<parts.length-1;i+=1) {
                                     var msg = {topic:node.topic, payload:parts[i],ip:socket.remoteAddress,port:socket.remotePort};
+                                    msg._session = {type:"tcp",id:id};
                                     node.send(msg);
                                 }
                                 buffer = parts[parts.length-1];
                             } else {
                                 var msg = {topic:node.topic, payload:data};
+                                msg._session = {type:"tcp",id:id};
                                 node.send(msg);
                             }
                         } else {
@@ -125,10 +140,14 @@ function TcpIn(n) {
                 socket.on('end', function() {
                         if (!node.stream || (node.datatype == "utf8" && node.newline != "" && buffer.length > 0)) {
                             var msg = {topic:node.topic,payload:buffer};
+                            msg._session = {type:"tcp",id:id};
                             node.send(msg);
                             buffer = null;
                         }
                 });
+                socket.on('close', function() {
+                    delete connectionPool[id];
+                });
                 socket.on('error',function(err) {
                     node.log(err);
                 });
@@ -206,6 +225,21 @@ function TcpOut(n) {
             clearTimeout(reconnectTimeout);
         });
         
+    } else if (node.beserver == "reply") {
+        node.on("input",function(msg) {
+            if (msg._session && msg._session.type == "tcp") {
+                var client = connectionPool[msg._session.id];
+                if (client) {
+                    if (Buffer.isBuffer(msg.payload)) {
+                        client.write(msg.payload);
+                    } else if (typeof msg.payload === "string" && node.base64) {
+                        client.write(new Buffer(msg.payload,'base64'));
+                    } else {
+                        client.write(new Buffer(""+msg.payload));
+                    }
+                }
+            }
+        });
     } else {
         var connectedSockets = [];
         var server = net.createServer(function (socket) {