From c7507e7e9dafa6d6e054978e720e4fce3abc9929 Mon Sep 17 00:00:00 2001
From: Brion Vibber <brion@pobox.com>
Date: Fri, 22 Jan 2010 12:52:36 -0800
Subject: XMPP queued output & initial retooling of DB queue manager to support
 non-Notice objects.

Queue handlers for XMPP individual & firehose output now send their XML stanzas
to another output queue instead of connecting directly to the chat server. This
lets us have as many general processing threads as we need, while all actual
XMPP input and output go through a single daemon with a single connection open.

This avoids problems with multiple connected resources:
* multiple windows shown in some chat clients (psi, gajim, kopete)
* extra load on server
* incoming message delivery forwarding issues

Database changes:
* queue_item drops 'notice_id' in favor of a 'frame' blob.
  This is based on Craig Andrews' work branch to generalize queues to take any
  object, but conservatively leaving out the serialization for now.
  Table updater (preserves any existing queued items) in db/rc3to09.sql

Code changes to watch out for:
* Queue handlers should now define a handle() method instead of handle_notice()
* QueueDaemon and XmppDaemon now share common i/o (IoMaster) and respawning
  thread management (RespawningDaemon) infrastructure.
* The polling XmppConfirmManager has been dropped, as the message is queued
  directly when saving IM settings.
* Enable $config['queue']['debug_memory'] to output current memory usage at
  each run through the event loop to watch for memory leaks

To do:
* Adapt XMPP i/o to component connection mode for multi-site support.
* XMPP input can also be broken out to a queue, which would allow the actual
  notice save etc to be handled by general queue threads.
* Make sure there are no problems with simply pushing serialized Notice objects
  to queues.
* Find a way to improve interactive performance of the database-backed queue
  handler; polling is pretty painful to XMPP.
* Possibly redo the way QueueHandlers are injected into a QueueManager. The
  grouping used to split out the XMPP output queue is a bit awkward.

Conflicts:

	scripts/xmppdaemon.php
---
 lib/dbqueuemanager.php      | 140 +++++++---------------
 lib/default.php             |   1 +
 lib/iomaster.php            |  39 +++---
 lib/jabber.php              |  43 ++++---
 lib/jabberqueuehandler.php  |   4 +-
 lib/ombqueuehandler.php     |   2 +-
 lib/pingqueuehandler.php    |   2 +-
 lib/pluginqueuehandler.php  |   2 +-
 lib/publicqueuehandler.php  |   6 +-
 lib/queued_xmpp.php         | 117 ++++++++++++++++++
 lib/queuehandler.php        |  95 ++-------------
 lib/queuemanager.php        | 125 +++++++++++++++----
 lib/smsqueuehandler.php     |   2 +-
 lib/spawningdaemon.php      | 159 ++++++++++++++++++++++++
 lib/stompqueuemanager.php   |  56 +++++----
 lib/util.php                |   3 +-
 lib/xmppconfirmmanager.php  | 168 --------------------------
 lib/xmppmanager.php         | 286 ++++++++++++++++++++++++++++++++++++++------
 lib/xmppoutqueuehandler.php |  55 +++++++++
 19 files changed, 821 insertions(+), 484 deletions(-)
 create mode 100644 lib/queued_xmpp.php
 create mode 100644 lib/spawningdaemon.php
 delete mode 100644 lib/xmppconfirmmanager.php
 create mode 100644 lib/xmppoutqueuehandler.php

(limited to 'lib')

