node-ipc/dao/client.js

238 lines
6.5 KiB
JavaScript
Raw Normal View History

2016-01-10 23:18:14 +11:00
'use strict';
const net = require('net'),
tls = require('tls'),
2016-01-11 15:59:40 +11:00
eventParser = require('./eventParser.js'),
2016-01-10 23:18:14 +11:00
Message = require('js-message'),
fs = require('fs'),
Queue = require('js-queue');
2014-02-22 20:13:31 +11:00
2016-10-04 07:34:27 +11:00
let Events = require('event-pubsub/es5');
if(process.version[1]>4){
Events = require('event-pubsub');
}
2016-09-30 23:00:28 +10:00
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
}
);
}
2014-02-22 20:13:31 +11:00
}
function emit(type,data){
this.log('dispatching event to ', this.id, this.path, ' : ', type, ',', data);
2016-01-10 23:18:14 +11:00
let message=new Message;
message.type=type;
message.data=data;
2015-08-23 15:46:55 +10:00
if(this.config.rawBuffer){
message=new Buffer(type,this.config.encoding);
2015-08-23 15:46:55 +10:00
}else{
message=eventParser.format(message);
2015-08-23 15:46:55 +10:00
}
if(!this.config.sync){
this.socket.write(message);
return;
}
this.queue.add(
syncEmit.bind(this,message)
);
2016-01-10 23:18:14 +11:00
}
2014-02-22 20:13:31 +11:00
function syncEmit(message){
this.log('dispatching event to ', this.id, this.path, ' : ', message);
this.socket.write(message);
}
2014-02-22 20:13:31 +11:00
function connect(){
//init client object for scope persistance especially inside of socket events.
2016-01-10 23:18:14 +11:00
let client=this;
client.log('requested connection to ', client.id, client.path);
2014-02-22 20:13:31 +11:00
if(!this.path){
client.log('\n\n######\nerror: ', client.id ,' client has not specified socket path it wishes to connect to.');
2014-02-22 20:13:31 +11:00
return;
}
2014-02-26 12:15:43 +11:00
if(!client.port){
client.log('Connecting client on Unix Socket :', client.path);
2016-01-10 23:18:14 +11:00
let path = client.path;
if (process.platform ==='win32' && !client.path.startsWith('\\\\.\\pipe\\')){
path = path.replace(/^\//, '');
path = path.replace(/\//g, '-');
2016-01-11 16:10:58 +11:00
path= `\\\\.\\pipe\\${path}`;
}
2014-02-26 12:15:43 +11:00
client.socket = net.connect(
{
path: path
2014-02-26 12:15:43 +11:00
}
);
}else{
if(!client.config.tls){
client.log('Connecting client via TCP to', client.path ,client.port);
client.socket = net.connect(
{
port:client.port,
host:client.path
}
);
}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);
2014-02-26 12:15:43 +11:00
}
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=[];
2016-01-10 23:18:14 +11:00
for(let i=0; i<client.config.tls.trustedConnections.length; i++){
client.config.tls.ca.push(
fs.readFileSync(client.config.tls.trustedConnections[i])
);
}
}
client.config.tls.host=client.path;
client.config.tls.port=client.port;
client.socket = tls.connect(
client.config.tls
);
}
2014-02-26 12:15:43 +11:00
}
2014-02-22 20:13:31 +11:00
client.socket.setEncoding(this.config.encoding);
2014-02-22 20:13:31 +11:00
client.socket.on(
'error',
function(err){
client.log('\n\n######\nerror: ', err);
2016-09-30 23:00:28 +10:00
client.publish('error', err);
2015-12-01 12:43:43 +11:00
2014-02-22 20:13:31 +11:00
}
);
2014-02-22 20:13:31 +11:00
client.socket.on(
'connect',
2016-01-10 23:18:14 +11:00
function connectionMade(){
2016-09-30 23:00:28 +10:00
client.publish('connect');
2015-08-23 16:16:20 +10:00
client.retriesRemaining=client.config.maxRetries;
2015-12-01 12:43:43 +11:00
client.log('retrying reset');
2014-02-22 20:13:31 +11:00
}
);
2014-02-22 20:13:31 +11:00
client.socket.on(
'close',
2016-01-10 23:18:14 +11:00
function connectionClosed(){
client.log('connection closed' ,client.id , client.path,
client.retriesRemaining, 'tries remaining of', client.config.maxRetries
2016-01-11 16:10:58 +11:00
);
if(
client.config.stopRetrying ||
client.retriesRemaining<1 ||
client.explicitlyDisconnected
2015-08-23 15:46:55 +10:00
){
2016-09-30 23:00:28 +10:00
client.publish('disconnect');
client.log(
(client.config.id),
'exceeded connection rety amount of',
2016-01-10 23:18:14 +11:00
' or stopRetrying flag set.'
);
client.socket.destroy();
2016-09-30 23:00:28 +10:00
client.publish('destroy');
client=undefined;
return;
}
2014-02-22 20:13:31 +11:00
setTimeout(
2016-01-10 23:18:14 +11:00
function retryTimeout(){
client.retriesRemaining--;
client.connect();
}.bind(null,client),
2014-02-22 20:13:31 +11:00
client.config.retry
);
2016-09-30 23:00:28 +10:00
client.publish('disconnect');
2014-02-22 20:13:31 +11:00
}
);
2014-02-22 20:13:31 +11:00
client.socket.on(
'data',
function(data) {
client.log('## received events ##');
2015-08-23 15:46:55 +10:00
if(client.config.rawBuffer){
2016-09-30 23:00:28 +10:00
client.publish(
2015-08-23 15:46:55 +10:00
'data',
new Buffer(data,client.config.encoding)
2015-08-23 15:46:55 +10:00
);
if(!client.config.sync){
return;
}
client.queue.next();
2015-08-23 15:46:55 +10:00
return;
}
2016-01-10 23:18:14 +11:00
if(!this.ipcBuffer){
2014-03-04 07:02:56 +11:00
this.ipcBuffer='';
2016-01-10 23:18:14 +11:00
}
2014-03-04 07:02:56 +11:00
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.');
2014-03-04 07:02:56 +11:00
return;
}
2014-03-04 07:02:56 +11:00
this.ipcBuffer='';
2016-01-10 23:18:14 +11:00
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);
2016-09-30 23:00:28 +10:00
client.publish(
message.type,
message.data
2014-02-22 20:13:31 +11:00
);
}
if(!client.config.sync){
return;
}
client.queue.next();
2014-02-22 20:13:31 +11:00
}
);
}
2016-09-30 23:00:28 +10:00
module.exports=Client;