Differential D9458 Diff 22705 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/AphlictFlashPolicyServer', __dirname);		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);

var debug = new JX.AphlictLog()		var debug = new JX.AphlictLog()
		epriestleyUnsubmitted Not Done Inline Actions What's this? epriestley: What's this?
.addConsole(console);		.addConsole(console);

var clients = new JX.AphlictListenerList();		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);
▲ Show 20 Lines • Show All 54 Lines • ▼ Show 20 Lines

var send_server = net.createServer(function(socket) {		var send_server = net.createServer(function(socket) {
var listener = clients.addListener(socket);		var listener = clients.addListener(socket);

debug.log('<%s> Connected from %s',		debug.log('<%s> Connected from %s',
listener.getDescription(),		listener.getDescription(),
socket.remoteAddress);		socket.remoteAddress);

		var buffer = new Buffer([]);
		var length = 0;

		epriestleyUnsubmitted Not Done Inline Actions I'd expect us to need to read lengths and parse messages off this stream? That is, this presumably only works by luck/accident/messages all being smaller than one packet right now? epriestley: I'd expect us to need to read lengths and parse messages off this stream? That is, this…
		socket.on('data', function(data) {
		buffer = Buffer.concat([buffer, new Buffer(data)]);

		while (buffer.length) {
		epriestleyUnsubmitted Not Done Inline Actions Probably needs a `return;` here if length is less than 2. epriestley: Probably needs a `return;` here if length is less than 2.
		if (!length) {
		epriestleyUnsubmitted Not Done Inline Actions Consider using 8-character padded ASCII text for consistency with transmissions and readability. epriestley: Consider using 8-character padded ASCII text for consistency with transmissions and readability.
		length = buffer.readUInt16BE(0);
		buffer = buffer.slice(2);
		}

		if (buffer.length < length) {
		// We need to wait for the rest of the data.
		return;
		epriestleyUnsubmitted Not Done Inline Actions Probably needs `return;`. epriestley: Probably needs `return;`.
		}

		var message;
		try {
		message = JSON.parse(buffer.toString('utf8', 0, length));
		joshuaspenceAuthorUnsubmitted Not Done Inline Actions I should `try`/`catch` this in case the data is not valid JSON. joshuaspence: I should `try`/`catch` this in case the data is not valid JSON.
		} catch (err) {
		debug.log('<%s> Received invalid data.', listener.getDescription());
		continue;
		} finally {
		buffer = buffer.slice(length);
		length = 0;
		}

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

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

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

		default:
		debug.log('<s> Unrecognized command.', listener.getDescription());
		}
		}
		});

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

socket.on('timeout', function() {		socket.on('timeout', function() {
debug.log('<%s> Timed Out', listener.getDescription());		debug.log('<%s> Timed Out', listener.getDescription());
});		});
Show All 20 Lines	if (request.method == 'POST') {

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('notification: ' + JSON.stringify(msg));
		joshuaspenceAuthorUnsubmitted Not Done Inline Actions Since the `AphlictMaster` needs to route notifications as well, perhaps the `subscribers` field should just be a part of `msg.data`? joshuaspence: Since the `AphlictMaster` needs to route notifications as well, perhaps the `subscribers` field…
		joshuaspenceAuthorUnsubmitted Not Done Inline Actions At the same time... maybe not. If we just blindly pass this stuff on then it would be possible for the client to determine what other clients are listening to. Whether or not this is an issue, I am not sure. joshuaspence: At the same time... maybe not. If we just blindly pass this stuff on then it would be possible…
		epriestleyUnsubmitted Not Done Inline Actions For now, I'd probably pass it as part of data. In the future, we could send just the intersection of the subscriber lists to the master, maybe, to split the difference? epriestley: For now, I'd probably pass it as part of data. In the future, we could send just the…
++messages_in;		++messages_in;
broadcast(msg);		transmit(msg);

response.writeHead(200, {'Content-Type': 'text/plain'});		response.writeHead(200, {'Content-Type': 'text/plain'});
} catch (err) {		} catch (err) {
response.statusCode = 400;		response.statusCode = 400;
response.write('400 Bad Request');		response.write('400 Bad Request');
} finally {		} finally {
response.end();		response.end();
}		}
Show All 22 Lines	var receive_server = http.createServer(function(request, response) {
} 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 transmit(msg) {
var listeners = clients.getListeners();		var listeners = clients.getListeners().filter(function(client) {
for (var id in listeners) {		return client.isSubscribedToAny(msg.subscribers);
var listener = listeners[id];		});
		epriestleyUnsubmitted Not Done Inline Actions (This is reasonable for now, but it would be vaguely nice to maintain a map eventually.) epriestley: (This is reasonable for now, but it would be vaguely nice to maintain a map eventually.)

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

try {		try {
listener.writeMessage(data);		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', error);		debug.log('<%s> Write Error: %s', error);
		epriestleyUnsubmitted Not Done Inline Actions (Is this missing an argument?) epriestley: (Is this missing an argument?)
		joshuaspenceAuthorUnsubmitted Not Done Inline Actions Yep, it is... let me fix. joshuaspence: Yep, it is... let me fix.
}		}
}		}
}		}

// 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);		debug.log('Started Server (PID %d)', process.pid);