Differential D8256 Diff 19669 support/aphlict/server/aphlict_server.js

Changeset View

Standalone View

support/aphlict/server/aphlict_server.js

/**		/**
* Notification server. Launch with:		* Notification server. Launch with:
*		*
* sudo node aphlict_server.js --user=aphlict		* sudo node aphlict_server.js --user=aphlict
*		*
* You can also specify `port`, `admin`, `host` and `log`.		* You can also specify `port`, `admin`, `host` and `log`.
*/		*/

var JX = require('./lib/javelin').JX;		var JX = require('./lib/javelin').JX;
JX.require('lib/AphlictIDGenerator', __dirname);

var id_generator = new JX.AphlictIDGenerator();		JX.require('lib/AphlictListenerList', __dirname);
		JX.require('lib/AphlictLog', __dirname);

		var debug = new JX.AphlictLog()
		.addConsole(console);

		var clients = new JX.AphlictListenerList();

var config = parse_command_line_arguments(process.argv);		var config = parse_command_line_arguments(process.argv);

		if (config.logfile) {
		debug.addLogfile(config.logfile);
		}

function parse_command_line_arguments(argv) {		function parse_command_line_arguments(argv) {
var config = {		var config = {
port : 22280,		port : 22280,
admin : 22281,		admin : 22281,
host : '127.0.0.1',		host : '127.0.0.1',
user : null,		user : null,
log: '/var/log/aphlict.log'		log: '/var/log/aphlict.log'
};		};
Show All 25 Lines	console.log(
"line arguments with '--user=alincoln'.");		"line arguments with '--user=alincoln'.");
process.exit(1);		process.exit(1);
}		}

var net = require('net');		var net = require('net');
var http = require('http');		var http = require('http');
var url = require('url');		var url = require('url');
var querystring = require('querystring');		var querystring = require('querystring');
var fs = require('fs');

// set up log file
var logfile = fs.createWriteStream(
config.log,
{
flags: 'a',
encoding: null,
mode: 0666
});

function log(str) {
console.log(str);
logfile.write(str + '\n');
}

process.on('uncaughtException', function (err) {		process.on('uncaughtException', function (err) {
log("\n<<< UNCAUGHT EXCEPTION! >>>\n\n" + err);		log("\n<<< UNCAUGHT EXCEPTION! >>>\n\n" + err);
process.exit(1);		process.exit(1);
});		});

log('----- ' + (new Date()).toLocaleString() + ' -----\n');

function getFlashPolicy() {		function getFlashPolicy() {
return [		return [
'<?xml version="1.0"?>',		'<?xml version="1.0"?>',
'<!DOCTYPE cross-domain-policy SYSTEM ' +		'<!DOCTYPE cross-domain-policy SYSTEM ' +
'"http://www.macromedia.com/xml/dtds/cross-domain-policy.dtd">',		'"http://www.macromedia.com/xml/dtds/cross-domain-policy.dtd">',
'<cross-domain-policy>',		'<cross-domain-policy>',
'<allow-access-from domain="*" to-ports="'+config.port+'"/>',		'<allow-access-from domain="*" to-ports="'+config.port+'"/>',
'</cross-domain-policy>'		'</cross-domain-policy>'
].join('\n');		].join('\n');
}		}

net.createServer(function(socket) {		net.createServer(function(socket) {
socket.write(getFlashPolicy() + '\0');		socket.write(getFlashPolicy() + '\0');
socket.end();		socket.end();

log('[' + socket.remoteAddress + '] Sent Flash Policy');		debug.log('[' + socket.remoteAddress + '] Sent Flash Policy');

socket.on('error', function (e) {		socket.on('error', function (e) {
log('Error in policy server: ' + e);		debug.log('Error in policy server: ' + e);
});		});
}).listen(843);		}).listen(843);


function write_json(socket, data) {
var serial = JSON.stringify(data);
var length = Buffer.byteLength(serial, 'utf8');
length = length.toString();
while (length.length < 8) {
length = '0' + length;
}
socket.write(length + serial);
}


var clients = {};
var current_connections = 0;

