Make the Aphlict server more resilient.

Summary:
Currently, the Aphlict server will crash if invalid JSON data is `POST`ed to it. I have fixed this to, instead, return a 400. Also made some minor formatting changes.

Ref T4324. Ref T5284. Also, modify the data structure that is passed around (i.e. `POST`ed to the Aphlict server and broadcast to the Aphlict clients) to include the subscribers. Initially, I figured that we shouldn't expose this information to the clients... however, it is necessary for T4324 that the `AphlictMaster` is able to route a notification to the appropriate clients.

Test Plan:
Making the following `curl` request: `curl --data "{" http://localhost:22281/`.

**Before**
```
sudo ./bin/aphlict debug
Starting Aphlict server in foreground...
Launching server:

    $ 'nodejs' '/usr/src/phabricator/src/applications/aphlict/management/../../../../support/aphlict/server/aphlict_server.js' --port='22280' --admin='22281' --host='localhost' --user='aphlict'

[Wed Jun 11 2014 17:07:51 GMT+0000 (UTC)] Started Server (PID 2033)
[Wed Jun 11 2014 17:07:55 GMT+0000 (UTC)]
<<< UNCAUGHT EXCEPTION! >>>

SyntaxError: Unexpected end of input
>>> Server exited!
```

**After**
(No output... the bad JSON is caught and a 400 is returned)

Reviewers: #blessed_reviewers, epriestley

Reviewed By: #blessed_reviewers, epriestley

Subscribers: epriestley, Korvin

Maniphest Tasks: T4324, T5284

Differential Revision: https://secure.phabricator.com/D9480

Author: joshuaspence

resources/celerity/map.php

@@ -476,7 +476,7 @@
     'rsrc/js/phuix/PHUIXActionListView.js' => 'b5c256b8',
     'rsrc/js/phuix/PHUIXActionView.js' => '6e8cefa4',
     'rsrc/js/phuix/PHUIXDropdownMenu.js' => 'bd4c8dca',
