node-ipc/dao/client.js

'use strict';

const net = require('net'),
    tls = require('tls'),
    EventParser = new require('../entities/EventParser.js'),
    Message = require('js-message'),
    fs = require('fs'),
    Queue = require('js-queue');

let Events = require('event-pubsub/es5');
if(process.version[1]>4){
    Events = require('event-pubsub');
}

let eventParser = new EventParser();

class Client extends Events{
    constructor(config,log){
        super();
        Object.assign(
            this,
            {
                Client  : Client,
                config  : config,
                queue   : new Queue,
                socket  : false,
                connect : connect,
                emit    : emit,
                log     : log,
                retriesRemaining:config.maxRetries||0,
                explicitlyDisconnected: false
            }
        );

        eventParser=new EventParser(this.config);
    }
}

function emit(type,data){
    this.log('dispatching event to ', this.id, this.path, ' : ', type, ',', data);

    let message=new Message;
    message.type=type;
    message.data=data;

    if(this.config.rawBuffer){
        message=new Buffer(type,this.config.encoding);
    }else{
        message=eventParser.format(message);
    }

    if(!this.config.sync){
        this.socket.write(message);
        return;
    }

    this.queue.add(
        syncEmit.bind(this,message)
    );
}

function syncEmit(message){
    this.log('dispatching event to ', this.id, this.path, ' : ', message);
    this.socket.write(message);
}

function connect(){
    //init client object for scope persistance especially inside of socket events.
    let client=this;

    client.log('requested connection to ', client.id, client.path);
    if(!this.path){
        client.log('\n\n######\nerror: ', client.id ,' client has not specified socket path it wishes to connect to.');
        return;
    }

    const options={};

    if(!client.port){
        client.log('Connecting client on Unix Socket :', client.path);

        options.path=client.path;

        if (process.platform ==='win32' && !client.path.startsWith('\\\\.\\pipe\\')){
            options.path = options.path.replace(/^\//, '');
            options.path = options.path.replace(/\//g, '-');
            options.path= `\\\\.\\pipe\\${options.path}`;
        }

        client.socket = net.connect(options);
    }else{
        options.host=client.path;
        options.port=client.port;

        if(client.config.interface.localAddress){
          options.localAddress=client.config.interface.localAddress;
        }

        if(client.config.interface.localPort){
          options.localPort=client.config.interface.localPort;
        }

        if(client.config.interface.family){
          options.family=client.config.interface.family;
        }

        if(client.config.interface.hints){
          options.hints=client.config.interface.hints;
        }

        if(client.config.interface.lookup){
          options.lookup=client.config.interface.lookup;
        }

        if(!client.config.tls){
            client.log('Connecting client via TCP to', options);
            client.socket = net.connect(options);
        }else{
            client.log('Connecting client via TLS to', client.path ,client.port,client.config.tls);
            if(client.config.tls.private){
                client.config.tls.key=fs.readFileSync(client.config.tls.private);
            }
            if(client.config.tls.public){
                client.config.tls.cert=fs.readFileSync(client.config.tls.public);
            }
            if(client.config.tls.trustedConnections){
                if(typeof client.config.tls.trustedConnections === 'string'){
                    client.config.tls.trustedConnections=[client.config.tls.trustedConnections];
                }
                client.config.tls.ca=[];
                for(let i=0; i<client.config.tls.trustedConnections.length; i++){
                    client.config.tls.ca.push(
                        fs.readFileSync(client.config.tls.trustedConnections[i])
                    );
                }
            }

            Object.assign(client.config.tls,options);

            client.socket = tls.connect(
                client.config.tls
            );
        }
    }

    client.socket.setEncoding(this.config.encoding);

    client.socket.on(
        'error',
        function(err){
            client.log('\n\n######\nerror: ', err);
            client.publish('error', err);

        }
    );

    client.socket.on(
        'connect',
        function connectionMade(){
            client.publish('connect');
            client.retriesRemaining=client.config.maxRetries;
            client.log('retrying reset');
        }
    );

    client.socket.on(
        'close',
        function connectionClosed(){
            client.log('connection closed' ,client.id , client.path,
            client.retriesRemaining, 'tries remaining of', client.config.maxRetries
        );

            if(
                client.config.stopRetrying ||
                client.retriesRemaining<1 ||
                client.explicitlyDisconnected

            ){
                client.publish('disconnect');
                client.log(
                    (client.config.id),
                    'exceeded connection rety amount of',
                    ' or stopRetrying flag set.'
                );

                client.socket.destroy();
                client.publish('destroy');
                client=undefined;

                return;
            }

            setTimeout(
                function retryTimeout(){
                    client.retriesRemaining--;
                    client.connect();
                }.bind(null,client),
                client.config.retry
            );

            client.publish('disconnect');
        }
    );

    client.socket.on(
        'data',
        function(data) {
            client.log('## received events ##');
            if(client.config.rawBuffer){
                client.publish(
                   'data',
                   new Buffer(data,client.config.encoding)
                );
                if(!client.config.sync){
                    return;
                }

                client.queue.next();
                return;
            }

            if(!this.ipcBuffer){
                this.ipcBuffer='';
            }

            data=(this.ipcBuffer+=data);

            if(data.slice(-1)!=eventParser.delimiter || data.indexOf(eventParser.delimiter) == -1){
                client.log('Messages are large, You may want to consider smaller messages.');
                return;
            }

            this.ipcBuffer='';

            const events = eventParser.parse(data);
            const eCount = events.length;
            for(let i=0; i<eCount; i++){
                let message=new Message;
                message.load(events[i]);

                client.log('detected event', message.type, message.data);
                client.publish(
                   message.type,
                   message.data
                );
            }

            if(!client.config.sync){
                return;
            }

            client.queue.next();
        }
    );
}

module.exports=Client;