Skip to content

Commit

Permalink
Refactoring of the Aphlict server
Browse files Browse the repository at this point in the history 
Summary: Tidy the Aphlict server by splitting the functionality into two main modules, `AphlictClientServer` and `AphlictAdminServer. There is still further tidying that could be done here, but I feel that this puts us in a much better place.

Test Plan: Sent notifications via `/notification/status/`.

Reviewers: epriestley, #blessed_reviewers

Reviewed By: epriestley, #blessed_reviewers

Subscribers: Korvin, epriestley

Differential Revision: https://secure.phabricator.com/D11383
  • Loading branch information
@joshuaspence
joshuaspence committed Jan 18, 2015
1 parent 19be326 commit 9a7ad97
Show file tree
Hide file tree
Showing 6 changed files with 268 additions and 192 deletions.
176 changes: 19 additions & 157 deletions support/aphlict/server/aphlict_server.js
View file
Original file line number Diff line number Diff line change
Expand Up @@ -44,8 +44,7 @@ var config = parse_command_line_arguments(process.argv);

process.on('uncaughtException', function(err) {
var context = null;
if ((err.code == 'EACCES') &&
(err.path == config.log)) {
if (err.code == 'EACCES' && err.path == config.log) {
context = util.format(
'Unable to open logfile ("%s"). Check that permissions are set ' +
'correctly.',
Expand All @@ -65,9 +64,8 @@ process.on('uncaughtException', function(err) {
process.exit(1);
});

var WebSocket;
try {
WebSocket = require('ws');
require('ws');
} catch (ex) {
throw new Error(
'You need to install the Node.js "ws" module for websocket support. ' +
Expand All @@ -88,7 +86,7 @@ if (ssl_config.enabled) {

// Add the logfile so we'll fail if we can't write to it.
if (config.log) {
debug.addLogfile(config.log);
debug.addLog(config.log);
}

// If we're just doing a configuration test, exit here before starting any
Expand All @@ -98,166 +96,30 @@ if (config.test) {
process.exit(0);
}

var start_time = new Date().getTime();
var messages_out = 0;
var messages_in = 0;
JX.require('lib/AphlictAdminServer', __dirname);
JX.require('lib/AphlictClientServer', __dirname);

var clients = new JX.AphlictListenerList();

function https_discard_handler(req, res) {
res.writeHead(501);
res.end('HTTP/501 Use Websockets\n');
}

var ws;
var server;
if (ssl_config.enabled) {
var https_server = https.createServer({
server = https.createServer({
key: ssl_config.key,
cert: ssl_config.cert
}, https_discard_handler).listen(
config['client-port'],
config['client-host']);

ws = new WebSocket.Server({server: https_server});
} else {
ws = new WebSocket.Server({
port: config['client-port'],
host: config['client-host'],
});
}

ws.on('connection', function(ws) {
var listener = clients.addListener(ws);

function log() {
debug.log(
util.format('<%s>', listener.getDescription()) +
' ' +
util.format.apply(null, arguments));
}

log('Connected from %s.', ws._socket.remoteAddress);

ws.on('message', function(data) {
log('Received message: %s', data);

var message;
try {
message = JSON.parse(data);
} catch (err) {
log('Message is invalid: %s', err.message);
return;
}

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

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

default:
log('Unrecognized command "%s".', message.command || '<undefined>');
}
}, function(req, res) {
res.writeHead(501);
res.end('HTTP/501 Use Websockets\n');
});

ws.on('close', function() {
clients.removeListener(listener);
log('Disconnected.');
});

ws.on('error', function(err) {
log('Error: %s', err.message);
});
});

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

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

try {
listener.writeMessage(msg);

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

http.createServer(function(request, response) {
// Publishing a notification.
if (request.url == '/') {
if (request.method == 'POST') {
var body = '';

request.on('data', function(data) {
body += data;
});
var client_server = new JX.AphlictClientServer(server);
var admin_server = new JX.AphlictAdminServer();

request.on('end', function() {
try {
var msg = JSON.parse(body);
client_server.setLogger(debug);
admin_server.setLogger(debug);
admin_server.setClientServer(client_server);

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

try {
transmit(msg);
response.writeHead(200, {'Content-Type': 'text/plain'});
} catch (err) {
debug.log(
'<%s> Internal Server Error! %s',
request.socket.remoteAddress,
err);
response.writeHead(500, 'Internal Server Error');
}
} catch (err) {
debug.log(
'<%s> Bad Request! %s',
request.socket.remoteAddress,
err);
response.writeHead(400, 'Bad Request');
} finally {
response.end();
}
});
} else {
response.writeHead(405, 'Method Not Allowed');
response.end();
}
} else if (request.url == '/status/') {
var status = {
'uptime': (new Date().getTime() - start_time),
'clients.active': clients.getActiveListenerCount(),
'clients.total': clients.getTotalListenerCount(),
'messages.in': messages_in,
'messages.out': messages_out,
'log': config.log,
'version': 6
};

response.writeHead(200, {'Content-Type': 'application/json'});
response.write(JSON.stringify(status));
response.end();
} else {
response.writeHead(404, 'Not Found');
response.end();
}
}).listen(config['admin-port'], config['admin-host']);
client_server.listen(config['client-port'], config['client-host']);
admin_server.listen(config['admin-port'], config['admin-host']);

debug.log('Started Server (PID %d)', process.pid);
135 changes: 135 additions & 0 deletions support/aphlict/server/lib/AphlictAdminServer.js
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,135 @@
var JX = require('javelin').JX;

JX.require('AphlictListenerList', __dirname);

var http = require('http');

JX.install('AphlictAdminServer', {

construct: function() {
this.setLogger(new JX.AphlictLog());

this._startTime = new Date().getTime();
this._messagesIn = 0;
this._messagesOut = 0;

var handler = this._handler.bind(this);
this._server = http.createServer(handler);
},

members: {
_messagesIn: null,
_messagesOut: null,
_server: null,
_startTime: null,

getListeners: function() {
return this.getListenerList().getListeners();
},

getListenerList: function() {
return this.getClientServer().getListenerList();
},

listen: function() {
return this._server.listen.apply(this._server, arguments);
},

_handler: function(request, response) {
var self = this;

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

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

request.on('end', function() {
try {
var msg = JSON.parse(body);

self.getLogger().log(
'Received notification: ' + JSON.stringify(msg));
++this._messagesIn;

try {
self._transmit(msg);
response.writeHead(200, {'Content-Type': 'text/plain'});
} catch (err) {
self.getLogger().log(
'<%s> Internal Server Error! %s',
request.socket.remoteAddress,
err);
response.writeHead(500, 'Internal Server Error');
}
} catch (err) {
self.getLogger().log(
'<%s> Bad Request! %s',
request.socket.remoteAddress,
err);
response.writeHead(400, 'Bad Request');
} finally {
response.end();
}
});
} else {
response.writeHead(405, 'Method Not Allowed');
response.end();
}
} else if (request.url == '/status/') {
var status = {
'uptime': (new Date().getTime() - this._startTime),
'clients.active': this.getListenerList().getActiveListenerCount(),
'clients.total': this.getListenerList().getTotalListenerCount(),
'messages.in': this._messagesIn,
'messages.out': this._messagesOut,
'version': 6
};

response.writeHead(200, {'Content-Type': 'application/json'});
response.write(JSON.stringify(status));
response.end();
} else {
response.writeHead(404, 'Not Found');
response.end();
}
},

/**
* Transmits a message to all subscribed listeners.
*/
_transmit: function(message) {
var listeners = this.getListeners().filter(function(client) {
return client.isSubscribedToAny(message.subscribers);
});

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

try {
listener.writeMessage(message);

++this._messagesOut;
this.getLogger().log(
'<%s> Wrote Message',
listener.getDescription());
} catch (error) {
this.getListenerList().removeListener(listener);
this.getLogger().log(
'<%s> Write Error: %s',
listener.getDescription(),
error);
}
}
},
},

properties: {
clientServer: null,
logger: null,
}

});
Loading

0 comments on commit 9a7ad97

Please sign in to comment.