Merge branch 'master' into testing

lib/default.php

@@ -91,10 +91,13 @@ $default =
               'spawndelay' => 1, // Wait at least N seconds between (re)spawns of child processes to avoid slamming the queue server with subscription startup
               'debug_memory' => false, // true to spit memory usage to log
               'inboxes' => true, // true to do inbox distribution & output queueing from in background via 'distrib' queue
-              'breakout' => array('*' => 'shared'), // set global or per-handler queue breakout
+              'breakout' => array(), // List queue specifiers to break out when using Stomp queue.
-                                      // 'shared': use a shared queue for all sites
+                                     // Default will share all queues for all sites within each group.
-                                      // 'handler': share each/this handler over multiple sites
+                                     // Specify as <group>/<queue> or <group>/<queue>/<site>,
-                                      // 'site': break out for each/this handler on this site
+                                     // using nickname identifier as site.
+                                     //
+                                     // 'main/distrib' separate "distrib" queue covering all sites
+                                     // 'xmpp/xmppout/mysite' separate "xmppout" queue covering just 'mysite'
               'max_retries' => 10, // drop messages after N failed attempts to process (Stomp)
               'dead_letter_dir' => false, // set to directory to save dropped messages into (Stomp)
               ),

lib/iomaster.php

@@ -55,27 +55,18 @@ abstract class IoMaster
         if ($multiSite !== null) {
             $this->multiSite = $multiSite;
         }
-        if ($this->multiSite) {
-            $this->sites = StatusNet::findAllSites();
-        } else {
-            $this->sites = array(StatusNet::currentSite());
-        }
 
-        if (empty($this->sites)) {
+        $this->initManagers();
-            throw new Exception("Empty status_network table, cannot init");
-        }
-
-        foreach ($this->sites as $site) {
-            StatusNet::switchSite($site);
-            $this->initManagers();
-        }
     }
 
     /**
-     * Initialize IoManagers for the currently configured site
+     * Initialize IoManagers which are appropriate to this instance;
-     * which are appropriate to this instance.
+     * pass class names or instances into $this->instantiate().
      *
-     * Pass class names into $this->instantiate()
+     * If setup and configuration may vary between sites in multi-site
+     * mode, it's the subclass's responsibility to set them up here.
+     *
+     * Switching site configurations is an acceptable side effect.
      */
     abstract function initManagers();
 

lib/stompqueuemanager.php

@@ -63,7 +63,7 @@ class StompQueueManager extends QueueManager
         $this->password = common_config('queue', 'stomp_password');
         $this->base     = common_config('queue', 'queue_basename');
         $this->control  = common_config('queue', 'control_channel');