diff --git a/lib/dbqueuemanager.php b/lib/dbqueuemanager.php
index 889365b64..c6350fc66 100644
--- a/lib/dbqueuemanager.php
+++ b/lib/dbqueuemanager.php
@@ -31,19 +31,17 @@
 class DBQueueManager extends QueueManager
 {
     /**
-     * Saves a notice object reference into the queue item table.
+     * Saves an object reference into the queue item table.
      * @return boolean true on success
      * @throws ServerException on failure
      */
     public function enqueue($object, $queue)
     {
-        $notice = $object;
-
         $qi = new Queue_item();
 
-        $qi->notice_id = $notice->id;
+        $qi->frame     = $this->encode($object);
         $qi->transport = $queue;
-        $qi->created   = $notice->created;
+        $qi->created   = common_sql_now();
         $result        = $qi->insert();
 
         if (!$result) {
@@ -57,146 +55,92 @@ class DBQueueManager extends QueueManager
     }
 
     /**
-     * Poll every minute for new events during idle periods.
+     * Poll every 10 seconds for new events during idle periods.
      * We'll look in more often when there's data available.
      *
      * @return int seconds
      */
     public function pollInterval()
     {
-        return 60;
+        return 10;
     }
 
     /**
      * Run a polling cycle during idle processing in the input loop.
-     * @return boolean true if we had a hit
+     * @return boolean true if we should poll again for more data immediately
      */
     public function poll()
     {
         $this->_log(LOG_DEBUG, 'Checking for notices...');
-        $item = $this->_nextItem();
-        if ($item === false) {
+        $qi = Queue_item::top($this->getQueues());
+        if (empty($qi)) {
             $this->_log(LOG_DEBUG, 'No notices waiting; idling.');
             return false;
         }
-        if ($item === true) {
-            // We dequeued an entry for a deleted or invalid notice.
-            // Consider it a hit for poll rate purposes.
-            return true;
-        }
 
-        list($queue, $notice) = $item;
-        $this->_log(LOG_INFO, 'Got notice '. $notice->id . ' for transport ' . $queue);
-
-        // Yay! Got one!
-        $handler = $this->getHandler($queue);
-        if ($handler) {
-            if ($handler->handle_notice($notice)) {
-                $this->_log(LOG_INFO, "[$queue:notice $notice->id] Successfully handled notice");
-                $this->_done($notice, $queue);
+        $queue = $qi->transport;
+        $item = $this->decode($qi->frame);
+
+        if ($item) {
+            $rep = $this->logrep($item);
+            $this->_log(LOG_INFO, "Got $rep for transport $queue");
+            
+            $handler = $this->getHandler($queue);
+            if ($handler) {
+                if ($handler->handle($item)) {
+                    $this->_log(LOG_INFO, "[$queue:$rep] Successfully handled item");
+                    $this->_done($qi);
+                } else {
+                    $this->_log(LOG_INFO, "[$queue:$rep] Failed to handle item");
+                    $this->_fail($qi);
+                }
             } else {
-                $this->_log(LOG_INFO, "[$queue:notice $notice->id] Failed to handle notice");
-                $this->_fail($notice, $queue);
+                $this->_log(LOG_INFO, "[$queue:$rep] No handler for queue $queue; discarding.");
+                $this->_done($qi);
             }
         } else {
-            $this->_log(LOG_INFO, "[$queue:notice $notice->id] No handler for queue $queue; discarding.");
-            $this->_done($notice, $queue);
+            $this->_log(LOG_INFO, "[$queue] Got empty/deleted item, discarding");
+            $this->_fail($qi);
         }
         return true;
     }
 
-    /**
-     * Pop the oldest unclaimed item off the queue set and claim it.
-     *
-     * @return mixed false if no items; true if bogus hit; otherwise array(string, Notice)
-     *               giving the queue transport name.
-     */
-    protected function _nextItem()
-    {
-        $start = time();
-        $result = null;
-
-        $qi = Queue_item::top();
-        if (empty($qi)) {
-            return false;
-        }
-
-        $queue = $qi->transport;
-        $notice = Notice::staticGet('id', $qi->notice_id);
-        if (empty($notice)) {
-            $this->_log(LOG_INFO, "[$queue:notice $notice->id] dequeued non-existent notice");
-            $qi->delete();
-            return true;
-        }
-
-        $result = $notice;
-        return array($queue, $notice);
-    }
-
     /**
      * Delete our claimed item from the queue after successful processing.
      *
-     * @param Notice $object
-     * @param string $queue
+     * @param QueueItem $qi
      */
-    protected function _done($object, $queue)
+    protected function _done($qi)
     {
-        // XXX: right now, we only handle notices
-
-        $notice = $object;
-
-        $qi = Queue_item::pkeyGet(array('notice_id' => $notice->id,
-                                        'transport' => $queue));
+        $queue = $qi->transport;
 
-        if (empty($qi)) {
-            $this->_log(LOG_INFO, "[$queue:notice $notice->id] Cannot find queue item");
-        } else {
-            if (empty($qi->claimed)) {
-                $this->_log(LOG_WARNING, "[$queue:notice $notice->id] Reluctantly releasing unclaimed queue item");
-            }
-            $qi->delete();
-            $qi->free();
+        if (empty($qi->claimed)) {
+            $this->_log(LOG_WARNING, "Reluctantly releasing unclaimed queue item $qi->id from $qi->queue");
         }
+        $qi->delete();
 
-        $this->_log(LOG_INFO, "[$queue:notice $notice->id] done with item");
         $this->stats('handled', $queue);
-
-        $notice->free();
     }
 
     /**
      * Free our claimed queue item for later reprocessing in case of
      * temporary failure.
      *
-     * @param Notice $object
-     * @param string $queue
+     * @param QueueItem $qi
      */
-    protected function _fail($object, $queue)
+    protected function _fail($qi)
     {
-        // XXX: right now, we only handle notices
-
-        $notice = $object;
-
-        $qi = Queue_item::pkeyGet(array('notice_id' => $notice->id,
-                                        'transport' => $queue));
+        $queue = $qi->transport;
 
-        if (empty($qi)) {
-            $this->_log(LOG_INFO, "[$queue:notice $notice->id] Cannot find queue item");
+        if (empty($qi->claimed)) {
+            $this->_log(LOG_WARNING, "[$queue:item $qi->id] Ignoring failure for unclaimed queue item");
         } else {
-            if (empty($qi->claimed)) {
-                $this->_log(LOG_WARNING, "[$queue:notice $notice->id] Ignoring failure for unclaimed queue item");
-            } else {
-                $orig = clone($qi);
-                $qi->claimed = null;
-                $qi->update($orig);
-                $qi = null;
-            }
+            $orig = clone($qi);
+            $qi->claimed = null;
+            $qi->update($orig);
         }
 
-        $this->_log(LOG_INFO, "[$queue:notice $notice->id] done with queue item");
         $this->stats('error', $queue);
-
-        $notice->free();
     }
 
     protected function _log($level, $msg)
diff --git a/lib/default.php b/lib/default.php
index 5b2ae6c7c..790a5b387 100644
--- a/lib/default.php
+++ b/lib/default.php
@@ -81,6 +81,7 @@ $default =
               'stomp_password' => null,
               'monitor' => null, // URL to monitor ping endpoint (work in progress)
               'softlimit' => '90%', // total size or % of memory_limit at which to restart queue threads gracefully
+              'debug_memory' => false, // true to spit memory usage to log
               ),
         'license' =>
         array('type' => 'cc', # can be 'cc', 'allrightsreserved', 'private'
diff --git a/lib/iomaster.php b/lib/iomaster.php
index ce77b53b2..004e92b3e 100644
--- a/lib/iomaster.php
+++ b/lib/iomaster.php
@@ -27,7 +27,7 @@
  * @link      http://status.net/
  */
 
-class IoMaster
+abstract class IoMaster
 {
     public $id;
 
@@ -66,23 +66,18 @@ class IoMaster
             if ($site != common_config('site', 'server')) {
                 StatusNet::init($site);
             }
-
-            $classes = array();
-            if (Event::handle('StartIoManagerClasses', array(&$classes))) {
-                $classes[] = 'QueueManager';
-                if (common_config('xmpp', 'enabled') && !defined('XMPP_EMERGENCY_FLAG')) {
-                    $classes[] = 'XmppManager'; // handles pings/reconnects
-                    $classes[] = 'XmppConfirmManager'; // polls for outgoing confirmations
-                }
-            }
-            Event::handle('EndIoManagerClasses', array(&$classes));
-
-            foreach ($classes as $class) {
-                $this->instantiate($class);
-            }
+            $this->initManagers();
         }
     }
 
+    /**
+     * Initialize IoManagers for the currently configured site
+     * which are appropriate to this instance.
+     *
+     * Pass class names into $this->instantiate()
+     */
+    abstract function initManagers();
+
     /**
      * Pull all local sites from status_network table.
      * @return array of hostnames
@@ -170,7 +165,7 @@ class IoMaster
                 $write = array();
                 $except = array();
                 $this->logState('listening');
-                common_log(LOG_INFO, "Waiting up to $timeout seconds for socket data...");
+                common_log(LOG_DEBUG, "Waiting up to $timeout seconds for socket data...");
                 $ready = stream_select($read, $write, $except, $timeout, 0);
 
                 if ($ready === false) {
@@ -190,7 +185,7 @@ class IoMaster
 
             if ($timeout > 0 && empty($sockets)) {
                 // If we had no listeners, sleep until the pollers' next requested wakeup.
-                common_log(LOG_INFO, "Sleeping $timeout seconds until next poll cycle...");
+                common_log(LOG_DEBUG, "Sleeping $timeout seconds until next poll cycle...");
                 $this->logState('sleep');
                 sleep($timeout);
             }
@@ -207,6 +202,8 @@ class IoMaster
                 if ($usage > $memoryLimit) {
                     common_log(LOG_INFO, "Queue thread hit soft memory limit ($usage > $memoryLimit); gracefully restarting.");
                     break;
+                } else if (common_config('queue', 'debug_memory')) {
+                    common_log(LOG_DEBUG, "Memory usage $usage");
                 }
             }
         }
@@ -223,8 +220,7 @@ class IoMaster
     {
         $softLimit = trim(common_config('queue', 'softlimit'));
         if (substr($softLimit, -1) == '%') {
-            $limit = trim(ini_get('memory_limit'));
-            $limit = $this->parseMemoryLimit($limit);
+            $limit = $this->parseMemoryLimit(ini_get('memory_limit'));
             if ($limit > 0) {
                 return intval(substr($softLimit, 0, -1) * $limit / 100);
             } else {
@@ -242,9 +238,10 @@ class IoMaster
      * @param string $mem
      * @return int
      */
-    protected function parseMemoryLimit($mem)
+    public function parseMemoryLimit($mem)
     {
         // http://www.php.net/manual/en/faq.using.php#faq.using.shorthandbytes
+        $mem = strtolower(trim($mem));
         $size = array('k' => 1024,
                       'm' => 1024*1024,
                       'g' => 1024*1024*1024);
@@ -253,7 +250,7 @@ class IoMaster
         } else if (is_numeric($mem)) {
             return intval($mem);
         } else {
-            $mult = strtolower(substr($mem, -1));
+            $mult = substr($mem, -1);
             if (isset($size[$mult])) {
                 return substr($mem, 0, -1) * $size[$mult];
             } else {
diff --git a/lib/jabber.php b/lib/jabber.php
index 4cdfa6746..b6b23521b 100644
--- a/lib/jabber.php
+++ b/lib/jabber.php
@@ -85,6 +85,27 @@ class Sharing_XMPP extends XMPPHP_XMPP
     }
 }
 
+/**
+ * Build an XMPP proxy connection that'll save outgoing messages
+ * to the 'xmppout' queue to be picked up by xmppdaemon later.
+ */
+function jabber_proxy()
+{
+	$proxy = new Queued_XMPP(common_config('xmpp', 'host') ?
+                             common_config('xmpp', 'host') :
+                             common_config('xmpp', 'server'),
+                             common_config('xmpp', 'port'),
+                             common_config('xmpp', 'user'),
+                             common_config('xmpp', 'password'),
+                             common_config('xmpp', 'resource') . 'daemon',
+                             common_config('xmpp', 'server'),
+                             common_config('xmpp', 'debug') ?
+                             true : false,
+                             common_config('xmpp', 'debug') ?
+                             XMPPHP_Log::LEVEL_VERBOSE :  null);
+    return $proxy;
+}
+
 /**
  * Lazy-connect the configured Jabber account to the configured server;
  * if already opened, the same connection will be returned.
@@ -143,7 +164,7 @@ function jabber_connect($resource=null)
 }
 
 /**
- * send a single notice to a given Jabber address
+ * Queue send for a single notice to a given Jabber address
  *
  * @param string $to     JID to send the notice to
  * @param Notice $notice notice to send
@@ -153,10 +174,7 @@ function jabber_connect($resource=null)
 
 function jabber_send_notice($to, $notice)
 {
-    $conn = jabber_connect();
-    if (!$conn) {
-        return false;
-    }
+    $conn = jabber_proxy();
     $profile = Profile::staticGet($notice->profile_id);
     if (!$profile) {
         common_log(LOG_WARNING, 'Refusing to send notice with ' .
@@ -221,10 +239,7 @@ function jabber_format_entry($profile, $notice)
 
 function jabber_send_message($to, $body, $type='chat', $subject=null)
 {
-    $conn = jabber_connect();
-    if (!$conn) {
-        return false;
-    }
+    $conn = jabber_proxy();
     $conn->message($to, $body, $type, $subject);
     return true;
 }
@@ -319,7 +334,7 @@ function jabber_special_presence($type, $to=null, $show=null, $status=null)
 }
 
 /**
- * broadcast a notice to all subscribers and reply recipients
+ * Queue broadcast of a notice to all subscribers and reply recipients
  *
  * This function will send a notice to all subscribers on the local server
  * who have Jabber addresses, and have Jabber notification enabled, and
@@ -354,7 +369,7 @@ function jabber_broadcast_notice($notice)
 
     $sent_to = array();
 
-    $conn = jabber_connect();
+    $conn = jabber_proxy();
 
     $ni = $notice->whoGets();
 
@@ -389,14 +404,13 @@ function jabber_broadcast_notice($notice)
                    'Sending notice ' . $notice->id . ' to ' . $user->jabber,
                    __FILE__);
         $conn->message($user->jabber, $msg, 'chat', null, $entry);
-        $conn->processTime(0);
     }
 
     return true;
 }
 
 /**
- * send a notice to all public listeners
+ * Queue send of a notice to all public listeners
  *
  * For notices that are generated on the local system (by users), we can optionally
  * forward them to remote listeners by XMPP.
@@ -429,7 +443,7 @@ function jabber_public_notice($notice)
         $msg   = jabber_format_notice($profile, $notice);
         $entry = jabber_format_entry($profile, $notice);
 
-        $conn = jabber_connect();
+        $conn = jabber_proxy();
 
         foreach ($public as $address) {
             common_log(LOG_INFO,
@@ -437,7 +451,6 @@ function jabber_public_notice($notice)
                        ' to public listener ' . $address,
                        __FILE__);
             $conn->message($address, $msg, 'chat', null, $entry);
-            $conn->processTime(0);
         }
         $profile->free();
     }
diff --git a/lib/jabberqueuehandler.php b/lib/jabberqueuehandler.php
index b1518866d..83471f2df 100644
--- a/lib/jabberqueuehandler.php
+++ b/lib/jabberqueuehandler.php
@@ -34,14 +34,14 @@ class JabberQueueHandler extends QueueHandler
         return 'jabber';
     }
 
-    function handle_notice($notice)
+    function handle($notice)
     {
         require_once(INSTALLDIR.'/lib/jabber.php');
         try {
             return jabber_broadcast_notice($notice);
         } catch (XMPPHP_Exception $e) {
             $this->log(LOG_ERR, "Got an XMPPHP_Exception: " . $e->getMessage());
-            exit(1);
+            return false;
         }
     }
 }
diff --git a/lib/ombqueuehandler.php b/lib/ombqueuehandler.php
index 3ffc1313b..24896c784 100644
--- a/lib/ombqueuehandler.php
+++ b/lib/ombqueuehandler.php
@@ -36,7 +36,7 @@ class OmbQueueHandler extends QueueHandler
      * @fixme doesn't currently report failure back to the queue manager
      * because omb_broadcast_notice() doesn't report it to us
      */
-    function handle_notice($notice)
+    function handle($notice)
     {
         if ($this->is_remote($notice)) {
             $this->log(LOG_DEBUG, 'Ignoring remote notice ' . $notice->id);
diff --git a/lib/pingqueuehandler.php b/lib/pingqueuehandler.php
index 8bb218078..4e4d74cb1 100644
--- a/lib/pingqueuehandler.php
+++ b/lib/pingqueuehandler.php
@@ -30,7 +30,7 @@ class PingQueueHandler extends QueueHandler {
         return 'ping';
     }
 
-    function handle_notice($notice) {
+    function handle($notice) {
         require_once INSTALLDIR . '/lib/ping.php';
         return ping_broadcast_notice($notice);
     }
diff --git a/lib/pluginqueuehandler.php b/lib/pluginqueuehandler.php
index 24d504699..9653ccad4 100644
--- a/lib/pluginqueuehandler.php
+++ b/lib/pluginqueuehandler.php
@@ -42,7 +42,7 @@ class PluginQueueHandler extends QueueHandler
         return 'plugin';
     }
 
-    function handle_notice($notice)
+    function handle($notice)
     {
         Event::handle('HandleQueuedNotice', array(&$notice));
         return true;
diff --git a/lib/publicqueuehandler.php b/lib/publicqueuehandler.php
index 9ea9ee73a..c9edb8d5d 100644
--- a/lib/publicqueuehandler.php
+++ b/lib/publicqueuehandler.php
@@ -23,7 +23,6 @@ if (!defined('STATUSNET') && !defined('LACONICA')) {
 
 /**
  * Queue handler for pushing new notices to public XMPP subscribers.
- * @fixme correct this exception handling
  */
 class PublicQueueHandler extends QueueHandler
 {
@@ -33,15 +32,14 @@ class PublicQueueHandler extends QueueHandler
         return 'public';
     }
 
-    function handle_notice($notice)
+    function handle($notice)
     {
         require_once(INSTALLDIR.'/lib/jabber.php');
         try {
             return jabber_public_notice($notice);
         } catch (XMPPHP_Exception $e) {
             $this->log(LOG_ERR, "Got an XMPPHP_Exception: " . $e->getMessage());
-            die($e->getMessage());
+            return false;
         }
-        return true;
     }
 }
diff --git a/lib/queued_xmpp.php b/lib/queued_xmpp.php
new file mode 100644
index 000000000..4b890c4ca
--- /dev/null
+++ b/lib/queued_xmpp.php
@@ -0,0 +1,117 @@
+<?php
+/**
+ * StatusNet, the distributed open-source microblogging tool
+ *
+ * Queue-mediated proxy class for outgoing XMPP messages.
+ *
+ * PHP version 5
+ *
+ * LICENCE: This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * @category  Network
+ * @package   StatusNet
+ * @author    Brion Vibber <brion@status.net>
+ * @copyright 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/
+ */
+
+if (!defined('STATUSNET') && !defined('LACONICA')) {
+    exit(1);
+}
+
+require_once INSTALLDIR . '/lib/jabber.php';
+
+class Queued_XMPP extends XMPPHP_XMPP
+{
+	/**
+	 * Constructor
+	 *
+	 * @param string  $host
+	 * @param integer $port
+	 * @param string  $user
+	 * @param string  $password
+	 * @param string  $resource
+	 * @param string  $server
+	 * @param boolean $printlog
+	 * @param string  $loglevel
+	 */
+	public function __construct($host, $port, $user, $password, $resource, $server = null, $printlog = false, $loglevel = null)
+	{
+		parent::__construct($host, $port, $user, $password, $resource, $server, $printlog, $loglevel);
+		// Normally the fulljid isn't filled out until resource binding time;
+		// we need to save it here since we're not talking to a real server.
+		$this->fulljid = "{$this->basejid}/{$this->resource}";
+    }
+
+    /**
+     * Send a formatted message to the outgoing queue for later forwarding
+     * to a real XMPP connection.
+     *
+     * @param string $msg
+     */
+    public function send($msg, $timeout=NULL)
+    {
+        $qm = QueueManager::get();
+        $qm->enqueue(strval($msg), 'xmppout');
+    }
+
+    /**
+     * Since we'll be getting input through a queue system's run loop,
+     * we'll process one standalone message at a time rather than our
+     * own XMPP message pump.
+     *
+     * @param string $message
+     */
+    public function processMessage($message) {
+       $frame = array_shift($this->frames);
+       xml_parse($this->parser, $frame->body, false);
+    }
+
+    //@{
+    /**
+     * Stream i/o functions disabled; push input through processMessage()
+     */
+    public function connect($timeout = 30, $persistent = false, $sendinit = true)
+    {
+        throw new Exception("Can't connect to server from XMPP queue proxy.");
+    }
+
+    public function disconnect()
+    {
+        throw new Exception("Can't connect to server from XMPP queue proxy.");
+    }
+
+    public function process()
+    {
+        throw new Exception("Can't read stream from XMPP queue proxy.");
+    }
+
+    public function processUntil($event, $timeout=-1)
+    {
+        throw new Exception("Can't read stream from XMPP queue proxy.");
+    }
+
+    public function read()
+    {
+        throw new Exception("Can't read stream from XMPP queue proxy.");
+    }
+
+    public function readyToProcess()
+    {
+        throw new Exception("Can't read stream from XMPP queue proxy.");
+    }
+    //@}
+}
+
diff --git a/lib/queuehandler.php b/lib/queuehandler.php
index 613be6e33..2909cd83b 100644
--- a/lib/queuehandler.php
+++ b/lib/queuehandler.php
@@ -22,51 +22,20 @@ if (!defined('STATUSNET') && !defined('LACONICA')) { exit(1); }
 /**
  * Base class for queue handlers.
  *
- * As extensions of the Daemon class, each queue handler has the ability
- * to launch itself in the background, at which point it'll pass control
- * to the configured QueueManager class to poll for updates.
+ * As of 0.9, queue handlers are short-lived for items as they are
+ * dequeued by a QueueManager running in an IoMaster in a daemon
+ * such as queuedaemon.php.
+ *
+ * Extensions requiring long-running maintenance or polling should
+ * register an IoManager.
  *
  * Subclasses must override at least the following methods:
  * - transport
- * - handle_notice
+ * - handle
  */
-#class QueueHandler extends Daemon
 class QueueHandler
 {
 
-#    function __construct($id=null, $daemonize=true)
-#    {
-#        parent::__construct($daemonize);
-#
-#        if ($id) {
-#            $this->set_id($id);
-#        }
-#    }
-
-    /**
-     * How many seconds a polling-based queue manager should wait between
-     * checks for new items to handle.
-     *
-     * Defaults to 60 seconds; override to speed up or slow down.
-     *
-     * @fixme not really compatible with global queue manager
-     * @return int timeout in seconds
-     */
-#    function timeout()
-#    {
-#        return 60;
-#    }
-
-#    function class_name()
-#    {
-#        return ucfirst($this->transport()) . 'Handler';
-#    }
-
-#    function name()
-#    {
-#        return strtolower($this->class_name().'.'.$this->get_id());
-#    }
-
     /**
      * Return transport keyword which identifies items this queue handler
      * services; must be defined for all subclasses.
@@ -83,61 +52,17 @@ class QueueHandler
 
     /**
      * Here's the meat of your queue handler -- you're handed a Notice
-     * object, which you may do as you will with.
+     * or other object, which you may do as you will with.
      *
      * If this function indicates failure, a warning will be logged
      * and the item is placed back in the queue to be re-run.
      *
-     * @param Notice $notice
-     * @return boolean true on success, false on failure
-     */
-    function handle_notice($notice)
-    {
-        return true;
-    }
-
-    /**
-     * Setup and start of run loop for this queue handler as a daemon.
-     * Most of the heavy lifting is passed on to the QueueManager's service()
-     * method, which passes control back to our handle_notice() method for
-     * each notice that comes in on the queue.
-     *
-     * Most of the time this won't need to be overridden in a subclass.
-     *
+     * @param mixed $object
      * @return boolean true on success, false on failure
      */
-    function run()
+    function handle($object)
     {
-        if (!$this->start()) {
-            $this->log(LOG_WARNING, 'failed to start');
-            return false;
-        }
-
-        $this->log(LOG_INFO, 'checking for queued notices');
-
-        $queue   = $this->transport();
-        $timeout = $this->timeout();
-
-        $qm = QueueManager::get();
-
-        $qm->service($queue, $this);
-
-        $this->log(LOG_INFO, 'finished servicing the queue');
-
-        if (!$this->finish()) {
-            $this->log(LOG_WARNING, 'failed to clean up');
-            return false;
-        }
-
-        $this->log(LOG_INFO, 'terminating normally');
-
         return true;
     }
-
-
-    function log($level, $msg)
-    {
-        common_log($level, $this->class_name() . ' ('. $this->get_id() .'): '.$msg);
-    }
 }
 
diff --git a/lib/queuemanager.php b/lib/queuemanager.php
index 291174d3c..4eb39bfa8 100644
--- 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
@@ -109,6 +113,64 @@ 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 for queued storage.
+     * Next gen may use serialization.
+     *
+     * @param mixed $object
+     * @return string
+     */
+    protected function encode($object)
+    {
+        if ($object instanceof Notice) {
+            return $object->id;
+        } else if (is_string($object)) {
+            return $object;
+        } else {
+            throw new ServerException("Can't queue this type", 500);
+        }
+    }
+
+    /**
+     * Decode an object from queued storage.
+     * Accepts back-compat notice reference entries and strings for now.
+     *
+     * @param string
+     * @return mixed
+     */
+    protected function decode($frame)
+    {
+        if (is_numeric($frame)) {
+            return Notice::staticGet(intval($frame));
+        } else {
+            return $frame;
+        }
+    }
+
     /**
      * Instantiate the appropriate QueueHandler class for the given queue.
      *
@@ -131,13 +193,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,33 +212,29 @@ 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))) {
-            if (!defined('XMPP_ONLY_FLAG')) { // hack!
-                $this->connect('plugin', 'PluginQueueHandler');
-                $this->connect('omb', 'OmbQueueHandler');
-                $this->connect('ping', 'PingQueueHandler');
-                if (common_config('sms', 'enabled')) {
-                    $this->connect('sms', 'SmsQueueHandler');
-                }
+            $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') && !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');
 
-            if (!defined('XMPP_ONLY_FLAG')) { // hack!
-                // For compat with old plugins not registering their own handlers.
-                $this->connect('plugin', 'PluginQueueHandler');
-            }
-        }
-        if (!defined('XMPP_ONLY_FLAG')) { // hack!
-            Event::handle('EndInitializeQueueManager', array($this));
         }
+        Event::handle('EndInitializeQueueManager', array($this));
     }
 
     /**
@@ -183,10 +243,27 @@ abstract class QueueManager extends IoManager
      *
      * @param string $transport
      * @param string $class
+     * @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;
     }
 
     /**
diff --git a/lib/smsqueuehandler.php b/lib/smsqueuehandler.php
index 48a96409d..6085d2b4a 100644
--- a/lib/smsqueuehandler.php
+++ b/lib/smsqueuehandler.php
@@ -31,7 +31,7 @@ class SmsQueueHandler extends QueueHandler
         return 'sms';
     }
 
-    function handle_notice($notice)
+    function handle($notice)
     {
     	require_once(INSTALLDIR.'/lib/mail.php');
         return mail_broadcast_notice_sms($notice);
diff --git a/lib/spawningdaemon.php b/lib/spawningdaemon.php
new file mode 100644
index 000000000..8baefe88e
--- /dev/null
+++ b/lib/spawningdaemon.php
@@ -0,0 +1,159 @@
+<?php
+/*
+ * StatusNet - the distributed open-source microblogging tool
+ * Copyright (C) 2010, StatusNet, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+/**
+ * Base class for daemon that can launch one or more processing threads,
+ * respawning them if they exit.
+ *
+ * This is mainly intended for indefinite workloads such as monitoring
+ * a queue or maintaining an IM channel.
+ *
+ * Child classes should implement the 
+ *
+ * We can then pass individual items through the QueueHandler subclasses
+ * they belong to. We additionally can handle queues for multiple sites.
+ *
+ * @package QueueHandler
+ * @author Brion Vibber <brion@status.net>
+ */
+abstract class SpawningDaemon extends Daemon
+{
+    protected $threads=1;
+
+    function __construct($id=null, $daemonize=true, $threads=1)
+    {
+        parent::__construct($daemonize);
+
+        if ($id) {
+            $this->set_id($id);
+        }
+        $this->threads = $threads;
+    }
+
+    /**
+     * Perform some actual work!
+     *
+     * @return boolean true on success, false on failure
+     */
+    public abstract function runThread();
+
+    /**
+     * Spawn one or more background processes and let them start running.
+     * Each individual process will execute whatever's in the runThread()
+     * method, which should be overridden.
+     *
+     * Child processes will be automatically respawned when they exit.
+     *
+     * @todo possibly allow for not respawning on "normal" exits...
+     *       though ParallelizingDaemon is probably better for workloads
+     *       that have forseeable endpoints.
+     */
+    function run()
+    {
+        $children = array();
+        for ($i = 1; $i <= $this->threads; $i++) {
+            $pid = pcntl_fork();
+            if ($pid < 0) {
+                $this->log(LOG_ERROR, "Couldn't fork for thread $i; aborting\n");
+                exit(1);
+            } else if ($pid == 0) {
+                $this->initAndRunChild($i);
+            } else {
+                $this->log(LOG_INFO, "Spawned thread $i as pid $pid");
+                $children[$i] = $pid;
+            }
+        }
+        
+        $this->log(LOG_INFO, "Waiting for children to complete.");
+        while (count($children) > 0) {
+            $status = null;
+            $pid = pcntl_wait($status);
+            if ($pid > 0) {
+                $i = array_search($pid, $children);
+                if ($i === false) {
+                    $this->log(LOG_ERR, "Unrecognized child pid $pid exited!");
+                    continue;
+                }
+                unset($children[$i]);
+                $this->log(LOG_INFO, "Thread $i pid $pid exited.");
+                
+                $pid = pcntl_fork();
+                if ($pid < 0) {
+                    $this->log(LOG_ERROR, "Couldn't fork to respawn thread $i; aborting thread.\n");
+                } else if ($pid == 0) {
+                    $this->initAndRunChild($i);
+                } else {
+                    $this->log(LOG_INFO, "Respawned thread $i as pid $pid");
+                    $children[$i] = $pid;
+                }
+            }
+        }
+        $this->log(LOG_INFO, "All child processes complete.");
+        return true;
+    }
+
+    /**
+     * Initialize things for a fresh thread, call runThread(), and
+     * exit at completion with appropriate return value.
+     */
+    protected function initAndRunChild($thread)
+    {
+        $this->set_id($this->get_id() . "." . $thread);
+        $this->resetDb();
+        $ok = $this->runThread();
+        exit($ok ? 0 : 1);
+    }
+
+    /**
+     * Reconnect to the database for each child process,
+     * or they'll get very confused trying to use the
+     * same socket.
+     */
+    protected function resetDb()
+    {
+        // @fixme do we need to explicitly open the db too
+        // or is this implied?
+        global $_DB_DATAOBJECT;
+        unset($_DB_DATAOBJECT['CONNECTIONS']);
+
+        // Reconnect main memcached, or threads will stomp on
+        // each other and corrupt their requests.
+        $cache = common_memcache();
+        if ($cache) {
+            $cache->reconnect();
+        }
+
+        // Also reconnect memcached for status_network table.
+        if (!empty(Status_network::$cache)) {
+            Status_network::$cache->close();
+            Status_network::$cache = null;
+        }
+    }
+
+    function log($level, $msg)
+    {
+        common_log($level, get_class($this) . ' ('. $this->get_id() .'): '.$msg);
+    }
+
+    function name()
+    {
+        return strtolower(get_class($this).'.'.$this->get_id());
+    }
+}
+
diff --git a/lib/stompqueuemanager.php b/lib/stompqueuemanager.php
index 00590fdb6..f057bd9e4 100644
--- a/lib/stompqueuemanager.php
+++ b/lib/stompqueuemanager.php
@@ -39,7 +39,6 @@ class StompQueueManager extends QueueManager
     var $base = null;
     var $con = null;
     
-    protected $master = null;
     protected $sites = array();
 
     function __construct()
@@ -104,11 +103,12 @@ class StompQueueManager extends QueueManager
      */
     function getQueues()
     {
+        $group = $this->activeGroup();
         $site = common_config('site', 'server');
-        if (empty($this->handlers[$site])) {
+        if (empty($this->groups[$site][$group])) {
             return array();
         } else {
-            return array_keys($this->handlers[$site]);
+            return array_keys($this->groups[$site][$group]);
         }
     }
 
@@ -118,10 +118,12 @@ class StompQueueManager extends QueueManager
      *
      * @param string $transport
      * @param string $class
+     * @param string $group
      */
-    public function connect($transport, $class)
+    public function connect($transport, $class, $group='queuedaemon')
     {
         $this->handlers[common_config('site', 'server')][$transport] = $class;
+        $this->groups[common_config('site', 'server')][$group][$transport] = $class;
     }
 
     /**
@@ -130,23 +132,23 @@ class StompQueueManager extends QueueManager
      */
     public function enqueue($object, $queue)
     {
-        $notice = $object;
+        $msg = $this->encode($object);
+        $rep = $this->logrep($object);
 
         $this->_connect();
 
         // XXX: serialize and send entire notice
 
         $result = $this->con->send($this->queueName($queue),
-                                   $notice->id, 		// BODY of the message
-                                   array ('created' => $notice->created));
+                                   $msg, 		// BODY of the message
+                                   array ('created' => common_sql_now()));
 
         if (!$result) {
-            common_log(LOG_ERR, 'Error sending to '.$queue.' queue');
+            common_log(LOG_ERR, "Error sending $rep to $queue queue");
             return false;
         }
 
-        common_log(LOG_DEBUG, 'complete remote queueing notice ID = '
-                   . $notice->id . ' for ' . $queue);
+        common_log(LOG_DEBUG, "complete remote queueing $rep for $queue");
         $this->stats('enqueued', $queue);
     }
 
@@ -174,7 +176,7 @@ class StompQueueManager extends QueueManager
         $ok = true;
         $frames = $this->con->readFrames();
         foreach ($frames as $frame) {
-            $ok = $ok && $this->_handleNotice($frame);
+            $ok = $ok && $this->_handleItem($frame);
         }
         return $ok;
     }
@@ -265,7 +267,7 @@ class StompQueueManager extends QueueManager
     }
 
     /**
-     * Handle and acknowledge a notice event that's come in through a queue.
+     * Handle and acknowledge an event that's come in through a queue.
      *
      * If the queue handler reports failure, the message is requeued for later.
      * Missing notices or handler classes will drop the message.
@@ -276,7 +278,7 @@ class StompQueueManager extends QueueManager
      * @param StompFrame $frame
      * @return bool
      */
-    protected function _handleNotice($frame)
+    protected function _handleItem($frame)
     {
         list($site, $queue) = $this->parseDestination($frame->headers['destination']);
         if ($site != common_config('site', 'server')) {
@@ -284,15 +286,23 @@ class StompQueueManager extends QueueManager
             StatusNet::init($site);
         }
 
-        $id = intval($frame->body);
-        $info = "notice $id posted at {$frame->headers['created']} in queue $queue";
+        if (is_numeric($frame->body)) {
+            $id = intval($frame->body);
+            $info = "notice $id posted at {$frame->headers['created']} in queue $queue";
 
-        $notice = Notice::staticGet('id', $id);
-        if (empty($notice)) {
-            $this->_log(LOG_WARNING, "Skipping missing $info");
-            $this->con->ack($frame);
-            $this->stats('badnotice', $queue);
-            return false;
+            $notice = Notice::staticGet('id', $id);
+            if (empty($notice)) {
+                $this->_log(LOG_WARNING, "Skipping missing $info");
+                $this->con->ack($frame);
+                $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;
         }
 
         $handler = $this->getHandler($queue);
@@ -303,7 +313,7 @@ class StompQueueManager extends QueueManager
             return false;
         }
 
-        $ok = $handler->handle_notice($notice);
+        $ok = $handler->handle($item);
 
         if (!$ok) {
             $this->_log(LOG_WARNING, "Failed handling $info");
@@ -311,7 +321,7 @@ class StompQueueManager extends QueueManager
             // this kind of queue management ourselves;
             // if we don't ack, it should resend...
             $this->con->ack($frame);
-            $this->enqueue($notice, $queue);
+            $this->enqueue($item, $queue);
             $this->stats('requeued', $queue);
             return false;
         }
diff --git a/lib/util.php b/lib/util.php
index ef8a5d1f0..fb3b8be87 100644
--- a/lib/util.php
+++ b/lib/util.php
@@ -1130,7 +1130,8 @@ function common_request_id()
     $pid = getmypid();
     $server = common_config('site', 'server');
     if (php_sapi_name() == 'cli') {
-        return "$server:$pid";
+        $script = basename($_SERVER['PHP_SELF']);
+        return "$server:$script:$pid";
     } else {
         static $req_id = null;
         if (!isset($req_id)) {
diff --git a/lib/xmppconfirmmanager.php b/lib/xmppconfirmmanager.php
deleted file mode 100644
index ee4e294fd..000000000
--- a/lib/xmppconfirmmanager.php
+++ /dev/null
@@ -1,168 +0,0 @@
-<?php
-/*
- * StatusNet - the distributed open-source microblogging tool
- * Copyright (C) 2008-2010 StatusNet, Inc.
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU Affero General Public License as published by
- * the Free Software Foundation, either version 3 of the License, or
- * (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- * GNU Affero General Public License for more details.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program.  If not, see <http://www.gnu.org/licenses/>.
- */
-
-if (!defined('STATUSNET') && !defined('LACONICA')) {
-    exit(1);
-}
-
-/**
- * Event handler for pushing new confirmations to Jabber users.
- * @fixme recommend redoing this on a queue-trigger model
- * @fixme expiration of old items got dropped in the past, put it back?
- */
-class XmppConfirmManager extends IoManager
-{
-
-    /**
-     * @return mixed XmppConfirmManager, or false if unneeded
-     */
-    public static function get()
-    {
-        if (common_config('xmpp', 'enabled')) {
-            $site = common_config('site', 'server');
-            return new XmppConfirmManager();
-        } else {
-            return false;
-        }
-    }
-
-    /**
-     * Tell the i/o master we need one instance for each supporting site
-     * being handled in this process.
-     */
-    public static function multiSite()
-    {
-        return IoManager::INSTANCE_PER_SITE;
-    }
-
-    function __construct()
-    {
-        $this->site = common_config('site', 'server');
-    }
-
-    /**
-     * 10 seconds? Really? That seems a bit frequent.
-     */
-    function pollInterval()
-    {
-        return 10;
-    }
-
-    /**
-     * Ping!
-     * @return boolean true if we found something
-     */
-    function poll()
-    {
-        $this->switchSite();
-        $confirm = $this->next_confirm();
-        if ($confirm) {
-            $this->handle_confirm($confirm);
-            return true;
-        } else {
-            return false;
-        }
-    }
-
-    protected function handle_confirm($confirm)
-    {
-        require_once INSTALLDIR . '/lib/jabber.php';
-
-        common_log(LOG_INFO, 'Sending confirmation for ' . $confirm->address);
-        $user = User::staticGet($confirm->user_id);
-        if (!$user) {
-            common_log(LOG_WARNING, 'Confirmation for unknown user ' . $confirm->user_id);
-            return;
-        }
-        $success = jabber_confirm_address($confirm->code,
-                                          $user->nickname,
-                                          $confirm->address);
-        if (!$success) {
-            common_log(LOG_ERR, 'Confirmation failed for ' . $confirm->address);
-            # Just let the claim age out; hopefully things work then
-            return;
-        } else {
-            common_log(LOG_INFO, 'Confirmation sent for ' . $confirm->address);
-            # Mark confirmation sent; need a dupe so we don't have the WHERE clause
-            $dupe = Confirm_address::staticGet('code', $confirm->code);
-            if (!$dupe) {
-                common_log(LOG_WARNING, 'Could not refetch confirm', __FILE__);
-                return;
-            }
-            $orig = clone($dupe);
-            $dupe->sent = $dupe->claimed;
-            $result = $dupe->update($orig);
-            if (!$result) {
-                common_log_db_error($dupe, 'UPDATE', __FILE__);
-                # Just let the claim age out; hopefully things work then
-                return;
-            }
-        }
-        return true;
-    }
-
-    protected function next_confirm()
-    {
-        $confirm = new Confirm_address();
-        $confirm->whereAdd('claimed IS null');
-        $confirm->whereAdd('sent IS null');
-        # XXX: eventually we could do other confirmations in the queue, too
-        $confirm->address_type = 'jabber';
-        $confirm->orderBy('modified DESC');
-        $confirm->limit(1);
-        if ($confirm->find(true)) {
-            common_log(LOG_INFO, 'Claiming confirmation for ' . $confirm->address);
-                # working around some weird DB_DataObject behaviour
-            $confirm->whereAdd(''); # clears where stuff
-            $original = clone($confirm);
-            $confirm->claimed = common_sql_now();
-            $result = $confirm->update($original);
-            if ($result) {
-                common_log(LOG_INFO, 'Succeeded in claim! '. $result);
-                return $confirm;
-            } else {
-                common_log(LOG_INFO, 'Failed in claim!');
-                return false;
-            }
-        }
-        return null;
-    }
-
-    protected function clear_old_confirm_claims()
-    {
-        $confirm = new Confirm();
-        $confirm->claimed = null;
-        $confirm->whereAdd('now() - claimed > '.CLAIM_TIMEOUT);
-        $confirm->update(DB_DATAOBJECT_WHEREADD_ONLY);
-        $confirm->free();
-        unset($confirm);
-    }
-
-    /**
-     * Make sure we're on the right site configuration
-     */
-    protected function switchSite()
-    {
-        if ($this->site != common_config('site', 'server')) {
-            common_log(LOG_DEBUG, __METHOD__ . ": switching to site $this->site");
-            $this->stats('switch');
-            StatusNet::init($this->site);
-        }
-    }
-}
diff --git a/lib/xmppmanager.php b/lib/xmppmanager.php
index dfff63a30..299175dd7 100644
--- a/lib/xmppmanager.php
+++ b/lib/xmppmanager.php
@@ -70,6 +70,7 @@ class XmppManager extends IoManager
     function __construct()
     {
         $this->site = common_config('site', 'server');
+        $this->resource = common_config('xmpp', 'resource') . 'daemon';
     }
 
     /**
@@ -86,15 +87,19 @@ class XmppManager extends IoManager
         # Low priority; we don't want to receive messages
 
         common_log(LOG_INFO, "INITIALIZE");
-        $this->conn = jabber_connect($this->resource());
+        $this->conn = jabber_connect($this->resource);
 
         if (empty($this->conn)) {
             common_log(LOG_ERR, "Couldn't connect to server.");
             return false;
         }
 
-        $this->conn->addEventHandler('message', 'forward_message', $this);
+        $this->log(LOG_DEBUG, "Initializing stanza handlers.");
+
+        $this->conn->addEventHandler('message', 'handle_message', $this);
+        $this->conn->addEventHandler('presence', 'handle_presence', $this);
         $this->conn->addEventHandler('reconnect', 'handle_reconnect', $this);
+
         $this->conn->setReconnectTimeout(600);
         jabber_send_presence("Send me a message to post a notice", 'available', null, 'available', -1);
 
@@ -175,12 +180,37 @@ class XmppManager extends IoManager
         }
     }
 
+    /**
+     * For queue handlers to pass us a message to push out,
+     * if we're active.
+     *
+     * @fixme should this be blocking etc?
+     *
+     * @param string $msg XML stanza to send
+     * @return boolean success
+     */
+    public function send($msg)
+    {
+        if ($this->conn && !$this->conn->isDisconnected()) {
+            $bytes = $this->conn->send($msg);
+            if ($bytes > 0) {
+                $this->conn->processTime(0);
+                return true;
+            } else {
+                return false;
+            }
+        } else {
+            // Can't send right now...
+            return false;
+        }
+    }
+
     /**
      * Send a keepalive ping to the XMPP server.
      */
     protected function sendPing()
     {
-        $jid = jabber_daemon_address().'/'.$this->resource();
+        $jid = jabber_daemon_address().'/'.$this->resource;
         $server = common_config('xmpp', 'server');
 
         if (!isset($this->pingid)) {
@@ -206,61 +236,239 @@ class XmppManager extends IoManager
         $this->conn->presence(null, 'available', null, 'available', -1);
     }
 
+
+    function get_user($from)
+    {
+        $user = User::staticGet('jabber', jabber_normalize_jid($from));
+        return $user;
+    }
+
     /**
-     * Callback for Jabber message event.
-     *
-     * This connection handles output; if we get a message straight to us,
-     * forward it on to our XmppDaemon listener for processing.
-     *
-     * @param $pl
+     * XMPP callback for handling message input...
+     * @param array $pl XMPP payload
      */
-    function forward_message(&$pl)
+    function handle_message(&$pl)
     {
+        $from = jabber_normalize_jid($pl['from']);
+
         if ($pl['type'] != 'chat') {
-            common_log(LOG_DEBUG, 'Ignoring message of type ' . $pl['type'] . ' from ' . $pl['from']);
+            $this->log(LOG_WARNING, "Ignoring message of type ".$pl['type']." from $from.");
             return;
         }
-        $listener = $this->listener();
-        if (strtolower($listener) == strtolower($pl['from'])) {
-            common_log(LOG_WARNING, 'Ignoring loop message.');
+
+        if (mb_strlen($pl['body']) == 0) {
+            $this->log(LOG_WARNING, "Ignoring message with empty body from $from.");
             return;
         }
-        common_log(LOG_INFO, 'Forwarding message from ' . $pl['from'] . ' to ' . $listener);
-        $this->conn->message($this->listener(), $pl['body'], 'chat', null, $this->ofrom($pl['from']));
+
+        // Forwarded from another daemon for us to handle; this shouldn't
+        // happen any more but we might get some legacy items.
+        if ($this->is_self($from)) {
+            $this->log(LOG_INFO, "Got forwarded notice from self ($from).");
+            $from = $this->get_ofrom($pl);
+            $this->log(LOG_INFO, "Originally sent by $from.");
+            if (is_null($from) || $this->is_self($from)) {
+                $this->log(LOG_INFO, "Ignoring notice originally sent by $from.");
+                return;
+            }
+        }
+
+        $user = $this->get_user($from);
+
+        // For common_current_user to work
+        global $_cur;
+        $_cur = $user;
+
+        if (!$user) {
+            $this->from_site($from, 'Unknown user; go to ' .
+                             common_local_url('imsettings') .
+                             ' to add your address to your account');
+            $this->log(LOG_WARNING, 'Message from unknown user ' . $from);
+            return;
+        }
+        if ($this->handle_command($user, $pl['body'])) {
+            $this->log(LOG_INFO, "Command message by $from handled.");
+            return;
+        } else if ($this->is_autoreply($pl['body'])) {
+            $this->log(LOG_INFO, 'Ignoring auto reply from ' . $from);
+            return;
+        } else if ($this->is_otr($pl['body'])) {
+            $this->log(LOG_INFO, 'Ignoring OTR from ' . $from);
+            return;
+        } else {
+
+            $this->log(LOG_INFO, 'Posting a notice from ' . $user->nickname);
+
+            $this->add_notice($user, $pl);
+        }
+
+        $user->free();
+        unset($user);
+        unset($_cur);
+
+        unset($pl['xml']);
+        $pl['xml'] = null;
+
+        $pl = null;
+        unset($pl);
     }
 
-    /**
-     * Build an <addresses> block with an ofrom entry for forwarded messages
-     *
-     * @param string $from Jabber ID of original sender
-     * @return string XML fragment
-     */
-    protected function ofrom($from)
+
+    function is_self($from)
     {
-        $address = "<addresses xmlns='http://jabber.org/protocol/address'>\n";
-        $address .= "<address type='ofrom' jid='$from' />\n";
-        $address .= "</addresses>\n";
-        return $address;
+        return preg_match('/^'.strtolower(jabber_daemon_address()).'/', strtolower($from));
     }
 
-    /**
-     * Build the complete JID of the XmppDaemon process which
-     * handles primary XMPP input for this site.
-     *
-     * @return string Jabber ID
-     */
-    protected function listener()
+    function get_ofrom($pl)
+    {
+        $xml = $pl['xml'];
+        $addresses = $xml->sub('addresses');
+        if (!$addresses) {
+            $this->log(LOG_WARNING, 'Forwarded message without addresses');
+            return null;
+        }
+        $address = $addresses->sub('address');
+        if (!$address) {
+            $this->log(LOG_WARNING, 'Forwarded message without address');
+            return null;
+        }
+        if (!array_key_exists('type', $address->attrs)) {
+            $this->log(LOG_WARNING, 'No type for forwarded message');
+            return null;
+        }
+        $type = $address->attrs['type'];
+        if ($type != 'ofrom') {
+            $this->log(LOG_WARNING, 'Type of forwarded message is not ofrom');
+            return null;
+        }
+        if (!array_key_exists('jid', $address->attrs)) {
+            $this->log(LOG_WARNING, 'No jid for forwarded message');
+            return null;
+        }
+        $jid = $address->attrs['jid'];
+        if (!$jid) {
+            $this->log(LOG_WARNING, 'Could not get jid from address');
+            return null;
+        }
+        $this->log(LOG_DEBUG, 'Got message forwarded from jid ' . $jid);
+        return $jid;
+    }
+
+    function is_autoreply($txt)
+    {
+        if (preg_match('/[\[\(]?[Aa]uto[-\s]?[Rr]e(ply|sponse)[\]\)]/', $txt)) {
+            return true;
+        } else if (preg_match('/^System: Message wasn\'t delivered. Offline storage size was exceeded.$/', $txt)) {
+            return true;
+        } else {
+            return false;
+        }
+    }
+
+    function is_otr($txt)
+    {
+        if (preg_match('/^\?OTR/', $txt)) {
+            return true;
+        } else {
+            return false;
+        }
+    }
+
+    function from_site($address, $msg)
+    {
+        $text = '['.common_config('site', 'name') . '] ' . $msg;
+        jabber_send_message($address, $text);
+    }
+
+    function handle_command($user, $body)
     {
-        if (common_config('xmpp', 'listener')) {
-            return common_config('xmpp', 'listener');
+        $inter = new CommandInterpreter();
+        $cmd = $inter->handle_command($user, $body);
+        if ($cmd) {
+            $chan = new XMPPChannel($this->conn);
+            $cmd->execute($chan);
+            return true;
         } else {
-            return jabber_daemon_address() . '/' . common_config('xmpp','resource') . 'daemon';
+            return false;
+        }
+    }
+
+    function add_notice(&$user, &$pl)
+    {
+        $body = trim($pl['body']);
+        $content_shortened = common_shorten_links($body);
+        if (Notice::contentTooLong($content_shortened)) {
+          $from = jabber_normalize_jid($pl['from']);
+          $this->from_site($from, sprintf(_('Message too long - maximum is %1$d characters, you sent %2$d.'),
+                                          Notice::maxContent(),
+                                          mb_strlen($content_shortened)));
+          return;
+        }
+
+        try {
+            $notice = Notice::saveNew($user->id, $content_shortened, 'xmpp');
+        } catch (Exception $e) {
+            $this->log(LOG_ERR, $e->getMessage());
+            $this->from_site($user->jabber, $e->getMessage());
+            return;
         }
+
+        common_broadcast_notice($notice);
+        $this->log(LOG_INFO,
+                   'Added notice ' . $notice->id . ' from user ' . $user->nickname);
+        $notice->free();
+        unset($notice);
+    }
+
+    function handle_presence(&$pl)
+    {
+        $from = jabber_normalize_jid($pl['from']);
+        switch ($pl['type']) {
+         case 'subscribe':
+            # We let anyone subscribe
+            $this->subscribed($from);
+            $this->log(LOG_INFO,
+                       'Accepted subscription from ' . $from);
+            break;
+         case 'subscribed':
+         case 'unsubscribed':
+         case 'unsubscribe':
+            $this->log(LOG_INFO,
+                       'Ignoring  "' . $pl['type'] . '" from ' . $from);
+            break;
+         default:
+            if (!$pl['type']) {
+                $user = User::staticGet('jabber', $from);
+                if (!$user) {
+                    $this->log(LOG_WARNING, 'Presence from unknown user ' . $from);
+                    return;
+                }
+                if ($user->updatefrompresence) {
+                    $this->log(LOG_INFO, 'Updating ' . $user->nickname .
+                               ' status from presence.');
+                    $this->add_notice($user, $pl);
+                }
+                $user->free();
+                unset($user);
+            }
+            break;
+        }
+        unset($pl['xml']);
+        $pl['xml'] = null;
+
+        $pl = null;
+        unset($pl);
+    }
+
+    function log($level, $msg)
+    {
+        $text = 'XMPPDaemon('.$this->resource.'): '.$msg;
+        common_log($level, $text);
     }
 
-    protected function resource()
+    function subscribed($to)
     {
-        return 'queue' . posix_getpid(); // @fixme PIDs won't be host-unique
+        jabber_special_presence('subscribed', $to);
     }
 
     /**
diff --git a/lib/xmppoutqueuehandler.php b/lib/xmppoutqueuehandler.php
new file mode 100644
index 000000000..2afa260f1
--- /dev/null
+++ b/lib/xmppoutqueuehandler.php
@@ -0,0 +1,55 @@
+<?php
+/*
+ * StatusNet - the distributed open-source microblogging tool
+ * Copyright (C) 2010, StatusNet, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+/**
+ * Queue handler for pre-processed outgoing XMPP messages.
+ * Formatted XML stanzas will have been pushed into the queue
+ * via the Queued_XMPP connection proxy, probably from some
+ * other queue processor.
+ *
+ * Here, the XML stanzas are simply pulled out of the queue and
+ * pushed out over the wire; an XmppManager is needed to set up
+ * and maintain the actual server connection.
+ *
+ * This queue will be run via XmppDaemon rather than QueueDaemon.
+ *
+ * @author Brion Vibber <brion@status.net>
+ */
+class XmppOutQueueHandler extends QueueHandler
+{
+    function transport() {
+        return 'xmppout';
+    }
+
+    /**
+     * Take a previously-queued XMPP stanza and send it out ot the server.
+     * @param string $msg
+     * @return boolean true on success
+     */
+    function handle($msg)
+    {
+        assert(is_string($msg));
+
+        $xmpp = XmppManager::get();
+        $ok = $xmpp->send($msg);
+
+        return $ok;
+    }
+}
+
-- 
cgit v1.2.3-54-g00ecf


From 6e4cad71e546505b4bb4a168c1b11b84a45b7889 Mon Sep 17 00:00:00 2001
From: Brion Vibber <brion@pobox.com>
Date: Fri, 22 Jan 2010 12:35:05 -0800
Subject: Fix for stuck queue messages: wrap processing in stomp transactions
 so our lack of an ACK if PHP dies actually triggers redelivery.

Previously, messages once delivered would just get stuck in the queue seemingly forever if they never got ACKed.
Note this could lead to partial duplication, for instance if the OMB or Twitter queue handlers die after 1/2 of the outgoing sends.

Recommendations:
* catch exceptions more aggressively within queue handlers (so only PHP fatal errors are likely to kill in the middle)
* for processing that involves sending to multiple clients, consider a second queue similar to the XMPP output, eg for OMB:
 - first queue gets delivery list and builds message data, enqueueing it for each target address
 - second queue can handle each individual outgoing message (and attempt redelivery etc separately)

This would also protect better against a recurring error preventing delivery in the second part, and could spread out any slow sends over multiple threads.
---
 lib/stompqueuemanager.php | 68 ++++++++++++++++++++++++++++++++++++++++++++---
 1 file changed, 64 insertions(+), 4 deletions(-)

(limited to 'lib')

diff --git a/lib/stompqueuemanager.php b/lib/stompqueuemanager.php
index f057bd9e4..8f0091a13 100644
--- a/lib/stompqueuemanager.php
+++ b/lib/stompqueuemanager.php
@@ -41,6 +41,10 @@ class StompQueueManager extends QueueManager
     
     protected $sites = array();
 
+    protected $useTransactions = true;
+    protected $transaction = null;
+    protected $transactionCount = 0;
+
     function __construct()
     {
         parent::__construct();
@@ -201,6 +205,7 @@ class StompQueueManager extends QueueManager
         } else {
             $this->doSubscribe();
         }
+        $this->begin();
         return true;
     }
     
@@ -213,6 +218,9 @@ class StompQueueManager extends QueueManager
      */
     public function finish()
     {
+        // If there are any outstanding delivered messages we haven't processed,
+        // free them for another thread to take.
+        $this->rollback();
         if ($this->sites) {
             foreach ($this->sites as $server) {
                 StatusNet::init($server);
@@ -293,7 +301,9 @@ class StompQueueManager extends QueueManager
             $notice = Notice::staticGet('id', $id);
             if (empty($notice)) {
                 $this->_log(LOG_WARNING, "Skipping missing $info");
-                $this->con->ack($frame);
+                $this->ack($frame);
+                $this->commit();
+                $this->begin();
                 $this->stats('badnotice', $queue);
                 return false;
             }
@@ -308,7 +318,9 @@ class StompQueueManager extends QueueManager
         $handler = $this->getHandler($queue);
         if (!$handler) {
             $this->_log(LOG_ERROR, "Missing handler class; skipping $info");
-            $this->con->ack($frame);
+            $this->ack($frame);
+            $this->commit();
+            $this->begin();
             $this->stats('badhandler', $queue);
             return false;
         }
@@ -320,14 +332,18 @@ class StompQueueManager extends QueueManager
             // FIXME we probably shouldn't have to do
             // this kind of queue management ourselves;
             // if we don't ack, it should resend...
-            $this->con->ack($frame);
+            $this->ack($frame);
             $this->enqueue($item, $queue);
+            $this->commit();
+            $this->begin();
             $this->stats('requeued', $queue);
             return false;
         }
 
         $this->_log(LOG_INFO, "Successfully handled $info");
-        $this->con->ack($frame);
+        $this->ack($frame);
+        $this->commit();
+        $this->begin();
         $this->stats('handled', $queue);
         return true;
     }
@@ -369,5 +385,49 @@ class StompQueueManager extends QueueManager
     {
         common_log($level, 'StompQueueManager: '.$msg);
     }
+
+    protected function begin()
+    {
+        if ($this->useTransactions) {
+            if ($this->transaction) {
+                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);
+        }
+    }
+
+    protected function ack($frame)
+    {
+        if ($this->useTransactions) {
+            if (!$this->transaction) {
+                throw new Exception("Tried to ack but not in a transaction");
+            }
+        }
+        $this->con->ack($frame, $this->transaction);
+    }
+
+    protected function commit()
+    {
+        if ($this->useTransactions) {
+            if (!$this->transaction) {
+                throw new Exception("Tried to commit but not in a transaction");
+            }
+            $this->con->commit($this->transaction);
+            $this->transaction = null;
+        }
+    }
+
+    protected function rollback()
+    {
+        if ($this->useTransactions) {
+            if (!$this->transaction) {
+                throw new Exception("Tried to rollback but not in a transaction");
+            }
+            $this->con->commit($this->transaction);
+            $this->transaction = null;
+        }
+    }
 }
 
-- 
cgit v1.2.3-54-g00ecf


From a66181c900119bb0efb63939f3a0b3d50de72b8d Mon Sep 17 00:00:00 2001
From: Brion Vibber <brion@pobox.com>
Date: Fri, 22 Jan 2010 13:49:05 -0800
Subject: Fix unqueuemanager for updated QueueHandler interface

---
 lib/unqueuemanager.php | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

(limited to 'lib')

diff --git a/lib/unqueuemanager.php b/lib/unqueuemanager.php
index 5595eac05..785de7c8c 100644
--- a/lib/unqueuemanager.php
+++ b/lib/unqueuemanager.php
@@ -47,7 +47,7 @@ class UnQueueManager extends QueueManager
         
         $handler = $this->getHandler($queue);
         if ($handler) {
-            $handler->handle_notice($notice);
+            $handler->handle($notice);
         } else {
             if (Event::handle('UnqueueHandleNotice', array(&$notice, $queue))) {
                 throw new ServerException("UnQueueManager: Unknown queue: $queue");
-- 
cgit v1.2.3-54-g00ecf


From 49d500e3e18a1520ca9e34d76d2e421b57afcb07 Mon Sep 17 00:00:00 2001
From: Evan Prodromou <evan@status.net>
Date: Sun, 24 Jan 2010 18:19:13 -0500
Subject: defaults for nickname and wildcard

---
 lib/default.php | 2 ++
 1 file changed, 2 insertions(+)

(limited to 'lib')

diff --git a/lib/default.php b/lib/default.php
index 790a5b387..57199b356 100644
--- a/lib/default.php
+++ b/lib/default.php
@@ -30,6 +30,8 @@
 $default =
   array('site' =>
         array('name' => 'Just another StatusNet microblog',
+              'nickname' => 'statusnet',
+              'wildcard' => null,
               'server' => $_server,
               'theme' => 'default',
               'path' => $_path,
-- 
cgit v1.2.3-54-g00ecf


From cb07fd29a31d681dff838e95f3a96033332020f0 Mon Sep 17 00:00:00 2001
From: Brion Vibber <brion@pobox.com>
Date: Sun, 24 Jan 2010 15:44:09 -0800
Subject: Use new StatusNetwork->serverName() to get full domain for wildcard
 config until we rebuild queues to be based on nicknames. Fixes live bug with
 new *.status.net sites breaking queuedaemon.php

---
 lib/iomaster.php | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

(limited to 'lib')

diff --git a/lib/iomaster.php b/lib/iomaster.php
index 004e92b3e..3bf82bc6b 100644
--- a/lib/iomaster.php
+++ b/lib/iomaster.php
@@ -88,7 +88,7 @@ abstract class IoMaster
         $sn = new Status_network();
         $sn->find();
         while ($sn->fetch()) {
-            $hosts[] = $sn->hostname;
+            $hosts[] = $sn->getServerName();
         }
         return $hosts;
     }
-- 
cgit v1.2.3-54-g00ecf


From f3beed68898a59c40220f4c47cb81a7cd91fbd09 Mon Sep 17 00:00:00 2001
From: Brion Vibber <brion@pobox.com>
Date: Mon, 25 Jan 2010 13:48:24 -0800
Subject: Fix presence notification on XMPP thread (now foreground, not
 background)

---
 lib/xmppmanager.php | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

(limited to 'lib')

diff --git a/lib/xmppmanager.php b/lib/xmppmanager.php
index 299175dd7..985e7c32e 100644
--- a/lib/xmppmanager.php
+++ b/lib/xmppmanager.php
@@ -101,7 +101,7 @@ class XmppManager extends IoManager
         $this->conn->addEventHandler('reconnect', 'handle_reconnect', $this);
 
         $this->conn->setReconnectTimeout(600);
-        jabber_send_presence("Send me a message to post a notice", 'available', null, 'available', -1);
+        jabber_send_presence("Send me a message to post a notice", 'available', null, 'available', 100);
 
         return !is_null($this->conn);
     }
@@ -233,7 +233,7 @@ class XmppManager extends IoManager
         common_log(LOG_NOTICE, 'XMPP reconnected');
 
         $this->conn->processUntil('session_start');
-        $this->conn->presence(null, 'available', null, 'available', -1);
+        $this->conn->presence(null, 'available', null, 'available', 100);
     }
 
 
-- 
cgit v1.2.3-54-g00ecf


From e26a843caf9f6bb0d11a7128884db235ededcce0 Mon Sep 17 00:00:00 2001
From: Evan Prodromou <evan@status.net>
Date: Mon, 25 Jan 2010 18:08:21 -0500
Subject: Offload inbox updates to a queue handler to speed up posting online

Moved much of the writing that happens when posting a notice to a new
queuehandler, distribqueuehandler. This updates tags, groups, replies
and inboxes at queue time (or at Web time, if queues are disabled).

To make this work well, I had to break up the monolithic
Notice::blowCaches() and make cache blowing happen closer to where
data is updated.

Squashed commit of the following:

commit 5257626c62750ac4ac1db0ce2b71410c5711cfa3
Author: Evan Prodromou <evan@status.net>
Date:   Mon Jan 25 14:56:41 2010 -0500

    slightly better handling of blowing tag memory cache

commit 8a22a3cdf6ec28685da129a0313e7b2a0837c9ef
Author: Evan Prodromou <evan@status.net>
Date:   Mon Jan 25 01:42:56 2010 -0500

    change 'distribute' to 'distrib' so not too long for dbqueue

commit 7a063315b0f7fad27cb6fbd2bdd74e253af83e4f
Author: Evan Prodromou <evan@status.net>
Date:   Mon Jan 25 01:39:15 2010 -0500

    change handle_notice() to handle() in distributqueuehandler

commit 1a39ccd28b9994137d7bfd21bb4f230546938e77
Author: Evan Prodromou <evan@status.net>
Date:   Mon Jan 25 16:05:25 2010 -0500

    error with queuemanager

commit e6b3bb93f305cfd2de71a6340b8aa6fb890049b7
Author: Evan Prodromou <evan@status.net>
Date:   Mon Jan 25 01:11:34 2010 -0500

    Blow memcache at different point rather than one big function for Notice class

commit 94d557cdc016187d1d0647ae1794cd94d6fb8ac8
Author: Evan Prodromou <evan@status.net>
Date:   Mon Jan 25 00:48:44 2010 -0500

    Blow memcache at different point rather than one big function for Notice class

commit 1c781dd08c88a35dafc5c01230b4872fd6b95182
Author: Evan Prodromou <evan@status.net>
Date:   Wed Jan 20 08:54:18 2010 -0500

    move broadcasting and distributing to new queuehandler

commit da3e46d26b84e4f028f34a13fd2ee373e4c1b954
Author: Evan Prodromou <evan@status.net>
Date:   Wed Jan 20 08:53:12 2010 -0500

    Move distribution of notices to new distribute queue handler
---
 actions/apistatusesretweet.php      |   2 +-
 actions/apistatusesupdate.php       |   2 +-
 actions/newnotice.php               |   2 +-
 actions/repeat.php                  |   2 +-
 classes/Inbox.php                   |   6 +-
 classes/Memcached_DataObject.php    |  19 ++
 classes/Notice.php                  | 406 +++++++++++++-----------------------
 classes/Notice_tag.php              |  10 +-
 classes/User.php                    |   2 +-
 lib/command.php                     |   4 +-
 lib/distribqueuehandler.php         |  84 ++++++++
 lib/mailhandler.php                 |   2 +-
 lib/oauthstore.php                  |   2 +-
 lib/queuemanager.php                |   3 +-
 lib/util.php                        |   2 +-
 plugins/Facebook/facebookaction.php |   2 +-
 16 files changed, 269 insertions(+), 281 deletions(-)
 create mode 100644 lib/distribqueuehandler.php

(limited to 'lib')

diff --git a/actions/apistatusesretweet.php b/actions/apistatusesretweet.php
index d9d4820c0..128c881e2 100644
--- a/actions/apistatusesretweet.php
+++ b/actions/apistatusesretweet.php
@@ -112,7 +112,7 @@ class ApiStatusesRetweetAction extends ApiAuthAction
 
         $repeat = $this->original->repeat($this->user->id, $this->source);
 
-        common_broadcast_notice($repeat);
+
 
         $this->showNotice($repeat);
     }
diff --git a/actions/apistatusesupdate.php b/actions/apistatusesupdate.php
index f594bbf39..9d831b9db 100644
--- a/actions/apistatusesupdate.php
+++ b/actions/apistatusesupdate.php
@@ -250,7 +250,7 @@ class ApiStatusesUpdateAction extends ApiAuthAction
                 $upload->attachToNotice($this->notice);
             }
 
-            common_broadcast_notice($this->notice);
+
         }
 
         $this->showNotice();
diff --git a/actions/newnotice.php b/actions/newnotice.php
index a4ed87bb6..78480abab 100644
--- a/actions/newnotice.php
+++ b/actions/newnotice.php
@@ -201,7 +201,7 @@ class NewnoticeAction extends Action
             $upload->attachToNotice($notice);
         }
 
-        common_broadcast_notice($notice);
+
 
         if ($this->boolean('ajax')) {
             header('Content-Type: text/xml;charset=utf-8');
diff --git a/actions/repeat.php b/actions/repeat.php
index b75523498..e112496bc 100644
--- a/actions/repeat.php
+++ b/actions/repeat.php
@@ -106,7 +106,7 @@ class RepeatAction extends Action
     {
         $repeat = $this->notice->repeat($this->user->id, 'web');
 
-        common_broadcast_notice($repeat);
+
 
         if ($this->boolean('ajax')) {
             $this->startHTML('text/xml;charset=utf-8');
diff --git a/classes/Inbox.php b/classes/Inbox.php
index 086dba1c9..26b27d2b5 100644
--- a/classes/Inbox.php
+++ b/classes/Inbox.php
@@ -120,11 +120,7 @@ class Inbox extends Memcached_DataObject
                                         $notice_id, $user_id));
 
         if ($result) {
-            $c = self::memcache();
-
-            if (!empty($c)) {
-                $c->delete(self::cacheKey('inbox', 'user_id', $user_id));
-            }
+            self::blow('inbox:user_id:%d', $user_id);
         }
 
         return $result;
diff --git a/classes/Memcached_DataObject.php b/classes/Memcached_DataObject.php
index 6ddef4816..2c9dcf595 100644
--- a/classes/Memcached_DataObject.php
+++ b/classes/Memcached_DataObject.php
@@ -425,4 +425,23 @@ class Memcached_DataObject extends DB_DataObject
 
         return $dsn;
     }
+
+    static function blow()
+    {
+        $c = self::memcache();
+
+        if (empty($c)) {
+            return false;
+        }
+
+        $args = func_get_args();
+
+        $format = array_shift($args);
+
+        $keyPart = vsprintf($format, $args);
+
+        $cacheKey = common_cache_key($keyPart);
+
+        return $c->delete($cacheKey);
+    }
 }
diff --git a/classes/Notice.php b/classes/Notice.php
index 38b10db04..0966697e2 100644
--- a/classes/Notice.php
+++ b/classes/Notice.php
@@ -94,10 +94,6 @@ class Notice extends Memcached_DataObject
 
     function delete()
     {
-        $this->blowCaches(true);
-        $this->blowFavesCache(true);
-        $this->blowSubsCache(true);
-
         // For auditing purposes, save a record that the notice
         // was deleted.
 
@@ -109,31 +105,20 @@ class Notice extends Memcached_DataObject
         $deleted->created    = $this->created;
         $deleted->deleted    = common_sql_now();
 
-        $this->query('BEGIN');
-
         $deleted->insert();
 
-        //Null any notices that are replies to this notice
-        $this->query(sprintf("UPDATE notice set reply_to = null WHERE reply_to = %d", $this->id));
-
-        //Null any notices that are repeats of this notice
-        //XXX: probably need to uncache these, too
+        // Clear related records
 
-        $this->query(sprintf("UPDATE notice set repeat_of = null WHERE repeat_of = %d", $this->id));
+        $this->clearReplies();
+        $this->clearRepeats();
+        $this->clearFaves();
+        $this->clearTags();
+        $this->clearGroupInboxes();
 
-        $related = array('Reply',
-                         'Fave',
-                         'Notice_tag',
-                         'Group_inbox',
-                         'Queue_item');
+        // NOTE: we don't clear inboxes
+        // NOTE: we don't clear queue items
 
-        foreach ($related as $cls) {
-            $inst = new $cls();
-            $inst->notice_id = $this->id;
-            $inst->delete();
-        }
         $result = parent::delete();
-        $this->query('COMMIT');
     }
 
     function saveTags()
@@ -155,6 +140,7 @@ class Notice extends Memcached_DataObject
         foreach(array_unique($hashtags) as $hashtag) {
             /* elide characters we don't want in the tag */
             $this->saveTag($hashtag);
+            self::blow('profile:notice_ids_tagged:%d:%s', $this->profile_id, $tag->tag);
         }
         return true;
     }
@@ -172,6 +158,9 @@ class Notice extends Memcached_DataObject
                                               $last_error->message));
             return;
         }
+
+        // if it's saved, blow its cache
+        $tag->blowCache(false);
     }
 
     /**
@@ -331,27 +320,43 @@ class Notice extends Memcached_DataObject
                 }
             }
 
-            // XXX: do we need to change this for remote users?
+        }
+
+        # Clear the cache for subscribed users, so they'll update at next request
+        # XXX: someone clever could prepend instead of clearing the cache
+        $notice->blowOnInsert();
 
-            $notice->saveTags();
+        $qm = QueueManager::get();
 
-            $groups = $notice->saveGroups();
+        $qm->enqueue($notice, 'distrib');
 
-            $recipients = $notice->saveReplies();
+        return $notice;
+    }
 
-            $notice->addToInboxes($groups, $recipients);
+    function blowOnInsert()
+    {
+        self::blow('profile:notice_ids:%d', $this->profile_id);
+        self::blow('public');
 
-            $notice->saveUrls();
+        if ($this->conversation != $this->id) {
+            self::blow('notice:conversation_ids:%d', $this->conversation);
+        }
 
-            Event::handle('EndNoticeSave', array($notice));
+        if (!empty($this->repeat_of)) {
+            self::blow('notice:repeats:%d', $this->repeat_of);
         }
 
-        # Clear the cache for subscribed users, so they'll update at next request
-        # XXX: someone clever could prepend instead of clearing the cache
+        $original = Notice::staticGet('id', $this->repeat_of);
 
-        $notice->blowCaches();
+        if (!empty($original)) {
+            $originalUser = User::staticGet('id', $original->profile_id);
+            if (!empty($originalUser)) {
+                self::blow('user:repeats_of_me:%d', $originalUser->id);
+            }
+        }
 
-        return $notice;
+        $profile = Profile::staticGet($this->profile_id);
+        $profile->blowNoticeCount();
     }
 
     /** save all urls in the notice to the db
@@ -456,227 +461,6 @@ class Notice extends Memcached_DataObject
         return $att;
     }
 
-    function blowCaches($blowLast=false)
-    {
-        $this->blowSubsCache($blowLast);
-        $this->blowNoticeCache($blowLast);
-        $this->blowRepliesCache($blowLast);
-        $this->blowPublicCache($blowLast);
-        $this->blowTagCache($blowLast);
-        $this->blowGroupCache($blowLast);
-        $this->blowConversationCache($blowLast);
-        $this->blowRepeatCache();
-        $profile = Profile::staticGet($this->profile_id);
-        $profile->blowNoticeCount();
-    }
-
-    function blowRepeatCache()
-    {
-        if (!empty($this->repeat_of)) {
-            $cache = common_memcache();
-            if (!empty($cache)) {
-                // XXX: only blow if <100 in cache
-                $ck = common_cache_key('notice:repeats:'.$this->repeat_of);
-                $result = $cache->delete($ck);
-
-                $user = User::staticGet('id', $this->profile_id);
-
-                if (!empty($user)) {
-                    $uk = common_cache_key('user:repeated_by_me:'.$user->id);
-                    $cache->delete($uk);
-                    $user->free();
-                    unset($user);
-                }
-
-                $original = Notice::staticGet('id', $this->repeat_of);
-
-                if (!empty($original)) {
-                    $originalUser = User::staticGet('id', $original->profile_id);
-                    if (!empty($originalUser)) {
-                        $ouk = common_cache_key('user:repeats_of_me:'.$originalUser->id);
-                        $cache->delete($ouk);
-                        $originalUser->free();
-                        unset($originalUser);
-                    }
-                    $original->free();
-                    unset($original);
-                }
-            }
-        }
-    }
-
-    function blowConversationCache($blowLast=false)
-    {
-        $cache = common_memcache();
-        if ($cache) {
-            $ck = common_cache_key('notice:conversation_ids:'.$this->conversation);
-            $cache->delete($ck);
-            if ($blowLast) {
-                $cache->delete($ck.';last');
-            }
-        }
-    }
-
-    function blowGroupCache($blowLast=false)
-    {
-        $cache = common_memcache();
-        if ($cache) {
-            $group_inbox = new Group_inbox();
-            $group_inbox->notice_id = $this->id;
-            if ($group_inbox->find()) {
-                while ($group_inbox->fetch()) {
-                    $cache->delete(common_cache_key('user_group:notice_ids:' . $group_inbox->group_id));
-                    if ($blowLast) {
-                        $cache->delete(common_cache_key('user_group:notice_ids:' . $group_inbox->group_id.';last'));
-                    }
-                    $member = new Group_member();
-                    $member->group_id = $group_inbox->group_id;
-                    if ($member->find()) {
-                        while ($member->fetch()) {
-                            $cache->delete(common_cache_key('notice_inbox:by_user:' . $member->profile_id));
-                            $cache->delete(common_cache_key('notice_inbox:by_user_own:' . $member->profile_id));
-                            if (empty($this->repeat_of)) {
-                                $cache->delete(common_cache_key('user:friends_timeline:' . $member->profile_id));
-                                $cache->delete(common_cache_key('user:friends_timeline_own:' . $member->profile_id));
-                            }
-                            if ($blowLast) {
-                                $cache->delete(common_cache_key('notice_inbox:by_user:' . $member->profile_id . ';last'));
-                                $cache->delete(common_cache_key('notice_inbox:by_user_own:' . $member->profile_id . ';last'));
-                                if (empty($this->repeat_of)) {
-                                    $cache->delete(common_cache_key('user:friends_timeline:' . $member->profile_id . ';last'));
-                                    $cache->delete(common_cache_key('user:friends_timeline_own:' . $member->profile_id . ';last'));
-                                }
-                            }
-                        }
-                    }
-                }
-            }
-            $group_inbox->free();
-            unset($group_inbox);
-        }
-    }
-
-    function blowTagCache($blowLast=false)
-    {
-        $cache = common_memcache();
-        if ($cache) {
-            $tag = new Notice_tag();
-            $tag->notice_id = $this->id;
-            if ($tag->find()) {
-                while ($tag->fetch()) {
-                    $tag->blowCache($blowLast);
-                    $ck = 'profile:notice_ids_tagged:' . $this->profile_id . ':' . $tag->tag;
-
-                    $cache->delete($ck);
-                    if ($blowLast) {
-                        $cache->delete($ck . ';last');
-                    }
-                }
-            }
-            $tag->free();
-            unset($tag);
-        }
-    }
-
-    function blowSubsCache($blowLast=false)
-    {
-        $cache = common_memcache();
-        if ($cache) {
-            $user = new User();
-
-            $UT = common_config('db','type')=='pgsql'?'"user"':'user';
-            $user->query('SELECT id ' .
-
-                         "FROM $UT JOIN subscription ON $UT.id = subscription.subscriber " .
-                         'WHERE subscription.subscribed = ' . $this->profile_id);
-
-            while ($user->fetch()) {
-                $cache->delete(common_cache_key('notice_inbox:by_user:'.$user->id));
-                $cache->delete(common_cache_key('notice_inbox:by_user_own:'.$user->id));
-                if (empty($this->repeat_of)) {
-                    $cache->delete(common_cache_key('user:friends_timeline:'.$user->id));
-                    $cache->delete(common_cache_key('user:friends_timeline_own:'.$user->id));
-                }
-                if ($blowLast) {
-                    $cache->delete(common_cache_key('notice_inbox:by_user:'.$user->id.';last'));
-                    $cache->delete(common_cache_key('notice_inbox:by_user_own:'.$user->id.';last'));
-                    if (empty($this->repeat_of)) {
-                        $cache->delete(common_cache_key('user:friends_timeline:'.$user->id.';last'));
-                        $cache->delete(common_cache_key('user:friends_timeline_own:'.$user->id.';last'));
-                    }
-                }
-            }
-            $user->free();
-            unset($user);
-        }
-    }
-
-    function blowNoticeCache($blowLast=false)
-    {
-        if ($this->is_local) {
-            $cache = common_memcache();
-            if (!empty($cache)) {
-                $cache->delete(common_cache_key('profile:notice_ids:'.$this->profile_id));
-                if ($blowLast) {
-                    $cache->delete(common_cache_key('profile:notice_ids:'.$this->profile_id.';last'));
-                }
-            }
-        }
-    }
-
-    function blowRepliesCache($blowLast=false)
-    {
-        $cache = common_memcache();
-        if ($cache) {
-            $reply = new Reply();
-            $reply->notice_id = $this->id;
-            if ($reply->find()) {
-                while ($reply->fetch()) {
-                    $cache->delete(common_cache_key('reply:stream:'.$reply->profile_id));
-                    if ($blowLast) {
-                        $cache->delete(common_cache_key('reply:stream:'.$reply->profile_id.';last'));
-                    }
-                }
-            }
-            $reply->free();
-            unset($reply);
-        }
-    }
-
-    function blowPublicCache($blowLast=false)
-    {
-        if ($this->is_local == Notice::LOCAL_PUBLIC) {
-            $cache = common_memcache();
-            if ($cache) {
-                $cache->delete(common_cache_key('public'));
-                if ($blowLast) {
-                    $cache->delete(common_cache_key('public').';last');
-                }
-            }
-        }
-    }
-
-    function blowFavesCache($blowLast=false)
-    {
-        $cache = common_memcache();
-        if ($cache) {
-            $fave = new Fave();
-            $fave->notice_id = $this->id;
-            if ($fave->find()) {
-                while ($fave->fetch()) {
-                    $cache->delete(common_cache_key('fave:ids_by_user:'.$fave->user_id));
-                    $cache->delete(common_cache_key('fave:by_user_own:'.$fave->user_id));
-                    if ($blowLast) {
-                        $cache->delete(common_cache_key('fave:ids_by_user:'.$fave->user_id.';last'));
-                        $cache->delete(common_cache_key('fave:by_user_own:'.$fave->user_id.';last'));
-                    }
-                }
-            }
-            $fave->free();
-            unset($fave);
-        }
-    }
-
     function getStreamByIds($ids)
     {
         $cache = common_memcache();
@@ -999,7 +783,14 @@ class Notice extends Memcached_DataObject
             $gi->notice_id = $this->id;
             $gi->created   = $this->created;
 
-            return $gi->insert();
+            $result = $gi->insert();
+
+            if (!result) {
+                common_log_db_error($gi, 'INSERT', __FILE__);
+                throw new ServerException(_('Problem saving group inbox.'));
+            }
+
+            self::blow('user_group:notice_ids:%d', $gi->group_id);
         }
 
         return true;
@@ -1094,7 +885,8 @@ class Notice extends Memcached_DataObject
 
         foreach ($recipientIds as $recipientId) {
             $user = User::staticGet('id', $recipientId);
-            if ($user) {
+            if (!empty($user)) {
+                self::blow('reply:stream:%d', $reply->profile_id);
                 mail_notify_attn($user, $this);
             }
         }
@@ -1553,4 +1345,104 @@ class Notice extends Memcached_DataObject
 
         return $options;
     }
+
+    function clearReplies()
+    {
+        $replyNotice = new Notice();
+        $replyNotice->reply_to = $this->id;
+
+        //Null any notices that are replies to this notice
+
+        if ($replyNotice->find()) {
+            while ($replyNotice->fetch()) {
+                $orig = clone($replyNotice);
+                $replyNotice->reply_to = null;
+                $replyNotice->update($orig);
+            }
+        }
+
+        // Reply records
+
+        $reply = new Reply();
+        $reply->notice_id = $this->id;
+
+        if ($reply->find()) {
+            while($reply->fetch()) {
+                self::blow('reply:stream:%d', $reply->profile_id);
+                $reply->delete();
+            }
+        }
+
+        $reply->free();
+
+        return $ids;
+    }
+
+    function clearRepeats()
+    {
+        $repeatNotice = new Notice();
+        $repeatNotice->repeat_of = $this->id;
+
+        //Null any notices that are repeats of this notice
+
+        if ($repeatNotice->find()) {
+            while ($repeatNotice->fetch()) {
+                $orig = clone($repeatNotice);
+                $repeatNotice->repeat_of = null;
+                $repeatNotice->update($orig);
+            }
+        }
+    }
+
+    function clearFaves()
+    {
+        $fave = new Fave();
+        $fave->notice_id = $this->id;
+
+        if ($fave->find()) {
+            while ($fave->fetch()) {
+                self::blow('fave:ids_by_user_own:%d', $fave->user_id);
+                self::blow('fave:ids_by_user_own:%d;last', $fave->user_id);
+                self::blow('fave:ids_by_user:%d', $fave->user_id);
+                self::blow('fave:ids_by_user:%d;last', $fave->user_id);
+                $fave->delete();
+            }
+        }
+
+        $fave->free();
+    }
+
+    function clearTags()
+    {
+        $tag = new Notice_tag();
+        $tag->notice_id = $this->id;
+
+        if ($tag->find()) {
+            while ($tag->fetch()) {
+                self::blow('profile:notice_ids_tagged:%d:%s', $this->profile_id, common_keyize($tag->tag));
+                self::blow('profile:notice_ids_tagged:%d:%s;last', $this->profile_id, common_keyize($tag->tag));
+                self::blow('notice_tag:notice_ids:%s', common_keyize($tag->tag));
+                self::blow('notice_tag:notice_ids:%s;last', common_keyize($tag->tag));
+                $tag->delete();
+            }
+        }
+
+        $tag->free();
+    }
+
+    function clearGroupInboxes()
+    {
+        $gi = new Group_inbox();
+
+        $gi->notice_id = $this->id;
+
+        if ($gi->find()) {
+            while ($gi->fetch()) {
+                self::blow('user_group:notice_ids:%d', $gi->group_id);
+                $gi->delete();
+            }
+        }
+
+        $gi->free();
+    }
 }
diff --git a/classes/Notice_tag.php b/classes/Notice_tag.php
index 79231f0b0..4fd76e8ea 100644
--- a/classes/Notice_tag.php
+++ b/classes/Notice_tag.php
@@ -86,13 +86,9 @@ class Notice_tag extends Memcached_DataObject
 
     function blowCache($blowLast=false)
     {
-        $cache = common_memcache();
-        if ($cache) {
-            $idkey = common_cache_key('notice_tag:notice_ids:' . common_keyize($this->tag));
-            $cache->delete($idkey);
-            if ($blowLast) {
-                $cache->delete($idkey.';last');
-            }
+        self::blow('notice_tag:notice_ids:%s', common_keyize($this->tag));
+        if ($blowLast) {
+            self::blow('notice_tag:notice_ids:%s;last', common_keyize($this->tag));
         }
     }
 
diff --git a/classes/User.php b/classes/User.php
index d6b52be01..6ea975202 100644
--- a/classes/User.php
+++ b/classes/User.php
@@ -383,7 +383,7 @@ class User extends Memcached_DataObject
                                                   common_config('site', 'name'),
                                                   $user->nickname),
                                           'system');
-                common_broadcast_notice($notice);
+
             }
         }
 
diff --git a/lib/command.php b/lib/command.php
index c0a32e1b1..2a51fd687 100644
--- a/lib/command.php
+++ b/lib/command.php
@@ -422,7 +422,7 @@ class RepeatCommand extends Command
         $repeat = $notice->repeat($this->user->id, $channel->source);
 
         if ($repeat) {
-            common_broadcast_notice($repeat);
+
             $channel->output($this->user, sprintf(_('Notice from %s repeated'), $recipient->nickname));
         } else {
             $channel->error($this->user, _('Error repeating notice.'));
@@ -492,7 +492,7 @@ class ReplyCommand extends Command
         } else {
             $channel->error($this->user, _('Error saving notice.'));
         }
-        common_broadcast_notice($notice);
+
     }
 }
 
diff --git a/lib/distribqueuehandler.php b/lib/distribqueuehandler.php
new file mode 100644
index 000000000..f458d238d
--- /dev/null
+++ b/lib/distribqueuehandler.php
@@ -0,0 +1,84 @@
+<?php
+/*
+ * StatusNet - the distributed open-source microblogging tool
+ * Copyright (C) 2008, 2009, StatusNet, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+if (!defined('STATUSNET') && !defined('LACONICA')) { exit(1); }
+
+/**
+ * Base class for queue handlers.
+ *
+ * As extensions of the Daemon class, each queue handler has the ability
+ * to launch itself in the background, at which point it'll pass control
+ * to the configured QueueManager class to poll for updates.
+ *
+ * Subclasses must override at least the following methods:
+ * - transport
+ * - handle_notice
+ */
+
+class DistribQueueHandler
+{
+    /**
+     * Return transport keyword which identifies items this queue handler
+     * services; must be defined for all subclasses.
+     *
+     * Must be 8 characters or less to fit in the queue_item database.
+     * ex "email", "jabber", "sms", "irc", ...
+     *
+     * @return string
+     */
+
+    function transport()
+    {
+        return 'distrib';
+    }
+
+    /**
+     * Here's the meat of your queue handler -- you're handed a Notice
+     * object, which you may do as you will with.
+     *
+     * If this function indicates failure, a warning will be logged
+     * and the item is placed back in the queue to be re-run.
+     *
+     * @param Notice $notice
+     * @return boolean true on success, false on failure
+     */
+    function handle($notice)
+    {
+        // XXX: do we need to change this for remote users?
+
+        $notice->saveTags();
+
+        $groups = $notice->saveGroups();
+
+        $recipients = $notice->saveReplies();
+
+        $notice->addToInboxes($groups, $recipients);
+
+        $notice->saveUrls();
+
+        Event::handle('EndNoticeSave', array($notice));
+
+        // Enqueue for other handlers
+
+        common_enqueue_notice($notice);
+
+        return true;
+    }
+}
+
diff --git a/lib/mailhandler.php b/lib/mailhandler.php
index 85be89f18..890f6d5b4 100644
--- a/lib/mailhandler.php
+++ b/lib/mailhandler.php
@@ -160,7 +160,7 @@ class MailHandler
         foreach($mediafiles as $mf){
             $mf->attachToNotice($notice);
         }
-        common_broadcast_notice($notice);
+
         $this->log(LOG_INFO,
                    'Added notice ' . $notice->id . ' from user ' . $user->nickname);
         return true;
diff --git a/lib/oauthstore.php b/lib/oauthstore.php
index df63cc151..b30fb49d5 100644
--- a/lib/oauthstore.php
+++ b/lib/oauthstore.php
@@ -362,7 +362,7 @@ class StatusNetOAuthDataStore extends OAuthDataStore
                                   array('is_local' => Notice::REMOTE_OMB,
                                         'uri' => $omb_notice->getIdentifierURI()));
 
-        common_broadcast_notice($notice, true);
+
     }
 
     /**
diff --git a/lib/queuemanager.php b/lib/queuemanager.php
index 4eb39bfa8..e5cf8239e 100644
--- a/lib/queuemanager.php
+++ b/lib/queuemanager.php
@@ -217,6 +217,7 @@ abstract class QueueManager extends IoManager
             $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');
             }
@@ -224,7 +225,7 @@ abstract class QueueManager extends IoManager
             // XMPP output handlers...
             $this->connect('jabber', 'JabberQueueHandler');
             $this->connect('public', 'PublicQueueHandler');
-            
+
             // @fixme this should get an actual queue
             //$this->connect('confirm', 'XmppConfirmHandler');
 
diff --git a/lib/util.php b/lib/util.php
index fb3b8be87..4312f9876 100644
--- a/lib/util.php
+++ b/lib/util.php
@@ -987,7 +987,7 @@ function common_redirect($url, $code=307)
 
 function common_broadcast_notice($notice, $remote=false)
 {
-    return common_enqueue_notice($notice);
+    // DO NOTHING!
 }
 
 // Stick the notice on the queue
diff --git a/plugins/Facebook/facebookaction.php b/plugins/Facebook/facebookaction.php
index bf9c037a5..815fee094 100644
--- a/plugins/Facebook/facebookaction.php
+++ b/plugins/Facebook/facebookaction.php
@@ -397,7 +397,7 @@ class FacebookAction extends Action
             return;
         }
 
-        common_broadcast_notice($notice);
+
 
     }
 
-- 
cgit v1.2.3-54-g00ecf