Major refactoring of queue handlers to support running multiple sites in one daemon.

Key changes:
* Initialization code moved from common.php to StatusNet class;
  can now switch configurations during runtime.
* As a consequence, configuration files must now be idempotent...
  Be careful with constant, function or class definitions.
* Control structure for daemons/QueueManager/QueueHandler has been refactored;
  the run loop is now managed by IoMaster run via scripts/queuedaemon.php
  IoManager subclasses are woken to handle socket input or polling, and may
  cover multiple sites.
* Plugins can implement notice queue handlers more easily by registering a
  QueueHandler class; no more need to add a daemon.

The new QueueDaemon runs from scripts/queuedaemon.php:

* This replaces most of the old *handler.php scripts; they've been refactored
  to the bare handler classes.
* Spawns multiple child processes to spread load; defaults to CPU count on
  Linux and Mac OS X systems, or override with --threads=N
* When multithreaded, child processes are automatically respawned on failure.
* Threads gracefully shut down and restart when passing a soft memory limit
  (defaults to 90% of memory_limit), limiting damage from memory leaks.
* Support for UDP-based monitoring: http://www.gitorious.org/snqmon

Rough control flow diagram:
QueueDaemon -> IoMaster -> IoManager
                           QueueManager [listen or poll] -> QueueHandler
                           XmppManager [ping & keepalive]
                           XmppConfirmManager [poll updates]

Todo:

* Respawning features not currently available running single-threaded.
* When running single-site, configuration changes aren't picked up.
* New sites or config changes affecting queue subscriptions are not yet
  handled without a daemon restart.
* SNMP monitoring output to integrate with general tools (nagios, ganglia)
* Convert XMPP confirmation message sends to use stomp queue instead of polling
* Convert xmppdaemon.php to IoManager?
* Convert Twitter status, friends import polling daemons to IoManager
* Clean up some error reporting and failure modes
* May need to adjust queue priorities for best perf in backlog/flood cases

Detailed code history available in my daemon-work branch:
http://www.gitorious.org/~brion/statusnet/brion-fixes/commits/daemon-work

1 files changed, 141 insertions, 8 deletions

diff --git a/lib/queuemanager.php b/lib/queuemanager.php
index 43105b7a8..a98c0efff 100644
--- a/lib/queuemanager.php
+++ b/lib/queuemanager.php
@@ -2,7 +2,7 @@
 /**
  * StatusNet, the distributed open-source microblogging tool
  *
- * Abstract class for queue managers
+ * Abstract class for i/o managers
  *
  * PHP version 5
  *
@@ -23,16 +23,32 @@
  * @package   StatusNet
  * @author    Evan Prodromou <evan@status.net>
  * @author    Sarven Capadisli <csarven@status.net>
- * @copyright 2009 StatusNet, Inc.
+ * @author    Brion Vibber <brion@status.net>
+ * @copyright 2009-2010 StatusNet, Inc.
  * @license   http://www.fsf.org/licensing/licenses/agpl-3.0.html GNU Affero General Public License version 3.0
  * @link      http://status.net/
  */
 
-class QueueManager
+/**
+ * Completed child classes must implement the enqueue() method.
+ *
+ * For background processing, classes should implement either socket-based
+ * input (handleInput(), getSockets()) or idle-loop polling (idle()).
+ */
+abstract class QueueManager extends IoManager
 {
     static $qm = null;
 
-    static function get()
+    /**
+     * Factory function to pull the appropriate QueueManager object
+     * for this site's configuration. It can then be used to queue
+     * events for later processing or to spawn a processing loop.
+     *
+     * Plugins can add to the built-in types by hooking StartNewQueueManager.
+     *
+     * @return QueueManager
+     */
+    public static function get()
     {
         if (empty(self::$qm)) {
 
@@ -62,13 +78,130 @@ class QueueManager
         return self::$qm;
     }
 
-    function enqueue($object, $queue)
+    /**
+     * @fixme wouldn't necessarily work with other class types.
+     * Better to change the interface...?
+     */
+    public static function multiSite()
+    {
+        if (common_config('queue', 'subsystem') == 'stomp') {
+            return IoManager::INSTANCE_PER_PROCESS;
+        } else {
+            return IoManager::SINGLE_ONLY;
+        }
+    }
+
+    function __construct()
     {
-        throw ServerException("Unimplemented function 'enqueue' called");
+        $this->initialize();
     }
 
-    function service($queue, $handler)
+    /**
+     * Store an object (usually/always a Notice) into the given queue
+     * for later processing. No guarantee is made on when it will be
+     * processed; it could be immediately or at some unspecified point
+     * in the future.
+     *
+     * Must be implemented by any queue manager.
+     *
+     * @param Notice $object
+     * @param string $queue
+     */
+    abstract function enqueue($object, $queue);
+
+    /**
+     * Instantiate the appropriate QueueHandler class for the given queue.
+     *
+     * @param string $queue
+     * @return mixed QueueHandler or null
+     */
+    function getHandler($queue)
     {
-        throw ServerException("Unimplemented function 'service' called");
+        if (isset($this->handlers[$queue])) {
+            $class = $this->handlers[$queue];
+            if (class_exists($class)) {
+                return new $class();
+            } else {
+                common_log(LOG_ERR, "Nonexistent handler class '$class' for queue '$queue'");
+            }
+        } else {
+            common_log(LOG_ERR, "Requested handler for unkown queue '$queue'");
+        }
+        return null;
+    }
+
+    /**
+     * Get a list of all registered queue transport names.
+     *
+     * @return array of strings
+     */
+    function getQueues()
+    {
+        return array_keys($this->handlers);
+    }
+
+    /**
+     * Initialize the list of queue handlers
+     *
+     * @event StartInitializeQueueManager
+     * @event EndInitializeQueueManager
+     */
+    function initialize()
+    {
+        if (Event::handle('StartInitializeQueueManager', array($this))) {
+            $this->connect('plugin', 'PluginQueueHandler');
+            $this->connect('omb', 'OmbQueueHandler');
+            $this->connect('ping', 'PingQueueHandler');
+            if (common_config('sms', 'enabled')) {
+                $this->connect('sms', 'SmsQueueHandler');
+            }
+
+            // XMPP output handlers...
+            if (common_config('xmpp', 'enabled')) {
+                $this->connect('jabber', 'JabberQueueHandler');
+                $this->connect('public', 'PublicQueueHandler');
+                
+                // @fixme this should move up a level or should get an actual queue
+                $this->connect('confirm', 'XmppConfirmHandler');
+            }
+
+            // For compat with old plugins not registering their own handlers.
+            $this->connect('plugin', 'PluginQueueHandler');
+        }
+        Event::handle('EndInitializeQueueManager', array($this));
+    }
+
+    /**
+     * Register a queue transport name and handler class for your plugin.
+     * Only registered transports will be reliably picked up!
+     *
+     * @param string $transport
+     * @param string $class
+     */
+    public function connect($transport, $class)
+    {
+        $this->handlers[$transport] = $class;
+    }
+
+    /**
+     * Send a statistic ping to the queue monitoring system,
+     * optionally with a per-queue id.
+     *
+     * @param string $key
+     * @param string $queue
+     */
+    function stats($key, $queue=false)
+    {
+        $owners = array();
+        if ($queue) {
+            $owners[] = "queue:$queue";
+            $owners[] = "site:" . common_config('site', 'server');
+        }
+        if (isset($this->master)) {
+            $this->master->stats($key, $owners);
+        } else {
+            $monitor = new QueueMonitor();
+            $monitor->stats($key, $owners);
+        }
     }
 }