25-serial.js

module.exports = function(RED) {
    "use strict";
    var settings = RED.settings;
    var events = require("events");
    var serialp = require("serialport");
    var bufMaxSize = 32768;  // Max serial buffer size, for inputs...

    // TODO: 'serialPool' should be encapsulated in SerialPortNode

    function SerialPortNode(n) {
        RED.nodes.createNode(this,n);
        this.serialport = n.serialport;
        this.newline = n.newline;
        this.addchar = n.addchar || "false";
        this.serialbaud = parseInt(n.serialbaud) || 57600;
        this.databits = parseInt(n.databits) || 8;
        this.parity = n.parity || "none";
        this.stopbits = parseInt(n.stopbits) || 1;
        this.bin = n.bin || "false";
        this.out = n.out || "char";
    }
    RED.nodes.registerType("serial-port",SerialPortNode);

    function SerialOutNode(n) {
        RED.nodes.createNode(this,n);
        this.serial = n.serial;
        this.serialConfig = RED.nodes.getNode(this.serial);

        if (this.serialConfig) {
            var node = this;
            node.port = serialPool.get(this.serialConfig.serialport,
                this.serialConfig.serialbaud,
                this.serialConfig.databits,
                this.serialConfig.parity,
                this.serialConfig.stopbits,
                this.serialConfig.newline);
            node.addCh = "";
            if (node.serialConfig.addchar == "true" || node.serialConfig.addchar === true) {
                node.addCh = this.serialConfig.newline.replace("\\n","\n").replace("\\r","\r").replace("\\t","\t").replace("\\e","\e").replace("\\f","\f").replace("\\0","\0"); // jshint ignore:line
            }
            node.on("input",function(msg) {
                if (msg.hasOwnProperty("payload")) {
                    var payload = msg.payload;
                    if (!Buffer.isBuffer(payload)) {
                        if (typeof payload === "object") {
                            payload = JSON.stringify(payload);
                        }
                        else {
                            payload = payload.toString();
                        }
                        if (node.out === "char") { payload += node.addCh; }
                    }
                    else if ((node.addCh !== "") && (node.out === "char")) {
                        payload = Buffer.concat([payload,new Buffer(node.addCh)]);
                    }
                    node.port.write(msg,payload,null,function(err,res) {
                        if (err) {
                            var errmsg = err.toString().replace("Serialport","Serialport "+node.port.serial.path);
                            node.error(errmsg,msg);
                        }
                    });
                }
            });
            node.port.on('ready', function() {
                node.status({fill:"green",shape:"dot",text:"node-red:common.status.connected"});
            });
            node.port.on('closed', function() {
                node.status({fill:"red",shape:"ring",text:"node-red:common.status.not-connected"});
            });
        }
        else {
            this.error(RED._("serial.errors.missing-conf"));
        }

        this.on("close", function(done) {
            if (this.serialConfig) {
                serialPool.close(this.serialConfig.serialport,done);
            }
            else {
                done();
            }
        });
    }
    RED.nodes.registerType("serial out",SerialOutNode);


    function SerialInNode(n) {
        RED.nodes.createNode(this,n);
        this.serial = n.serial;
        this.serialConfig = RED.nodes.getNode(this.serial);

        if (this.serialConfig) {
            var node = this;
            node.tout = null;
            var buf;
            if (node.serialConfig.out != "count") { buf = new Buffer(bufMaxSize); }
            else { buf = new Buffer(Number(node.serialConfig.newline)); }
            var i = 0;
            node.status({fill:"grey",shape:"dot",text:"node-red:common.status.not-connected"});
            node.port = serialPool.get(this.serialConfig.serialport,
                this.serialConfig.serialbaud,
                this.serialConfig.databits,
                this.serialConfig.parity,
                this.serialConfig.stopbits,
                this.serialConfig.newline
            );

            var splitc;
            if (node.serialConfig.newline.substr(0,2) == "0x") {
                splitc = new Buffer([parseInt(node.serialConfig.newline)]);
            }
            else {
                splitc = new Buffer(node.serialConfig.newline.replace("\\n","\n").replace("\\r","\r").replace("\\t","\t").replace("\\e","\e").replace("\\f","\f").replace("\\0","\0")); // jshint ignore:line
            }

            this.port.on('data', function(msg, sender) {
                // single char buffer
                if ((node.serialConfig.newline === 0)||(node.serialConfig.newline === "")) {
                    if (node.serialConfig.bin !== "bin") { node.send({"payload": String.fromCharCode(msg), port:node.serialConfig.serialport}); }
                    else { node.send({"payload": new Buffer([msg]), port:node.serialConfig.serialport}); }
                }
                else {
                    // do the timer thing
                    if (node.serialConfig.out === "time") {
                        if (node.tout) {
                            i += 1;
                            buf[i] = msg;
                        }
                        else {
                            node.tout = setTimeout(function () {
                                node.tout = null;
                                var m = new Buffer(i+1);
                                buf.copy(m,0,0,i+1);
                                if (node.serialConfig.bin !== "bin") { m = m.toString(); }
                                node.send({"payload": m, port:node.serialConfig.serialport});
                                m = null;
                            }, node.serialConfig.newline);
                            i = 0;
                            buf[0] = msg;
                        }
                    }
                    // count bytes into a buffer...
                    else if (node.serialConfig.out === "count") {
                        buf[i] = msg;
                        i += 1;
                        if ( i >= parseInt(node.serialConfig.newline)) {
                            var m = new Buffer(i);
                            buf.copy(m,0,0,i);
                            if (node.serialConfig.bin !== "bin") { m = m.toString(); }
                            node.send({"payload":m, port:node.serialConfig.serialport});
                            m = null;
                            i = 0;
                        }
                    }
                    // look to match char...
                    else if (node.serialConfig.out === "char") {
                        buf[i] = msg;
                        i += 1;
                        if ((msg === splitc[0]) || (i === bufMaxSize)) {
                            var n = new Buffer(i);
                            buf.copy(n,0,0,i);
                            if (node.serialConfig.bin !== "bin") { n = n.toString(); }
                            node.send({"payload":n, port:node.serialConfig.serialport});
                            n = null;
                            i = 0;
                        }
                    }
                }
            });
            this.port.on('ready', function() {
                node.status({fill:"green",shape:"dot",text:"node-red:common.status.connected"});
            });
            this.port.on('closed', function() {
                node.status({fill:"red",shape:"ring",text:"node-red:common.status.not-connected"});
            });
        }
        else {
            this.error(RED._("serial.errors.missing-conf"));
        }

        this.on("close", function(done) {
            if (this.serialConfig) {
                serialPool.close(this.serialConfig.serialport,done);
            }
            else {
                done();
            }
        });
    }
    RED.nodes.registerType("serial in",SerialInNode);


    /******* REQUEST *********/
    function SerialRequestNode(n) {
        RED.nodes.createNode(this,n);
        this.serial = n.serial;
        this.serialConfig = RED.nodes.getNode(this.serial);

        if (this.serialConfig) {
            var node = this;
            node.tout = null;
            var buf;
            node.port = serialPool.get(this.serialConfig.serialport,
                this.serialConfig.serialbaud,
                this.serialConfig.databits,
                this.serialConfig.parity,
                this.serialConfig.stopbits,
                this.serialConfig.newline);
            node.addCh = "";
            if (node.serialConfig.addchar == "true" || node.serialConfig.addchar === true) {
                node.addCh = this.serialConfig.newline.replace("\\n","\n").replace("\\r","\r").replace("\\t","\t").replace("\\e","\e").replace("\\f","\f").replace("\\0","\0"); // jshint ignore:line
            }
            if (node.serialConfig.out != "count") { buf = new Buffer(bufMaxSize); }
            else { buf = new Buffer(Number(node.serialConfig.newline)); }
            var i = 0;
            node.on("input",function(msg) {
                if (msg.hasOwnProperty("payload")) {
                    var payload = msg.payload;
                    if (!Buffer.isBuffer(payload)) {
                        if (typeof payload === "object") {
                            payload = JSON.stringify(payload);
                        }
                        else {
                            payload = payload.toString();
                        }
                        if (node.out === "char") { payload += node.addCh; }
                    }
                    else if ((node.addCh !== "") && (node.out === "char")) {
                        payload = Buffer.concat([payload,new Buffer(node.addCh)]);
                    }
                    node.port.write(msg,payload,node,function(err,res) {
                        if (err) {
                            var errmsg = err.toString().replace("Serialport","Serialport "+node.port.serial.path);
                            node.error(errmsg,msg);
                        }
                    });
                }
            });
            var splitc;
            if (node.serialConfig.newline.substr(0,2) == "0x") {
                splitc = new Buffer([parseInt(node.serialConfig.newline)]);
            }
            else {
                splitc = new Buffer(node.serialConfig.newline.replace("\\n","\n").replace("\\r","\r").replace("\\t","\t").replace("\\e","\e").replace("\\f","\f").replace("\\0","\0")); // jshint ignore:line
            }
            this.port.on('data', function(msg, sender) {
            
                if (sender !== node)
                  return;
                // single char buffer
                if ((node.serialConfig.newline === 0)||(node.serialConfig.newline === "")) {
                    if (node.serialConfig.bin !== "bin") { node.send({"payload": String.fromCharCode(msg), port:node.serialConfig.serialport}); }
                    else { node.send({"payload": new Buffer([msg]), port:node.serialConfig.serialport}); }
                }
                else {
                    // do the timer thing
                    if (node.serialConfig.out === "time") {
                        if (node.tout) {
                            i += 1;
                            buf[i] = msg;
                        }
                        else {
                            node.tout = setTimeout(function () {
                                node.tout = null;
                                var m = new Buffer(i+1);
                                buf.copy(m,0,0,i+1);
                                if (node.serialConfig.bin !== "bin") { m = m.toString(); }
                                node.send({"payload": m, port:node.serialConfig.serialport});
                                m = null;
                            }, node.serialConfig.newline);
                            i = 0;
                            buf[0] = msg;
                        }
                    }
                    // count bytes into a buffer...
                    else if (node.serialConfig.out === "count") {
                        buf[i] = msg;
                        i += 1;
                        if ( i >= parseInt(node.serialConfig.newline)) {
                            var m = new Buffer(i);
                            buf.copy(m,0,0,i);
                            if (node.serialConfig.bin !== "bin") { m = m.toString(); }
                            node.send({"payload":m, port:node.serialConfig.serialport});
                            m = null;
                            i = 0;
                        }
                    }
                    // look to match char...
                    else if (node.serialConfig.out === "char") {
                        buf[i] = msg;
                        i += 1;
                        if ((msg === splitc[0]) || (i === bufMaxSize)) {
                            var n = new Buffer(i);
                            buf.copy(n,0,0,i);
                            if (node.serialConfig.bin !== "bin") { n = n.toString(); }
                            var msgin = node.port.unlock();
                            var msgout = Object.assign(
                              { 
                              }, msgin);
                              
                            msgout= Object.assign(msgout,
                            {
                              payload: n,
                              port:node.serialConfig.serialport,
                              request_payload: msgin.payload,
                              request_msgid: msgin._msgid,
                            });
                            node.send(msgout);
                            n = null;
                            i = 0;
                        }
                    }
                }
            });
            node.port.on('ready', function() {
                node.status({fill:"green",shape:"dot",text:"node-red:common.status.connected"});
            });
            node.port.on('closed', function() {
                node.status({fill:"red",shape:"ring",text:"node-red:common.status.not-connected"});
            });
        }
        else {
            this.error(RED._("serial.errors.missing-conf"));
        }

        this.on("close", function(done) {
            if (this.serialConfig) {
                serialPool.close(this.serialConfig.serialport,done);
            }
            else {
                done();
            }
        });
    }
    RED.nodes.registerType("serial request", SerialRequestNode);
    /******* REQUEST *********/


    var serialPool = (function() {
        var connections = {};
        return {
            get:function(port,baud,databits,parity,stopbits,newline,callback) {
                var id = port;
                if (!connections[id]) {
                    connections[id] = (function() {
                        var obj = {
                            _emitter: new events.EventEmitter(),
                            serial: null,
                            _closing: false,
                            tout: null,
                            queue: [],
                            on: function(a,b) { this._emitter.on(a,b); },
                            close: function(cb) { this.serial.close(cb); },
                            write: function(msg,payload,sender,cb) {
                              this.queue.push(
                                {sender: sender,
                                 msg: msg,
                                 payload: payload,
                                 cb: cb
                                 });
                              if (this.queue.length == 1)
                              {
                                  this.serial.write(payload,cb);
                              }
                            },
                            unlock: function() {
                              var msg1 = null;
                              if (this.queue.length)
                                msg1 = this.queue[0].msg;
                              this.queue.shift();
                              if (this.queue.length) {
                                var item = this.queue[0];
                                this.serial.write(item.payload,item.cb);
                              }
                              return msg1;
                            },
                        }
                        //newline = newline.replace("\\n","\n").replace("\\r","\r");
                        var olderr = "";
                        var setupSerial = function() {
                            obj.serial = new serialp(port,{
                                baudRate: baud,
                                dataBits: databits,
                                parity: parity,
                                stopBits: stopbits,
                                //parser: serialp.parsers.raw,
                                autoOpen: true
                            }, function(err, results) {
                                if (err) {
                                    if (err.toString() !== olderr) {
                                        olderr = err.toString();
                                        RED.log.error(RED._("serial.errors.error",{port:port,error:olderr}));
                                    }
                                    obj.tout = setTimeout(function() {
                                        setupSerial();
                                    }, settings.serialReconnectTime);
                                }
                            });
                            obj.serial.on('error', function(err) {
                                RED.log.error(RED._("serial.errors.error",{port:port,error:err.toString()}));
                                obj._emitter.emit('closed');
                                obj.tout = setTimeout(function() {
                                    setupSerial();
                                }, settings.serialReconnectTime);
                            });
                            obj.serial.on('close', function() {
                                if (!obj._closing) {
                                    RED.log.error(RED._("serial.errors.unexpected-close",{port:port}));
                                    obj._emitter.emit('closed');
                                    obj.tout = setTimeout(function() {
                                        setupSerial();
                                    }, settings.serialReconnectTime);
                                }
                            });
                            obj.serial.on('open',function() {
                                olderr = "";
                                RED.log.info(RED._("serial.onopen",{port:port,baud:baud,config: databits+""+parity.charAt(0).toUpperCase()+stopbits}));
                                if (obj.tout) { clearTimeout(obj.tout); }
                                //obj.serial.flush();
                                obj._emitter.emit('ready');
                            });
                            obj.serial.on('data',function(d) {
                                for (var z=0; z<d.length; z++) {
                                    var last_sender = null;
                                    if (obj.queue.length)
                                      last_sender = obj.queue[0].sender;
                                    obj._emitter.emit('data',d[z],last_sender);
                                }
                            });
                            // obj.serial.on("disconnect",function() {
                            //     RED.log.error(RED._("serial.errors.disconnected",{port:port}));
                            // });
                        }
                        setupSerial();
                        return obj;
                    }());
                }
                return connections[id];
            },
            close: function(port,done) {
                if (connections[port]) {
                    if (connections[port].tout != null) {
                        clearTimeout(connections[port].tout);
                    }
                    connections[port]._closing = true;
                    try {
                        connections[port].close(function() {
                            RED.log.info(RED._("serial.errors.closed",{port:port}));
                            done();
                        });
                    }
                    catch(err) { }
                    delete connections[port];
                }
                else {
                    done();
                }
            }
        }
    }());

    RED.httpAdmin.get("/serialports", RED.auth.needsPermission('serial.read'), function(req,res) {
        serialp.list(function (err, ports) {
            res.json(ports);
        });
    });
}