dataflo.ws/arex.js

var cluster   = require ('cluster'),
	WebSocket = require ('ws'),
	uuid      = require ('node-uuid'),
	util      = require ('util'),
	EventEmitter = require ('events').EventEmitter;

var data = require ('./test-data.json');

var testConf = {
	threads:       4,
	connPerThread: 1000,
	instruments:   1000,
};

function toJSONStingArray(orderArray) {
	var ret = [];
	for(var i = 0; i < orderArray.length; i++)
		ret[i] = JSON.stringify(orderArray[i]);
	return ret;
}

function launch () {
	if (cluster.isMaster) {
		for (var i = 0; i < testConf.threads; i++) {
			cluster.fork();
		}
	} else if (cluster.isWorker) {
		console.log ('I am worker #' + cluster.worker.id);

		var instruments = prepareInstruments (testConf.instruments);

		var wsc = new wsClient (cluster);

		wsc.registerAll (instruments);

		wsc.on ('register', function (err) {
			if (err) {
				// TODO: pass error to the script sute
				return;
			}
			// TODO: pass ok to the script suite
			wsc.executeAll (testConf.connPerThread);
		});

		wsc.on ('execute-connection', function (err) {
			if (err) {
				// TODO: pass error to the script sute
				return;
			}
			// TODO: pass ok to the script suite
		});

		setTimeout (function () {

		}, 1000);
	}
}

function prepareInstruments (instrumentCount) {
	var instruments = [];

	for (var k = 0; k < instrumentCount; k++) {
		instruments[k] = new Buffer (uuid.v1(null, [])).toString('base64');
	}

	return instruments;
}

/**
 * class for ws requests
 */
function wsClient (cluster) {
	this.cluster = cluster;
}

util.inherits (wsClient, EventEmitter);

wsClient.prototype.registerAll = function (instruments) {
	var socketRegister = new WebSocket('ws://127.0.0.1:8080/library/text/register');

	// here we start to send messages
	socketRegister.on ('open', function() {
		// actually we're trying to register instruments in parallel from every cluster worker
		// ???
		console.time ("register time "+this.cluster.worker.id);
		for (var k = 0; k < instruments.length; k++) {
			// console.log('register ' + instruments[k]);
			socketRegister.send (instruments[k]);
		}
		console.log("register: " + this.cluster.worker.id + ": " + instruments.length + ' messages was sent');
	}.bind (this));

	var l = 0;

	// this is called on every received message
	socketRegister.on('message', function(data, flags) {
		// TODO: check that we actually get all instruments registered
		// maybe using data?
		l++;
		if (l == instruments.length) {
			console.timeEnd("register time "+this.cluster.worker.id);
			console.log("register: " + this.cluster.worker.id + ": all messages was delivered.");
			this.emit ('register', null);
		}
	}.bind (this));

	// if we don't have confirmations for every single instrument?
	setTimeout (function () {
		// TODO: set test to fail
		this.emit ('register', true);
	}.bind (this), 1000);
}

wsClient.prototype.execute = function (connId) {
	var execute = new WebSocket ('ws://127.0.0.1:8080/library/text/execute');
	var m = 0;
	var failure = null;

	// Ensure real instrumentId is properly escaped
	var sellObject =
		{id: "9g8G2mofRRiexY+b6bCYeg==", instruction: "sell",type: "limit", attributes:["AON"], price: 1000000,quantity: 1100000, instrumentId: "!!!SPLIT!!!"};
	var sellData = JSON.stringify (sellObject).split ('"!!!SPLIT!!!"');
	var buyObject = [
		{id: "dYo+hyMASEiCnT5+RVLWAw==", instruction: "buy", type: "limit", attributes:["AON"], price: 100000, quantity: 110000, instrumentId: "!!!SPLIT!!!"},
		{id: "KjGjpMmDQ4WaHwLg3w6prQ==", instruction: "buy", type: "limit", attributes:["AON"], price: 100000, quantity: 110000, instrumentId: "!!!SPLIT!!!"},
		{id: "zRDnn1ELRU2oBvLUIvjf8g==", instruction: "buy", type: "limit", attributes:["AON"], price: 100000, quantity: 110000, instrumentId: "!!!SPLIT!!!"},
		{id: "2LNF0tA6QbqDSMi1HcTBgA==", instruction: "buy", type: "limit", attributes:["AON"], price: 100000, quantity: 110000, instrumentId: "!!!SPLIT!!!"},
		{id: "bXHhJts9TtCP/ADcpZOYEg==", instruction: "buy", type: "limit", attributes:["AON"], price: 100000, quantity: 110000, instrumentId: "!!!SPLIT!!!"},
		{id: "Xj1k+TVlQweEzU5hB1M8ug==", instruction: "buy", type: "limit", attributes:["AON"], price: 100000, quantity: 110000, instrumentId: "!!!SPLIT!!!"},
		{id: "UhnSakG3RCuSeQuFcY+y4g==", instruction: "buy", type: "limit", attributes:["AON"], price: 100000, quantity: 110000, instrumentId: "!!!SPLIT!!!"},
		{id: "jf1xQQEoS6ifvMC7uk08eQ==", instruction: "buy", type: "limit", attributes:["AON"], price: 100000, quantity: 110000, instrumentId: "!!!SPLIT!!!"},
		{id: "2LKzvosFSzSQKYkgk0iEuA==", instruction: "buy", type: "limit", attributes:["AON"], price: 100000, quantity: 110000, instrumentId: "!!!SPLIT!!!"},
		{id: "+Nm5lkZeRjuiha0KrZHcEg==", instruction: "buy", type: "limit", attributes:["AON"], price: 100000, quantity: 110000, instrumentId: "!!!SPLIT!!!"}
	];
	var buyData = buyObject.map (function (o) {return JSON.stringify (o).split ('"!!!SPLIT!!!"')});

	execute.on ('message', function (data, flags) {
		// console.log('data', data);

		if (
			data === "null"
			|| data.length === 0
			|| data === "[]"
			|| JSON.parse(data)[0].message !== 'accept'
		) {
			failure = true;
			// throw new Error("FAILED: execute: data=" + data);
		}

		m++;

		if (m == K * 10) {
			console.timeEnd("execute time " + this.cluster.worker.id);
			console.log("execute: " + this.cluster.worker.id + ": all messages was delivered.");
			this.emit ('execute-connection', failure);
		}

	}.bind (this));

	execute.on ('open',  function() {
		console.time("execute time " + this.cluster.worker.id);
		for (var k = 0; k < K; k++) {
			var  instrumentId = instruments[k];
			// console.log("execute sell: " + sellData);
			execute.send(sellData[0] + instrumentId + sellData[1]);
			for (var i = 0; i < buyData.length; i++) {
				// console.log("execute buy: " + b);
				execute.send(buyData[i][0] + instrumentId + buyData[i][1]);
			}
		}
		console.log("execute: " + cluster.worker.id + ": " + (K * buyData.length) + ' messages was sended');
	}.bind (this));
}

wsClient.prototype.executeAll = function (connections) {
	for (var connId = 0; connId < connections; connId ++) {
		// create as much concurrent connections as you can
		this.execute (connId);
	}

}