1
0
Fork 0
mirror of https://we.phorge.it/source/phorge.git synced 2024-12-05 21:26:14 +01:00
phorge-phorge/support/aphlict/server/lib/AphlictAdminServer.js
epriestley ebcab8edb6 Namespace Aphlict clients by request path, plus other fixes
Summary:
Fixes T7130. Fixes T7041. Fixes T7012.

Major change here is partitioning clients. In the Phacility cluster, being able to get a huge pile of instances on a single server -- without needing to run a process per instance -- is desirable.

To accomplish this, just bucket clients by the path they connect with. This will let us set client URIs to `/instancename/` and then route connections to a small set of servers. This degrades cleanly in the common case and has no effect on installs which don't do instancing.

Also fix two unrelated issues:

  - Fix the timeouts, which were incorrectly initializing in `open()` (which is called during reconnect, causing them to reset every time). Instead, initialize in the constructor. Cap timeout at 5 minutes.
  - Probably fix subscriptions, which were using a property with an object definition. Since this is by-ref, all concrete instances of the object share the same property, so all users would be subscribed to everything. Probably.

Test Plan:
  - Hit notification status page, saw version bump and instance/path name.
  - Saw instance/path name in client and server logs.
  - Stopped server, saw reconnects after 2, 4, 16, ... seconds.
  - Sent test notification; received test notification.
  - Didn't explicitly test the subscription thing but it should be obvious by looking at `/notification/status/` shortly after a push.

Reviewers: joshuaspence, btrahan

Reviewed By: btrahan

Subscribers: epriestley

Maniphest Tasks: T7041, T7012, T7130

Differential Revision: https://secure.phabricator.com/D11769
2015-02-16 11:31:15 -08:00

142 lines
3.8 KiB
JavaScript

'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());
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,
getListenerList: function(instance) {
return this.getClientServer().getListenerList(instance);
},
listen: function() {
return this._server.listen.apply(this._server, arguments);
},
_handler: 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(
'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(
'<%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 (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();
} else {
response.writeHead(404, 'Not Found');
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);
});
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(instance).removeListener(listener);
this.getLogger().log(
'<%s> Write Error: %s',
listener.getDescription(),
error);
}
}
},
},
properties: {
clientServer: null,
logger: null,
}
});