Differential D11143 Diff 27053 support/aphlict/server/aphlict_server.js

Changeset View

Standalone View

support/aphlict/server/aphlict_server.js

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

var JX = require('./lib/javelin').JX;		var JX = require('./lib/javelin').JX;
		var http = require('http');
		var https = require('https');
		var util = require('util');
		var WebSocket = require('ws');
		joshuaspenceUnsubmitted Not Done Inline Actions Maybe we should `try`/`catch` this and provide installation instructions if it is missing. Otherwise, users who blindly update may not understand why Aphlict stopped working. joshuaspence: Maybe we should `try`/`catch` this and provide installation instructions if it is missing.
		var fs = require('fs');

JX.require('lib/AphlictFlashPolicyServer', __dirname);
JX.require('lib/AphlictListenerList', __dirname);		JX.require('lib/AphlictListenerList', __dirname);
JX.require('lib/AphlictLog', __dirname);		JX.require('lib/AphlictLog', __dirname);

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,		log: '/var/log/aphlict.log',
log: '/var/log/aphlict.log'		'ssl-key': null,
		'ssl-certificate': null
		joshuaspenceUnsubmitted Not Done Inline Actions Sort of ugly, maybe just `cert` and `key` would suffice. joshuaspence: Sort of ugly, maybe just `cert` and `key` would suffice.
};		};

for (var ii = 2; ii < argv.length; ii++) {		for (var ii = 2; ii < argv.length; ii++) {
var arg = argv[ii];		var arg = argv[ii];
var matches = arg.match(/^--([^=]+)=(.*)$/);		var matches = arg.match(/^--([^=]+)=(.*)$/);
if (!matches) {		if (!matches) {
throw new Error("Unknown argument '" + arg + "'!");		throw new Error("Unknown argument '" + arg + "'!");
}		}
if (!(matches[1] in config)) {		if (!(matches[1] in config)) {
throw new Error("Unknown argument '" + matches[1] + "'!");		throw new Error("Unknown argument '" + matches[1] + "'!");
}		}
config[matches[1]] = matches[2];		config[matches[1]] = matches[2];
}		}

config.port = parseInt(config.port, 10);		config.port = parseInt(config.port, 10);
config.admin = parseInt(config.admin, 10);		config.admin = parseInt(config.admin, 10);

return config;		return config;
}		}

var debug = new JX.AphlictLog()		var debug = new JX.AphlictLog()
.addConsole(console);		.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) {		if (config.logfile) {
debug.addLogfile(config.logfile);		debug.addLogfile(config.logfile);
}		}

if (process.getuid() !== 0) {
console.log(
"ERROR: " +
"This server must be run as root because it needs to bind to privileged " +
"port 843 to start a Flash policy server. It will downgrade to run as a " +
"less-privileged user after binding if you pass a user in the command " +
"line arguments with '--user=alincoln'.");
process.exit(1);
}

var net = require('net');
var http = require('http');

process.on('uncaughtException', function(err) {		process.on('uncaughtException', function(err) {
debug.log('\n<<< UNCAUGHT EXCEPTION! >>>\n' + err.stack);		debug.log('\n<<< UNCAUGHT EXCEPTION! >>>\n' + err.stack);

process.exit(1);		process.exit(1);
});		});

new JX.AphlictFlashPolicyServer()		var start_time = new Date().getTime();
.setDebugLog(debug)		var messages_out = 0;
.setAccessPort(config.port)		var messages_in = 0;
.start();

		var clients = new JX.AphlictListenerList();