-        $this->subscriptions = array($this->control => $this->control);
+        $this->breakout = common_config('queue', 'breakout');
     }
 
     /**
@@ -75,28 +75,6 @@ class StompQueueManager extends QueueManager
         return IoManager::INSTANCE_PER_PROCESS;
     }
 
-    /**
-     * Record queue subscriptions we'll need to handle the current site.
-     */
-    public function addSite()
-    {
-        $this->sites[] = StatusNet::currentSite();
-
-        // Set up handlers active for this site...
-        $this->initialize();
-
-        foreach ($this->activeGroups as $group) {
-            if (isset($this->groups[$group])) {
-                // Actual queues may be broken out or consolidated...
-                // Subscribe to all the target queues we'll need.
-                foreach ($this->groups[$group] as $transport => $class) {
-                    $target = $this->queueName($transport);
-                    $this->subscriptions[$target] = $target;
-                }
-            }
-        }
-    }
-
     /**
      * Optional; ping any running queue handler daemons with a notification
      * such as announcing a new site to handle or requesting clean shutdown.
@@ -166,14 +144,15 @@ class StompQueueManager extends QueueManager
 
         $con = $this->cons[$idx];
         $host = $con->getServer();
-        $result = $con->send($this->queueName($queue), $msg, $props);
+        $target = $this->queueName($queue);
+        $result = $con->send($target, $msg, $props);
 
         if (!$result) {
-            $this->_log(LOG_ERR, "Error sending $rep to $queue queue on $host");
+            $this->_log(LOG_ERR, "Error sending $rep to $queue queue on $host $target");
             return false;
         }
 
-        $this->_log(LOG_DEBUG, "complete remote queueing $rep for $queue on $host");
+        $this->_log(LOG_DEBUG, "complete remote queueing $rep for $queue on $host $target");
         $this->stats('enqueued', $queue);
         return true;
     }
@@ -432,11 +411,42 @@ class StompQueueManager extends QueueManager
     protected function doSubscribe(LiberalStomp $con)
     {
         $host = $con->getServer();
-        foreach ($this->subscriptions as $queue) {
+        foreach ($this->subscriptions() as $sub) {
-            $this->_log(LOG_INFO, "Subscribing to $queue on $host");
+            $this->_log(LOG_INFO, "Subscribing to $sub on $host");
-            $con->subscribe($queue);
+            $con->subscribe($sub);
         }
     }
+    
+    /**
+     * Grab a full list of stomp-side queue subscriptions.
+     * Will include:
+     *  - control broadcast channel
+     *  - shared group queues for active groups
+     *  - per-handler and per-site breakouts from $config['queue']['breakout']
+     *    that are rooted in the active groups.
+     *
+     * @return array of strings
+     */
+    protected function subscriptions()
+    {
+        $subs = array();
+        $subs[] = $this->control;
+
+        foreach ($this->activeGroups as $group) {
+            $subs[] = $this->base . $group;
+        }
+
+        foreach ($this->breakout as $spec) {
+            $parts = explode('/', $spec);
+            if (count($parts) < 2 || count($parts) > 3) {
+                common_log(LOG_ERR, "Bad queue breakout specifier $spec");
+            }
+            if (in_array($parts[0], $this->activeGroups)) {
+                $subs[] = $this->base . $spec;
+            }
+        }
+        return array_unique($subs);
+    }
 
     /**
      * Handle and acknowledge an event that's come in through a queue.
@@ -612,32 +622,26 @@ class StompQueueManager extends QueueManager
     }
 
     /**
-     * Set us up with queue subscriptions for a new site added at runtime,
+     * (Re)load runtime configuration for a given site by nickname,
      * triggered by a broadcast to the 'statusnet-control' topic.
      *
+     * Configuration changes in database should update, but config
+     * files might not.
+     *
      * @param array $frame Stomp frame
      * @return bool true to continue; false to stop further processing.
      */
     protected function updateSiteConfig($nickname)
     {
-        if (empty($this->sites)) {
+        $sn = Status_network::staticGet($nickname);
-            if ($nickname == common_config('site', 'nickname')) {
+        if ($sn) {
-                StatusNet::init(common_config('site', 'server'));
+            $this->switchSite($nickname);
-            } else {
+            if (!in_array($nickname, $this->sites)) {
-                $this->_log(LOG_INFO, "Ignoring update ping for other site $nickname");
+                $this->addSite();
             }
+            $this->stats('siteupdate');
         } else {
-            $sn = Status_network::staticGet($nickname);
+            $this->_log(LOG_ERR, "Ignoring ping for unrecognized new site $nickname");
-            if ($sn) {
-                $this->switchSite($nickname);
-                if (!in_array($nickname, $this->sites)) {
-                    $this->addSite();
-                }
-                // @fixme update subscriptions, if applicable
-                $this->stats('siteupdate');
-            } else {
-                $this->_log(LOG_ERR, "Ignoring ping for unrecognized new site $nickname");
-            }
         }
     }
 
@@ -646,24 +650,25 @@ class StompQueueManager extends QueueManager
      * group name for this queue to give eg:
      *
      * /queue/statusnet/main
+     * /queue/statusnet/main/distrib
+     * /queue/statusnet/xmpp/xmppout/site01
      *
      * @param string $queue
      * @return string
      */
     protected function queueName($queue)
     {
-        $base = common_config('queue', 'queue_basename');
         $group = $this->queueGroup($queue);
-        $breakout = $this->breakoutMode($queue);
+        $site = StatusNet::currentSite();
-        if ($breakout == 'shared') {
+
-            return $base . "$group";
+        $specs = array("$group/$queue/$site",
-        } else if ($breakout == 'handler') {
+                       "$group/$queue");
-            return $base . "$group/$queue";
+        foreach ($specs as $spec) {
-        } else if ($breakout == 'site') {
+            if (in_array($spec, $this->breakout)) {
-            $site = StatusNet::currentSite();
+                return $this->base . $spec;
-            return $base . "$group/$queue/$site";
+            }
         }
-        throw Exception("Unrecognized queue breakout mode '$breakout' for '$queue'");
+        return $this->base . $group;
     }
 
     /**

scripts/queuedaemon.php

@@ -126,8 +126,7 @@ class QueueDaemon extends SpawningDaemon
 class QueueMaster extends IoMaster
 {
     /**
-     * Initialize IoManagers for the currently configured site
+     * Initialize IoManagers which are appropriate to this instance.
-     * which are appropriate to this instance.
      */
     function initManagers()
     {