X-Git-Url: https://git.mxchange.org/?a=blobdiff_plain;f=lib%2Fstompqueuemanager.php;h=fc98c77d4086cffea3a90076790c19f39f8bbced;hb=1fdc72f5956f00c736f7761aead4abd8da192eda;hp=da70d9ae31e70296cc046b512c23d0aebab7f033;hpb=427ac3a3a6eafa9a82006e9f0a9c2ab19a09fb4b;p=quix0rs-gnu-social.git

diff --git a/lib/stompqueuemanager.php b/lib/stompqueuemanager.php
index da70d9ae31..fc98c77d40 100644
--- a/lib/stompqueuemanager.php
+++ b/lib/stompqueuemanager.php
@@ -29,32 +29,44 @@
  */
 
 require_once 'Stomp.php';
-
+require_once 'Stomp/Exception.php';
 
 class StompQueueManager extends QueueManager
 {
-    var $server = null;
-    var $username = null;
-    var $password = null;
-    var $base = null;
-    var $con = null;
+    protected $servers;
+    protected $username;
+    protected $password;
+    protected $base;
     protected $control;
-    
+
+    protected $useTransactions;
+    protected $useAcks;
+
     protected $sites = array();
     protected $subscriptions = array();
 
-    protected $useTransactions = true;
-    protected $transaction = null;
-    protected $transactionCount = 0;
+    protected $cons = array(); // all open connections
+    protected $disconnect = array();
+    protected $transaction = array();
+    protected $transactionCount = array();
+    protected $defaultIdx = 0;
 
     function __construct()
     {
         parent::__construct();
-        $this->server   = common_config('queue', 'stomp_server');
-        $this->username = common_config('queue', 'stomp_username');
-        $this->password = common_config('queue', 'stomp_password');
-        $this->base     = common_config('queue', 'queue_basename');
-        $this->control  = common_config('queue', 'control_channel');
+        $server = common_config('queue', 'stomp_server');
+        if (is_array($server)) {
+            $this->servers = $server;
+        } else {
+            $this->servers = array($server);
+        }
+        $this->username        = common_config('queue', 'stomp_username');
+        $this->password        = common_config('queue', 'stomp_password');
+        $this->base            = common_config('queue', 'queue_basename');
+        $this->control         = common_config('queue', 'control_channel');
+        $this->breakout        = common_config('queue', 'breakout');
+        $this->useTransactions = common_config('queue', 'stomp_transactions');
+        $this->useAcks         = common_config('queue', 'stomp_acks');
     }
 
     /**
@@ -66,20 +78,6 @@ class StompQueueManager extends QueueManager
         return IoManager::INSTANCE_PER_PROCESS;
     }
 
-    /**
-     * Record each site we'll be handling input for in this process,
-     * so we can listen to the necessary queues for it.
-     *
-     * @fixme possibly actually do subscription here to save another
-     *        loop over all sites later?
-     * @fixme possibly don't assume it's the current site
-     */
-    public function addSite($server)
-    {
-        $this->sites[] = $server;
-        $this->initialize();
-    }
-
     /**
      * Optional; ping any running queue handler daemons with a notification
      * such as announcing a new site to handle or requesting clean shutdown.
@@ -99,9 +97,10 @@ class StompQueueManager extends QueueManager
             $message .= ':' . $param;
         }
         $this->_connect();
-        $result = $this->con->send($this->control,
-                                   $message,
-                                   array ('created' => common_sql_now()));
+        $con = $this->cons[$this->defaultIdx];
+        $result = $con->send($this->control,
+                             $message,
+                             array ('created' => common_sql_now()));
         if ($result) {
             $this->_log(LOG_INFO, "Sent control ping to queue daemons: $message");
             return true;
@@ -112,82 +111,82 @@ class StompQueueManager extends QueueManager
     }
 
     /**
-     * Instantiate the appropriate QueueHandler class for the given queue.
+     * Saves an object into the queue item table.
      *
+     * @param mixed $object
      * @param string $queue
-     * @return mixed QueueHandler or null
-     */
-    function getHandler($queue)
-    {
-        $handlers = $this->handlers[$this->currentSite()];
-        if (isset($handlers[$queue])) {
-            $class = $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.
+     * @param string $siteNickname optional override to drop into another site's queue
      *
-     * @return array of strings
+     * @return boolean true on success
+     * @throws StompException on connection or send error
      */
-    function getQueues()
+    public function enqueue($object, $queue, $siteNickname=null)
     {
-        $group = $this->activeGroup();
-        $site = $this->currentSite();
-        if (empty($this->groups[$site][$group])) {
-            return array();
+        $this->_connect();
+        if (common_config('queue', 'stomp_enqueue_on')) {
+            // We're trying to force all writes to a single server.
+            // WARNING: this might do odd things if that server connection dies.
+            $idx = array_search(common_config('queue', 'stomp_enqueue_on'),
+                                $this->servers);
+            if ($idx === false) {
+                common_log(LOG_ERR, 'queue stomp_enqueue_on setting does not match our server list.');
+                $idx = $this->defaultIdx;
+            }
         } else {
-            return array_keys($this->groups[$site][$group]);
+            $idx = $this->defaultIdx;
         }
+        return $this->_doEnqueue($object, $queue, $idx, $siteNickname);
     }
 
     /**
-     * Register a queue transport name and handler class for your plugin.
-     * Only registered transports will be reliably picked up!
+     * Saves a notice object reference into the queue item table
+     * on the given connection.
      *
-     * @param string $transport
-     * @param string $class
-     * @param string $group
-     */
-    public function connect($transport, $class, $group='queuedaemon')
-    {
-        $this->handlers[$this->currentSite()][$transport] = $class;
-        $this->groups[$this->currentSite()][$group][$transport] = $class;
-    }
-
-    /**
-     * Saves a notice object reference into the queue item table.
      * @return boolean true on success
+     * @throws StompException on connection or send error
      */
-    public function enqueue($object, $queue)
+    protected function _doEnqueue($object, $queue, $idx, $siteNickname=null)
     {
-        $msg = $this->encode($object);
         $rep = $this->logrep($object);
+        $envelope = array('site' => $siteNickname ? $siteNickname : common_config('site', 'nickname'),
+                          'handler' => $queue,
+                          'payload' => $this->encode($object));
+        $msg = serialize($envelope);
+
+        $props = array('created' => common_sql_now());
+        if ($this->isPersistent($queue)) {
+            $props['persistent'] = 'true';
+        }
 
-        $this->_connect();
-
-        // XXX: serialize and send entire notice
-
-        $result = $this->con->send($this->queueName($queue),
-                                   $msg, 		// BODY of the message
-                                   array ('created' => common_sql_now(),
-                                          'persistent' => 'true'));
+        $con = $this->cons[$idx];
+        $host = $con->getServer();
+        $target = $this->queueName($queue);
+        $result = $con->send($target, $msg, $props);
 
         if (!$result) {
-            common_log(LOG_ERR, "Error sending $rep to $queue queue");
+            $this->_log(LOG_ERR, "Error sending $rep to $queue queue on $host $target");
             return false;
         }
 
-        common_log(LOG_DEBUG, "complete remote queueing $rep for $queue");
+        $this->_log(LOG_DEBUG, "complete remote queueing $rep for $queue on $host $target");
         $this->stats('enqueued', $queue);
+        return true;
+    }
+
+    /**
+     * Determine whether messages to this queue should be marked as persistent.
+     * Actual persistent storage depends on the queue server's configuration.
+     * @param string $queue
+     * @return bool
+     */
+    protected function isPersistent($queue)
+    {
+        $mode = common_config('queue', 'stomp_persistent');
+        if (is_array($mode)) {
+            return in_array($queue, $mode);
+        } else {
+            return (bool)$mode;
+        }
     }
 
     /**
@@ -198,7 +197,29 @@ class StompQueueManager extends QueueManager
      */
     public function getSockets()
     {
-        return array($this->con->getSocket());
+        $sockets = array();
+        foreach ($this->cons as $con) {
+            if ($con) {
+                $sockets[] = $con->getSocket();
+            }
+        }
+        return $sockets;
+    }
+
+    /**
+     * Get the Stomp connection object associated with the given socket.
+     * @param resource $socket
+     * @return int index into connections list
+     * @throws Exception
+     */
+    protected function connectionFromSocket($socket)
+    {
+        foreach ($this->cons as $i => $con) {
+            if ($con && $con->getSocket() === $socket) {
+                return $i;
+            }
+        }
+        throw new Exception(__CLASS__ . " asked to read from unrecognized socket");
     }
 
     /**
@@ -210,9 +231,22 @@ class StompQueueManager extends QueueManager
      */
     public function handleInput($socket)
     {
-        assert($socket === $this->con->getSocket());
+        $idx = $this->connectionFromSocket($socket);
+        $con = $this->cons[$idx];
+        $host = $con->getServer();
+        $this->defaultIdx = $idx;
+
         $ok = true;
-        $frames = $this->con->readFrames();
+        try {
+            $frames = $con->readFrames();
+        } catch (StompException $e) {
+            $this->_log(LOG_ERR, "Lost connection to $host: " . $e->getMessage());
+            fclose($socket); // ???
+            $this->cons[$idx] = null;
+            $this->transaction[$idx] = null;
+            $this->disconnect[$idx] = time();
+            return false;
+        }
         foreach ($frames as $frame) {
             $dest = $frame->headers['destination'];
             if ($dest == $this->control) {
@@ -222,15 +256,36 @@ class StompQueueManager extends QueueManager
                     break;
                 }
             } else {
-                $ok = $ok && $this->handleItem($frame);
+                $ok = $this->handleItem($frame) && $ok;
             }
+            $this->ack($idx, $frame);
+            $this->commit($idx);
+            $this->begin($idx);
         }
         return $ok;
     }
 
+    /**
+     * Attempt to reconnect in background if we lost a connection.
+     */
+    function idle()
+    {
+        $now = time();
+        foreach ($this->cons as $idx => $con) {
+            if (empty($con)) {
+                $age = $now - $this->disconnect[$idx];
+                if ($age >= 60) {
+                    $this->_reconnect($idx);
+                }
+            }
+        }
+        return true;
+    }
+
     /**
      * Initialize our connection and subscribe to all the queues
-     * we're going to need to handle...
+     * we're going to need to handle... If multiple queue servers
+     * are configured for failover, we'll listen to all of them.
      *
      * Side effects: in multi-site mode, may reset site configuration.
      *
@@ -240,26 +295,19 @@ class StompQueueManager extends QueueManager
     public function start($master)
     {
         parent::start($master);
-        $this->_connect();
+        $this->_connectAll();
 
-        common_log(LOG_INFO, "Subscribing to $this->control");
-        $this->con->subscribe($this->control);
-        if ($this->sites) {
-            foreach ($this->sites as $server) {
-                StatusNet::init($server);
-                $this->doSubscribe();
+        foreach ($this->cons as $i => $con) {
+            if ($con) {
+                $this->doSubscribe($con);
+                $this->begin($i);
             }
-        } else {
-            $this->doSubscribe();
         }
-        $this->begin();
         return true;
     }
-    
+
     /**
-     * Subscribe to all the queues we're going to need to handle...
-     *
-     * Side effects: in multi-site mode, may reset site configuration.
+     * Close out any active connections.
      *
      * @return bool return false on failure
      */
@@ -267,75 +315,153 @@ class StompQueueManager extends QueueManager
     {
         // If there are any outstanding delivered messages we haven't processed,
         // free them for another thread to take.
-        $this->rollback();
-        $this->con->unsubscribe($this->control);
-        if ($this->sites) {
-            foreach ($this->sites as $server) {
-                StatusNet::init($server);
-                $this->doUnsubscribe();
+        foreach ($this->cons as $i => $con) {
+            if ($con) {
+                $this->rollback($i);
+                $con->disconnect();
+                $this->cons[$i] = null;
             }
-        } else {
-            $this->doUnsubscribe();
         }
         return true;
     }
 
     /**
-     * Get identifier of the currently active site configuration
-     * @return string
+     * Lazy open a single connection to Stomp queue server.
+     * If multiple servers are configured, we let the Stomp client library
+     * worry about finding a working connection among them.
      */
-    protected function currentSite()
+    protected function _connect()
     {
-        return common_config('site', 'server'); // @fixme switch to nickname
+        if (empty($this->cons)) {
+            $list = $this->servers;
+            if (count($list) > 1) {
+                shuffle($list); // Randomize to spread load
+                $url = 'failover://(' . implode(',', $list) . ')';
+            } else {
+                $url = $list[0];
+            }
+            $con = $this->_doConnect($url);
+            $this->cons = array($con);
+            $this->transactionCount = array(0);
+            $this->transaction = array(null);
+            $this->disconnect = array(null);
+        }
     }
 
     /**
-     * Lazy open connection to Stomp queue server.
+     * Lazy open connections to all Stomp servers, if in manual failover
+     * mode. This means the queue servers don't speak to each other, so
+     * we have to listen to all of them to make sure we get all events.
      */
-    protected function _connect()
+    protected function _connectAll()
     {
-        if (empty($this->con)) {
-            $this->_log(LOG_INFO, "Connecting to '$this->server' as '$this->username'...");
-            $this->con = new LiberalStomp($this->server);
-
-            if ($this->con->connect($this->username, $this->password)) {
-                $this->_log(LOG_INFO, "Connected.");
-            } else {
-                $this->_log(LOG_ERR, 'Failed to connect to queue server');
-                throw new ServerException('Failed to connect to queue server');
+        if (!common_config('queue', 'stomp_manual_failover')) {
+            return $this->_connect();
+        }
+        if (empty($this->cons)) {
+            $this->cons = array();
+            $this->transactionCount = array();
+            $this->transaction = array();
+            foreach ($this->servers as $idx => $server) {
+                try {
+                    $this->cons[] = $this->_doConnect($server);
+                    $this->disconnect[] = null;
+                } catch (Exception $e) {
+                    // s'okay, we'll live
+                    $this->cons[] = null;
+                    $this->disconnect[] = time();
+                }
+                $this->transactionCount[] = 0;
+                $this->transaction[] = null;
+            }
+            if (empty($this->cons)) {
+                throw new ServerException("No queue servers reachable...");
+                return false;
             }
         }
     }
 
     /**
-     * Subscribe to all enabled notice queues for the current site.
+     * Attempt to manually reconnect to the Stomp server for the given
+     * slot. If successful, set up our subscriptions on it.
      */
-    protected function doSubscribe()
+    protected function _reconnect($idx)
     {
-        $site = $this->currentSite();
-        $this->_connect();
-        foreach ($this->getQueues() as $queue) {
-            $rawqueue = $this->queueName($queue);
-            $this->subscriptions[$site][$queue] = $rawqueue;
-            $this->_log(LOG_INFO, "Subscribing to $rawqueue");
-            $this->con->subscribe($rawqueue);
+        try {
+            $con = $this->_doConnect($this->servers[$idx]);
+        } catch (Exception $e) {
+            $this->_log(LOG_ERR, $e->getMessage());
+            $con = null;
+        }
+        if ($con) {
+            $this->cons[$idx] = $con;
+            $this->disconnect[$idx] = null;
+
+            $this->doSubscribe($con);
+            $this->begin($idx);
+        } else {
+            // Try again later...
+            $this->disconnect[$idx] = time();
+        }
+    }
+
+    protected function _doConnect($server)
+    {
+        $this->_log(LOG_INFO, "Connecting to '$server' as '$this->username'...");
+        $con = new LiberalStomp($server);
+
+        if ($con->connect($this->username, $this->password)) {
+            $this->_log(LOG_INFO, "Connected.");
+        } else {
+            $this->_log(LOG_ERR, 'Failed to connect to queue server');
+            throw new ServerException('Failed to connect to queue server');
         }
+
+        return $con;
     }
 
     /**
-     * Subscribe from all enabled notice queues for the current site.
+     * Set up all our raw queue subscriptions on the given connection
+     * @param LiberalStomp $con
      */
-    protected function doUnsubscribe()
+    protected function doSubscribe(LiberalStomp $con)
     {
-        $site = $this->currentSite();
-        $this->_connect();
-        if (!empty($this->subscriptions[$site])) {
-            foreach ($this->subscriptions[$site] as $queue => $rawqueue) {
-                $this->_log(LOG_INFO, "Unsubscribing from $rawqueue");
-                $this->con->unsubscribe($rawqueue);
-                unset($this->subscriptions[$site][$queue]);
+        $host = $con->getServer();
+        foreach ($this->subscriptions() as $sub) {
+            $this->_log(LOG_INFO, "Subscribing to $sub on $host");
+            $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);
     }
 
     /**
@@ -348,77 +474,131 @@ class StompQueueManager extends QueueManager
      * match the site that queued the event.
      *
      * @param StompFrame $frame
-     * @return bool
+     * @return bool success
      */
     protected function handleItem($frame)
     {
-        list($site, $queue) = $this->parseDestination($frame->headers['destination']);
-        if ($site != $this->currentSite()) {
-            $this->stats('switch');
-            StatusNet::init($site);
+        $host = $this->cons[$this->defaultIdx]->getServer();
+        $message = unserialize($frame->body);
+        $site = $message['site'];
+        $queue = $message['handler'];
+
+        if ($this->isDeadletter($frame, $message)) {
+            $this->stats('deadletter', $queue);
+	        return false;
         }
 
-        if (is_numeric($frame->body)) {
-            $id = intval($frame->body);
-            $info = "notice $id posted at {$frame->headers['created']} in queue $queue";
+        // @fixme detect failing site switches
+        $this->switchSite($site);
 
-            $notice = Notice::staticGet('id', $id);
-            if (empty($notice)) {
-                $this->_log(LOG_WARNING, "Skipping missing $info");
-                $this->ack($frame);
-                $this->commit();
-                $this->begin();
-                $this->stats('badnotice', $queue);
-                return false;
-            }
-
-            $item = $notice;
-        } else {
-            // @fixme should we serialize, or json, or what here?
-            $info = "string posted at {$frame->headers['created']} in queue $queue";
-            $item = $frame->body;
+        $item = $this->decode($message['payload']);
+        if (empty($item)) {
+            $this->_log(LOG_ERR, "Skipping empty or deleted item in queue $queue from $host");
+            $this->stats('baditem', $queue);
+            return false;
         }
+        $info = $this->logrep($item) . " posted at " .
+                $frame->headers['created'] . " in queue $queue from $host";
+        $this->_log(LOG_DEBUG, "Dequeued $info");
 
         $handler = $this->getHandler($queue);
         if (!$handler) {
             $this->_log(LOG_ERR, "Missing handler class; skipping $info");
-            $this->ack($frame);
-            $this->commit();
-            $this->begin();
             $this->stats('badhandler', $queue);
             return false;
         }
 
-        $ok = $handler->handle($item);
+        try {
+            $ok = $handler->handle($item);
+        } catch (Exception $e) {
+            $this->_log(LOG_ERR, "Exception on queue $queue: " . $e->getMessage());
+            $ok = false;
+        }
 
-        if (!$ok) {
+        if ($ok) {
+            $this->_log(LOG_INFO, "Successfully handled $info");
+            $this->stats('handled', $queue);
+        } else {
             $this->_log(LOG_WARNING, "Failed handling $info");
-            // FIXME we probably shouldn't have to do
-            // this kind of queue management ourselves;
-            // if we don't ack, it should resend...
-            $this->ack($frame);
+            // Requeing moves the item to the end of the line for its next try.
+            // @fixme add a manual retry count
             $this->enqueue($item, $queue);
-            $this->commit();
-            $this->begin();
             $this->stats('requeued', $queue);
-            return false;
         }
 
-        $this->_log(LOG_INFO, "Successfully handled $info");
-        $this->ack($frame);
-        $this->commit();
-        $this->begin();
-        $this->stats('handled', $queue);
-        return true;
+        return $ok;
+    }
+
+    /**
+     * Check if a redelivered message has been run through enough
+     * that we're going to give up on it.
+     *
+     * @param StompFrame $frame
+     * @param array $message unserialized message body
+     * @return boolean true if we should discard
+     */
+    protected function isDeadLetter($frame, $message)
+    {
+        if (isset($frame->headers['redelivered']) && $frame->headers['redelivered'] == 'true') {
+	        // Message was redelivered, possibly indicating a previous failure.
+            $msgId = $frame->headers['message-id'];
+            $site = $message['site'];
+            $queue = $message['handler'];
+	        $msgInfo = "message $msgId for $site in queue $queue";
+
+	        $deliveries = $this->incDeliveryCount($msgId);
+	        if ($deliveries > common_config('queue', 'max_retries')) {
+		        $info = "DEAD-LETTER FILE: Gave up after retry $deliveries on $msgInfo";
+
+		        $outdir = common_config('queue', 'dead_letter_dir');
+		        if ($outdir) {
+    		        $filename = $outdir . "/$site-$queue-" . rawurlencode($msgId);
+    		        $info .= ": dumping to $filename";
+    		        file_put_contents($filename, $message['payload']);
+		        }
+
+		        common_log(LOG_ERR, $info);
+		        return true;
+	        } else {
+	            common_log(LOG_INFO, "retry $deliveries on $msgInfo");
+	        }
+        }
+        return false;
+    }
+
+    /**
+     * Update count of times we've re-encountered this message recently,
+     * triggered when we get a message marked as 'redelivered'.
+     *
+     * Requires a CLI-friendly cache configuration.
+     *
+     * @param string $msgId message-id header from message
+     * @return int number of retries recorded
+     */
+    function incDeliveryCount($msgId)
+    {
+	    $count = 0;
+	    $cache = common_memcache();
+	    if ($cache) {
+		    $key = 'statusnet:stomp:message-retries:' . $msgId;
+		    $count = $cache->increment($key);
+		    if (!$count) {
+			    $count = 1;
+			    $cache->set($key, $count, null, 3600);
+			    $got = $cache->get($key);
+		    }
+	    }
+	    return $count;
     }
 
     /**
      * Process a control signal broadcast.
      *
+     * @param int $idx connection index
      * @param array $frame Stomp frame
      * @return bool true to continue; false to stop further processing.
      */
-    protected function handleControlSignal($frame)
+    protected function handleControlSignal($idx, $frame)
     {
         $message = trim($frame->body);
         if (strpos($message, ':') !== false) {
@@ -441,127 +621,135 @@ class StompQueueManager extends QueueManager
         } else {
             $this->_log(LOG_ERR, "Ignoring unrecognized control message: $message");
         }
-
-        $this->ack($frame);
-        $this->commit();
-        $this->begin();
         return $shutdown;
     }
-    
+
+    /**
+     * Switch site, if necessary, and reset current handler assignments
+     * @param string $site
+     */
+    function switchSite($site)
+    {
+        if ($site != StatusNet::currentSite()) {
+            $this->stats('switch');
+            StatusNet::switchSite($site);
+            $this->initialize();
+        }
+    }
+
     /**
-     * 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)) {
-            if ($nickname == common_config('site', 'nickname')) {
-                StatusNet::init(common_config('site', 'server'));
-                $this->doUnsubscribe();
-                $this->doSubscribe();
-            } else {
-                $this->_log(LOG_INFO, "Ignoring update ping for other site $nickname");
+        $sn = Status_network::staticGet('nickname', $nickname);
+        if ($sn) {
+            $this->switchSite($nickname);
+            if (!in_array($nickname, $this->sites)) {
+                $this->addSite();
             }
+            $this->stats('siteupdate');
         } else {
-            $sn = Status_network::staticGet($nickname);
-            if ($sn) {
-                $server = $sn->getServerName(); // @fixme do config-by-nick
-                StatusNet::init($server);
-                if (empty($this->sites[$server])) {
-                    $this->addSite($server);
-                }
-                $this->_log(LOG_INFO, "(Re)subscribing to queues for site $nickname / $server");
-                $this->doUnsubscribe();
-                $this->doSubscribe();
-                $this->stats('siteupdate');
-            } else {
-                $this->_log(LOG_ERR, "Ignoring ping for unrecognized new site $nickname");
-            }
+            $this->_log(LOG_ERR, "Ignoring ping for unrecognized new site $nickname");
         }
     }
 
     /**
      * Combines the queue_basename from configuration with the
-     * site server name and queue name to give eg:
+     * group name for this queue to give eg:
      *
-     * /queue/statusnet/identi.ca/sms
+     * /queue/statusnet/main
+     * /queue/statusnet/main/distrib
+     * /queue/statusnet/xmpp/xmppout/site01
      *
      * @param string $queue
      * @return string
      */
     protected function queueName($queue)
     {
-        return common_config('queue', 'queue_basename') .
-            $this->currentSite() . '/' . $queue;
+        $group = $this->queueGroup($queue);
+        $site = StatusNet::currentSite();
+
+        $specs = array("$group/$queue/$site",
+                       "$group/$queue");
+        foreach ($specs as $spec) {
+            if (in_array($spec, $this->breakout)) {
+                return $this->base . $spec;
+            }
+        }
+        return $this->base . $group;
     }
 
     /**
-     * Returns the site and queue name from the server-side queue.
+     * Get the breakout mode for the given queue on the current site.
      *
-     * @param string queue destination (eg '/queue/statusnet/identi.ca/sms')
-     * @return array of site and queue: ('identi.ca','sms') or false if unrecognized
+     * @param string $queue
+     * @return string one of 'shared', 'handler', 'site'
      */
-    protected function parseDestination($dest)
+    protected function breakoutMode($queue)
     {
-        $prefix = common_config('queue', 'queue_basename');
-        if (substr($dest, 0, strlen($prefix)) == $prefix) {
-            $rest = substr($dest, strlen($prefix));
-            return explode("/", $rest, 2);
+        $breakout = common_config('queue', 'breakout');
+        if (isset($breakout[$queue])) {
+            return $breakout[$queue];
+        } else if (isset($breakout['*'])) {
+            return $breakout['*'];
         } else {
-            common_log(LOG_ERR, "Got a message from unrecognized stomp queue: $dest");
-            return array(false, false);
+            return 'shared';
         }
     }
 
-    function _log($level, $msg)
-    {
-        common_log($level, 'StompQueueManager: '.$msg);
-    }
-
-    protected function begin()
+    protected function begin($idx)
     {
         if ($this->useTransactions) {
-            if ($this->transaction) {
+            if (!empty($this->transaction[$idx])) {
                 throw new Exception("Tried to start transaction in the middle of a transaction");
             }
-            $this->transactionCount++;
-            $this->transaction = $this->master->id . '-' . $this->transactionCount . '-' . time();
-            $this->con->begin($this->transaction);
+            $this->transactionCount[$idx]++;
+            $this->transaction[$idx] = $this->master->id . '-' . $this->transactionCount[$idx] . '-' . time();
+            $this->cons[$idx]->begin($this->transaction[$idx]);
         }
     }
 
-    protected function ack($frame)
+    protected function ack($idx, $frame)
     {
-        if ($this->useTransactions) {
-            if (!$this->transaction) {
-                throw new Exception("Tried to ack but not in a transaction");
+        if ($this->useAcks) {
+            if ($this->useTransactions) {
+                if (empty($this->transaction[$idx])) {
+                    throw new Exception("Tried to ack but not in a transaction");
+                }
+                $this->cons[$idx]->ack($frame, $this->transaction[$idx]);
+            } else {
+                $this->cons[$idx]->ack($frame);
             }
         }
-        $this->con->ack($frame, $this->transaction);
     }
 
-    protected function commit()
+    protected function commit($idx)
     {
         if ($this->useTransactions) {
-            if (!$this->transaction) {
+            if (empty($this->transaction[$idx])) {
                 throw new Exception("Tried to commit but not in a transaction");
             }
-            $this->con->commit($this->transaction);
-            $this->transaction = null;
+            $this->cons[$idx]->commit($this->transaction[$idx]);
+            $this->transaction[$idx] = null;
         }
     }
 
-    protected function rollback()
+    protected function rollback($idx)
     {
         if ($this->useTransactions) {
-            if (!$this->transaction) {
+            if (empty($this->transaction[$idx])) {
                 throw new Exception("Tried to rollback but not in a transaction");
             }
-            $this->con->commit($this->transaction);
-            $this->transaction = null;
+            $this->cons[$idx]->commit($this->transaction[$idx]);
+            $this->transaction[$idx] = null;
         }
     }
 }