diff --git a/nodes/core/io/31-tcpin.js b/nodes/core/io/31-tcpin.js index c3bc2ab8c..63eb4a478 100644 --- a/nodes/core/io/31-tcpin.js +++ b/nodes/core/io/31-tcpin.js @@ -18,10 +18,36 @@ module.exports = function(RED) { "use strict"; var reconnectTime = RED.settings.socketReconnectTime||10000; var socketTimeout = RED.settings.socketTimeout||null; + const msgQueueSize = RED.settings.tcpMsgQueueSize || 1000; + const Denque = require('denque'); var net = require('net'); var connectionPool = {}; + /** + * Enqueue `item` in `queue` + * @param {Denque} queue - Queue + * @param {*} item - Item to enqueue + * @private + * @returns {Denque} `queue` + */ + const enqueue = (queue, item) => { + // drop msgs from front of queue if size is going to be exceeded + if (queue.size() === msgQueueSize) { + queue.shift(); + } + queue.push(item); + return queue; + }; + + /** + * Shifts item off front of queue + * @param {Deque} queue - Queue + * @private + * @returns {*} Item previously at front of queue + */ + const dequeue = queue => queue.shift(); + function TcpIn(n) { RED.nodes.createNode(this,n); this.host = n.host; @@ -435,11 +461,14 @@ module.exports = function(RED) { // the clients object will have: // clients[id].client, clients[id].msg, clients[id].timeout var connection_id = host + ":" + port; - clients[connection_id] = clients[connection_id] || {}; - clients[connection_id].msg = msg; - clients[connection_id].connected = clients[connection_id].connected || false; + clients[connection_id] = clients[connection_id] || { + msgQueue: new Denque(), + connected: false, + connecting: false + }; + enqueue(clients[connection_id].msgQueue, msg); - if (!clients[connection_id].connected) { + if (!clients[connection_id].connecting && !clients[connection_id].connected) { var buf; if (this.out == "count") { if (this.splitc === 0) { buf = Buffer.alloc(1); } @@ -451,14 +480,19 @@ module.exports = function(RED) { if (socketTimeout !== null) { clients[connection_id].client.setTimeout(socketTimeout);} if (host && port) { + clients[connection_id].connecting = true; clients[connection_id].client.connect(port, host, function() { //node.log(RED._("tcpin.errors.client-connected")); node.status({fill:"green",shape:"dot",text:"common.status.connected"}); if (clients[connection_id] && clients[connection_id].client) { clients[connection_id].connected = true; - clients[connection_id].client.write(clients[connection_id].msg.payload); + clients[connection_id].connecting = false; + let msg; + while (msg = dequeue(clients[connection_id].msgQueue)) { + clients[connection_id].client.write(msg.payload); + } if (node.out === "time" && node.splitc < 0) { - clients[connection_id].connected = false; + clients[connection_id].connected = clients[connection_id].connecting = false; clients[connection_id].client.end(); delete clients[connection_id]; node.status({}); @@ -473,9 +507,10 @@ module.exports = function(RED) { clients[connection_id].client.on('data', function(data) { if (node.out === "sit") { // if we are staying connected just send the buffer if (clients[connection_id]) { - if (!clients[connection_id].hasOwnProperty("msg")) { clients[connection_id].msg = {}; } - clients[connection_id].msg.payload = data; - node.send(RED.util.cloneMessage(clients[connection_id].msg)); + let msg = dequeue(clients[connection_id].msgQueue) || {}; + clients[connection_id].msgQueue.unshift(msg); + msg.payload = data; + node.send(RED.util.cloneMessage(msg)); } } // else if (node.splitc === 0) { @@ -495,9 +530,11 @@ module.exports = function(RED) { clients[connection_id].timeout = setTimeout(function () { if (clients[connection_id]) { clients[connection_id].timeout = null; - clients[connection_id].msg.payload = Buffer.alloc(i+1); - buf.copy(clients[connection_id].msg.payload,0,0,i+1); - node.send(clients[connection_id].msg); + let msg = dequeue(clients[connection_id].msgQueue) || {}; + clients[connection_id].msgQueue.unshift(msg); + msg.payload = Buffer.alloc(i+1); + buf.copy(msg.payload,0,0,i+1); + node.send(msg); if (clients[connection_id].client) { node.status({}); clients[connection_id].client.destroy(); @@ -516,9 +553,11 @@ module.exports = function(RED) { i += 1; if ( i >= node.splitc) { if (clients[connection_id]) { - clients[connection_id].msg.payload = Buffer.alloc(i); - buf.copy(clients[connection_id].msg.payload,0,0,i); - node.send(clients[connection_id].msg); + let msg = dequeue(clients[connection_id].msgQueue) || {}; + clients[connection_id].msgQueue.unshift(msg); + msg.payload = Buffer.alloc(i); + buf.copy(msg.payload,0,0,i); + node.send(msg); if (clients[connection_id].client) { node.status({}); clients[connection_id].client.destroy(); @@ -534,9 +573,11 @@ module.exports = function(RED) { i += 1; if (data[j] == node.splitc) { if (clients[connection_id]) { - clients[connection_id].msg.payload = Buffer.alloc(i); - buf.copy(clients[connection_id].msg.payload,0,0,i); - node.send(clients[connection_id].msg); + let msg = dequeue(clients[connection_id].msgQueue) || {}; + clients[connection_id].msgQueue.unshift(msg); + msg.payload = Buffer.alloc(i); + buf.copy(msg.payload,0,0,i); + node.send(msg); if (clients[connection_id].client) { node.status({}); clients[connection_id].client.destroy(); @@ -554,7 +595,7 @@ module.exports = function(RED) { //console.log("END"); node.status({fill:"grey",shape:"ring",text:"common.status.disconnected"}); if (clients[connection_id] && clients[connection_id].client) { - clients[connection_id].connected = false; + clients[connection_id].connected = clients[connection_id].connecting = false; clients[connection_id].client = null; } }); @@ -562,7 +603,7 @@ module.exports = function(RED) { clients[connection_id].client.on('close', function() { //console.log("CLOSE"); if (clients[connection_id]) { - clients[connection_id].connected = false; + clients[connection_id].connected = clients[connection_id].connecting = false; } var anyConnected = false; @@ -592,21 +633,23 @@ module.exports = function(RED) { clients[connection_id].client.on('timeout',function() { //console.log("TIMEOUT"); if (clients[connection_id]) { - clients[connection_id].connected = false; + clients[connection_id].connected = clients[connection_id].connecting = false; node.status({fill:"grey",shape:"dot",text:"tcpin.errors.connect-timeout"}); //node.warn(RED._("tcpin.errors.connect-timeout")); if (clients[connection_id].client) { + clients[connection_id].connecting = true; clients[connection_id].client.connect(port, host, function() { clients[connection_id].connected = true; + clients[connection_id].connecting = false; node.status({fill:"green",shape:"dot",text:"common.status.connected"}); }); } } }); } - else { + else if (!clients[connection_id].connecting && clients[connection_id].connected) { if (clients[connection_id] && clients[connection_id].client) { - clients[connection_id].client.write(clients[connection_id].msg.payload); + clients[connection_id].client.write(dequeue(clients[connection_id].msgQueue)); } } }); diff --git a/package.json b/package.json index cae321d1e..b1031f886 100644 --- a/package.json +++ b/package.json @@ -43,6 +43,7 @@ "cookie-parser": "1.4.3", "cors": "2.8.4", "cron": "1.3.0", + "denque": "1.3.0", "express": "4.16.3", "express-session": "1.15.6", "fs-extra": "5.0.0", diff --git a/settings.js b/settings.js index dc79223fe..1082332cc 100644 --- a/settings.js +++ b/settings.js @@ -41,6 +41,10 @@ module.exports = { // defaults to no timeout //socketTimeout: 120000, + // Maximum number of messages to wait in queue while attempting to connect to TCP socket + // defaults to 1000 + //tcpMsgQueueSize: 2000, + // Timeout in milliseconds for HTTP request connections // defaults to 120 seconds //httpRequestTimeout: 120000, diff --git a/test/nodes/core/io/31-tcprequest_spec.js b/test/nodes/core/io/31-tcprequest_spec.js index 651384ad5..3cc0e1224 100644 --- a/test/nodes/core/io/31-tcprequest_spec.js +++ b/test/nodes/core/io/31-tcprequest_spec.js @@ -19,6 +19,7 @@ var should = require("should"); var stoppable = require('stoppable'); var helper = require("node-red-node-test-helper"); var tcpinNode = require("../../../../nodes/core/io/31-tcpin.js"); +const RED = require("../../../../red/red.js"); describe('TCP Request Node', function() { @@ -28,12 +29,12 @@ describe('TCP Request Node', function() { function startServer(done) { port += 1; server = stoppable(net.createServer(function(c) { - c.on('data', function(data) { - var rdata = "ACK:"+data.toString(); - c.write(rdata); - }); + c.on('data', function(data) { + var rdata = "ACK:"+data.toString(); + c.write(rdata); + }); c.on('error', function(err) { - startServer(done); + startServer(done); }); })).listen(port, "127.0.0.1", function(err) { done(); @@ -64,48 +65,108 @@ describe('TCP Request Node', function() { done(err); } }); - if((typeof val0) === 'object') { - n1.receive(val0); - } else { - n1.receive({payload:val0}); - } + if((typeof val0) === 'object') { + n1.receive(val0); + } else { + n1.receive({payload:val0}); + } }); } - it('should send & recv data', function(done) { - var flow = [{id:"n1", type:"tcp request", server:"localhost", port:port, out:"time", splitc: "0", wires:[["n2"]] }, - {id:"n2", type:"helper"}]; - testTCP(flow, "foo", "ACK:foo", done) + function testTCPMany(flow, values, result, done) { + helper.load(tcpinNode, flow, () => { + const n1 = helper.getNode("n1"); + const n2 = helper.getNode("n2"); + n2.on("input", msg => { + try { + msg.should.have.property('payload', Buffer(result)); + done(); + } catch(err) { + done(err); + } + }); + values.forEach(value => { + n1.receive(typeof value === 'object' ? value : {payload: value}); + }); + }); + } + + describe('single message', function () { + it('should send & recv data', function(done) { + var flow = [{id:"n1", type:"tcp request", server:"localhost", port:port, out:"time", splitc: "0", wires:[["n2"]] }, + {id:"n2", type:"helper"}]; + testTCP(flow, "foo", "ACK:foo", done) + }); + + it('should send & recv data when specified character received', function(done) { + var flow = [{id:"n1", type:"tcp request", server:"localhost", port:port, out:"char", splitc: "0", wires:[["n2"]] }, + {id:"n2", type:"helper"}]; + testTCP(flow, "foo0bar0", "ACK:foo0", done); + }); + + it('should send & recv data after fixed number of chars received', function(done) { + var flow = [{id:"n1", type:"tcp request", server:"localhost", port:port, out:"count", splitc: "7", wires:[["n2"]] }, + {id:"n2", type:"helper"}]; + testTCP(flow, "foo bar", "ACK:foo", done); + }); + + it('should send & receive, then keep connection', function(done) { + var flow = [{id:"n1", type:"tcp request", server:"localhost", port:port, out:"sit", splitc: "5", wires:[["n2"]] }, + {id:"n2", type:"helper"}]; + testTCP(flow, "foo", "ACK:foo", done); + }); + + it('should send & recv data to/from server:port from msg', function(done) { + var flow = [{id:"n1", type:"tcp request", server:"", port:"", out:"time", splitc: "0", wires:[["n2"]] }, + {id:"n2", type:"helper"}]; + testTCP(flow, {payload:"foo", host:"localhost", port:port}, "ACK:foo", done) + }); }); - it('should send & recv data when specified character received', function(done) { - var flow = [{id:"n1", type:"tcp request", server:"localhost", port:port, out:"char", splitc: "0", wires:[["n2"]] }, - {id:"n2", type:"helper"}]; - testTCP(flow, "foo0bar0", "ACK:foo0", done); - }); + describe('many messages', function () { + it('should send & recv data', function(done) { + var flow = [{id:"n1", type:"tcp request", server:"localhost", port:port, out:"time", splitc: "0", wires:[["n2"]] }, + {id:"n2", type:"helper"}]; - it('should send & recv data after fixed number of chars received', function(done) { - var flow = [{id:"n1", type:"tcp request", server:"localhost", port:port, out:"count", splitc: "7", wires:[["n2"]] }, - {id:"n2", type:"helper"}]; - testTCP(flow, "foo bar", "ACK:foo", done); - }); + testTCPMany(flow, ['f', 'o', 'o'], 'ACK:foo', done); + }); - it('should send & receive, then keep connection', function(done) { - var flow = [{id:"n1", type:"tcp request", server:"localhost", port:port, out:"sit", splitc: "5", wires:[["n2"]] }, - {id:"n2", type:"helper"}]; - testTCP(flow, "foo", "ACK:foo", done); - }); + it('should send & recv data when specified character received', function(done) { + var flow = [{id:"n1", type:"tcp request", server:"localhost", port:port, out:"char", splitc: "0", wires:[["n2"]] }, + {id:"n2", type:"helper"}]; + testTCPMany(flow, ["foo0","bar0"], "ACK:foo0", done); + }); - it('should send & close', function(done) { - var flow = [{id:"n1", type:"tcp request", server:"localhost", port:port, out:"sit", splitc: "5", wires:[["n2"]] }, - {id:"n2", type:"helper"}]; - testTCP(flow, "foo", "ACK:foo", done); - }); + it('should send & recv data after fixed number of chars received', function(done) { + var flow = [{id:"n1", type:"tcp request", server:"localhost", port:port, out:"count", splitc: "7", wires:[["n2"]] }, + {id:"n2", type:"helper"}]; + testTCPMany(flow, ["fo", "ob", "ar"], "ACK:foo", done); + }); - it('should send & recv data to/from server:port from msg', function(done) { - var flow = [{id:"n1", type:"tcp request", server:"", port:"", out:"time", splitc: "0", wires:[["n2"]] }, - {id:"n2", type:"helper"}]; - testTCP(flow, {payload:"foo", host:"localhost", port:port}, "ACK:foo", done) - }); + it('should send & receive, then keep connection', function(done) { + var flow = [{id:"n1", type:"tcp request", server:"localhost", port:port, out:"sit", splitc: "5", wires:[["n2"]] }, + {id:"n2", type:"helper"}]; + testTCPMany(flow, ["foo", "bar", "baz"], "ACK:foobarbaz", done); + }); + + it('should send & recv data to/from server:port from msg', function(done) { + var flow = [{id:"n1", type:"tcp request", server:"", port:"", out:"time", splitc: "0", wires:[["n2"]] }, + {id:"n2", type:"helper"}]; + testTCPMany(flow, [ + {payload:"f", host:"localhost", port:port}, + {payload:"o", host:"localhost", port:port}, + {payload:"o", host:"localhost", port:port}], "ACK:foo", done); + }); + + it('should limit the queue size', function (done) { + RED.settings.tcpMsgQueueSize = 10; + var flow = [{id:"n1", type:"tcp request", server:"localhost", port:port, out:"sit", splitc: "5", wires:[["n2"]] }, + {id:"n2", type:"helper"}]; + // create one more msg than is allowed + const msgs = new Array(RED.settings.tcpMsgQueueSize + 1).fill('x'); + const expected = msgs.slice(0, -1); + testTCPMany(flow, msgs, "ACK:" + expected.join(''), done); + }); + }); });