Make TCP nodes more intuitive

The edit dialog for the TCP nodes is (IMHO) clearer to use now
and provides options for how to handle the incoming messages.

The general error handling to much improved as well.

nodes/io/31-tcpin.html

@@ -16,22 +16,36 @@
 
 <script type="text/x-red" data-template-name="tcp in">
     <div class="form-row">
-        <label for="node-input-host"><i class="icon-bookmark"></i> Host</label>
+        <label for="node-input-server"><i class="icon-resize-small"></i> Type</label>
-        <input type="text" id="node-input-host" placeholder="localhost" style="width: 40%;" >
+        <select id="node-input-server" style="width:120px; margin-right:5px;">
+            <option value="server">Listen on</option>
+            <option value="client">Connect to</option>
+        </select>
+        port <input type="text" id="node-input-port" style="width: 50px">
+    </div>
+    <div class="form-row hidden" id="node-input-host-row" style="padding-left: 110px;"> 
+        at host <input type="text" id="node-input-host" placeholder="localhost" style="width: 40%;">
+    </div>
 
-        <label for="node-input-port" style="margin-left: 10px; width: 35px;"> Port</label>
-        <input type="text" id="node-input-port" placeholder="Port" style="width: 45px">
-    </div>
     <div class="form-row">
-        <label>&nbsp;</label>
+        <label><i class="icon-th"></i> Output</label>
-        <input type="checkbox" id="node-input-server" placeholder="server" style="display: inline-block; width: auto; vertical-align: top;">
+        a 
-        <label for="node-input-server" style="width: 70%;">Be a server socket ?</label>
+        <select id="node-input-datamode" style="width:110px;">
+            <option value="stream">stream of</option>
+            <option value="single">single</option>
+        </select>
+        <select id="node-input-datatype" style="width:140px;">
+            <option value="buffer">Buffer</option>
+            <option value="utf8">String</option>
+            <option value="base64">Base64 String</option>
+        </select>
+        payload<span id="node-input-datamode-plural">s</span>
     </div>
-    <div class="form-row">
+
-        <label>&nbsp;</label>
+    <div id="node-row-newline" class="form-row hidden" style="padding-left: 110px;">
-        <input type="checkbox" id="node-input-base64" placeholder="base64" style="display: inline-block; width: auto; vertical-align: top;">
+        delimited by <input type="text" id="node-input-newline" style="width: 110px;">
-        <label for="node-input-base64" style="width: 70%;">Base64 encode payload ?</label>
     </div>
+    
     <div class="form-row">
         <label for="node-input-topic"><i class="icon-tasks"></i> Topic</label>
         <input type="text" id="node-input-topic" placeholder="Topic">
@@ -40,15 +54,11 @@
         <label for="node-input-name"><i class="icon-tag"></i> Name</label>
         <input type="text" id="node-input-name" placeholder="Name">
     </div>
-    <div class="form-tips">Tip: sends the received data as a Buffer object (not a String).<br/>If you select server socket the host defaults to localhost.</div>
 </script>
 
 <script type="text/x-red" data-help-name="tcp in">
-	<p>Provides a choice of tcp input connections. Can either be a client - or provide a listening socket.</p>
+	<p>Provides a choice of tcp inputs. Can either connect to a remote tcp port, 
-	<p>The TCP node produces a <i>BUFFER</i> object <b></b>msg.payload</b> and NOT a String. If you need a String then use <i>.toString()</i> on <b>msg.payload</b> in your next function block.</p>
+	   or accept incoming connections.</p>
-	<p>It also provides <b>msg.fromip</b> of the form ipaddress:port .</p>
-	<p>You can select Base64 encoding if you want to make it easy to preserve the complete message as a string.</p>
-	<p>In case of disconnection the client trys to reconnect every 10 secs. Topic is optional.</p>
 </script>
 
 <script type="text/javascript">
