crypto-stream.js

/***
 * Node modules
 */

var events = require('events');
var Stream = require('stream');
var util = require('util');
var path = require('path');
var crypto = require('crypto');
var fs = require('fs');
//
module.exports = {}

/**
 *
 */
var Write = module.exports.Write = function() {
	Stream.call(this);
	this.key = ''

	this.readable = this.writable = true;
}
/***
 * Make it an event
 */
util.inherits(Write, Stream);

/**
 *
 */
Write.prototype.setKey = function(key) {
	this.key = key;
	return this
}
/**
 *
 */
Write.prototype.write = function(str) {
	//console.log('crypto.Wrtie', str)
	var cipher = crypto.createCipher('aes-256-cbc', this.key)
	this.emit('data', cipher.update(str, 'utf8', 'hex') + cipher['final']('hex'))
}
/**
 *
 */
Write.prototype.end = function(str) {
	if(str) {
		this.write(str)
	}
	this.emit('end')
};
/***
 *
 *
 *
 */
var Read = module.exports.Read = function() {
	Stream.call(this);
	this.readable = this.writable = true;
	this.key = ''

};
/***
 * Make it an event
 */
util.inherits(Read, Stream);

/**
 *
 */
Read.prototype.setKey = function(key) {
	this.key = key;
	return this
}
/**
 *
 */
Read.prototype.write = function(str) {
	//console.log('crypto.Read', str)
	var decipher = crypto.createDecipher('aes-256-cbc', this.key);
	this.emit('data', decipher.update(str, 'hex', 'utf8') + decipher['final']('utf8'));
}
/**
 *
 */
Read.prototype.end = function(str) {
	if(str) {
		this.write(str)
	}
	this.emit('end')
};

json-stream.js

/***
 * Node modules
 */

var events = require('events');
var Stream = require('stream');
var util = require('util');
var path = require('path');
var crypto = require('crypto');
var fs = require('fs');
//
module.exports = {}

/**
 *
 */
var Read = module.exports.Read = function() {
	Stream.call(this);
	this.kwy = ''

	this.readable = this.writable = true;

}
/***
 * Make it an event
 */
util.inherits(Read, Stream);

/**
 *
 */
Read.prototype.write = function(str) {
	//console.log('json.Read', str)
	this.emit('data', JSON.parse(str))
}
/**
 *
 */
Read.prototype.end = function(str) {
	if(str) {
		this.emit('data', JSON.parse(str))
	}
	this.emit('end')
};

/**
 *
 */
var Write = module.exports.Write = function() {
	Stream.call(this);
	this.kwy = ''

	this.readable = this.writable = true;

}
/***
 * Make it an event
 */
util.inherits(Write, Stream);

/**
 *
 */
Write.prototype.write = function(str) {
	//console.log('json.Write', str)
	this.emit('data', JSON.stringify(str))
}
/**
 *
 */
Write.prototype.end = function(str) {
	if(str) {
		this.emit('data', JSON.stringify(str))
	}
	this.emit('end')
};

socket-split-stream.js

/***
 * Node modules
 */

var events = require('events');
var Stream = require('stream');
var util = require('util');
var path = require('path');
var crypto = require('crypto');
var fs = require('fs');
//
module.exports = {}

/**
 *
 */
var Read = module.exports.Read = function() {
	Stream.call(this);
	this.buffer = []

	this.readable = this.writable = true;

}
/***
 * Make it an event
 */
util.inherits(Read, Stream);

/**
 *
 */
Read.prototype.write = function(str) {
	if(str.indexOf('\n') > -1) {

		var message = this.buffer.join('');
		var data = str.split('\n');
		message += data.shift();

		this.buffer = [];

		this.emit('data', message);

		data = data.join('\n');

		if(data.length) {
			this.write(data);
		}
	} else {
		this.buffer.push(str);
	}
}
/**
 *
 */
Read.prototype.end = function(str) {
	if(str) {
		this.write(str)
	}
	this.emit('end')
};

/**
 *
 */
var Write = module.exports.Write = function() {
	Stream.call(this);
	this.buffer = []

	this.readable = this.writable = true;

}
/***
 * Make it an event
 */
util.inherits(Write, Stream);

/**
 *
 */
Write.prototype.write = function(str) {
	this.emit('data', str + '\n')
}
/**
 *
 */
Write.prototype.end = function(str) {
	if(str) {
		this.write(str)
	}
	this.emit('end')
};

test.js

var net = require('net');

var Split = require('./socket-split-stream')

var Cyrpt = require('./crypto-stream')

var json = require('./json-stream')

var event = 'rpc';
var key = 'somegreatkey';
var port = 8124;

var server = net.createServer(function(socket) {//'connection' listener
	socket.setEncoding('utf8')

	socket.pipe(new Split.Read()).pipe((new Cyrpt.Read()).setKey(key)).pipe(new json.Read());

	var jStream = new json.Write()
	jStream.pipe((new Cyrpt.Write()).setKey(key)).pipe(new Split.Write()).pipe(socket)
	setTimeout(function() {
		jStream.write({
			a : true
		})
	}, 1000)

});
server.listen(port, function() {//'listening' listener
	console.log('server bound');

	var socket = net.connect(port, function() {//'connect' listener
		socket.setEncoding('utf8')
		console.log('client connected');
		socket.pipe(new Split.Read()).pipe((new Cyrpt.Read()).setKey(key)).pipe(new json.Read()).on('data', function(data) {
			console.log('client', data)
		});

		var jStream = new json.Write()
		jStream.pipe((new Cyrpt.Write()).setKey(key)).pipe(new Split.Write()).pipe(socket)

	});

});