Make Notifications Realtime

Summary:
Adds the node.js Aphlict server, the flash Aphlict client, and some
supporting javascript. Built on top of - and requires - D2703 (which is still
in progress).  Will likely work with no modification on top of the final
version, though.

The node server is currently run with

   sudo node support/aphlict/server/aphlict_server.js

Test Plan: tested locally

Reviewers: epriestley

Reviewed By: epriestley

CC: allenjohnashton, keebuhm, aran, Korvin

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

src/__celerity_resource_map__.php

@@ -772,7 +772,7 @@ celerity_register_resource_map(array(
   ),
   'javelin-behavior-aphlict-listen' =>
   array(
-    'uri' => '/res/6388e057/rsrc/js/application/aphlict/behavior-aphlict-listen.js',
+    'uri' => '/res/7f4bc63b/rsrc/js/application/aphlict/behavior-aphlict-listen.js',
     'type' => 'js',
     'requires' =>
     array(
@@ -780,6 +780,7 @@ celerity_register_resource_map(array(
       1 => 'javelin-aphlict',
       2 => 'javelin-util',
       3 => 'javelin-stratcom',
+      4 => 'javelin-behavior-aphlict-dropdown',
     ),
     'disk' => '/rsrc/js/application/aphlict/behavior-aphlict-listen.js',
   ),

src/__phutil_library_map__.php

@@ -736,6 +736,7 @@ phutil_register_library_map(array(
     'PhabricatorMySQLFileStorageEngine' => 'applications/files/engine/PhabricatorMySQLFileStorageEngine.php',
     'PhabricatorNotificationBuilder' => 'applications/notification/builder/PhabricatorNotificationBuilder.php',
     'PhabricatorNotificationController' => 'applications/notification/controller/PhabricatorNotificationController.php',
+    'PhabricatorNotificationIndividualController' => 'applications/notification/controller/PhabricatorNotificationIndividualController.php',
     'PhabricatorNotificationPanelController' => 'applications/notification/controller/PhabricatorNotificationPanelController.php',
     'PhabricatorNotificationQuery' => 'applications/notification/PhabricatorNotificationQuery.php',
     'PhabricatorNotificationStoryView' => 'applications/notification/view/PhabricatorNotificationStoryView.php',
@@ -1698,6 +1699,7 @@ phutil_register_library_map(array(
     'PhabricatorMustVerifyEmailController' => 'PhabricatorAuthController',
     'PhabricatorMySQLFileStorageEngine' => 'PhabricatorFileStorageEngine',
     'PhabricatorNotificationController' => 'PhabricatorController',
+    'PhabricatorNotificationIndividualController' => 'PhabricatorNotificationController',
     'PhabricatorNotificationPanelController' => 'PhabricatorNotificationController',
     'PhabricatorNotificationStoryView' => 'PhabricatorNotificationView',
     'PhabricatorNotificationTestController' => 'PhabricatorNotificationController',

src/aphront/configuration/AphrontDefaultApplicationConfiguration.php

@@ -423,6 +423,8 @@ class AphrontDefaultApplicationConfiguration
 
       '/notification/test/' => 'PhabricatorNotificationTestController',
       '/notification/panel/' => 'PhabricatorNotificationPanelController',
+      '/notification/individual/'
+            => 'PhabricatorNotificationIndividualController',
       '/flag/' => array(
         '' => 'PhabricatorFlagListController',
         'view/(?P<view>[^/]+)/' => 'PhabricatorFlagListController',

src/applications/feed/PhabricatorFeedStoryPublisher.php

@@ -98,6 +98,7 @@ final class PhabricatorFeedStoryPublisher {
 
     if (PhabricatorEnv::getEnvConfig('notification.enabled')) {
       $this->insertNotifications($chrono_key);
+      $this->sendNotification($chrono_key);
     }
     return $story;
   }
@@ -136,6 +137,17 @@ final class PhabricatorFeedStoryPublisher {
       implode(', ', $sql));
   }
 
+  private function sendNotification($chrono_key) {
+    $aphlict_url = 'http://127.0.0.1:22281/push?'; //TODO: make configurable
+    $future = new HTTPFuture($aphlict_url, array(
+      "key" => (string)$chrono_key,
+      // TODO: fix. \r\n appears to be appended to the final value here.
+      //       this is a temporary workaround
+      "nothing" => "",
+    ));
+    $future->setMethod('POST');
+    $future->resolve();
+  }
 
   /**
    * We generate a unique chronological key for each story type because we want

src/applications/notification/controller/PhabricatorNotificationIndividualController.php

@@ -0,0 +1,43 @@
+<?php
+
+/*
+ * Copyright 2012 Facebook, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+final class PhabricatorNotificationIndividualController
+  extends PhabricatorNotificationController {
+
+  public function processRequest() {
+    $request = $this->getRequest();
+    $user = $request->getUser();
+
+    $chron_key = $request->getStr('key');
+    $story = id(new PhabricatorFeedStoryNotification())
+      ->loadOneWhere('userPHID = %s AND chronologicalKey = %s',
+                     $user->getPHID(),
+                     $chron_key);
+
+    if ($story == null) {
+      $json = array( "pertinent" => false );
+    } else {
+      $json = array(
+        "pertinent" => true,
+        "primaryObjectPHID" => $story->getPrimaryObjectPHID(),
+      );
+    }
+
+    return id(new AphrontAjaxResponse())->setContent($json);
+  }
+}

src/applications/notification/controller/PhabricatorNotificationPanelController.php

@@ -33,10 +33,18 @@ final class PhabricatorNotificationPanelController
     $builder = new PhabricatorNotificationBuilder($stories);
     $notifications_view = $builder->buildView();
 
+    $num_unconsumed = 0;
+    foreach ($stories as $story) {
+      if (!$story->getHasViewed()) {
+        $num_unconsumed++;
+      }
+    }
+
     $json = array(
       "content" => $stories ?
         $notifications_view->render() :
         "<b>You currently have no notifications<b>",
+      "number" => $num_unconsumed,
     );
 
     return id(new AphrontAjaxResponse())->setContent($json);

src/view/page/PhabricatorStandardPageView.php

@@ -376,18 +376,19 @@ final class PhabricatorStandardPageView extends AphrontPageView {
 
     if (PhabricatorEnv::getEnvConfig('notification.enabled') &&
       $user->isLoggedIn()) {
+
       $aphlict_object_id = 'aphlictswfobject';
 
-      $aphlict_content = phutil_render_tag(
-        'object',
+      $server_uri = new PhutilURI(PhabricatorEnv::getURI(''));
+      $server_domain = $server_uri->getDomain();
+
+      Javelin::initBehavior(
+        'aphlict-listen',
         array(
-          'classid' => 'clsid:d27cdb6e-ae6d-11cf-96b8-444553540000',
-              ),
-        '<param name="movie" value="/rsrc/swf/aphlict.swf" />'.
-        '<param name="allowScriptAccess" value="always" />'.
-        '<param name="wmode" value="opaque" />'.
-        '<embed src="/rsrc/swf/aphlict.swf" wmode="opaque" id="'.
-        $aphlict_object_id.'"></embed>');
+          'id'           => $aphlict_object_id,
+          'server'       => $server_domain,
+          'port'         => 2600,
+        ));
 
       Javelin::initBehavior('aphlict-dropdown', array());
 
@@ -405,8 +406,7 @@ final class PhabricatorStandardPageView extends AphrontPageView {
       $notification_header =
         $notification_indicator.
         '<td>'.
-        '<div style="height:1px; width:1px;">'.
-        $aphlict_content.
+        '<div id="aphlictswf-container" style="height:1px; width:1px;">'.
         '</div>'.
         '</td>';
       $notification_dropdown =

support/aphlict/client/src/Aphlict.as

@@ -7,23 +7,12 @@ package {
   import flash.events.*;
   import flash.external.ExternalInterface;
 
-  import com.phabricator.*;
-
   import vegas.strings.JSON;
 
   public class Aphlict extends Sprite {
 
     private var client:String;
 
-    private var master:LocalConnection;
-    private var recv:LocalConnection;
-    private var send:LocalConnection;
-
-    private var receiver:AphlictReceiver;
-    private var loyalUntil:Number = 0;
-    private var subjects:Array;
-    private var frequency:Number = 100;
-
     private var socket:Socket;
     private var readBuffer:ByteArray;
 
@@ -47,60 +36,10 @@ package {
       this.remoteServer = server;
       this.remotePort   = port;
 
-      this.master = null;
-      this.receiver = new AphlictReceiver(this);
-      this.subjects = [];
-
-      this.send = new LocalConnection();
-
-      this.recv = new LocalConnection();
-      this.recv.client = this.receiver;
-      for (var ii:Number = 0; ii < 32; ii++) {
-        try {
-          this.recv.connect('aphlict_subject_' + ii);
-          this.client = 'aphlict_subject_' + ii;
-        } catch (x:Error) {
-          // Some other Aphlict client is holding that ID.
-        }
-      }
-
-      if (!this.client) {
-        // Too many clients open already, just exit.
+      this.connectToServer();
       return;
     }
 
-      this.usurp();
-    }
-
-    private function usurp():void {
-      if (this.master) {
-        for (var ii:Number = 0; ii < this.subjects.length; ii++) {
-          if (this.subjects[ii] == this.client) {
-            continue;
-          }
-          this.send.send(this.subjects[ii], 'remainLoyal');
-        }
-      } else if (this.loyalUntil < new Date().getTime()) {
-        var recv:LocalConnection = new LocalConnection();
-        recv.client = this.receiver;
-        try {
-          recv.connect('aphlict_master');
-          this.master = recv;
-          this.subjects = [this.client];
-
-          this.connectToServer();
-
-        } catch (x:Error) {
-          // Can't become the master.
-        }
-
-        if (!this.master) {
-          this.send.send('aphlict_master', 'becomeLoyal', this.client);
-          this.remainLoyal();
-        }
-      }
-      setTimeout(this.usurp, this.frequency);
-    }
 
     public function connectToServer():void {
       var socket:Socket = new Socket();
@@ -156,9 +95,7 @@ package {
           t.writeBytes(b, msg_len + 8);
           this.readBuffer = t;
 
-          for (var ii:Number = 0; ii < this.subjects.length; ii++) {
-            this.send.send(this.subjects[ii], 'receiveMessage', data);
-          }
+          this.receiveMessage(data);
         } else {
           break;
         }
@@ -166,14 +103,6 @@ package {
 
     }
 
-    public function remainLoyal():void {
-      this.loyalUntil = new Date().getTime() + (2 * this.frequency);
-    }
-
-    public function becomeLoyal(subject:String):void {
-      this.subjects.push(subject);
-    }
-
     public function receiveMessage(msg:Object):void {
       this.externalInvoke('receive', msg);
     }

support/aphlict/client/src/com/phabricator/AphlictReceiver.as

@@ -1,25 +0,0 @@
-package com.phabricator {
-
-  public class AphlictReceiver {
-
-    private var core:Object;
-
-    public function AphlictReceiver(core:Object) {
-      this.core = core;
-    }
-
-    public function remainLoyal():void {
-      this.core.remainLoyal();
-    }
-
-    public function becomeLoyal(subject:String):void {
-      this.core.becomeLoyal(subject);
-    }
-
-    public function receiveMessage(msg:Object):void {
-      this.core.receiveMessage(msg);
-    }
-
-  }
-
-}

support/aphlict/server/aphlict_server.js

@@ -1,4 +1,21 @@
 var net = require('net');
+var http  = require('http');
+var url = require('url');
+var querystring = require('querystring');
+var fs = require('fs');
+
+// set up log file
+logfile = fs.createWriteStream('/var/log/aphlict.log',
+        { flags: 'a',
+          encoding: null,
+          mode: 0666 });
+logfile.write('----- ' + (new Date()).toLocaleString() + ' -----\n');
+
+function log(str) {
+    console.log(str);
+    logfile.write(str + '\n');
+}
+
 
 function getFlashPolicy() {
   return [
@@ -8,35 +25,113 @@ function getFlashPolicy() {
     '<cross-domain-policy>',
     '<allow-access-from domain="*" to-ports="2600"/>',
     '</cross-domain-policy>'
-  ].join("\n");
+  ].join('\n');
 }
 
 net.createServer(function(socket) {
   socket.on('data', function() {
     socket.write(getFlashPolicy() + '\0');
   });
+
+  socket.on('error', function (e) {
+    log('Error in policy server: ' + e);
+  });
 }).listen(843);
 
-var sp_server = net.createServer(function(socket) {
-  function xwrite() {
-    var data = {hi: "hello"};
-    var serial = JSON.stringify(data);
 
+
+function write_json(socket, data) {
+  var serial = JSON.stringify(data);
   var length = Buffer.byteLength(serial, 'utf8');
   length = length.toString();
   while (length.length < 8) {
-      length = "0" + length;
+    length = '0' + length;
   }
-
   socket.write(length + serial);
-
-    console.log('write : ' + length + serial);
 }
 
+
+var clients = {};
+var current_connections = 0;
+// According to the internet up to 2^53 can
+// be stored in javascript, this is less than that
+var MAX_ID = 9007199254740991;//2^53 -1
+
+// If we get one connections per millisecond this will
+// be fine as long as someone doesn't maintain a
+// connection for longer than 6854793 years.  If
+// you want to write something pretty be my guest
+
+function generate_id() {
+  if (typeof generate_id.current_id == 'undefined'
+      || generate_id.current_id > MAX_ID) {
+    generate_id.current_id = 0;
+  }
+  return generate_id.current_id++;
+}
+
+var send_server = net.createServer(function(socket) {
+  var client_id = generate_id();
+
   socket.on('connect', function() {
+    clients[client_id] = socket;
+    current_connections++;
+    log(client_id + ': connected\t\t('
+        + current_connections + ' current connections)');
+  });
 
-    xwrite();
-    setInterval(xwrite, 1000);
+  socket.on('close', function() {
+    delete clients[client_id];
+    current_connections--;
+    log(client_id + ': closed\t\t('
+        + current_connections + ' current connections)');
+  });
 
+  socket.on('timeout', function() {
+    log(client_id + ': timed out!');
+  });
+
+  socket.on('end', function() {
+    log(client_id + ': ended the connection');
+    // node automatically closes half-open connections
+  });
+
+  socket.on('error', function (e) {
+    console.log('Uncaught error in send server: ' + e);
   });
 }).listen(2600);
+
+
+
+var receive_server = http.createServer(function(request, response) {
+  response.writeHead(200, {'Content-Type' : 'text/plain'});
+
+  if (request.method == 'POST') { // Only pay attention to POST requests
+    var body = '';
+
+    request.on('data', function (data) {
+      body += data;
+    });
+
+    request.on('end', function () {
+      var data = querystring.parse(body);
+      log('notification: ' + JSON.stringify(data));
+      broadcast(data);
+      response.end();
+    });
+  }
+}).listen(22281, '127.0.0.1');
+
+function broadcast(data) {
+  for(var client_id in clients) {
+    try {
+      write_json(clients[client_id], data);
+      log(' wrote to client ' + client_id);
+    } catch (error) {
+      delete clients[client_id];
+      current_connections--;
+      log(' ERROR: could not write to client ' + client_id);
+    }
+  }
+}
+

webroot/rsrc/js/application/aphlict/behavior-aphlict-dropdown.js

@@ -11,13 +11,28 @@ JX.behavior('aphlict-dropdown', function(config) {
   var dropdown = JX.$('phabricator-notification-dropdown');
   var indicator = JX.$('phabricator-notification-indicator');
   var visible = false;
+  var request = null;
+
+  function refresh() {
+    if (request) { //already fetching
+      return;
+    }
+
+    request = new JX.Request('/notification/panel/', function(response) {
+      indicator.textContent = '' + response.number;
+      if (response.number == 0) {
+        indicator.style.fontWeight = "";
+      } else {
+        indicator.style.fontWeight = "bold";
+      }
+      JX.DOM.setContent(dropdown, JX.$H(response.content));
+      request = null;
+    });
+    request.send();
+  }
 
   //populate panel
-  (new JX.Request('/notification/panel/',
-    function(response) {
-      JX.DOM.setContent(dropdown, JX.$H(response.content));
-    })).send();
-
+  refresh();
 
   JX.Stratcom.listen(
     'click',
@@ -48,4 +63,5 @@ JX.behavior('aphlict-dropdown', function(config) {
     }
   )
 
+  JX.Stratcom.listen('notification-panel-update', null, refresh);
 });

webroot/rsrc/js/application/aphlict/behavior-aphlict-listen.js

@@ -4,19 +4,24 @@
  *           javelin-aphlict
  *           javelin-util
  *           javelin-stratcom
+ *           javelin-behavior-aphlict-dropdown
  */
 
 JX.behavior('aphlict-listen', function(config) {
   function onready() {
-    JX.log("The flash component is ready!");
-
     var client = new JX.Aphlict(config.id, config.server, config.port)
       .setHandler(function(type, message) {
         if (message) {
-          JX.log("Got aphlict event '" + type + "':");
-          JX.log(message);
-        } else {
-          JX.log("Got aphlict event '" + type + "'.");
+          if (type == 'receive') {
+            var request = new JX.Request('/notification/individual/',
+              function(response) {
+                if (response.pertinent) {
+                  JX.Stratcom.invoke('notification-panel-update', null, {});
+                }
+              });
+            request.addData({ "key": message.key });
+            request.send();
+          }
         }
       })
       .start();
@@ -27,4 +32,13 @@ JX.behavior('aphlict-listen', function(config) {
   // If we just go crazy and start making calls to it before it loads, its
   // interfaces won't be registered yet.
   JX.Stratcom.listen('aphlict-component-ready', null, onready);
+
+  // Add Flash object to page
+  JX.$("aphlictswf-container").innerHTML =
+    '<object classid="clsid:d27cdb6e-ae6d-11cf-96b8-444553540000">'
+    + '<param name="movie" value="/rsrc/swf/aphlict.swf" />'
+    + '<param name="allowScriptAccess" value="always" />'
+    + '<param name="wmode" value="opaque" />'
+    + '<embed src="/rsrc/swf/aphlict.swf" wmode="opaque" id="aphlictswfobject">'
+    + '</embed></object>'; //Evan sanctioned
 });