var send_server = net.createServer(function(socket) {		var send_server = net.createServer(function(socket) {
var client_id = id_generator.generateNext();		var listener = clients.addListener(socket);
var client_name = '[' + socket.remoteAddress + '] [#' + client_id + '] ';

clients[client_id] = socket;		debug.log('<%s> Connected from %s',
current_connections++;		listener.getDescription(),
log(client_name + 'connected\t\t(' +		socket.remoteAddress);
current_connections + ' current connections)');

socket.on('close', function() {		socket.on('close', function() {
delete clients[client_id];		clients.removeListener(listener);
current_connections--;		debug.log('<%s> Disconnected', listener.getDescription());
log(client_name + 'closed\t\t(' +
current_connections + ' current connections)');
});		});

socket.on('timeout', function() {		socket.on('timeout', function() {
log(client_name + 'timed out!');		debug.log('<%s> Timed Out', listener.getDescription());
});		});

socket.on('end', function() {		socket.on('end', function() {
log(client_name + 'ended the connection');		debug.log('<%s> Ended Connection', listener.getDescription());
// node automatically closes half-open connections
});		});

socket.on('error', function (e) {		socket.on('error', function (e) {
log(client_name + 'Uncaught error in send server: ' + e);		debug.log('<%s> Error: %s', listener.getDescription(), e);
});		});

}).listen(config.port);		}).listen(config.port);


var messages_out = 0;		var messages_out = 0;
var messages_in = 0;		var messages_in = 0;
var start_time = new Date().getTime();		var start_time = new Date().getTime();

var receive_server = http.createServer(function(request, response) {		var receive_server = http.createServer(function(request, response) {
response.writeHead(200, {'Content-Type' : 'text/plain'});		response.writeHead(200, {'Content-Type' : 'text/plain'});

// Publishing a notification.		// Publishing a notification.
if (request.method == 'POST') {		if (request.method == 'POST') {
var body = '';		var body = '';

request.on('data', function (data) {		request.on('data', function (data) {
body += data;		body += data;
});		});

request.on('end', function () {		request.on('end', function () {
++messages_in;		++messages_in;

var data = querystring.parse(body);		var data = querystring.parse(body);
log('notification: ' + JSON.stringify(data));		debug.log('notification: ' + JSON.stringify(data));
broadcast(data);		broadcast(data);
response.end();		response.end();
});		});
} else if (request.url == '/status/') {		} else if (request.url == '/status/') {
request.on('data', function(data) {		request.on('data', function(data) {
// We just ignore the request data, but newer versions of Node don't		// We just ignore the request data, but newer versions of Node don't
// get to 'end' if we don't process the data. See T2953.		// get to 'end' if we don't process the data. See T2953.
});		});

request.on('end', function() {		request.on('end', function() {
var status = {		var status = {
'uptime': (new Date().getTime() - start_time),		'uptime': (new Date().getTime() - start_time),
'clients.active': current_connections,		'clients.active': clients.getActiveListenerCount(),
'clients.total': id_generator.getTotalCount(),		'clients.total': clients.getTotalListenerCount(),
'messages.in': messages_in,		'messages.in': messages_in,
'messages.out': messages_out,		'messages.out': messages_out,
'log': config.log,		'log': config.log,
'version': 3		'version': 3
};		};

response.write(JSON.stringify(status));		response.write(JSON.stringify(status));
response.end();		response.end();
});		});
} else {		} else {
response.statusCode = 400;		response.statusCode = 400;
response.write('400 Bad Request');		response.write('400 Bad Request');
response.end();		response.end();
}		}

}).listen(config.admin, config.host);		}).listen(config.admin, config.host);

function broadcast(data) {		function broadcast(data) {
for (var client_id in clients) {		var listeners = clients.getListeners();
		for (var id in listeners) {
		var listener = listeners[id];
try {		try {
write_json(clients[client_id], data);		listener.writeMessage(data);

++messages_out;		++messages_out;
log('wrote to client ' + client_id);		debug.log('<%s> Wrote Message', listener.getDescription());
} catch (error) {		} catch (error) {
delete clients[client_id];		clients.removeListener(listener);
current_connections--;		debug.log('<%s> Write Error: %s', error);
log('ERROR: could not write to client ' + client_id);
}		}
}		}
}		}

// If we're configured to drop permissions, get rid of them now that we've		// If we're configured to drop permissions, get rid of them now that we've
// bound to the ports we need and opened logfiles.		// bound to the ports we need and opened logfiles.
if (config.user) {		if (config.user) {
process.setuid(config.user);		process.setuid(config.user);
}		}

		debug.log('Started Server (PID %d)', process.pid);