@@ -56,24 +66,51 @@
         category: 'input',
         color:"Silver",
         defaults: {
-            host: {value:"127.0.0.1",required:true},
+            server: {value:"server",required:true},
+            host: {value:"",validate:function(v) { return (this.server == "server")||v.length > 0;} },
             port: {value:"",required:true,validate:RED.validators.number()},
-            base64: {value:false,required:true},
+            datamode:{value:"stream"},
-            server: {value:false,required:true},
+            datatype:{value:"buffer"},
+            newline:{value:""},
             topic: {value:""},
-            name: {value:""}
+            name: {value:""},
+            base64: {/*deprecated*/ value:false,required:true}
         },
         inputs:0,
         outputs:1,
         icon: "bridge-dash.png",
         label: function() {
-			if ((this.host!="") & (this.port!="")) {
+            return this.name || "tcp:"+(this.host?this.host+":":"")+this.port;
-				return this.name||(this.host+":"+this.port);
-			}
-			else { return "tcp in"; }
         },
         labelStyle: function() {
             return this.name?"node_label_italic":"";
+        },
+        oneditprepare: function() {
+            var updateOptions = function() {
+                var sockettype = $("#node-input-server option:selected").val();
+                if (sockettype == "client") {
+                    $("#node-input-host-row").show();
+                } else {
+                    $("#node-input-host-row").hide();
+                }
+                var datamode = $("#node-input-datamode option:selected").val();
+                var datatype = $("#node-input-datatype option:selected").val();
+                if (datamode == "stream") {
+                    $("#node-input-datamode-plural").show();
+                    if (datatype == "utf8") {
+                        $("#node-row-newline").show();
+                    } else {
+                        $("#node-row-newline").hide();
+                    }
+                } else {
+                    $("#node-input-datamode-plural").hide();
+                    $("#node-row-newline").hide();
+                }
+            };
+            updateOptions();
+            $("#node-input-server").change(updateOptions);
+            $("#node-input-datatype").change(updateOptions);
+            $("#node-input-datamode").change(updateOptions);
         }
     });
 </script>

nodes/io/31-tcpin.js

@@ -15,7 +15,7 @@
  **/
 
 var RED = require("../../red/red");
