1
0
Fork 0
mirror of https://we.phorge.it/source/phorge.git synced 2024-11-09 16:32:39 +01:00

Refactoring of the Aphlict server

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
This commit is contained in:
Joshua Spence 2015-01-19 07:48:48 +11:00
parent 19be32656f
commit 9a7ad972cd
6 changed files with 268 additions and 192 deletions

View file

@ -44,8 +44,7 @@ var config = parse_command_line_arguments(process.argv);
process.on('uncaughtException', function(err) { process.on('uncaughtException', function(err) {
var context = null; var context = null;
if ((err.code == 'EACCES') && if (err.code == 'EACCES' && err.path == config.log) {
(err.path == config.log)) {
context = util.format( context = util.format(
'Unable to open logfile ("%s"). Check that permissions are set ' + 'Unable to open logfile ("%s"). Check that permissions are set ' +
'correctly.', 'correctly.',
@ -65,9 +64,8 @@ process.on('uncaughtException', function(err) {
process.exit(1); process.exit(1);
}); });
var WebSocket;
try { try {
WebSocket = require('ws'); require('ws');
} catch (ex) { } catch (ex) {
throw new Error( throw new Error(
'You need to install the Node.js "ws" module for websocket support. ' + 'You need to install the Node.js "ws" module for websocket support. ' +
@ -88,7 +86,7 @@ if (ssl_config.enabled) {
// Add the logfile so we'll fail if we can't write to it. // Add the logfile so we'll fail if we can't write to it.
if (config.log) { if (config.log) {
debug.addLogfile(config.log); debug.addLog(config.log);
} }
// If we're just doing a configuration test, exit here before starting any // If we're just doing a configuration test, exit here before starting any
@ -98,166 +96,30 @@ if (config.test) {
process.exit(0); process.exit(0);
} }
var start_time = new Date().getTime(); JX.require('lib/AphlictAdminServer', __dirname);
var messages_out = 0; JX.require('lib/AphlictClientServer', __dirname);
var messages_in = 0;
var clients = new JX.AphlictListenerList(); var server;
function https_discard_handler(req, res) {
res.writeHead(501);
res.end('HTTP/501 Use Websockets\n');
}
var ws;
if (ssl_config.enabled) { if (ssl_config.enabled) {
var https_server = https.createServer({ server = https.createServer({
key: ssl_config.key, key: ssl_config.key,
cert: ssl_config.cert cert: ssl_config.cert
}, https_discard_handler).listen( }, function(req, res) {
config['client-port'], res.writeHead(501);
config['client-host']); res.end('HTTP/501 Use Websockets\n');
});
ws = new WebSocket.Server({server: https_server});
} else { } else {
ws = new WebSocket.Server({ server = http.createServer(function() {});
port: config['client-port'],
host: config['client-host'],
});
} }
ws.on('connection', function(ws) { var client_server = new JX.AphlictClientServer(server);
var listener = clients.addListener(ws); var admin_server = new JX.AphlictAdminServer();
function log() { client_server.setLogger(debug);
debug.log( admin_server.setLogger(debug);
util.format('<%s>', listener.getDescription()) + admin_server.setClientServer(client_server);
' ' +
util.format.apply(null, arguments));
}
log('Connected from %s.', ws._socket.remoteAddress); client_server.listen(config['client-port'], config['client-host']);
admin_server.listen(config['admin-port'], config['admin-host']);
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>');
}
});
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);
}
}
}
http.createServer(function(request, response) {
// 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);
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']);
debug.log('Started Server (PID %d)', process.pid); debug.log('Started Server (PID %d)', process.pid);

View file

@ -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,
}
});

View file

@ -0,0 +1,90 @@
var JX = require('javelin').JX;
JX.require('AphlictListenerList', __dirname);
JX.require('AphlictLog', __dirname);
var util = require('util');
var WebSocket = require('ws');
JX.install('AphlictClientServer', {
construct: function(server) {
this.setListenerList(new JX.AphlictListenerList());
this.setLogger(new JX.AphlictLog());
this._server = server;
},
members: {
_server: null,
listen: function() {
var self = this;
var server = this._server.listen.apply(this._server, arguments);
var wss = new WebSocket.Server({server: server});
wss.on('connection', function(ws) {
var listener = self.getListenerList().addListener(ws);
function log() {
self.getLogger().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>');
}
});
wss.on('close', function() {
self.getListenerList().removeListener(listener);
log('Disconnected.');
});
wss.on('error', function(err) {
log('Error: %s', err.message);
});
});
},
},
properties: {
listenerList: null,
logger: null,
}
});

View file

@ -50,8 +50,6 @@ JX.install('AphlictListener', {
writeMessage: function(message) { writeMessage: function(message) {
this._socket.send(JSON.stringify(message)); this._socket.send(JSON.stringify(message));
} },
},
}
}); });

View file

@ -39,7 +39,7 @@ JX.install('AphlictListenerList', {
}, },
getActiveListenerCount: function() { getActiveListenerCount: function() {
return Object.keys(this._listeners).length; return this._listeners.length;
}, },
getTotalListenerCount: function() { getTotalListenerCount: function() {
@ -48,12 +48,10 @@ JX.install('AphlictListenerList', {
_generateNextID: function() { _generateNextID: function() {
do { do {
this._nextID = ((this._nextID + 1) % 1000000000000); this._nextID = (this._nextID + 1) % 1000000000000;
} while (this._nextID in this._listeners); } while (this._nextID in this._listeners);
return this._nextID; return this._nextID;
} },
},
}
}); });

View file

@ -5,30 +5,25 @@ var util = require('util');
JX.install('AphlictLog', { JX.install('AphlictLog', {
construct: function() { construct: function() {
this._writeToLogs = []; this._consoles = [];
this._writeToConsoles = []; this._logs = [];
}, },
members: { members: {
_writeToConsoles: null, _consoles: null,
_writeToLogs: null, _logs: null,
addLogfile: function(path) {
var options = {
flags: 'a',
encoding: 'utf8',
mode: 0664,
};
var logfile = fs.createWriteStream(path, options);
this._writeToLogs.push(logfile);
addConsole: function(console) {
this._consoles.push(console);
return this; return this;
}, },
addConsole: function(console) { addLog: function(path) {
this._writeToConsoles.push(console); this._logs.push(fs.createWriteStream(path, {
flags: 'a',
encoding: 'utf8',
mode: 0664,
}));
return this; return this;
}, },
@ -38,15 +33,13 @@ JX.install('AphlictLog', {
str = '[' + date + '] ' + str; str = '[' + date + '] ' + str;
var ii; var ii;
for (ii = 0; ii < this._writeToConsoles.length; ii++) { for (ii = 0; ii < this._consoles.length; ii++) {
this._writeToConsoles[ii].log(str); this._consoles[ii].log(str);
} }
for (ii = 0; ii < this._writeToLogs.length; ii++) { for (ii = 0; ii < this._logs.length; ii++) {
this._writeToLogs[ii].write(str + '\n'); this._logs[ii].write(str + '\n');
} }
} },
},
}
}); });