net.createServer(function(socket) {		function https_discard_handler(req, res) {
var listener = clients.addListener(socket);		res.writeHead(501);
		res.end('HTTP/501 Use Websockets\n');
		}

debug.log('<%s> Connected from %s',		var ws;
listener.getDescription(),		if (config['ssl-key'] \|\| config['ssl-cert']) {
socket.remoteAddress);		var https_server = https.createServer({
		key: fs.readFileSync(config['ssl-key']),
		cert: fs.readFileSync(config['ssl-cert'])
		}, https_discard_handler).listen(config.port);

var buffer = new Buffer([]);		ws = new WebSocket.Server({server: https_server});
var length = 0;		} else {
		ws = new WebSocket.Server({host: config.host, port: config.port});
		}

socket.on('data', function(data) {		ws.on('connection', function(ws) {
buffer = Buffer.concat([buffer, new Buffer(data)]);		var listener = clients.addListener(ws);

while (buffer.length) {		function log() {
if (!length) {		debug.log(
length = buffer.readUInt16BE(0);		util.format('<%s>', listener.getDescription()) +
buffer = buffer.slice(2);		' ' +
		util.format.apply(null, arguments));
}		}

if (buffer.length < length) {		log('Connected from %s.', ws._socket.remoteAddress);
		joshuaspenceUnsubmitted Not Done Inline Actions Maybe include `ws._socket.remotePort` as well? joshuaspence: Maybe include `ws._socket.remotePort` as well?
// We need to wait for the rest of the data.
return;		ws.on('message', function(data) {
}		log('Received message: %s', data);

var message;		var message;
try {		try {
message = JSON.parse(buffer.toString('utf8', 0, length));		message = JSON.parse(data);
} catch (err) {		} catch (err) {
debug.log('<%s> Received invalid data.', listener.getDescription());		log('Message is invalid: %s', err.message);
continue;		return;
} finally {
buffer = buffer.slice(length);
length = 0;
}		}

debug.log('<%s> Received data: %s',
listener.getDescription(),
JSON.stringify(message));

switch (message.command) {		switch (message.command) {
case 'subscribe':		case 'subscribe':
debug.log(		log(
'<%s> Subscribed to: %s',		'Subscribed to: %s',
listener.getDescription(),
JSON.stringify(message.data));		JSON.stringify(message.data));
listener.subscribe(message.data);		listener.subscribe(message.data);
break;		break;

case 'unsubscribe':		case 'unsubscribe':
debug.log(		log(
'<%s> Unsubscribed from: %s',		'Unsubscribed from: %s',
listener.getDescription(),
JSON.stringify(message.data));		JSON.stringify(message.data));
listener.unsubscribe(message.data);		listener.unsubscribe(message.data);
break;		break;

default:		default:
debug.log('<s> Unrecognized command.', listener.getDescription());		log('Unrecognized command "%s".', message.command \|\| '<undefined>');
}
}		}
});		});

socket.on('close', function() {		ws.on('close', function() {
clients.removeListener(listener);		clients.removeListener(listener);
debug.log('<%s> Disconnected', listener.getDescription());		log('Disconnected.');
});		});

socket.on('timeout', function() {		ws.on('error', function(err) {
debug.log('<%s> Timed Out', listener.getDescription());		log('Error: %s', err.message);
});		});

socket.on('end', function() {
debug.log('<%s> Ended Connection', listener.getDescription());
});		});

socket.on('error', function(e) {
debug.log('<%s> Error: %s', listener.getDescription(), e);
});

}).listen(config.port);


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

function transmit(msg) {		function transmit(msg) {
var listeners = clients.getListeners().filter(function(client) {		var listeners = clients.getListeners().filter(function(client) {
return client.isSubscribedToAny(msg.subscribers);		return client.isSubscribedToAny(msg.subscribers);
});		});

for (var i = 0; i < listeners.length; i++) {		for (var i = 0; i < listeners.length; i++) {
var listener = listeners[i];		var listener = listeners[i];

try {		try {
listener.writeMessage(msg);		listener.writeMessage(msg);

++messages_out;		++messages_out;
debug.log('<%s> Wrote Message', listener.getDescription());		debug.log('<%s> Wrote Message', listener.getDescription());
} catch (error) {		} catch (error) {
clients.removeListener(listener);		clients.removeListener(listener);
debug.log('<%s> Write Error: %s', listener.getDescription(), error);		debug.log('<%s> Write Error: %s', listener.getDescription(), error);
}		}
}		}
}		}

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

http.createServer(function(request, response) {		http.createServer(function(request, response) {
// Publishing a notification.		// Publishing a notification.
if (request.url == '/') {		if (request.url == '/') {
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() {
try {		try {
var msg = JSON.parse(body);		var msg = JSON.parse(body);

debug.log('notification: ' + JSON.stringify(msg));		debug.log('Received notification: ' + JSON.stringify(msg));
++messages_in;		++messages_in;

try {		try {
transmit(msg);		transmit(msg);
response.writeHead(200, {'Content-Type': 'text/plain'});		response.writeHead(200, {'Content-Type': 'text/plain'});
} catch (err) {		} catch (err) {
debug.log(		debug.log(
'<%s> Internal Server Error! %s',		'<%s> Internal Server Error! %s',
Show All 30 Lines	if (request.url == '/') {
response.write(JSON.stringify(status));		response.write(JSON.stringify(status));
response.end();		response.end();
} else {		} else {
response.writeHead(404, 'Not Found');		response.writeHead(404, 'Not Found');
response.end();		response.end();
}		}
}).listen(config.admin, config.host);		}).listen(config.admin, config.host);

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

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