mirror of
https://we.phorge.it/source/phorge.git
synced 2024-12-05 05:02:44 +01:00
ebcab8edb6
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
142 lines
3.8 KiB
JavaScript
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,
|
|
}
|
|
|
|
});
|