Page MenuHomestyx hydra
Files F10048779

No OneTemporary
Actions

View Options

diff --git a/support/aphlict/server/aphlict_server.js b/support/aphlict/server/aphlict_server.js
index a567b0874f..4d6e975dd3 100644
--- a/support/aphlict/server/aphlict_server.js
+++ b/support/aphlict/server/aphlict_server.js
@@ -1,135 +1,178 @@
'use strict';
var JX = require('./lib/javelin').JX;
var http = require('http');
var https = require('https');
var util = require('util');
var fs = require('fs');
function parse_command_line_arguments(argv) {
var config = {
'client-port': 22280,
'admin-port': 22281,
'client-host': '0.0.0.0',
'admin-host': '127.0.0.1',
log: '/var/log/aphlict.log',
'ssl-key': null,
'ssl-cert': null,
test: false
};
for (var ii = 2; ii < argv.length; ii++) {
var arg = argv[ii];
var matches = arg.match(/^--([^=]+)=(.*)$/);
if (!matches) {
throw new Error('Unknown argument "' + arg + '"!');
}
if (!(matches[1] in config)) {
throw new Error('Unknown argument "' + matches[1] + '"!');
}
config[matches[1]] = matches[2];
}
config['client-port'] = parseInt(config['client-port'], 10);
config['admin-port'] = parseInt(config['admin-port'], 10);
return config;
}
require('./lib/AphlictLog');
var debug = new JX.AphlictLog()
.addConsole(console);
var config = parse_command_line_arguments(process.argv);
function set_exit_code(code) {
process.on('exit', function() {
process.exit(code);
});
}
process.on('uncaughtException', function(err) {
var context = null;
if (err.code == 'EACCES' && err.path == config.log) {
context = util.format(
'Unable to open logfile ("%s"). Check that permissions are set ' +
'correctly.',
err.path);
}
var message = [
'\n<<< UNCAUGHT EXCEPTION! >>>',
];
if (context) {
message.push(context);
}
message.push(err.stack);
debug.log(message.join('\n\n'));
set_exit_code(1);
});
// Add the logfile so we'll fail if we can't write to it.
if (config.log) {
debug.addLog(config.log);
}
try {
require('ws');
} catch (ex) {
throw new Error(
'You need to install the Node.js "ws" module for websocket support. ' +
'See "Notifications User Guide: Setup and Configuration" in the ' +
'documentation for instructions. ' + ex.toString());
}
// NOTE: Require these only after checking for the "ws" module, since they
// depend on it.
require('./lib/AphlictAdminServer');
require('./lib/AphlictClientServer');
var ssl_config = {
enabled: (config['ssl-key'] || config['ssl-cert'])
};
// Load the SSL certificates (if any were provided) now, so that runs with
// `--test` will see any errors.
if (ssl_config.enabled) {
ssl_config.key = fs.readFileSync(config['ssl-key']);
ssl_config.cert = fs.readFileSync(config['ssl-cert']);
+} else {
+ ssl_config.key = null;
+ ssl_config.cert = null;
}
+var servers = [];
+
+servers.push({
+ type: 'client',
+ port: config['client-port'],
+ listen: config['client-host'],
+ 'ssl.key': ssl_config.key,
+ 'ssl.certificate': ssl_config.cert
+});
+
+servers.push({
+ type: 'admin',
+ port: config['admin-port'],
+ listen: config['admin-host'],
+ 'ssl.key': null,
+ 'ssl.cert': null
+});
+
// If we're just doing a configuration test, exit here before starting any
// servers.
if (config.test) {
debug.log('Configuration test OK.');
set_exit_code(0);
return;
}
-var server;
-if (ssl_config.enabled) {
- server = https.createServer({
- key: ssl_config.key,
- cert: ssl_config.cert
- }, function(req, res) {
- res.writeHead(501);
- res.end('HTTP/501 Use Websockets\n');
- });
-} else {
- server = http.createServer(function() {});
-}
+var aphlict_servers = [];
+var aphlict_clients = [];
+var aphlict_admins = [];
+
+var ii;
+for (ii = 0; ii < servers.length; ii++) {
+ var server = servers[ii];
+ var is_client = (server.type == 'client');
+
+ var http_server;
+ if (server['ssl.key']) {
+ var https_config = {
+ key: server['ssl.key'],
+ cert: server['ssl.cert']
+ };
+
+ http_server = https.createServer(https_config);
+ } else {
+ http_server = http.createServer();
+ }
+
+ var aphlict_server;
+ if (is_client) {
+ aphlict_server = new JX.AphlictClientServer(http_server);
+ } else {
+ aphlict_server = new JX.AphlictAdminServer(http_server);
+ }
-var client_server = new JX.AphlictClientServer(server);
-var admin_server = new JX.AphlictAdminServer();
+ aphlict_server.setLogger(debug);
+ aphlict_server.listen(server.port, server.listen);
-client_server.setLogger(debug);
-admin_server.setLogger(debug);
-admin_server.setClientServer(client_server);
+ aphlict_servers.push(aphlict_server);
-client_server.listen(config['client-port'], config['client-host']);
-admin_server.listen(config['admin-port'], config['admin-host']);
+ if (is_client) {
+ aphlict_clients.push(aphlict_server);
+ } else {
+ aphlict_admins.push(aphlict_server);
+ }
+}
+
+for (ii = 0; ii < aphlict_admins.length; ii++) {
+ var admin_server = aphlict_admins[ii];
+ admin_server.setClientServers(aphlict_clients);
+}
debug.log('Started Server (PID %d)', process.pid);
diff --git a/support/aphlict/server/lib/AphlictAdminServer.js b/support/aphlict/server/lib/AphlictAdminServer.js
index f9caddc2bb..c08f09b70c 100644
--- a/support/aphlict/server/lib/AphlictAdminServer.js
+++ b/support/aphlict/server/lib/AphlictAdminServer.js
@@ -1,142 +1,179 @@
'use strict';
var JX = require('./javelin').JX;
require('./AphlictListenerList');
var http = require('http');
var url = require('url');
JX.install('AphlictAdminServer', {
- construct: function() {
- this.setLogger(new JX.AphlictLog());
-
+ construct: function(server) {
this._startTime = new Date().getTime();
this._messagesIn = 0;
this._messagesOut = 0;
- var handler = this._handler.bind(this);
- this._server = http.createServer(handler);
+ server.on('request', JX.bind(this, this._onrequest));
+ this._server = server;
+ this._clientServers = [];
+ },
+
+ properties: {
+ clientServers: null,
+ logger: null,
},
members: {
_messagesIn: null,
_messagesOut: null,
_server: null,
_startTime: null,
- getListenerList: function(instance) {
- return this.getClientServer().getListenerList(instance);
+ getListenerLists: function(instance) {
+ var clients = this.getClientServers();
+
+ var lists = [];
+ for (var ii = 0; ii < clients.length; ii++) {
+ lists.push(clients[ii].getListenerList(instance));
+ }
+ return lists;
+ },
+
+ log: function() {
+ var logger = this.getLogger();
+ if (!logger) {
+ return;
+ }
+
+ logger.log.apply(logger, arguments);
+
+ return this;
},
listen: function() {
return this._server.listen.apply(this._server, arguments);
},
- _handler: function(request, response) {
+ _onrequest: function(request, response) {
var self = this;
var u = url.parse(request.url, true);
var instance = u.query.instance || '/';
// Publishing a notification.
if (u.pathname == '/') {
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(
+ self.log(
'Received notification (' + instance + '): ' +
JSON.stringify(msg));
++self._messagesIn;
try {
self._transmit(instance, msg);
response.writeHead(200, {'Content-Type': 'text/plain'});
} catch (err) {
- self.getLogger().log(
+ self.log(
'<%s> Internal Server Error! %s',
request.socket.remoteAddress,
err);
response.writeHead(500, 'Internal Server Error');
}
} catch (err) {
- self.getLogger().log(
+ self.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 (u.pathname == '/status/') {
- var status = {
- 'instance': instance,
- 'uptime': (new Date().getTime() - this._startTime),
- 'clients.active': this.getListenerList(instance)
- .getActiveListenerCount(),
- 'clients.total': this.getListenerList(instance)
- .getTotalListenerCount(),
- 'messages.in': this._messagesIn,
- 'messages.out': this._messagesOut,
- 'version': 7
- };
-
- response.writeHead(200, {'Content-Type': 'application/json'});
- response.write(JSON.stringify(status));
- response.end();
+ this._handleStatusRequest(request, response, instance);
} else {
response.writeHead(404, 'Not Found');
response.end();
}
},
+ _handleStatusRequest: function(request, response, instance) {
+ var active_count = 0;
+ var total_count = 0;
+
+ var lists = this.getListenerLists(instance);
+ for (var ii = 0; ii < lists.length; ii++) {
+ var list = lists[ii];
+ active_count += list.getActiveListenerCount();
+ total_count += list.getTotalListenerCount();
+ }
+
+ var server_status = {
+ 'instance': instance,
+ 'uptime': (new Date().getTime() - this._startTime),
+ 'clients.active': active_count,
+ 'clients.total': total_count,
+ 'messages.in': this._messagesIn,
+ 'messages.out': this._messagesOut,
+ 'version': 7
+ };
+
+ response.writeHead(200, {'Content-Type': 'application/json'});
+ response.write(JSON.stringify(server_status));
+ response.end();
+ },
+
/**
* Transmits a message to all subscribed listeners.
*/
_transmit: function(instance, message) {
- var listeners = this.getListenerList(instance)
- .getListeners()
- .filter(function(client) {
- return client.isSubscribedToAny(message.subscribers);
- });
+ var lists = this.getListenerLists(instance);
+
+ for (var ii = 0; ii < lists.length; ii++) {
+ var list = lists[ii];
+ var listeners = list.getListeners();
+ this._transmitToListeners(list, listeners, message);
+ }
+ },
+
+ _transmitToListeners: function(list, listeners, message) {
+ for (var ii = 0; ii < listeners.length; ii++) {
+ var listener = listeners[ii];
- for (var i = 0; i < listeners.length; i++) {
- var listener = listeners[i];
+ if (!listener.isSubscribedToAny(message.subscribers)) {
+ continue;
+ }
try {
listener.writeMessage(message);
++this._messagesOut;
- this.getLogger().log(
+ this.log(
'<%s> Wrote Message',
listener.getDescription());
} catch (error) {
- this.getListenerList(instance).removeListener(listener);
- this.getLogger().log(
+ list.removeListener(listener);
+
+ this.log(
'<%s> Write Error: %s',
listener.getDescription(),
error);
}
}
- },
- },
-
- properties: {
- clientServer: null,
- logger: null,
+ }
}
});
diff --git a/support/aphlict/server/lib/AphlictClientServer.js b/support/aphlict/server/lib/AphlictClientServer.js
index 0d54297c12..1d8051a65c 100644
--- a/support/aphlict/server/lib/AphlictClientServer.js
+++ b/support/aphlict/server/lib/AphlictClientServer.js
@@ -1,106 +1,126 @@
'use strict';
var JX = require('./javelin').JX;
require('./AphlictListenerList');
require('./AphlictLog');
var url = require('url');
var util = require('util');
var WebSocket = require('ws');
JX.install('AphlictClientServer', {
construct: function(server) {
- this.setLogger(new JX.AphlictLog());
+ server.on('request', JX.bind(this, this._onrequest));
+
this._server = server;
this._lists = {};
},
+ properties: {
+ logger: null,
+ },
+
members: {
_server: null,
_lists: null,
getListenerList: function(path) {
if (!this._lists[path]) {
this._lists[path] = new JX.AphlictListenerList(path);
}
return this._lists[path];
},
+ log: function() {
+ var logger = this.getLogger();
+ if (!logger) {
+ return;
+ }
+
+ logger.log.apply(logger, arguments);
+
+ return this;
+ },
+
+ _onrequest: function(request, response) {
+ // The websocket code upgrades connections before they get here, so
+ // this only handles normal HTTP connections. We just fail them with
+ // a 501 response.
+ response.writeHead(501);
+ response.end('HTTP/501 Use Websockets\n');
+ },
+
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 path = url.parse(ws.upgradeReq.url).pathname;
var listener = self.getListenerList(path).addListener(ws);
function log() {
- self.getLogger().log(
+ self.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>');
}
});
ws.on('close', function() {
self.getListenerList(path).removeListener(listener);
log('Disconnected.');
});
wss.on('close', function() {
self.getListenerList(path).removeListener(listener);
log('Disconnected.');
});
wss.on('error', function(err) {
log('Error: %s', err.message);
});
});
},
- },
-
- properties: {
- logger: null,
}
});

File Metadata

Mime Type
text/x-diff
Expires
Mon, Mar 16, 10:36 PM (23 h, 18 m)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
963466
Default Alt Text
(15 KB)

Event Timeline

styx copyright 2020 sirocyl / ellie (kymkdd) / ncommander / styx contributors
Use of other names and brands is not intended to reflect a claim of, or right to, use the name or brand.
styx is an operating system using the Linux kernel and components, and "styx Linux" as a phrase does not reflect the inclusion of the Linux brand or trademark in the styx project.
Linux® is the registered trademark of Linus Torvalds in the U.S. and other countries.
Terms and Conditions (TODO: fix this!)