-var reConnect = RED.settings.socketReconnectTime||10000;
+var reconnectTime = RED.settings.socketReconnectTime||10000;
 var net = require('net');
 
 function TcpIn(n) {
@@ -23,73 +23,123 @@ function TcpIn(n) {
 	this.host = n.host;
 	this.port = n.port * 1;
 	this.topic = n.topic;
+	this.stream = (!n.datamode||n.datamode=='stream'); /* stream,single*/
+	this.datatype = n.datatype||'buffer'; /* buffer,utf8,base64 */
+	this.newline = (n.newline||"").replace("\\n","\n").replace("\\r","\r");
 	this.base64 = n.base64;
-	this.server = n.server;
+	this.server = (typeof n.server == 'boolean')?n.server:(n.server == "server");
+	this.closing = false;
 	var node = this;
     
 	if (!node.server) {
+        var buffer = null;
 		var client;
-		var to;
+		var reconnectTimeout;
-
 		function setupTcpClient() {
-			node.log('connecting to port '+node.port);
+			node.log("connecting to "+node.host+":"+node.port);
 			client = net.connect(node.port, node.host, function() {
-				node.log("input connected to "+node.host+":"+node.port);
+			        buffer = (node.datatype == 'buffer')? new Buffer(0):"";
+			        node.log("connected to "+node.host+":"+node.port);
 			});
             
 			client.on('data', function (data) {
-				var msg;
+			        if (node.datatype != 'buffer') {
-				if (node.base64) { msg = { topic:node.topic, payload:new Buffer(data).toString('base64') }; }
+			            data = data.toString(node.datatype);
-				else { msg = {topic:node.topic, payload:data}; }
+			        }
+			        if (node.stream) {
+			            if ((typeof data) === "string" && node.newline != "") {
+			                buffer = buffer+data;
+			                var parts = buffer.split(node.newline);
+			                for (var i = 0;i<parts.length-1;i+=1) {
+			                    var msg = {topic:node.topic, payload:parts[i]};
 			                    node.send(msg);
+			                }
+			                buffer = parts[parts.length-1];
+			            } else {
+			                var msg = {topic:node.topic, payload:data};
+			                node.send(msg);
+			            }
+			        } else {
+			            if ((typeof data) === "string") {
+			                buffer = buffer+data;
+			            } else {
+			                buffer = Buffer.concat([buffer,data],buffer.length+data.length);
+			            }
+			        }
 			});
-
 			client.on('end', function() {
-				node.log("ended");
+			        if (!node.stream || (node.datatype == "utf8" && node.newline != "" && buffer.length > 0)) {
+			            var msg = {topic:node.topic,payload:buffer};
+			            node.send(msg);
+			            buffer = null;
+			        }
 			});
 
 			client.on('close', function() {
-				client.destroy();
+			        node.log("connection lost to "+node.host+":"+node.port);
-				node.log('closed');
+			        if (!node.closing) {
-				to = setTimeout(setupTcpClient, reConnect);
+			            reconnectTimeout = setTimeout(setupTcpClient, reconnectTime);
+			        }
 			});
 
 			client.on('error', function(err) {
-				node.log('error : '+err);
+			        node.log(err);
-				//to = setTimeout(setupTcpClient, reConnect);
 			});
 		}
 		setupTcpClient();
 
 		this._close = function() {
+		    this.closing = true;
 			client.end();
-			clearTimeout(to);
+			clearTimeout(reconnectTimeout);
-			node.log('input stopped');
 		}
-	}
+	} else {
-	else {
 		var server = net.createServer(function (socket) {
-			var buffer = null;
+		        var buffer = (node.datatype == 'buffer')? new Buffer(0):"";
-			socket.on('data', function (chunk) {
+		        socket.on('data', function (data) {
-			//if (buffer == null) {
+		                if (node.datatype != 'buffer') {
-			//	buffer = chunk;
+		                    data = data.toString(node.datatype);
-			//} else {
+		                }
-				//buffer = Buffer.concat([buffer,chunk]);
+		                    
-				var msg = {topic:node.topic, payload:chunk, fromip:socket.remoteAddress+':'+socket.remotePort};
+		                if (node.stream) {
+		                    if ((typeof data) === "string" && node.newline != "") {
+		                        buffer = buffer+data;
+		                        var parts = buffer.split(node.newline);
+		                        for (var i = 0;i<parts.length-1;i+=1) {
+		                            var msg = {topic:node.topic, payload:parts[i]};
 		                            node.send(msg);
-			//}
+		                        }
+		                        buffer = parts[parts.length-1];
+		                    } else {
+		                        var msg = {topic:node.topic, payload:data};
+		                        node.send(msg);
+		                    }
+		                } else {
+		                    if ((typeof data) === "string") {
+		                        buffer = buffer+data;
+		                    } else {
+		                        buffer = Buffer.concat([buffer,data],buffer.length+data.length);
+		                    }
+		                }
 		        });
 		        socket.on('end', function() {
-				var msg = {topic:node.topic, payload:buffer, fromip:socket.remoteAddress+':'+socket.remotePort};
+		                if (!node.stream || (node.datatype == "utf8" && node.newline != "" && buffer.length > 0)) {
+		                    var msg = {topic:node.topic,payload:buffer};
 		                    node.send(msg);
+		                    buffer = null;
+		                }
+		        });		   
+		        socket.on('error',function(err) {
+		            node.log(err);
 		        });
 		});
 		server.listen(node.port);
-		node.log('socket input on port '+node.port);
+		node.log('listening on port '+node.port);
 		
 		this._close = function() {
+		    this.closing = true;
 			server.close();
-			node.log('socket input stopped');
+			node.log('stopped listening on port '+node.port);
 		}
 	}
 	

nodes/io/31-tcpout.html

@@ -16,34 +16,37 @@
 
 <script type="text/x-red" data-template-name="tcp out">
     <div class="form-row">
-        <label for="node-input-host"><i class="icon-bookmark"></i> Host</label>
+        <label for="node-input-beserver"><i class="icon-resize-small"></i> Type</label>
-        <input type="text" id="node-input-host" placeholder="localhost" style="width: 40%;" >
+        <select id="node-input-beserver" style="width:120px; margin-right:5px;">
+            <option value="server">Listen on</option>
+            <option value="client">Connect to</option>
+        </select>
+        port <input type="text" id="node-input-port" style="width: 50px">
+    </div>
     
-        <label for="node-input-port" style="margin-left: 10px; width: 35px;"> Port</label>
+    <div class="form-row hidden" id="node-input-host-row" style="padding-left: 110px;"> 
-        <input type="text" id="node-input-port" placeholder="Port" style="width: 45px">
+        at host <input type="text" id="node-input-host" placeholder="localhost" style="width: 40%;">
-    </div>
-    <div class="form-row">
-        <label>&nbsp;</label>
-        <input type="checkbox" id="node-input-beserver" placeholder="server" style="display: inline-block; width: auto; vertical-align: top;">
-        <label for="node-input-beserver" style="width: 70%;">Be a server socket ?</label>
     </div>
+
     <div class="form-row">
         <label>&nbsp;</label>
         <input type="checkbox" id="node-input-base64" placeholder="base64" style="display: inline-block; width: auto; vertical-align: top;">
         <label for="node-input-base64" style="width: 70%;">Decode Base64 message ?</label>
     </div>
+    
     <div class="form-row">
         <label for="node-input-name"><i class="icon-tag"></i> Name</label>
         <input type="text" id="node-input-name" placeholder="Name">
     </div>
-    <div class="form-tips">Tip: If you select server socket the host defaults to localhost.</div>
-
 </script>
 
 <script type="text/x-red" data-help-name="tcp out">
-	<p>Provides a choice of tcp output connections. Can either connect out - or provide a socket connection.</p>
+	<p>Provides a choice of tcp outputs. Can either connect to a remote tcp port,
+	or accept incoming connections.</p>
 	<p>Only <b>msg.payload</b> is sent.</p>
-	<p>You can select Base64 decoding if you want to decode a message encoded by the input socket.</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
+	before being sent.</p>
 </script>
 
 <script type="text/javascript">
@@ -51,9 +54,9 @@
         category: 'output',
         color:"Silver",
         defaults: {
-            host: {value:"127.0.0.1",required:true},
+            host: {value:"",validate:function(v) { return (this.beserver == "server")||v.length > 0;} },
             port: {value:"",required:true},
-            beserver: {value:false,required:true},
+            beserver: {value:"client",required:true},
             base64: {value:false,required:true},
             name: {value:""}
         },
@@ -62,12 +65,22 @@
         icon: "bridge-dash.png",
         align: "right",
         label: function() {
-			var lab = this.host+":"+this.port;
+            return this.name || "tcp:"+(this.host?this.host+":":"")+this.port;
-			if (this.server) lab = "tcp out:"+this.port;
-            return this.name||lab;
         },
         labelStyle: function() {
             return (this.name)?"node_label_italic":"";
+        },
+        oneditprepare: function() {
+            var updateOptions = function() {
+                var sockettype = $("#node-input-beserver option:selected").val();
+                if (sockettype == "client") {
+                    $("#node-input-host-row").show();
+                } else {
+                    $("#node-input-host-row").hide();
+                }
+            };
+            updateOptions();
+            $("#node-input-beserver").change(updateOptions);
         }
     });
 </script>

nodes/io/31-tcpout.js

@@ -15,7 +15,7 @@
  **/
 
 var RED = require("../../red/red");
-var reConnect = RED.settings.socketReconnectTime||10000;
+var reconnectTime = RED.settings.socketReconnectTime||10000;
 var net = require('net');
 
 function TcpOut(n) {
@@ -25,67 +25,93 @@ function TcpOut(n) {
 	this.base64 = n.base64;
 	this.beserver = n.beserver;
 	this.name = n.name;
+	this.closing = false;
 	var node = this;
 
-	if (!node.beserver) {
+	if (!node.beserver||node.beserver=="client") {
-		var client = new net.Socket();
+		var reconnectTimeout;
-		var to;
+		var client = null;
+		var connected = false;
 		
 		function setupTcpClient() {
-			client.connect(node.port, node.host, function() {
+			node.log("connecting to "+node.host+":"+node.port);
-				node.log("output connected to "+node.host+":"+node.port);
+			client = net.connect(node.port, node.host, function() {
+			        connected = true;
+			        node.log("connected to "+node.host+":"+node.port);
 			});
 			
 			client.on('error', function (err) {
-				node.error('error : '+err);
+			        node.log('error : '+err);
-				to = setTimeout(setupTcpClient, reConnect);
 			});
 			
 			client.on('end', function (err) {
-				node.log("output disconnected");
-				to = setTimeout(setupTcpClient, reConnect);
 			});
 			
 			client.on('close', function() {
+			        node.log("connection lost to "+node.host+":"+node.port);
+			        connected = false;
 			        client.destroy();
-				node.log('closed');
+			        if (!node.closing) {
-				to = setTimeout(setupTcpClient, reConnect);
+			            reconnectTimeout = setTimeout(setupTcpClient,reconnectTime);
-			});
-
-			node.on("input", function(msg) {
-				if (msg.payload != null) {
-					if (node.base64) { client.write(new Buffer(msg.payload,'base64')); }
-					else { client.write(msg.payload);}
 			        }
 			});
 		}
 		setupTcpClient();
 		
+		
+		node.on("input", function(msg) {
+				if (connected && msg.payload != null) {
+				    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));
+				    }
+				}
+		});
+		
+		
 		this._close = function() {
+		    this.closing = true;
 			client.end();
-			clearTimeout(to);
+			clearTimeout(reconnectTimeout);
-			node.log('output stopped');
 		}
 	}
 
 	else {
+	    var connectedSockets = [];
 		var server = net.createServer(function (socket) {
-			socket.on("connect",function() {
+				var remoteDetails = socket.remoteAddress+":"+socket.remotePort;
-				node.log("Connection from "+socket.remoteAddress);
+				node.log("connection from "+remoteDetails);
+				connectedSockets.push(socket);
+				socket.on('close',function() {
+				        node.log("connection closed from "+remoteDetails);
+				        connectedSockets.splice(connectedSockets.indexOf(socket),1);
+				});
 		});
 		node.on("input", function(msg) {
 				if (msg.payload != null) {
-					if (node.base64) { socket.write(new Buffer(msg.payload,'base64')); }
+				    var buffer;
-					else { socket.write(msg.payload);}
+				    if (Buffer.isBuffer(msg.payload)) {
+				        buffer = msg.payload;
+				    } else if (typeof msg.payload === "string" && node.base64) {
+				        buffer = new Buffer(msg.payload,'base64');
+				    } else {
+				        buffer = new Buffer(""+msg.payload);
+				    }
+				    for (var i = 0; i<connectedSockets.length;i+=1) {
+				        connectedSockets[i].write(buffer);
+				    }
 				}
 		});
-		});
+
 		server.listen(node.port);
-		node.log('socket output on port '+node.port);
+		node.log('listening on port '+node.port);
 
 		this._close = function() {
 			server.close();
-			node.log('output stopped');
+			node.log('stopped listening on port '+node.port);
 		}
 	}
 }