-    'rsrc/swf/aphlict.swf' => 'b7c2d7aa',
+    'rsrc/swf/aphlict.swf' => 'f45c3edc',
   ),
   'symbols' =>
   array(

src/applications/feed/PhabricatorFeedStoryPublisher.php

@@ -176,10 +176,8 @@ private function insertNotifications($chrono_key) {
 
   private function sendNotification($chrono_key) {
     $data = array(
-      'data' => array(
-        'key'  => (string)$chrono_key,
-        'type' => 'notification',
-      ),
+      'key'         => (string)$chrono_key,
+      'type'        => 'notification',
       'subscribers' => $this->subscribedPHIDs,
     );
 

src/applications/notification/client/PhabricatorNotificationClient.php

@@ -2,7 +2,7 @@
 
 final class PhabricatorNotificationClient {
 
-  const EXPECT_VERSION = 5;
+  const EXPECT_VERSION = 6;
 
   public static function getServerStatus() {
     $uri = PhabricatorEnv::getEnvConfig('notification.server-uri');

support/aphlict/server/aphlict_server.js

@@ -25,21 +25,21 @@ if (config.logfile) {
 
 function parse_command_line_arguments(argv) {
   var config = {
-    port : 22280,
-    admin : 22281,
-    host : '127.0.0.1',
-    user : null,
+    port: 22280,
+    admin: 22281,
+    host: '127.0.0.1',
+    user: null,
     log: '/var/log/aphlict.log'
   };
 
   for (var ii = 2; ii < argv.length; ii++) {
     var arg = argv[ii];
     var matches = arg.match(/^--([^=]+)=(.*)$/);
     if (!matches) {
-      throw new Error("Unknown argument '"+arg+"'!");
+      throw new Error("Unknown argument '" + arg + "'!");
     }
     if (!(matches[1] in config)) {
-      throw new Error("Unknown argument '"+matches[1]+"'!");
+      throw new Error("Unknown argument '" + matches[1] + "'!");
     }
     config[matches[1]] = matches[2];
   }
@@ -52,19 +52,19 @@ function parse_command_line_arguments(argv) {
 
 if (process.getuid() !== 0) {
   console.log(
-    "ERROR: "+
-    "This server must be run as root because it needs to bind to privileged "+
-    "port 843 to start a Flash policy server. It will downgrade to run as a "+
-    "less-privileged user after binding if you pass a user in the command "+
+    "ERROR: " +
+    "This server must be run as root because it needs to bind to privileged " +
+    "port 843 to start a Flash policy server. It will downgrade to run as a " +
+    "less-privileged user after binding if you pass a user in the command " +
     "line arguments with '--user=alincoln'.");
   process.exit(1);
 }
 
 var net = require('net');
-var http  = require('http');
+var http = require('http');
 var url = require('url');
 
-process.on('uncaughtException', function (err) {
+process.on('uncaughtException', function(err) {
   debug.log("\n<<< UNCAUGHT EXCEPTION! >>>\n\n" + err);
   process.exit(1);
 });
@@ -95,7 +95,7 @@ var send_server = net.createServer(function(socket) {
     debug.log('<%s> Ended Connection', listener.getDescription());
   });
 
-  socket.on('error', function (e) {
+  socket.on('error', function(e) {
     debug.log('<%s> Error: %s', listener.getDescription(), e);
   });
 
@@ -107,23 +107,29 @@ var messages_in = 0;
 var start_time = new Date().getTime();
 
 var receive_server = http.createServer(function(request, response) {
-  response.writeHead(200, {'Content-Type' : 'text/plain'});
-
   // Publishing a notification.
   if (request.method == 'POST') {
     var body = '';
 
-    request.on('data', function (data) {
+    request.on('data', function(data) {
       body += data;
     });
 
-    request.on('end', function () {
-      ++messages_in;
-
-      var msg = JSON.parse(body);
-      debug.log('notification: ' + JSON.stringify(msg));
-      broadcast(msg.data);
-      response.end();
+    request.on('end', function() {
+      try {
+        var msg = JSON.parse(body);
+
+        debug.log('notification: ' + JSON.stringify(msg));
+        ++messages_in;
+        broadcast(msg);
+
+        response.writeHead(200, {'Content-Type': 'text/plain'});
+      } catch (err) {
+        response.statusCode = 400;
+        response.write('400 Bad Request');
+      } finally {
+        response.end();
+      }
     });
   } else if (request.url == '/status/') {
     request.on('data', function(data) {
@@ -139,9 +145,10 @@ var receive_server = http.createServer(function(request, response) {
         'messages.in': messages_in,
         'messages.out': messages_out,
         'log': config.log,
-        'version': 5
+        'version': 6
       };
 
+      response.writeHead(200, {'Content-Type': 'text/plain'});
       response.write(JSON.stringify(status));
       response.end();
     });

support/aphlict/server/lib/AphlictFlashPolicyServer.js

@@ -17,12 +17,12 @@ JX.install('AphlictFlashPolicyServer', {
     _accessPort: null,
     _debug: null,
 
-    setDebugLog : function(log) {
+    setDebugLog: function(log) {
       this._debug = log;
       return this;
     },
 
-    setAccessPort : function(port) {
+    setAccessPort: function(port) {
       this._accessPort = port;
       return this;
     },

support/aphlict/server/lib/AphlictListener.js

@@ -1,28 +1,28 @@
 var JX = require('javelin').JX;
 
 JX.install('AphlictListener', {
-  construct : function(id, socket) {
+  construct: function(id, socket) {
     this._id = id;
     this._socket = socket;
   },
 
-  members : {
-    _id : null,
-    _socket : null,
+  members: {
+    _id: null,
+    _socket: null,
 
-    getID : function() {
+    getID: function() {
       return this._id;
     },
 
-    getSocket : function() {
+    getSocket: function() {
       return this._socket;
     },
 
-    getDescription : function() {
+    getDescription: function() {
       return 'Listener/' + this.getID();
     },
 
-    writeMessage : function(message) {
+    writeMessage: function(message) {
       var serial = JSON.stringify(message);
 
       var length = Buffer.byteLength(serial, 'utf8');

support/aphlict/server/lib/AphlictListenerList.js

@@ -2,20 +2,18 @@ var JX = require('javelin').JX;
 JX.require('AphlictListener', __dirname);
 
 JX.install('AphlictListenerList', {
-  construct : function() {
+  construct: function() {
     this._listeners = {};
   },
 
-  members : {
-    _listeners : null,
-    _nextID : 0,
-    _activeListenerCount : 0,
-    _totalListenerCount : 0,
+  members: {
+    _listeners: null,
+    _nextID: 0,
+    _activeListenerCount: 0,
+    _totalListenerCount: 0,
 
-    addListener : function(socket) {
-      var listener = new JX.AphlictListener(
-        this._generateNextID(),
-        socket);
+    addListener: function(socket) {
+      var listener = new JX.AphlictListener(this._generateNextID(), socket);
 
       this._listeners[listener.getID()] = listener;
       this._activeListenerCount++;
@@ -24,27 +22,27 @@ JX.install('AphlictListenerList', {
       return listener;
     },
 
-    removeListener : function(listener) {
+    removeListener: function(listener) {
       var id = listener.getID();
       if (id in this._listeners) {
         delete this._listeners[id];
         this._activeListenerCount--;
       }
     },
 
-    getListeners : function() {
+    getListeners: function() {
       return this._listeners;
     },
 
-    getActiveListenerCount : function() {
+    getActiveListenerCount: function() {
       return this._activeListenerCount;
     },
 
-    getTotalListenerCount : function() {
+    getTotalListenerCount: function() {
       return this._totalListenerCount;
     },
 
-    _generateNextID : function() {
+    _generateNextID: function() {
       do {
         this._nextID = ((this._nextID + 1) % 1000000000000);
       } while (this._nextID in this._listeners);

support/aphlict/server/lib/AphlictLog.js

@@ -4,16 +4,16 @@ var fs = require('fs');
 var util = require('util');
 
 JX.install('AphlictLog', {
-  construct : function() {
+  construct: function() {
     this._writeToLogs = [];
     this._writeToConsoles = [];
   },
 
-  members : {
-    _writeToConsoles : null,
-    _writeToLogs : null,
+  members: {
+    _writeToConsoles: null,
+    _writeToLogs: null,
 
-    addLogfile : function(path) {
+    addLogfile: function(path) {
       var options = {
         flags: 'a',
         encoding: 'utf8',
@@ -27,12 +27,12 @@ JX.install('AphlictLog', {
       return this;
     },
 
-    addConsole : function(console) {
+    addConsole: function(console) {
       this._writeToConsoles.push(console);
       return this;
     },
 
-    log : function(pattern) {
+    log: function(pattern) {
       var str = util.format.apply(null, arguments);
       var date = new Date().toLocaleString();
       str = '[' + date + '] ' + str;

support/aphlict/server/lib/javelin.js

@@ -6,7 +6,7 @@ JX.require('core/install');
 
 // NOTE: This is faking out a piece of code in JX.install which waits for
 // Stratcom before running static initializers.
-JX.Stratcom = {ready : true};
+JX.Stratcom = {ready: true};
 JX.require('core/Event');
 JX.require('core/Stratcom');