Merge branch 'testing' of gitorious.org:statusnet/mainline into 0.9.x

[quix0rs-gnu-social.git] / lib / queuemanager.php

diff --git a/lib/queuemanager.php b/lib/queuemanager.php
index b98e57a1ffd1e5f9d7a2ae2d833a4efbf9770518..64bb52e106ee7289bf03934ebb32b145f33315d5 100644 (file)
--- a/lib/queuemanager.php
+++ b/lib/queuemanager.php
@@ -39,6 +39,10 @@ abstract class QueueManager extends IoManager
 {
     static $qm = null;
 
+    public $master = null;
+    public $handlers = array();
+    public $groups = array();
+
     /**
      * Factory function to pull the appropriate QueueManager object
      * for this site's configuration. It can then be used to queue
@@ -96,6 +100,23 @@ abstract class QueueManager extends IoManager
         $this->initialize();
     }
 
+    /**
+     * Optional; ping any running queue handler daemons with a notification
+     * such as announcing a new site to handle or requesting clean shutdown.
+     * This avoids having to restart all the daemons manually to update configs
+     * and such.
+     *
+     * Called from scripts/queuectl.php controller utility.
+     *
+     * @param string $event event key
+     * @param string $param optional parameter to append to key
+     * @return boolean success
+     */
+    public function sendControlSignal($event, $param='')
+    {
+        throw new Exception(get_class($this) . " does not support control signals.");
+    }
+
     /**
      * Store an object (usually/always a Notice) into the given queue
      * for later processing. No guarantee is made on when it will be
@@ -109,6 +130,78 @@ abstract class QueueManager extends IoManager
      */
     abstract function enqueue($object, $queue);
 
+    /**
+     * Build a representation for an object for logging
+     * @param mixed
+     * @return string
+     */
+    function logrep($object) {
+        if (is_object($object)) {
+            $class = get_class($object);
+            if (isset($object->id)) {
+                return "$class $object->id";
+            }
+            return $class;
+        }
+        if (is_string($object)) {
+            $len = strlen($object);
+            $fragment = mb_substr($object, 0, 32);
+            if (mb_strlen($object) > 32) {
+                $fragment .= '...';
+            }
+            return "string '$fragment' ($len bytes)";
+        }
+        return strval($object);
+    }
+
+    /**
+     * Encode an object or variable for queued storage.
+     * Notice objects are currently stored as an id reference;
+     * other items are serialized.
+     *
+     * @param mixed $item
+     * @return string
+     */
+    protected function encode($item)
+    {
+        if ($item instanceof Notice) {
+            // Backwards compat
+            return $item->id;
+        } else {
+            return serialize($item);
+        }
+    }
+
+    /**
+     * Decode an object from queued storage.
+     * Accepts notice reference entries and serialized items.
+     *
+     * @param string
+     * @return mixed
+     */
+    protected function decode($frame)
+    {
+        if (is_numeric($frame)) {
+            // Back-compat for notices...
+            return Notice::staticGet(intval($frame));
+        } elseif (substr($frame, 0, 1) == '<') {
+            // Back-compat for XML source
+            return $frame;
+        } else {
+            // Deserialize!
+            #$old = error_reporting();
+            #error_reporting($old & ~E_NOTICE);
+            $out = unserialize($frame);
+            #error_reporting($old);
+
+            if ($out === false && $frame !== 'b:0;') {
+                common_log(LOG_ERR, "Couldn't unserialize queued frame: $frame");
+                return false;
+            }
+            return $out;
+        }
+    }
+
     /**
      * Instantiate the appropriate QueueHandler class for the given queue.
      *
@@ -119,7 +212,9 @@ abstract class QueueManager extends IoManager
     {
         if (isset($this->handlers[$queue])) {
             $class = $this->handlers[$queue];
-            if (class_exists($class)) {
+            if(is_object($class)) {
+                return $class;
+            } else if (class_exists($class)) {
                 return new $class();
             } else {
                 common_log(LOG_ERR, "Nonexistent handler class '$class' for queue '$queue'");
@@ -131,13 +226,15 @@ abstract class QueueManager extends IoManager
     }
 
     /**
-     * Get a list of all registered queue transport names.
+     * Get a list of registered queue transport names to be used
+     * for this daemon.
      *
      * @return array of strings
      */
     function getQueues()
     {
-        return array_keys($this->handlers);
+        $group = $this->activeGroup();
+        return array_keys($this->groups[$group]);
     }
 
     /**
@@ -148,25 +245,27 @@ abstract class QueueManager extends IoManager
      */
     function initialize()
     {
+        // @fixme we'll want to be able to listen to particular queues...
         if (Event::handle('StartInitializeQueueManager', array($this))) {
             $this->connect('plugin', 'PluginQueueHandler');
             $this->connect('omb', 'OmbQueueHandler');
             $this->connect('ping', 'PingQueueHandler');
+            $this->connect('distrib', 'DistribQueueHandler');
             if (common_config('sms', 'enabled')) {
                 $this->connect('sms', 'SmsQueueHandler');
             }
 
             // XMPP output handlers...
-            if (common_config('xmpp', 'enabled') && !defined('XMPP_EMERGENCY_FLAG')) {
-                $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');
-            }
+            $this->connect('jabber', 'JabberQueueHandler');
+            $this->connect('public', 'PublicQueueHandler');
+            // @fixme this should get an actual queue
+            //$this->connect('confirm', 'XmppConfirmHandler');
 
             // For compat with old plugins not registering their own handlers.
             $this->connect('plugin', 'PluginQueueHandler');
+
+            $this->connect('xmppout', 'XmppOutQueueHandler', 'xmppdaemon');
+
         }
         Event::handle('EndInitializeQueueManager', array($this));
     }
@@ -176,11 +275,28 @@ abstract class QueueManager extends IoManager
      * Only registered transports will be reliably picked up!
      *
      * @param string $transport
-     * @param string $class
+     * @param string $class class name or object instance
+     * @param string $group
      */
-    public function connect($transport, $class)
+    public function connect($transport, $class, $group='queuedaemon')
     {
         $this->handlers[$transport] = $class;
+        $this->groups[$group][$transport] = $class;
+    }
+
+    /**
+     * @return string queue group to use for this request
+     */
+    function activeGroup()
+    {
+        $group = 'queuedaemon';
+        if ($this->master) {
+            // hack hack
+            if ($this->master instanceof XmppMaster) {
+                return 'xmppdaemon';
+            }
+        }
+        return $group;
     }
 
     /**