From 8b2f6c01fb3e15007ec7466f6a8f53e3b13ca41b Mon Sep 17 00:00:00 2001
From: Evan Prodromou <evan@status.net>
Date: Tue, 29 Dec 2009 22:02:46 -0800
Subject: add inbox data class

---
 classes/Inbox.php     | 51 +++++++++++++++++++++++++++++++++++++++++++++++++++
 classes/statusnet.ini |  7 +++++++
 2 files changed, 58 insertions(+)
 create mode 100755 classes/Inbox.php

(limited to 'classes')

diff --git a/classes/Inbox.php b/classes/Inbox.php
new file mode 100755
index 000000000..35f532c06
--- /dev/null
+++ b/classes/Inbox.php
@@ -0,0 +1,51 @@
+<?php
+/**
+ * StatusNet, the distributed open-source microblogging tool
+ *
+ * Data class for user location preferences
+ *
+ * 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  Data
+ * @package   StatusNet
+ * @author    Evan Prodromou <evan@status.net>
+ * @copyright 2009 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/
+ */
+
+require_once INSTALLDIR.'/classes/Memcached_DataObject.php';
+
+class Inbox extends Memcached_DataObject
+{
+    ###START_AUTOCODE
+    /* the code below is auto generated do not remove the above tag */
+
+    public $__table = 'inbox';                           // table name
+    public $user_id;                         // int(4)  primary_key not_null
+    public $notice_ids;                      // blob
+
+    /* Static get */
+    function staticGet($k,$v=NULL) { return Memcached_DataObject::staticGet('Inbox',$k,$v); }
+
+    /* the code above is auto generated do not remove the tag below */
+    ###END_AUTOCODE
+
+    function sequenceKey()
+    {
+        return array(false, false, false);
+    }
+}
diff --git a/classes/statusnet.ini b/classes/statusnet.ini
index 0db2c5d6e..73727a6d6 100644
--- a/classes/statusnet.ini
+++ b/classes/statusnet.ini
@@ -241,6 +241,13 @@ address = 130
 address_type = 130
 created = 142
 
+[inbox]
+user_id = 129
+notice_ids = 66
+
+[inbox__keys]
+user_id = K
+
 [invitation__keys]
 code = K
 
-- 
cgit v1.2.3-54-g00ecf


From 475bf6010f18a17042d9171998c82e787c5abf74 Mon Sep 17 00:00:00 2001
From: Evan Prodromou <evan@status.net>
Date: Tue, 29 Dec 2009 22:03:06 -0800
Subject: flip exe bit

---
 classes/Inbox.php | 0
 1 file changed, 0 insertions(+), 0 deletions(-)
 mode change 100755 => 100644 classes/Inbox.php

(limited to 'classes')

diff --git a/classes/Inbox.php b/classes/Inbox.php
old mode 100755
new mode 100644
-- 
cgit v1.2.3-54-g00ecf


From 7ec27b657e6b7ad9416e531cf70fffcf48614708 Mon Sep 17 00:00:00 2001
From: Evan Prodromou <evan@status.net>
Date: Wed, 30 Dec 2009 09:03:06 -1000
Subject: Replace Notice_inbox with Inbox

---
 classes/Notice_inbox.php | 193 -----------------------------------------------
 scripts/inbox_users.php  | 107 --------------------------
 scripts/triminboxes.php  |  76 -------------------
 3 files changed, 376 deletions(-)
 delete mode 100644 classes/Notice_inbox.php
 delete mode 100755 scripts/inbox_users.php
 delete mode 100644 scripts/triminboxes.php

(limited to 'classes')

diff --git a/classes/Notice_inbox.php b/classes/Notice_inbox.php
deleted file mode 100644
index e350e6e2f..000000000
--- a/classes/Notice_inbox.php
+++ /dev/null
@@ -1,193 +0,0 @@
-<?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); }
-
-require_once INSTALLDIR.'/classes/Memcached_DataObject.php';
-
-// We keep 5 pages of inbox notices in memcache, +1 for pagination check
-
-define('INBOX_CACHE_WINDOW', 101);
-define('NOTICE_INBOX_GC_BOXCAR', 128);
-define('NOTICE_INBOX_GC_MAX', 12800);
-define('NOTICE_INBOX_LIMIT', 1000);
-define('NOTICE_INBOX_SOFT_LIMIT', 1000);
-
-define('NOTICE_INBOX_SOURCE_SUB', 1);
-define('NOTICE_INBOX_SOURCE_GROUP', 2);
-define('NOTICE_INBOX_SOURCE_REPLY', 3);
-define('NOTICE_INBOX_SOURCE_FORWARD', 4);
-define('NOTICE_INBOX_SOURCE_GATEWAY', -1);
-
-class Notice_inbox extends Memcached_DataObject
-{
-    ###START_AUTOCODE
-    /* the code below is auto generated do not remove the above tag */
-
-    public $__table = 'notice_inbox';                    // table name
-    public $user_id;                         // int(4)  primary_key not_null
-    public $notice_id;                       // int(4)  primary_key not_null
-    public $created;                         // datetime()   not_null
-    public $source;                          // tinyint(1)   default_1
-
-    /* Static get */
-    function staticGet($k,$v=null)
-    { return Memcached_DataObject::staticGet('Notice_inbox',$k,$v); }
-
-    /* the code above is auto generated do not remove the tag below */
-    ###END_AUTOCODE
-
-    function stream($user_id, $offset, $limit, $since_id, $max_id, $since, $own=false)
-    {
-        return Notice::stream(array('Notice_inbox', '_streamDirect'),
-                              array($user_id, $own),
-                              ($own) ? 'notice_inbox:by_user:'.$user_id :
-                              'notice_inbox:by_user_own:'.$user_id,
-                              $offset, $limit, $since_id, $max_id, $since);
-    }
-
-    function _streamDirect($user_id, $own, $offset, $limit, $since_id, $max_id, $since)
-    {
-        $inbox = new Notice_inbox();
-
-        $inbox->user_id = $user_id;
-
-        if (!$own) {
-            $inbox->whereAdd('source != ' . NOTICE_INBOX_SOURCE_GATEWAY);
-        }
-
-        if ($since_id != 0) {
-            $inbox->whereAdd('notice_id > ' . $since_id);
-        }
-
-        if ($max_id != 0) {
-            $inbox->whereAdd('notice_id <= ' . $max_id);
-        }
-
-        if (!is_null($since)) {
-            $inbox->whereAdd('created > \'' . date('Y-m-d H:i:s', $since) . '\'');
-        }
-
-        $inbox->orderBy('created DESC');
-
-        if (!is_null($offset)) {
-            $inbox->limit($offset, $limit);
-        }
-
-        $ids = array();
-
-        if ($inbox->find()) {
-            while ($inbox->fetch()) {
-                $ids[] = $inbox->notice_id;
-            }
-        }
-
-        return $ids;
-    }
-
-    function pkeyGet($kv)
-    {
-        return Memcached_DataObject::pkeyGet('Notice_inbox', $kv);
-    }
-
-    /**
-     * Trim inbox for a given user to latest NOTICE_INBOX_LIMIT items
-     * (up to NOTICE_INBOX_GC_MAX will be deleted).
-     *
-     * @param int $user_id
-     * @return int count of notices dropped from the inbox, if any
-     */
-    static function gc($user_id)
-    {
-        $entry = new Notice_inbox();
-        $entry->user_id = $user_id;
-        $entry->orderBy('created DESC');
-        $entry->limit(NOTICE_INBOX_LIMIT - 1, NOTICE_INBOX_GC_MAX);
-
-        $total = $entry->find();
-
-        if ($total > 0) {
-            $notices = array();
-            $cnt = 0;
-            while ($entry->fetch()) {
-                $notices[] = $entry->notice_id;
-                $cnt++;
-                if ($cnt >= NOTICE_INBOX_GC_BOXCAR) {
-                    self::deleteMatching($user_id, $notices);
-                    $notices = array();
-                    $cnt = 0;
-                }
-            }
-
-            if ($cnt > 0) {
-                self::deleteMatching($user_id, $notices);
-                $notices = array();
-            }
-        }
-
-        return $total;
-    }
-
-    static function deleteMatching($user_id, $notices)
-    {
-        $entry = new Notice_inbox();
-        return $entry->query('DELETE FROM notice_inbox '.
-                             'WHERE user_id = ' . $user_id . ' ' .
-                             'AND notice_id in ('.implode(',', $notices).')');
-    }
-
-    static function bulkInsert($notice_id, $created, $ni)
-    {
-        $cnt = 0;
-
-        $qryhdr = 'INSERT INTO notice_inbox (user_id, notice_id, source, created) VALUES ';
-        $qry = $qryhdr;
-
-        foreach ($ni as $id => $source) {
-            if ($cnt > 0) {
-                $qry .= ', ';
-            }
-            $qry .= '('.$id.', '.$notice_id.', '.$source.", '".$created. "') ";
-            $cnt++;
-            if (rand() % NOTICE_INBOX_SOFT_LIMIT == 0) {
-                // FIXME: Causes lag in replicated servers
-                // Notice_inbox::gc($id);
-            }
-            if ($cnt >= MAX_BOXCARS) {
-                $inbox = new Notice_inbox();
-                $result = $inbox->query($qry);
-                if (PEAR::isError($result)) {
-                    common_log_db_error($inbox, $qry);
-                }
-                $qry = $qryhdr;
-                $cnt = 0;
-            }
-        }
-
-        if ($cnt > 0) {
-            $inbox = new Notice_inbox();
-            $result = $inbox->query($qry);
-            if (PEAR::isError($result)) {
-                common_log_db_error($inbox, $qry);
-            }
-        }
-
-        return;
-    }
-}
diff --git a/scripts/inbox_users.php b/scripts/inbox_users.php
deleted file mode 100755
index 32adcea21..000000000
--- a/scripts/inbox_users.php
+++ /dev/null
@@ -1,107 +0,0 @@
-#!/usr/bin/env php
-<?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/>.
- */
-
-# Abort if called from a web server
-
-define('INSTALLDIR', realpath(dirname(__FILE__) . '/..'));
-
-$helptext = <<<ENDOFHELP
-inbox_users.php <idfile>
-
-Update users to use inbox table. Listed in an ID file, default 'ids.txt'.
-
-ENDOFHELP;
-
-require_once INSTALLDIR.'/scripts/commandline.inc';
-
-$id_file = (count($args) > 1) ? $args[0] : 'ids.txt';
-
-common_log(LOG_INFO, 'Updating user inboxes.');
-
-$ids = file($id_file);
-
-foreach ($ids as $id) {
-
-	$user = User::staticGet('id', $id);
-
-	if (!$user) {
-		common_log(LOG_WARNING, 'No such user: ' . $id);
-		continue;
-	}
-
-	if ($user->inboxed) {
-		common_log(LOG_WARNING, 'Already inboxed: ' . $id);
-		continue;
-	}
-
-    common_log(LOG_INFO, 'Updating inbox for user ' . $user->id);
-
-	$user->query('BEGIN');
-
-	$old_inbox = new Notice_inbox();
-	$old_inbox->user_id = $user->id;
-
-	$result = $old_inbox->delete();
-
-	if (is_null($result) || $result === false) {
-		common_log_db_error($old_inbox, 'DELETE', __FILE__);
-		continue;
-	}
-
-	$old_inbox->free();
-
-	$inbox = new Notice_inbox();
-
-	$result = $inbox->query('INSERT INTO notice_inbox (user_id, notice_id, created) ' .
-							'SELECT ' . $user->id . ', notice.id, notice.created ' .
-							'FROM subscription JOIN notice ON subscription.subscribed = notice.profile_id ' .
-							'WHERE subscription.subscriber = ' . $user->id . ' ' .
-							'AND notice.created >= subscription.created ' .
-							'AND NOT EXISTS (SELECT user_id, notice_id ' .
-							'FROM notice_inbox ' .
-							'WHERE user_id = ' . $user->id . ' ' .
-							'AND notice_id = notice.id) ' .
-				                        'ORDER BY notice.created DESC ' .
-							'LIMIT 0, 1000');
-
-	if (is_null($result) || $result === false) {
-		common_log_db_error($inbox, 'INSERT', __FILE__);
-		continue;
-	}
-
-	$orig = clone($user);
-	$user->inboxed = 1;
-	$result = $user->update($orig);
-
-	if (!$result) {
-		common_log_db_error($user, 'UPDATE', __FILE__);
-		continue;
-	}
-
-	$user->query('COMMIT');
-
-	$inbox->free();
-	unset($inbox);
-
-	if ($cache) {
-		$cache->delete(common_cache_key('user:notices_with_friends:' . $user->id));
-		$cache->delete(common_cache_key('user:notices_with_friends:' . $user->id . ';last'));
-	}
-}
diff --git a/scripts/triminboxes.php b/scripts/triminboxes.php
deleted file mode 100644
index ea4751305..000000000
--- a/scripts/triminboxes.php
+++ /dev/null
@@ -1,76 +0,0 @@
-#!/usr/bin/env php
-<?php
-/*
- * StatusNet - the distributed open-source microblogging tool
- * Copyright (C) 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/>.
- */
-
-define('INSTALLDIR', realpath(dirname(__FILE__) . '/..'));
-
-$shortoptions = 'u::';
-$longoptions = array('start-user-id=', 'sleep-time=');
-
-$helptext = <<<END_OF_TRIM_HELP
-Batch script for trimming notice inboxes to a reasonable size.
-
-    -u <id>
-    --start-user-id=<id>   User ID to start after. Default is all.
-    --sleep-time=<integer> Amount of time to wait (in seconds) between trims. Default is zero.
-
-END_OF_TRIM_HELP;
-
-require_once INSTALLDIR.'/scripts/commandline.inc';
-
-$id = null;
-$sleep_time = 0;
-
-if (have_option('u')) {
-    $id = get_option_value('u');
-} else if (have_option('--start-user-id')) {
-    $id = get_option_value('--start-user-id');
-} else {
-    $id = null;
-}
-
-if (have_option('--sleep-time')) {
-    $sleep_time = intval(get_option_value('--sleep-time'));
-}
-
-$quiet = have_option('q') || have_option('--quiet');
-
-$user = new User();
-
-if (!empty($id)) {
-    $user->whereAdd('id > ' . $id);
-}
-
-$cnt = $user->find();
-
-while ($user->fetch()) {
-    if (!$quiet) {
-        print "Trimming inbox for user $user->id";
-    }
-    $count = Notice_inbox::gc($user->id);
-    if ($count) {
-        if (!$quiet) {
-            print ": $count trimmed...";
-        }
-        sleep($sleep_time);
-    }
-    if (!$quiet) {
-        print "\n";
-    }
-}
-- 
cgit v1.2.3-54-g00ecf


From f2a403589cc8858ec3d3ce70f09709f36277c348 Mon Sep 17 00:00:00 2001
From: Evan Prodromou <evan@status.net>
Date: Wed, 30 Dec 2009 09:06:07 -1000
Subject: Use inbox instead of notice_inbox

---
 classes/Inbox.php  | 53 +++++++++++++++++++++++++++++++++++
 classes/Notice.php | 20 ++------------
 classes/User.php   | 81 ++++++++++++------------------------------------------
 3 files changed, 72 insertions(+), 82 deletions(-)

(limited to 'classes')

diff --git a/classes/Inbox.php b/classes/Inbox.php
index 35f532c06..de48d7381 100644
--- a/classes/Inbox.php
+++ b/classes/Inbox.php
@@ -31,6 +31,8 @@ require_once INSTALLDIR.'/classes/Memcached_DataObject.php';
 
 class Inbox extends Memcached_DataObject
 {
+    const BOXCAR = 128;
+
     ###START_AUTOCODE
     /* the code below is auto generated do not remove the above tag */
 
@@ -48,4 +50,55 @@ class Inbox extends Memcached_DataObject
     {
         return array(false, false, false);
     }
+
+    static function insertNotice($user_id, $notice_id)
+    {
+        $inbox = new Inbox();
+
+        $inbox->query(sprintf('UPDATE inbox '.
+                              'set notice_ids = concat(cast(%08x as binary(4)), '.
+                              'substr(notice_ids, 1, 4092)) '.
+                              'WHERE user_id = %d',
+                              $notice_id, $user_id));
+    }
+
+    static function bulkInsert($notice_id, $user_ids)
+    {
+        $cnt = count($user_ids);
+
+        for ($off = 0; $off < $cnt; $off += self::BOXCAR) {
+
+            $boxcar = array_slice($user_ids, $off, self::BOXCAR);
+
+            if (empty($boxcar)) { // jump in, hobo!
+                break;
+            }
+
+            $inbox = new Inbox();
+
+            $inbox->query(sprintf('UPDATE inbox '.
+                                  'set notice_ids = concat(cast(%08x as binary(4)), '.
+                                  'substr(notice_ids, 1, 4092)) '.
+                                  'WHERE user_id in (%s)',
+                                  $notice_id, implode(',', $boxcar)));
+
+            $inbox->free();
+        }
+    }
+
+    function stream($user_id, $offset, $limit, $since_id, $max_id, $since, $own=false)
+    {
+        $inbox = Inbox::staticGet('user_id', $user_id);
+
+        if (empty($inbox)) {
+            return array();
+        }
+
+        $ids = unpack('L*', $inbox->notice_ids);
+
+        // XXX: handle since_id
+        // XXX: handle max_id
+
+        $ids = array_slice($ids, $offset, $limit);
+    }
 }
diff --git a/classes/Notice.php b/classes/Notice.php
index 9bda47827..8783912e8 100644
--- a/classes/Notice.php
+++ b/classes/Notice.php
@@ -125,8 +125,7 @@ class Notice extends Memcached_DataObject
                          'Fave',
                          'Notice_tag',
                          'Group_inbox',
-                         'Queue_item',
-                         'Notice_inbox');
+                         'Queue_item');
 
         foreach ($related as $cls) {
             $inst = new $cls();
@@ -504,17 +503,6 @@ class Notice extends Memcached_DataObject
                     unset($original);
                 }
 
-                $ni = new Notice_inbox();
-
-                $ni->notice_id = $this->id;
-
-                if ($ni->find()) {
-                    while ($ni->fetch()) {
-                        $tmk = common_cache_key('user:repeated_to_me:'.$ni->user_id);
-                        $cache->delete($tmk);
-                    }
-                }
-
                 $ni->free();
                 unset($ni);
             }
@@ -844,10 +832,6 @@ class Notice extends Memcached_DataObject
 
     function addToInboxes()
     {
-        // XXX: loads constants
-
-        $inbox = new Notice_inbox();
-
         $users = $this->getSubscribedUsers();
 
         // FIXME: kind of ignoring 'transitional'...
@@ -887,7 +871,7 @@ class Notice extends Memcached_DataObject
             }
         }
 
-        Notice_inbox::bulkInsert($this->id, $this->created, $ni);
+        Inbox::bulkInsert($this->id, array_keys($ni));
 
         return;
     }
diff --git a/classes/User.php b/classes/User.php
index 34151778c..773723da4 100644
--- a/classes/User.php
+++ b/classes/User.php
@@ -291,6 +291,19 @@ class User extends Memcached_DataObject
             return false;
         }
 
+        // Everyone gets an inbox
+
+        $inbox = new Inbox();
+
+        $inbox->user_id = $user->id;
+
+        $result = $inbox->insert();
+
+        if (!$result) {
+            common_log_db_error($inbox, 'INSERT', __FILE__);
+            return false;
+        }
+
         // Everyone is subscribed to themself
 
         $subscription = new Subscription();
@@ -482,89 +495,30 @@ class User extends Memcached_DataObject
 
     function noticesWithFriends($offset=0, $limit=NOTICES_PER_PAGE, $since_id=0, $before_id=0, $since=null)
     {
-        $ids = Notice_inbox::stream($this->id, $offset, $limit, $since_id, $before_id, $since, false);
-
+        $ids = Inbox::stream($this->id, $offset, $limit, $since_id, $before_id, $since, false);
         return Notice::getStreamByIds($ids);
     }
 
     function noticeInbox($offset=0, $limit=NOTICES_PER_PAGE, $since_id=0, $before_id=0, $since=null)
     {
-        $ids = Notice_inbox::stream($this->id, $offset, $limit, $since_id, $before_id, $since, true);
-
+        $ids = Inbox::stream($this->id, $offset, $limit, $since_id, $before_id, $since, true);
         return Notice::getStreamByIds($ids);
     }
 
     function friendsTimeline($offset=0, $limit=NOTICES_PER_PAGE, $since_id=0, $before_id=0, $since=null)
     {
-        $ids = Notice::stream(array($this, '_friendsTimelineDirect'),
-                              array(false),
-                              'user:friends_timeline:'.$this->id,
-                              $offset, $limit, $since_id, $before_id, $since);
+        $ids = Inbox::stream($this->id, $offset, $limit, $since_id, $before_id, $since, false);
 
         return Notice::getStreamByIds($ids);
     }
 
     function ownFriendsTimeline($offset=0, $limit=NOTICES_PER_PAGE, $since_id=0, $before_id=0, $since=null)
     {
-        $ids = Notice::stream(array($this, '_friendsTimelineDirect'),
-                              array(true),
-                              'user:friends_timeline_own:'.$this->id,
-                              $offset, $limit, $since_id, $before_id, $since);
+        $ids = Inbox::stream($this->id, $offset, $limit, $since_id, $before_id, $since, true);
 
         return Notice::getStreamByIds($ids);
     }
 
-    function _friendsTimelineDirect($own, $offset, $limit, $since_id, $max_id, $since)
-    {
-        $qry =
-          'SELECT notice.id AS id ' .
-          'FROM notice JOIN notice_inbox ON notice.id = notice_inbox.notice_id ' .
-          'WHERE notice_inbox.user_id = ' . $this->id . ' ' .
-          'AND notice.repeat_of IS NULL ';
-
-        if (!$own) {
-            // XXX: autoload notice inbox for constant
-            $inbox = new Notice_inbox();
-
-            $qry .= 'AND notice_inbox.source != ' . NOTICE_INBOX_SOURCE_GATEWAY . ' ';
-        }
-
-        if ($since_id != 0) {
-            $qry .= 'AND notice.id > ' . $since_id . ' ';
-        }
-
-        if ($max_id != 0) {
-            $qry .= 'AND notice.id <= ' . $max_id . ' ';
-        }
-
-        if (!is_null($since)) {
-            $qry .= 'AND notice.modified > \'' . date('Y-m-d H:i:s', $since) . '\' ';
-        }
-
-        // NOTE: we sort by fave time, not by notice time!
-
-        $qry .= 'ORDER BY notice_id DESC ';
-
-        if (!is_null($offset)) {
-            $qry .= "LIMIT $limit OFFSET $offset";
-        }
-
-        $ids = array();
-
-        $notice = new Notice();
-
-        $notice->query($qry);
-
-        while ($notice->fetch()) {
-            $ids[] = $notice->id;
-        }
-
-        $notice->free();
-        $notice = NULL;
-
-        return $ids;
-    }
-
     function blowFavesCache()
     {
         $cache = common_memcache();
@@ -777,7 +731,6 @@ class User extends Memcached_DataObject
                          'Remember_me',
                          'Foreign_link',
                          'Invitation',
-                         'Notice_inbox',
                          );
         Event::handle('UserDeleteRelated', array($this, &$related));
 
-- 
cgit v1.2.3-54-g00ecf


From 1e7ec69190e089f89b57181d561e2df2c8bdd3b5 Mon Sep 17 00:00:00 2001
From: Evan Prodromou <evan@status.net>
Date: Thu, 31 Dec 2009 09:09:07 -1000
Subject: some formatting changes to make inblobs work

---
 classes/Inbox.php | 8 +++++---
 classes/User.php  | 1 +
 2 files changed, 6 insertions(+), 3 deletions(-)

(limited to 'classes')

diff --git a/classes/Inbox.php b/classes/Inbox.php
index de48d7381..610b5fceb 100644
--- a/classes/Inbox.php
+++ b/classes/Inbox.php
@@ -56,7 +56,7 @@ class Inbox extends Memcached_DataObject
         $inbox = new Inbox();
 
         $inbox->query(sprintf('UPDATE inbox '.
-                              'set notice_ids = concat(cast(%08x as binary(4)), '.
+                              'set notice_ids = concat(cast(0x%08x as binary(4)), '.
                               'substr(notice_ids, 1, 4092)) '.
                               'WHERE user_id = %d',
                               $notice_id, $user_id));
@@ -77,7 +77,7 @@ class Inbox extends Memcached_DataObject
             $inbox = new Inbox();
 
             $inbox->query(sprintf('UPDATE inbox '.
-                                  'set notice_ids = concat(cast(%08x as binary(4)), '.
+                                  'set notice_ids = concat(cast(0x%08x as binary(4)), '.
                                   'substr(notice_ids, 1, 4092)) '.
                                   'WHERE user_id in (%s)',
                                   $notice_id, implode(',', $boxcar)));
@@ -94,11 +94,13 @@ class Inbox extends Memcached_DataObject
             return array();
         }
 
-        $ids = unpack('L*', $inbox->notice_ids);
+        $ids = unpack('N*', $inbox->notice_ids);
 
         // XXX: handle since_id
         // XXX: handle max_id
 
         $ids = array_slice($ids, $offset, $limit);
+
+        return $ids;
     }
 }
diff --git a/classes/User.php b/classes/User.php
index 773723da4..bde3f71b9 100644
--- a/classes/User.php
+++ b/classes/User.php
@@ -296,6 +296,7 @@ class User extends Memcached_DataObject
         $inbox = new Inbox();
 
         $inbox->user_id = $user->id;
+        $inbox->notice_ids = '';
 
         $result = $inbox->insert();
 
-- 
cgit v1.2.3-54-g00ecf


From 72934e9f50ee8970372b39b3ff27dc4ded30e0c5 Mon Sep 17 00:00:00 2001
From: Evan Prodromou <evan@status.net>
Date: Sat, 9 Jan 2010 09:31:57 -0800
Subject: Revert "Replace Notice_inbox with Inbox"

We use Notice_inbox to transition to Inbox.

This reverts commit 7640d3f07bad0710d69575efc7ceda115f24a60a.
---
 classes/Notice_inbox.php | 184 +++++++++++++++++++++++++++++++++++++++++++++++
 scripts/inbox_users.php  | 107 +++++++++++++++++++++++++++
 scripts/triminboxes.php  |  56 +++++++++++++++
 3 files changed, 347 insertions(+)
 create mode 100644 classes/Notice_inbox.php
 create mode 100755 scripts/inbox_users.php
 create mode 100644 scripts/triminboxes.php

(limited to 'classes')

diff --git a/classes/Notice_inbox.php b/classes/Notice_inbox.php
new file mode 100644
index 000000000..b39006542
--- /dev/null
+++ b/classes/Notice_inbox.php
@@ -0,0 +1,184 @@
+<?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); }
+
+require_once INSTALLDIR.'/classes/Memcached_DataObject.php';
+
+// We keep 5 pages of inbox notices in memcache, +1 for pagination check
+
+define('INBOX_CACHE_WINDOW', 101);
+define('NOTICE_INBOX_GC_BOXCAR', 128);
+define('NOTICE_INBOX_GC_MAX', 12800);
+define('NOTICE_INBOX_LIMIT', 1000);
+define('NOTICE_INBOX_SOFT_LIMIT', 1000);
+
+define('NOTICE_INBOX_SOURCE_SUB', 1);
+define('NOTICE_INBOX_SOURCE_GROUP', 2);
+define('NOTICE_INBOX_SOURCE_REPLY', 3);
+define('NOTICE_INBOX_SOURCE_FORWARD', 4);
+define('NOTICE_INBOX_SOURCE_GATEWAY', -1);
+
+class Notice_inbox extends Memcached_DataObject
+{
+    ###START_AUTOCODE
+    /* the code below is auto generated do not remove the above tag */
+
+    public $__table = 'notice_inbox';                    // table name
+    public $user_id;                         // int(4)  primary_key not_null
+    public $notice_id;                       // int(4)  primary_key not_null
+    public $created;                         // datetime()   not_null
+    public $source;                          // tinyint(1)   default_1
+
+    /* Static get */
+    function staticGet($k,$v=null)
+    { return Memcached_DataObject::staticGet('Notice_inbox',$k,$v); }
+
+    /* the code above is auto generated do not remove the tag below */
+    ###END_AUTOCODE
+
+    function stream($user_id, $offset, $limit, $since_id, $max_id, $since, $own=false)
+    {
+        return Notice::stream(array('Notice_inbox', '_streamDirect'),
+                              array($user_id, $own),
+                              ($own) ? 'notice_inbox:by_user:'.$user_id :
+                              'notice_inbox:by_user_own:'.$user_id,
+                              $offset, $limit, $since_id, $max_id, $since);
+    }
+
+    function _streamDirect($user_id, $own, $offset, $limit, $since_id, $max_id, $since)
+    {
+        $inbox = new Notice_inbox();
+
+        $inbox->user_id = $user_id;
+
+        if (!$own) {
+            $inbox->whereAdd('source != ' . NOTICE_INBOX_SOURCE_GATEWAY);
+        }
+
+        if ($since_id != 0) {
+            $inbox->whereAdd('notice_id > ' . $since_id);
+        }
+
+        if ($max_id != 0) {
+            $inbox->whereAdd('notice_id <= ' . $max_id);
+        }
+
+        if (!is_null($since)) {
+            $inbox->whereAdd('created > \'' . date('Y-m-d H:i:s', $since) . '\'');
+        }
+
+        $inbox->orderBy('created DESC');
+
+        if (!is_null($offset)) {
+            $inbox->limit($offset, $limit);
+        }
+
+        $ids = array();
+
+        if ($inbox->find()) {
+            while ($inbox->fetch()) {
+                $ids[] = $inbox->notice_id;
+            }
+        }
+
+        return $ids;
+    }
+
+    function &pkeyGet($kv)
+    {
+        return Memcached_DataObject::pkeyGet('Notice_inbox', $kv);
+    }
+
+    static function gc($user_id)
+    {
+        $entry = new Notice_inbox();
+        $entry->user_id = $user_id;
+        $entry->orderBy('created DESC');
+        $entry->limit(NOTICE_INBOX_LIMIT - 1, NOTICE_INBOX_GC_MAX);
+
+        $total = $entry->find();
+
+        if ($total > 0) {
+            $notices = array();
+            $cnt = 0;
+            while ($entry->fetch()) {
+                $notices[] = $entry->notice_id;
+                $cnt++;
+                if ($cnt >= NOTICE_INBOX_GC_BOXCAR) {
+                    self::deleteMatching($user_id, $notices);
+                    $notices = array();
+                    $cnt = 0;
+                }
+            }
+
+            if ($cnt > 0) {
+                self::deleteMatching($user_id, $notices);
+                $notices = array();
+            }
+        }
+    }
+
+    static function deleteMatching($user_id, $notices)
+    {
+        $entry = new Notice_inbox();
+        return $entry->query('DELETE FROM notice_inbox '.
+                             'WHERE user_id = ' . $user_id . ' ' .
+                             'AND notice_id in ('.implode(',', $notices).')');
+    }
+
+    static function bulkInsert($notice_id, $created, $ni)
+    {
+        $cnt = 0;
+
+        $qryhdr = 'INSERT INTO notice_inbox (user_id, notice_id, source, created) VALUES ';
+        $qry = $qryhdr;
+
+        foreach ($ni as $id => $source) {
+            if ($cnt > 0) {
+                $qry .= ', ';
+            }
+            $qry .= '('.$id.', '.$notice_id.', '.$source.", '".$created. "') ";
+            $cnt++;
+            if (rand() % NOTICE_INBOX_SOFT_LIMIT == 0) {
+                // FIXME: Causes lag in replicated servers
+                // Notice_inbox::gc($id);
+            }
+            if ($cnt >= MAX_BOXCARS) {
+                $inbox = new Notice_inbox();
+                $result = $inbox->query($qry);
+                if (PEAR::isError($result)) {
+                    common_log_db_error($inbox, $qry);
+                }
+                $qry = $qryhdr;
+                $cnt = 0;
+            }
+        }
+
+        if ($cnt > 0) {
+            $inbox = new Notice_inbox();
+            $result = $inbox->query($qry);
+            if (PEAR::isError($result)) {
+                common_log_db_error($inbox, $qry);
+            }
+        }
+
+        return;
+    }
+}
diff --git a/scripts/inbox_users.php b/scripts/inbox_users.php
new file mode 100755
index 000000000..32adcea21
--- /dev/null
+++ b/scripts/inbox_users.php
@@ -0,0 +1,107 @@
+#!/usr/bin/env php
+<?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/>.
+ */
+
+# Abort if called from a web server
+
+define('INSTALLDIR', realpath(dirname(__FILE__) . '/..'));
+
+$helptext = <<<ENDOFHELP
+inbox_users.php <idfile>
+
+Update users to use inbox table. Listed in an ID file, default 'ids.txt'.
+
+ENDOFHELP;
+
+require_once INSTALLDIR.'/scripts/commandline.inc';
+
+$id_file = (count($args) > 1) ? $args[0] : 'ids.txt';
+
+common_log(LOG_INFO, 'Updating user inboxes.');
+
+$ids = file($id_file);
+
+foreach ($ids as $id) {
+
+	$user = User::staticGet('id', $id);
+
+	if (!$user) {
+		common_log(LOG_WARNING, 'No such user: ' . $id);
+		continue;
+	}
+
+	if ($user->inboxed) {
+		common_log(LOG_WARNING, 'Already inboxed: ' . $id);
+		continue;
+	}
+
+    common_log(LOG_INFO, 'Updating inbox for user ' . $user->id);
+
+	$user->query('BEGIN');
+
+	$old_inbox = new Notice_inbox();
+	$old_inbox->user_id = $user->id;
+
+	$result = $old_inbox->delete();
+
+	if (is_null($result) || $result === false) {
+		common_log_db_error($old_inbox, 'DELETE', __FILE__);
+		continue;
+	}
+
+	$old_inbox->free();
+
+	$inbox = new Notice_inbox();
+
+	$result = $inbox->query('INSERT INTO notice_inbox (user_id, notice_id, created) ' .
+							'SELECT ' . $user->id . ', notice.id, notice.created ' .
+							'FROM subscription JOIN notice ON subscription.subscribed = notice.profile_id ' .
+							'WHERE subscription.subscriber = ' . $user->id . ' ' .
+							'AND notice.created >= subscription.created ' .
+							'AND NOT EXISTS (SELECT user_id, notice_id ' .
+							'FROM notice_inbox ' .
+							'WHERE user_id = ' . $user->id . ' ' .
+							'AND notice_id = notice.id) ' .
+				                        'ORDER BY notice.created DESC ' .
+							'LIMIT 0, 1000');
+
+	if (is_null($result) || $result === false) {
+		common_log_db_error($inbox, 'INSERT', __FILE__);
+		continue;
+	}
+
+	$orig = clone($user);
+	$user->inboxed = 1;
+	$result = $user->update($orig);
+
+	if (!$result) {
+		common_log_db_error($user, 'UPDATE', __FILE__);
+		continue;
+	}
+
+	$user->query('COMMIT');
+
+	$inbox->free();
+	unset($inbox);
+
+	if ($cache) {
+		$cache->delete(common_cache_key('user:notices_with_friends:' . $user->id));
+		$cache->delete(common_cache_key('user:notices_with_friends:' . $user->id . ';last'));
+	}
+}
diff --git a/scripts/triminboxes.php b/scripts/triminboxes.php
new file mode 100644
index 000000000..da09817e5
--- /dev/null
+++ b/scripts/triminboxes.php
@@ -0,0 +1,56 @@
+#!/usr/bin/env php
+<?php
+/*
+ * StatusNet - the distributed open-source microblogging tool
+ * Copyright (C) 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/>.
+ */
+
+define('INSTALLDIR', realpath(dirname(__FILE__) . '/..'));
+
+$shortoptions = 'u::';
+$longoptions = array('start-user-id::');
+
+$helptext = <<<END_OF_TRIM_HELP
+Batch script for trimming notice inboxes to a reasonable size.
+
+    -u <id>
+    --start-user-id=<id>   User ID to start after. Default is all.
+
+END_OF_TRIM_HELP;
+
+require_once INSTALLDIR.'/scripts/commandline.inc';
+
+$id = null;
+
+if (have_option('u')) {
+    $id = get_option_value('u');
+} else if (have_option('--start-user-id')) {
+    $id = get_option_value('--start-user-id');
+} else {
+    $id = null;
+}
+
+$user = new User();
+
+if (!empty($id)) {
+    $user->whereAdd('id > ' . $id);
+}
+
+$cnt = $user->find();
+
+while ($user->fetch()) {
+    Notice_inbox::gc($user->id);
+}
-- 
cgit v1.2.3-54-g00ecf


From 1de0f37f0d6e4706fe9f1ce086df730ad4a05b78 Mon Sep 17 00:00:00 2001
From: Evan Prodromou <evan@status.net>
Date: Sat, 9 Jan 2010 09:59:58 -0800
Subject: disable all Notice_inbox functions

---
 classes/Notice_inbox.php | 115 ++++-------------------------------------------
 1 file changed, 9 insertions(+), 106 deletions(-)

(limited to 'classes')

diff --git a/classes/Notice_inbox.php b/classes/Notice_inbox.php
index b39006542..6c328e685 100644
--- a/classes/Notice_inbox.php
+++ b/classes/Notice_inbox.php
@@ -1,7 +1,7 @@
 <?php
 /*
  * StatusNet - the distributed open-source microblogging tool
- * Copyright (C) 2008, 2009, StatusNet, Inc.
+ * 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
@@ -17,7 +17,9 @@
  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  */
 
-if (!defined('STATUSNET') && !defined('LACONICA')) { exit(1); }
+if (!defined('STATUSNET')) {
+    exit(1);
+}
 
 require_once INSTALLDIR.'/classes/Memcached_DataObject.php';
 
@@ -55,50 +57,12 @@ class Notice_inbox extends Memcached_DataObject
 
     function stream($user_id, $offset, $limit, $since_id, $max_id, $since, $own=false)
     {
-        return Notice::stream(array('Notice_inbox', '_streamDirect'),
-                              array($user_id, $own),
-                              ($own) ? 'notice_inbox:by_user:'.$user_id :
-                              'notice_inbox:by_user_own:'.$user_id,
-                              $offset, $limit, $since_id, $max_id, $since);
+        throw new Exception('Notice_inbox no longer used; use Inbox');
     }
 
     function _streamDirect($user_id, $own, $offset, $limit, $since_id, $max_id, $since)
     {
-        $inbox = new Notice_inbox();
-
-        $inbox->user_id = $user_id;
-
-        if (!$own) {
-            $inbox->whereAdd('source != ' . NOTICE_INBOX_SOURCE_GATEWAY);
-        }
-
-        if ($since_id != 0) {
-            $inbox->whereAdd('notice_id > ' . $since_id);
-        }
-
-        if ($max_id != 0) {
-            $inbox->whereAdd('notice_id <= ' . $max_id);
-        }
-
-        if (!is_null($since)) {
-            $inbox->whereAdd('created > \'' . date('Y-m-d H:i:s', $since) . '\'');
-        }
-
-        $inbox->orderBy('created DESC');
-
-        if (!is_null($offset)) {
-            $inbox->limit($offset, $limit);
-        }
-
-        $ids = array();
-
-        if ($inbox->find()) {
-            while ($inbox->fetch()) {
-                $ids[] = $inbox->notice_id;
-            }
-        }
-
-        return $ids;
+        throw new Exception('Notice_inbox no longer used; use Inbox');
     }
 
     function &pkeyGet($kv)
@@ -108,77 +72,16 @@ class Notice_inbox extends Memcached_DataObject
 
     static function gc($user_id)
     {
-        $entry = new Notice_inbox();
-        $entry->user_id = $user_id;
-        $entry->orderBy('created DESC');
-        $entry->limit(NOTICE_INBOX_LIMIT - 1, NOTICE_INBOX_GC_MAX);
-
-        $total = $entry->find();
-
-        if ($total > 0) {
-            $notices = array();
-            $cnt = 0;
-            while ($entry->fetch()) {
-                $notices[] = $entry->notice_id;
-                $cnt++;
-                if ($cnt >= NOTICE_INBOX_GC_BOXCAR) {
-                    self::deleteMatching($user_id, $notices);
-                    $notices = array();
-                    $cnt = 0;
-                }
-            }
-
-            if ($cnt > 0) {
-                self::deleteMatching($user_id, $notices);
-                $notices = array();
-            }
-        }
+        throw new Exception('Notice_inbox no longer used; use Inbox');
     }
 
     static function deleteMatching($user_id, $notices)
     {
-        $entry = new Notice_inbox();
-        return $entry->query('DELETE FROM notice_inbox '.
-                             'WHERE user_id = ' . $user_id . ' ' .
-                             'AND notice_id in ('.implode(',', $notices).')');
+        throw new Exception('Notice_inbox no longer used; use Inbox');
     }
 
     static function bulkInsert($notice_id, $created, $ni)
     {
-        $cnt = 0;
-
-        $qryhdr = 'INSERT INTO notice_inbox (user_id, notice_id, source, created) VALUES ';
-        $qry = $qryhdr;
-
-        foreach ($ni as $id => $source) {
-            if ($cnt > 0) {
-                $qry .= ', ';
-            }
-            $qry .= '('.$id.', '.$notice_id.', '.$source.", '".$created. "') ";
-            $cnt++;
-            if (rand() % NOTICE_INBOX_SOFT_LIMIT == 0) {
-                // FIXME: Causes lag in replicated servers
-                // Notice_inbox::gc($id);
-            }
-            if ($cnt >= MAX_BOXCARS) {
-                $inbox = new Notice_inbox();
-                $result = $inbox->query($qry);
-                if (PEAR::isError($result)) {
-                    common_log_db_error($inbox, $qry);
-                }
-                $qry = $qryhdr;
-                $cnt = 0;
-            }
-        }
-
-        if ($cnt > 0) {
-            $inbox = new Notice_inbox();
-            $result = $inbox->query($qry);
-            if (PEAR::isError($result)) {
-                common_log_db_error($inbox, $qry);
-            }
-        }
-
-        return;
+        throw new Exception('Notice_inbox no longer used; use Inbox');
     }
 }
-- 
cgit v1.2.3-54-g00ecf


From 920c878221cf87c1cf74baabbf0636337550eeab Mon Sep 17 00:00:00 2001
From: Evan Prodromou <evan@status.net>
Date: Sat, 9 Jan 2010 10:01:07 -0800
Subject: initialize an Inbox from Notice_inbox records

---
 classes/Inbox.php | 75 +++++++++++++++++++++++++++++++++++++++++++++++++------
 1 file changed, 68 insertions(+), 7 deletions(-)

(limited to 'classes')

diff --git a/classes/Inbox.php b/classes/Inbox.php
index 610b5fceb..de52e1e0c 100644
--- a/classes/Inbox.php
+++ b/classes/Inbox.php
@@ -51,15 +51,73 @@ class Inbox extends Memcached_DataObject
         return array(false, false, false);
     }
 
-    static function insertNotice($user_id, $notice_id)
+    /**
+     * Create a new inbox from existing Notice_inbox stuff
+     */
+
+    static function initialize($user_id)
     {
+        $ids = array();
+
+        $ni = new Notice_inbox();
+
+        $ni->user_id = $user_id;
+        $ni->selectAdd();
+        $ni->selectAdd('notice_id');
+        $ni->orderBy('notice_id DESC');
+        $ni->limit(0, 1024);
+
+        if ($ni->find()) {
+            while($ni->fetch()) {
+                $ids[] = $ni->notice_id;
+            }
+        }
+
+        $ni->free();
+        unset($ni);
+
         $inbox = new Inbox();
 
-        $inbox->query(sprintf('UPDATE inbox '.
-                              'set notice_ids = concat(cast(0x%08x as binary(4)), '.
-                              'substr(notice_ids, 1, 4092)) '.
-                              'WHERE user_id = %d',
-                              $notice_id, $user_id));
+        $inbox->user_id = $user_id;
+        $inbox->notice_ids = pack('N*', $ids);
+
+        $result = $inbox->insert();
+
+        if (!$result) {
+            common_log_db_error($inbox, 'INSERT', __FILE__);
+            return null;
+        }
+
+        return $inbox;
+    }
+
+    static function insertNotice($user_id, $notice_id)
+    {
+        $inbox = Inbox::staticGet('user_id', $user_id);
+
+        if (empty($inbox)) {
+            $inbox = Inbox::initialize($user_id);
+        }
+
+        if (empty($inbox)) {
+            return false;
+        }
+
+        $result = $inbox->query(sprintf('UPDATE inbox '.
+                                        'set notice_ids = concat(cast(0x%08x as binary(4)), '.
+                                        'substr(notice_ids, 1, 4092)) '.
+                                        'WHERE user_id = %d',
+                                        $notice_id, $user_id));
+
+        if ($result) {
+            $c = $this->memcache();
+
+            if (!empty($c)) {
+                $c->delete($this->cacheKey($this->tableName(), 'user_id', $user_id));
+            }
+        }
+
+        return $result;
     }
 
     static function bulkInsert($notice_id, $user_ids)
@@ -91,7 +149,10 @@ class Inbox extends Memcached_DataObject
         $inbox = Inbox::staticGet('user_id', $user_id);
 
         if (empty($inbox)) {
-            return array();
+            $inbox = Inbox::initialize($user_id);
+            if (empty($inbox)) {
+                return array();
+            }
         }
 
         $ids = unpack('N*', $inbox->notice_ids);
-- 
cgit v1.2.3-54-g00ecf


From 5e81149e4773799b2dcf426987e6263b8841c5cc Mon Sep 17 00:00:00 2001
From: Evan Prodromou <evan@status.net>
Date: Sat, 9 Jan 2010 10:26:50 -0800
Subject: create new Inbox from Notice_inbox if not exists at read/write time

---
 classes/Inbox.php | 28 ++++++----------------------
 1 file changed, 6 insertions(+), 22 deletions(-)

(limited to 'classes')

diff --git a/classes/Inbox.php b/classes/Inbox.php
index de52e1e0c..e14d4f4e7 100644
--- a/classes/Inbox.php
+++ b/classes/Inbox.php
@@ -79,7 +79,7 @@ class Inbox extends Memcached_DataObject
         $inbox = new Inbox();
 
         $inbox->user_id = $user_id;
-        $inbox->notice_ids = pack('N*', $ids);
+        $inbox->notice_ids = call_user_func_array('pack', array_merge(array('N*'), $ids));
 
         $result = $inbox->insert();
 
@@ -110,10 +110,10 @@ class Inbox extends Memcached_DataObject
                                         $notice_id, $user_id));
 
         if ($result) {
-            $c = $this->memcache();
+            $c = self::memcache();
 
             if (!empty($c)) {
-                $c->delete($this->cacheKey($this->tableName(), 'user_id', $user_id));
+                $c->delete(self::cacheKey('inbox', 'user_id', $user_id));
             }
         }
 
@@ -122,25 +122,9 @@ class Inbox extends Memcached_DataObject
 
     static function bulkInsert($notice_id, $user_ids)
     {
-        $cnt = count($user_ids);
-
-        for ($off = 0; $off < $cnt; $off += self::BOXCAR) {
-
-            $boxcar = array_slice($user_ids, $off, self::BOXCAR);
-
-            if (empty($boxcar)) { // jump in, hobo!
-                break;
-            }
-
-            $inbox = new Inbox();
-
-            $inbox->query(sprintf('UPDATE inbox '.
-                                  'set notice_ids = concat(cast(0x%08x as binary(4)), '.
-                                  'substr(notice_ids, 1, 4092)) '.
-                                  'WHERE user_id in (%s)',
-                                  $notice_id, implode(',', $boxcar)));
-
-            $inbox->free();
+        foreach ($user_ids as $user_id)
+        {
+            Inbox::insertNotice($user_id, $notice_id);
         }
     }
 
-- 
cgit v1.2.3-54-g00ecf


From 96e51dad4b9ff2208cfef2e578e4bc85a8937680 Mon Sep 17 00:00:00 2001
From: Evan Prodromou <evan@status.net>
Date: Sat, 9 Jan 2010 13:55:54 -0800
Subject: whoGets() method for Notice

---
 classes/Notice.php | 9 ++++++++-
 1 file changed, 8 insertions(+), 1 deletion(-)

(limited to 'classes')

diff --git a/classes/Notice.php b/classes/Notice.php
index 8783912e8..6284b8ca5 100644
--- a/classes/Notice.php
+++ b/classes/Notice.php
@@ -830,7 +830,7 @@ class Notice extends Memcached_DataObject
         return $ids;
     }
 
-    function addToInboxes()
+    function whoGets()
     {
         $users = $this->getSubscribedUsers();
 
@@ -871,6 +871,13 @@ class Notice extends Memcached_DataObject
             }
         }
 
+        return $ni;
+    }
+
+    function addToInboxes()
+    {
+        $ni = $this->whoGets();
+
         Inbox::bulkInsert($this->id, array_keys($ni));
 
         return;
-- 
cgit v1.2.3-54-g00ecf


From 4cc9b183d797cbd3d1f6230a46cb5df6e6161dbb Mon Sep 17 00:00:00 2001
From: Brion Vibber <brion@pobox.com>
Date: Tue, 12 Jan 2010 12:20:45 -0800
Subject: Avoid spewing warnings about returning by reference on memcached
 cache misses by setting a var and returning it instead of trying to return
 false directly. This hacky workaround is brought to you by DB_DataObject's
 PHP 4 roots.

---
 classes/Memcached_DataObject.php | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

(limited to 'classes')

diff --git a/classes/Memcached_DataObject.php b/classes/Memcached_DataObject.php
index b68a4af8e..9c9994cea 100644
--- a/classes/Memcached_DataObject.php
+++ b/classes/Memcached_DataObject.php
@@ -98,14 +98,16 @@ class Memcached_DataObject extends DB_DataObject
         } else {
             $i = DB_DataObject::factory($cls);
             if (empty($i)) {
-                return false;
+                $i = false;
+                return $i;
             }
             $result = $i->get($k, $v);
             if ($result) {
                 $i->encache();
                 return $i;
             } else {
-                return false;
+                $i = false;
+                return $i;
             }
         }
     }
-- 
cgit v1.2.3-54-g00ecf


From ec145b73fc91dd54695dd374c8a71a11e233b8c0 Mon Sep 17 00:00:00 2001
From: Brion Vibber <brion@status.net>
Date: Tue, 12 Jan 2010 19:57:15 -0800
Subject: Major refactoring of queue handlers to support running multiple sites
 in one daemon.

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

The new QueueDaemon runs from scripts/queuedaemon.php:

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

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

Todo:

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

Detailed code history available in my daemon-work branch:
http://www.gitorious.org/~brion/statusnet/brion-fixes/commits/daemon-work
---
 classes/Memcached_DataObject.php                   |  23 ++
 classes/Queue_item.php                             |   9 +-
 classes/Status_network.php                         |  23 +-
 lib/cache.php                                      |  19 ++
 lib/common.php                                     | 194 +----------
 lib/dbqueuemanager.php                             | 161 +++++----
 lib/default.php                                    |   2 +
 lib/event.php                                      |   8 +
 lib/iomanager.php                                  | 193 +++++++++++
 lib/iomaster.php                                   | 361 +++++++++++++++++++++
 lib/jabber.php                                     |  25 +-
 lib/jabberqueuehandler.php                         |  47 +++
 lib/liberalstomp.php                               | 133 ++++++++
 lib/ombqueuehandler.php                            |  56 ++++
 lib/pingqueuehandler.php                           |  37 +++
 lib/pluginqueuehandler.php                         |  50 +++
 lib/publicqueuehandler.php                         |  47 +++
 lib/queuehandler.php                               | 100 ++----
 lib/queuemanager.php                               | 149 ++++++++-
 lib/queuemonitor.php                               | 116 +++++++
 lib/smsqueuehandler.php                            |  39 +++
 lib/statusnet.php                                  | 295 +++++++++++++++++
 lib/stompqueuemanager.php                          | 286 ++++++++++++----
 lib/unqueuemanager.php                             |  58 +---
 lib/util.php                                       |   5 +-
 lib/xmppconfirmmanager.php                         | 168 ++++++++++
 lib/xmppmanager.php                                | 273 ++++++++++++++++
 lib/xmppqueuehandler.php                           | 142 --------
 plugins/Enjit/README                               |   5 +
 plugins/Enjit/enjitqueuehandler.php                |  91 ++++++
 plugins/Facebook/FacebookPlugin.php                |  46 +--
 plugins/Facebook/facebookqueuehandler.php          |  52 +--
 plugins/MemcachePlugin.php                         |  17 +
 plugins/TwitterBridge/TwitterBridgePlugin.php      |  50 +--
 .../TwitterBridge/daemons/twitterqueuehandler.php  |  73 -----
 plugins/TwitterBridge/twitterqueuehandler.php      |  35 ++
 scripts/enjitqueuehandler.php                      | 121 -------
 scripts/getvaliddaemons.php                        |  11 +-
 scripts/handlequeued.php                           |  56 ++++
 scripts/jabberqueuehandler.php                     |  78 -----
 scripts/ombqueuehandler.php                        |  87 -----
 scripts/pingqueuehandler.php                       |  69 ----
 scripts/pluginqueuehandler.php                     |  64 ----
 scripts/publicqueuehandler.php                     |  76 -----
 scripts/queuedaemon.php                            | 265 +++++++++++++++
 scripts/smsqueuehandler.php                        |  73 -----
 scripts/xmppconfirmhandler.php                     | 161 ---------
 47 files changed, 2929 insertions(+), 1520 deletions(-)
 create mode 100644 lib/iomanager.php
 create mode 100644 lib/iomaster.php
 create mode 100644 lib/jabberqueuehandler.php
 create mode 100644 lib/liberalstomp.php
 create mode 100644 lib/ombqueuehandler.php
 create mode 100644 lib/pingqueuehandler.php
 create mode 100644 lib/pluginqueuehandler.php
 create mode 100644 lib/publicqueuehandler.php
 create mode 100644 lib/queuemonitor.php
 create mode 100644 lib/smsqueuehandler.php
 create mode 100644 lib/statusnet.php
 create mode 100644 lib/xmppconfirmmanager.php
 create mode 100644 lib/xmppmanager.php
 delete mode 100644 lib/xmppqueuehandler.php
 create mode 100644 plugins/Enjit/README
 create mode 100644 plugins/Enjit/enjitqueuehandler.php
 mode change 100755 => 100644 plugins/Facebook/facebookqueuehandler.php
 delete mode 100755 plugins/TwitterBridge/daemons/twitterqueuehandler.php
 create mode 100644 plugins/TwitterBridge/twitterqueuehandler.php
 delete mode 100755 scripts/enjitqueuehandler.php
 create mode 100755 scripts/handlequeued.php
 delete mode 100755 scripts/jabberqueuehandler.php
 delete mode 100755 scripts/ombqueuehandler.php
 delete mode 100755 scripts/pingqueuehandler.php
 delete mode 100755 scripts/pluginqueuehandler.php
 delete mode 100755 scripts/publicqueuehandler.php
 create mode 100755 scripts/queuedaemon.php
 delete mode 100755 scripts/smsqueuehandler.php
 delete mode 100755 scripts/xmppconfirmhandler.php

(limited to 'classes')

diff --git a/classes/Memcached_DataObject.php b/classes/Memcached_DataObject.php
index 9c9994cea..4ecab9db6 100644
--- a/classes/Memcached_DataObject.php
+++ b/classes/Memcached_DataObject.php
@@ -331,6 +331,29 @@ class Memcached_DataObject extends DB_DataObject
             $exists = false;
        }
 
+        // @fixme horrible evil hack!
+        //
+        // In multisite configuration we don't want to keep around a separate
+        // connection for every database; we could end up with thousands of
+        // connections open per thread. In an ideal world we might keep
+        // a connection per server and select different databases, but that'd
+        // be reliant on having the same db username/pass as well.
+        //
+        // MySQL connections are cheap enough we're going to try just
+        // closing out the old connection and reopening when we encounter
+        // a new DSN.
+        //
+        // WARNING WARNING if we end up actually using multiple DBs at a time
+        // we'll need some fancier logic here.
+        if (!$exists && !empty($_DB_DATAOBJECT['CONNECTIONS'])) {
+            foreach ($_DB_DATAOBJECT['CONNECTIONS'] as $index => $conn) {
+                if (!empty($conn)) {
+                    $conn->disconnect();
+                }
+                unset($_DB_DATAOBJECT['CONNECTIONS'][$index]);
+            }
+        }
+        
         $result = parent::_connect();
 
         if ($result && !$exists) {
diff --git a/classes/Queue_item.php b/classes/Queue_item.php
index 9c673540d..cf805a606 100644
--- a/classes/Queue_item.php
+++ b/classes/Queue_item.php
@@ -25,10 +25,12 @@ class Queue_item extends Memcached_DataObject
     function sequenceKey()
     { return array(false, false); }
 
-    static function top($transport) {
+    static function top($transport=null) {
 
         $qi = new Queue_item();
-        $qi->transport = $transport;
+        if ($transport) {
+            $qi->transport = $transport;
+        }
         $qi->orderBy('created');
         $qi->whereAdd('claimed is null');
 
@@ -40,7 +42,8 @@ class Queue_item extends Memcached_DataObject
             # XXX: potential race condition
             # can we force it to only update if claimed is still null
             # (or old)?
-            common_log(LOG_INFO, 'claiming queue item = ' . $qi->notice_id . ' for transport ' . $transport);
+            common_log(LOG_INFO, 'claiming queue item = ' . $qi->notice_id .
+                ' for transport ' . $qi->transport);
             $orig = clone($qi);
             $qi->claimed = common_sql_now();
             $result = $qi->update($orig);
diff --git a/classes/Status_network.php b/classes/Status_network.php
index 776f6abb0..ef8e1ed43 100644
--- a/classes/Status_network.php
+++ b/classes/Status_network.php
@@ -49,6 +49,13 @@ class Status_network extends DB_DataObject
     static $cache = null;
     static $base = null;
 
+    /**
+     * @param string $dbhost
+     * @param string $dbuser
+     * @param string $dbpass
+     * @param string $dbname
+     * @param array $servers memcached servers to use for caching config info
+     */
     static function setupDB($dbhost, $dbuser, $dbpass, $dbname, $servers)
     {
         global $config;
@@ -60,12 +67,17 @@ class Status_network extends DB_DataObject
         if (class_exists('Memcache')) {
             self::$cache = new Memcache();
 
+            // Can't close persistent connections, making forking painful.
+            //
+            // @fixme only do this in *parent* CLI processes.
+            // single-process and child-processes *should* use persistent.
+            $persist = php_sapi_name() != 'cli';
             if (is_array($servers)) {
                 foreach($servers as $server) {
-                    self::$cache->addServer($server);
+                    self::$cache->addServer($server, 11211, $persist);
                 }
             } else {
-                self::$cache->addServer($servers);
+                self::$cache->addServer($servers, 11211, $persist);
             }
         }
 
@@ -89,7 +101,7 @@ class Status_network extends DB_DataObject
         if (empty($sn)) {
             $sn = self::staticGet($k, $v);
             if (!empty($sn)) {
-                self::$cache->set($ck, $sn);
+                self::$cache->set($ck, clone($sn));
             }
         }
 
@@ -121,6 +133,11 @@ class Status_network extends DB_DataObject
         return parent::delete();
     }
 
+    /**
+     * @param string $servername hostname
+     * @param string $pathname URL base path
+     * @param string $wildcard hostname suffix to match wildcard config
+     */
     static function setupSite($servername, $pathname, $wildcard)
     {
         global $config;
diff --git a/lib/cache.php b/lib/cache.php
index b7b34c050..635c96ad4 100644
--- a/lib/cache.php
+++ b/lib/cache.php
@@ -179,4 +179,23 @@ class Cache
 
         return $success;
     }
+
+    /**
+     * Close or reconnect any remote connections, such as to give
+     * daemon processes a chance to reconnect on a fresh socket.
+     *
+     * @return boolean success flag
+     */
+
+    function reconnect()
+    {
+        $success = false;
+
+        if (Event::handle('StartCacheReconnect', array(&$success))) {
+            $success = true;
+            Event::handle('EndCacheReconnect', array());
+        }
+
+        return $success;
+    }
 }
diff --git a/lib/common.php b/lib/common.php
index b280afec0..61decebb7 100644
--- a/lib/common.php
+++ b/lib/common.php
@@ -76,159 +76,14 @@ require_once(INSTALLDIR.'/lib/language.php');
 require_once(INSTALLDIR.'/lib/event.php');
 require_once(INSTALLDIR.'/lib/plugin.php');
 
-function _sn_to_path($sn)
-{
-    $past_root = substr($sn, 1);
-    $last_slash = strrpos($past_root, '/');
-    if ($last_slash > 0) {
-        $p = substr($past_root, 0, $last_slash);
-    } else {
-        $p = '';
-    }
-    return $p;
-}
-
-// Save our sanity when code gets loaded through subroutines such as PHPUnit tests
-global $default, $config, $_server, $_path;
-
-// try to figure out where we are. $server and $path
-// can be set by including module, else we guess based
-// on HTTP info.
-
-if (isset($server)) {
-    $_server = $server;
-} else {
-    $_server = array_key_exists('SERVER_NAME', $_SERVER) ?
-      strtolower($_SERVER['SERVER_NAME']) :
-    null;
-}
-
-if (isset($path)) {
-    $_path = $path;
-} else {
-    $_path = (array_key_exists('SERVER_NAME', $_SERVER) && array_key_exists('SCRIPT_NAME', $_SERVER)) ?
-      _sn_to_path($_SERVER['SCRIPT_NAME']) :
-    null;
-}
-
-require_once(INSTALLDIR.'/lib/default.php');
-
-// Set config values initially to default values
-
-$config = $default;
-
-// default configuration, overwritten in config.php
-
-$config['db'] = &PEAR::getStaticProperty('DB_DataObject','options');
-
-$config['db'] = $default['db'];
-
-// Backward compatibility
-
-$config['site']['design'] =& $config['design'];
-
-if (function_exists('date_default_timezone_set')) {
-    /* Work internally in UTC */
-    date_default_timezone_set('UTC');
-}
-
 function addPlugin($name, $attrs = null)
 {
-    $name = ucfirst($name);
-    $pluginclass = "{$name}Plugin";
-
-    if (!class_exists($pluginclass)) {
-
-        $files = array("local/plugins/{$pluginclass}.php",
-                       "local/plugins/{$name}/{$pluginclass}.php",
-                       "local/{$pluginclass}.php",
-                       "local/{$name}/{$pluginclass}.php",
-                       "plugins/{$pluginclass}.php",
-                       "plugins/{$name}/{$pluginclass}.php");
-
-        foreach ($files as $file) {
-            $fullpath = INSTALLDIR.'/'.$file;
-            if (@file_exists($fullpath)) {
-                include_once($fullpath);
-                break;
-            }
-        }
-    }
-
-    $inst = new $pluginclass();
-
-    if (!empty($attrs)) {
-        foreach ($attrs as $aname => $avalue) {
-            $inst->$aname = $avalue;
-        }
-    }
-    return $inst;
-}
-
-// From most general to most specific:
-// server-wide, then vhost-wide, then for a path,
-// finally for a dir (usually only need one of the last two).
-
-if (isset($conffile)) {
-    $_config_files = array($conffile);
-} else {
-    $_config_files = array('/etc/statusnet/statusnet.php',
-                           '/etc/statusnet/laconica.php',
-                           '/etc/laconica/laconica.php',
-                           '/etc/statusnet/'.$_server.'.php',
-                           '/etc/laconica/'.$_server.'.php');
-
-    if (strlen($_path) > 0) {
-        $_config_files[] = '/etc/statusnet/'.$_server.'_'.$_path.'.php';
-        $_config_files[] = '/etc/laconica/'.$_server.'_'.$_path.'.php';
-    }
-
-    $_config_files[] = INSTALLDIR.'/config.php';
-}
-
-global $_have_a_config;
-$_have_a_config = false;
-
-foreach ($_config_files as $_config_file) {
-    if (@file_exists($_config_file)) {
-        include_once($_config_file);
-        $_have_a_config = true;
-    }
+    return StatusNet::addPlugin($name, $attrs);
 }
 
 function _have_config()
 {
-    global $_have_a_config;
-    return $_have_a_config;
-}
-
-// XXX: Throw a conniption if database not installed
-// XXX: Find a way to use htmlwriter for this instead of handcoded markup
-if (!_have_config()) {
-  echo '<p>'. _('No configuration file found. ') .'</p>';
-  echo '<p>'. _('I looked for configuration files in the following places: ') .'<br /> '. implode($_config_files, '<br />');
-  echo '<p>'. _('You may wish to run the installer to fix this.') .'</p>';
-  echo '<a href="install.php">'. _('Go to the installer.') .'</a>';
-  exit;
-}
-// Fixup for statusnet.ini
-
-$_db_name = substr($config['db']['database'], strrpos($config['db']['database'], '/') + 1);
-
-if ($_db_name != 'statusnet' && !array_key_exists('ini_'.$_db_name, $config['db'])) {
-    $config['db']['ini_'.$_db_name] = INSTALLDIR.'/classes/statusnet.ini';
-}
-
-// Backwards compatibility
-
-if (array_key_exists('memcached', $config)) {
-    if ($config['memcached']['enabled']) {
-        addPlugin('Memcache', array('servers' => $config['memcached']['server']));
-    }
-
-    if (!empty($config['memcached']['base'])) {
-        $config['cache']['base'] = $config['memcached']['base'];
-    }
+    return StatusNet::haveConfig();
 }
 
 function __autoload($cls)
@@ -247,27 +102,6 @@ function __autoload($cls)
     }
 }
 
-// Load default plugins
-
-foreach ($config['plugins']['default'] as $name => $params) {
-    if (is_null($params)) {
-        addPlugin($name);
-    } else if (is_array($params)) {
-        if (count($params) == 0) {
-            addPlugin($name);
-        } else {
-            $keys = array_keys($params);
-            if (is_string($keys[0])) {
-                addPlugin($name, $params);
-            } else {
-                foreach ($params as $paramset) {
-                    addPlugin($name, $paramset);
-                }
-            }
-        }
-    }
-}
-
 // XXX: how many of these could be auto-loaded on use?
 // XXX: note that these files should not use config options
 // at compile time since DB config options are not yet loaded.
@@ -283,20 +117,20 @@ require_once INSTALLDIR.'/lib/subs.php';
 require_once INSTALLDIR.'/lib/clientexception.php';
 require_once INSTALLDIR.'/lib/serverexception.php';
 
-// Load settings from database; note we need autoload for this
-
-Config::loadSettings();
-
-// XXX: if plugins should check the schema at runtime, do that here.
-
-if ($config['db']['schemacheck'] == 'runtime') {
-    Event::handle('CheckSchema');
+try {
+    StatusNet::init(@$server, @$path, @$conffile);
+} catch (NoConfigException $e) {
+    // XXX: Throw a conniption if database not installed
+    // XXX: Find a way to use htmlwriter for this instead of handcoded markup
+    echo '<p>'. _('No configuration file found. ') .'</p>';
+    echo '<p>'. _('I looked for configuration files in the following places: ') .'<br/> ';
+    echo implode($e->configFiles, '<br/>');
+    echo '<p>'. _('You may wish to run the installer to fix this.') .'</p>';
+    echo '<a href="install.php">'. _('Go to the installer.') .'</a>';
+    exit;
 }
 
+
 // XXX: other formats here
 
 define('NICKNAME_FMT', VALIDATE_NUM.VALIDATE_ALPHA_LOWER);
-
-// Give plugins a chance to initialize in a fully-prepared environment
-
-Event::handle('InitializePlugin');
diff --git a/lib/dbqueuemanager.php b/lib/dbqueuemanager.php
index 750300928..a5c6fd28b 100644
--- a/lib/dbqueuemanager.php
+++ b/lib/dbqueuemanager.php
@@ -22,16 +22,20 @@
  * @category  QueueManager
  * @package   StatusNet
  * @author    Evan Prodromou <evan@status.net>
- * @copyright 2009 StatusNet, Inc.
+ * @author    Brion Vibber <brion@status.net>
+ * @copyright 2009-2010 StatusNet, Inc.
  * @license   http://www.fsf.org/licensing/licenses/agpl-3.0.html GNU Affero General Public License version 3.0
  * @link      http://status.net/
  */
 
 class DBQueueManager extends QueueManager
 {
-    var $qis = array();
-
-    function enqueue($object, $queue)
+    /**
+     * Saves a notice object reference into the queue item table.
+     * @return boolean true on success
+     * @throws ServerException on failure
+     */
+    public function enqueue($object, $queue)
     {
         $notice = $object;
 
@@ -47,70 +51,95 @@ class DBQueueManager extends QueueManager
             throw new ServerException('DB error inserting queue item');
         }
 
+        $this->stats('enqueued', $queue);
+
         return true;
     }
 
-    function service($queue, $handler)
+    /**
+     * Poll every minute 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;
+    }
+
+    /**
+     * Run a polling cycle during idle processing in the input loop.
+     * @return boolean true if we had a hit
+     */
+    public function poll()
     {
-        while (true) {
-            $this->_log(LOG_DEBUG, 'Checking for notices...');
-            $timeout = $handler->timeout();
-            $notice = $this->_nextItem($queue, $timeout);
-            if (empty($notice)) {
-                $this->_log(LOG_DEBUG, 'No notices waiting; idling.');
-                // Nothing in the queue. Do you
-                // have other tasks, like servicing your
-                // XMPP connection, to do?
-                $handler->idle(QUEUE_HANDLER_MISS_IDLE);
+        $this->_log(LOG_DEBUG, 'Checking for notices...');
+        $item = $this->_nextItem();
+        if ($item === false) {
+            $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);
             } else {
-                $this->_log(LOG_INFO, 'Got notice '. $notice->id);
-                // Yay! Got one!
-                if ($handler->handle_notice($notice)) {
-                    $this->_log(LOG_INFO, 'Successfully handled notice '. $notice->id);
-                    $this->_done($notice, $queue);
-                } else {
-                    $this->_log(LOG_INFO, 'Failed to handle notice '. $notice->id);
-                    $this->_fail($notice, $queue);
-                }
-                // Chance to e.g. service your XMPP connection
-                $this->_log(LOG_DEBUG, 'Idling after success.');
-                $handler->idle(QUEUE_HANDLER_HIT_IDLE);
+                $this->_log(LOG_INFO, "[$queue:notice $notice->id] Failed to handle notice");
+                $this->_fail($notice, $queue);
             }
-            // XXX: when do we give up?
+        } else {
+            $this->_log(LOG_INFO, "[$queue:notice $notice->id] No handler for queue $queue");
+            $this->_fail($notice, $queue);
         }
+        return true;
     }
 
-    function _nextItem($queue, $timeout=null)
+    /**
+     * 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;
 
-        $sleeptime = 1;
+        $qi = Queue_item::top();
+        if (empty($qi)) {
+            return false;
+        }
 
-        do {
-            $qi = Queue_item::top($queue);
-            if (empty($qi)) {
-                $this->_log(LOG_DEBUG, "No new queue items, sleeping $sleeptime seconds.");
-                sleep($sleeptime);
-                $sleeptime *= 2;
-            } else {
-                $notice = Notice::staticGet('id', $qi->notice_id);
-                if (!empty($notice)) {
-                    $result = $notice;
-                } else {
-                    $this->_log(LOG_INFO, 'dequeued non-existent notice ' . $notice->id);
-                    $qi->delete();
-                    $qi->free();
-                    $qi = null;
-                }
-                $sleeptime = 1;
-            }
-        } while (empty($result) && (is_null($timeout) || (time() - $start) < $timeout));
+        $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;
+        }
 
-        return $result;
+        $result = $notice;
+        return array($queue, $notice);
     }
 
-    function _done($object, $queue)
+    /**
+     * Delete our claimed item from the queue after successful processing.
+     *
+     * @param Notice $object
+     * @param string $queue
+     */
+    protected function _done($object, $queue)
     {
         // XXX: right now, we only handle notices
 
@@ -120,24 +149,29 @@ class DBQueueManager extends QueueManager
                                         'transport' => $queue));
 
         if (empty($qi)) {
-            $this->_log(LOG_INFO, 'Cannot find queue item for notice '.$notice->id.', queue '.$queue);
+            $this->_log(LOG_INFO, "[$queue:notice $notice->id] Cannot find queue item");
         } else {
             if (empty($qi->claimed)) {
-                $this->_log(LOG_WARNING, 'Reluctantly releasing unclaimed queue item '.
-                            'for '.$notice->id.', queue '.$queue);
+                $this->_log(LOG_WARNING, "[$queue:notice $notice->id] Reluctantly releasing unclaimed queue item");
             }
             $qi->delete();
             $qi->free();
-            $qi = null;
         }
 
-        $this->_log(LOG_INFO, 'done with notice ID = ' . $notice->id);
+        $this->_log(LOG_INFO, "[$queue:notice $notice->id] done with item");
+        $this->stats('handled', $queue);
 
         $notice->free();
-        $notice = null;
     }
 
-    function _fail($object, $queue)
+    /**
+     * Free our claimed queue item for later reprocessing in case of
+     * temporary failure.
+     *
+     * @param Notice $object
+     * @param string $queue
+     */
+    protected function _fail($object, $queue)
     {
         // XXX: right now, we only handle notices
 
@@ -147,11 +181,10 @@ class DBQueueManager extends QueueManager
                                         'transport' => $queue));
 
         if (empty($qi)) {
-            $this->_log(LOG_INFO, 'Cannot find queue item for notice '.$notice->id.', queue '.$queue);
+            $this->_log(LOG_INFO, "[$queue:notice $notice->id] Cannot find queue item");
         } else {
             if (empty($qi->claimed)) {
-                $this->_log(LOG_WARNING, 'Ignoring failure for unclaimed queue item '.
-                            'for '.$notice->id.', queue '.$queue);
+                $this->_log(LOG_WARNING, "[$queue:notice $notice->id] Ignoring failure for unclaimed queue item");
             } else {
                 $orig = clone($qi);
                 $qi->claimed = null;
@@ -160,13 +193,13 @@ class DBQueueManager extends QueueManager
             }
         }
 
-        $this->_log(LOG_INFO, 'done with notice ID = ' . $notice->id);
+        $this->_log(LOG_INFO, "[$queue:notice $notice->id] done with queue item");
+        $this->stats('error', $queue);
 
         $notice->free();
-        $notice = null;
     }
 
-    function _log($level, $msg)
+    protected function _log($level, $msg)
     {
         common_log($level, 'DBQueueManager: '.$msg);
     }
diff --git a/lib/default.php b/lib/default.php
index fa862f3ff..f7f4777a2 100644
--- a/lib/default.php
+++ b/lib/default.php
@@ -79,6 +79,8 @@ $default =
               'queue_basename' => '/queue/statusnet/',
               'stomp_username' => null,
               '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
               ),
         'license' =>
         array('url' => 'http://creativecommons.org/licenses/by/3.0/',
diff --git a/lib/event.php b/lib/event.php
index 4819b71b4..41fb53ffe 100644
--- a/lib/event.php
+++ b/lib/event.php
@@ -138,4 +138,12 @@ class Event {
         }
         return false;
     }
+
+    /**
+     * Disables any and all handlers that have been set up so far;
+     * use only if you know it's safe to reinitialize all plugins.
+     */
+    public static function clearHandlers() {
+        Event::$_handlers = array();
+    }
 }
diff --git a/lib/iomanager.php b/lib/iomanager.php
new file mode 100644
index 000000000..ee2ff958b
--- /dev/null
+++ b/lib/iomanager.php
@@ -0,0 +1,193 @@
+<?php
+/**
+ * StatusNet, the distributed open-source microblogging tool
+ *
+ * Abstract class for i/o managers
+ *
+ * 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  QueueManager
+ * @package   StatusNet
+ * @author    Evan Prodromou <evan@status.net>
+ * @author    Sarven Capadisli <csarven@status.net>
+ * @author    Brion Vibber <brion@status.net>
+ * @copyright 2009-2010 StatusNet, Inc.
+ * @license   http://www.fsf.org/licensing/licenses/agpl-3.0.html GNU Affero General Public License version 3.0
+ * @link      http://status.net/
+ */
+
+abstract class IoManager
+{
+    const SINGLE_ONLY = 0;
+    const INSTANCE_PER_SITE = 1;
+    const INSTANCE_PER_PROCESS = 2;
+
+    /**
+     * Factory function to get an appropriate subclass.
+     */
+    public abstract static function get();
+
+    /**
+     * Tell the i/o queue master if and how we can handle multi-site
+     * processes.
+     *
+     * Return one of:
+     *   IoManager::SINGLE_ONLY
+     *   IoManager::INSTANCE_PER_SITE
+     *   IoManager::INSTANCE_PER_PROCESS
+     */
+    public static function multiSite()
+    {
+        return IoManager::SINGLE_ONLY;
+    }
+    
+    /**
+     * If in a multisite configuration, the i/o master will tell
+     * your manager about each site you'll have to handle so you
+     * can do any necessary per-site setup.
+     *
+     * @param string $site target site server name
+     */
+    public function addSite($site)
+    {
+        /* no-op */
+    }
+
+    /**
+     * This method is called when data is available on one of your
+     * i/o manager's sockets. The socket with data is passed in,
+     * in case you have multiple sockets.
+     *
+     * If your i/o manager is based on polling during idle processing,
+     * you don't need to implement this.
+     *
+     * @param resource $socket
+     * @return boolean true on success, false on failure
+     */
+    public function handleInput($socket)
+    {
+        return true;
+    }
+
+    /**
+     * Return any open sockets that the run loop should listen
+     * for input on. If input comes in on a listed socket,
+     * the matching manager's handleInput method will be called.
+     *
+     * @return array of resources
+     */
+    function getSockets()
+    {
+        return array();
+    }
+
+    /**
+     * Maximum planned time between poll() calls when input isn't waiting.
+     * Actual time may vary!
+     *
+     * When we get a polling hit, the timeout will be cut down to 0 while
+     * input is coming in, then will back off to this amount if no further
+     * input shows up.
+     *
+     * By default polling is disabled; you must override this to enable
+     * polling for this manager.
+     *
+     * @return int max poll interval in seconds, or 0 to disable polling
+     */
+    function pollInterval()
+    {
+        return 0;
+    }
+
+    /**
+     * Request a maximum timeout for listeners before the next idle period.
+     * Actual wait may be shorter, so don't go crazy in your idle()!
+     * Wait could be longer if other handlers performed some slow activity.
+     *
+     * Return 0 to request that listeners return immediately if there's no
+     * i/o and speed up the idle as much as possible; but don't do that all
+     * the time as this will burn CPU.
+     *
+     * @return int seconds
+     */
+    function timeout()
+    {
+        return 60;
+    }
+
+    /**
+     * Called by IoManager after each handled item or empty polling cycle.
+     * This is a good time to e.g. service your XMPP connection.
+     *
+     * Doesn't need to be overridden if there's no maintenance to do.
+     */
+    function idle()
+    {
+        return true;
+    }
+
+    /**
+     * The meat of a polling manager... check for something to do
+     * and do it! Note that you should not take too long, as other
+     * i/o managers may need to do some work too!
+     *
+     * On a successful hit, the next poll() call will come as soon
+     * as possible followed by exponential backoff up to pollInterval()
+     * if no more data is available.
+     *
+     * @return boolean true if events were hit
+     */
+    public function poll()
+    {
+        return false;
+    }
+
+    /**
+     * Initialization, run when the queue manager starts.
+     * If this function indicates failure, the handler run will be aborted.
+     *
+     * @param IoMaster $master process/event controller
+     * @return boolean true on success, false on failure
+     */
+    public function start($master)
+    {
+        $this->master = $master;
+        return true;
+    }
+
+    /**
+     * Cleanup, run when the queue manager ends.
+     * If this function indicates failure, a warning will be logged.
+     *
+     * @return boolean true on success, false on failure
+     */
+    public function finish()
+    {
+        return true;
+    }
+
+    /**
+     * Ping iomaster's queue status monitor with a stats update.
+     * Only valid during input loop!
+     *
+     * @param string $counter keyword for counter to increment
+     */
+    public function stats($counter, $owners=array())
+    {
+        $this->master->stats($counter, $owners);
+    }
+}
+
diff --git a/lib/iomaster.php b/lib/iomaster.php
new file mode 100644
index 000000000..aff5b145c
--- /dev/null
+++ b/lib/iomaster.php
@@ -0,0 +1,361 @@
+<?php
+/**
+ * StatusNet, the distributed open-source microblogging tool
+ *
+ * I/O manager to wrap around socket-reading and polling queue & connection managers.
+ *
+ * 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  QueueManager
+ * @package   StatusNet
+ * @author    Brion Vibber <brion@status.net>
+ * @copyright 2009 StatusNet, Inc.
+ * @license   http://www.fsf.org/licensing/licenses/agpl-3.0.html GNU Affero General Public License version 3.0
+ * @link      http://status.net/
+ */
+
+class IoMaster
+{
+    public $id;
+
+    protected $multiSite = false;
+    protected $managers = array();
+    protected $singletons = array();
+
+    protected $pollTimeouts = array();
+    protected $lastPoll = array();
+
+    /**
+     * @param string $id process ID to use in logging/monitoring
+     */
+    public function __construct($id)
+    {
+        $this->id = $id;
+        $this->monitor = new QueueMonitor();
+    }
+
+    public function init($multiSite=null)
+    {
+        if ($multiSite !== null) {
+            $this->multiSite = $multiSite;
+        }
+        if ($this->multiSite) {
+            $this->sites = $this->findAllSites();
+        } else {
+            $this->sites = array(common_config('site', 'server'));
+        }
+
+        if (empty($this->sites)) {
+            throw new Exception("Empty status_network table, cannot init");
+        }
+
+        foreach ($this->sites as $site) {
+            if ($site != common_config('site', 'server')) {
+                StatusNet::init($site);
+            }
+
+            $classes = array();
+            if (Event::handle('StartIoManagerClasses', array(&$classes))) {
+                $classes[] = 'QueueManager';
+                if (common_config('xmpp', 'enabled')) {
+                    $classes[] = 'XmppManager'; // handles pings/reconnects
+                    $classes[] = 'XmppConfirmManager'; // polls for outgoing confirmations
+                }
+            }
+            Event::handle('EndIoManagerClasses', array(&$classes));
+
+            foreach ($classes as $class) {
+                $this->instantiate($class);
+            }
+        }
+    }
+
+    /**
+     * Pull all local sites from status_network table.
+     * @return array of hostnames
+     */
+    protected function findAllSites()
+    {
+        $hosts = array();
+        $sn = new Status_network();
+        $sn->find();
+        while ($sn->fetch()) {
+            $hosts[] = $sn->hostname;
+        }
+        return $hosts;
+    }
+
+    /**
+     * Instantiate an i/o manager class for the current site.
+     * If a multi-site capable handler is already present,
+     * we don't need to build a new one.
+     *
+     * @param string $class
+     */
+    protected function instantiate($class)
+    {
+        if (isset($this->singletons[$class])) {
+            // Already instantiated a multi-site-capable handler.
+            // Just let it know it should listen to this site too!
+            $this->singletons[$class]->addSite(common_config('site', 'server'));
+            return;
+        }
+
+        $manager = $this->getManager($class);
+
+        if ($this->multiSite) {
+            $caps = $manager->multiSite();
+            if ($caps == IoManager::SINGLE_ONLY) {
+                throw new Exception("$class can't run with --all; aborting.");
+            }
+            if ($caps == IoManager::INSTANCE_PER_PROCESS) {
+                // Save this guy for later!
+                // We'll only need the one to cover multiple sites.
+                $this->singletons[$class] = $manager;
+                $manager->addSite(common_config('site', 'server'));
+            }
+        }
+
+        $this->managers[] = $manager;
+    }
+    
+    protected function getManager($class)
+    {
+        return call_user_func(array($class, 'get'));
+    }
+
+    /**
+     * Basic run loop...
+     *
+     * Initialize all io managers, then sit around waiting for input.
+     * Between events or timeouts, pass control back to idle() method
+     * to allow for any additional background processing.
+     */
+    function service()
+    {
+        $this->logState('init');
+        $this->start();
+
+        while (true) {
+            $timeouts = array_values($this->pollTimeouts);
+            $timeouts[] = 60; // default max timeout
+
+            // Wait for something on one of our sockets
+            $sockets = array();
+            $managers = array();
+            foreach ($this->managers as $manager) {
+                foreach ($manager->getSockets() as $socket) {
+                    $sockets[] = $socket;
+                    $managers[] = $manager;
+                }
+                $timeouts[] = intval($manager->timeout());
+            }
+
+            $timeout = min($timeouts);
+            if ($sockets) {
+                $read = $sockets;
+                $write = array();
+                $except = array();
+                $this->logState('listening');
+                common_log(LOG_INFO, "Waiting up to $timeout seconds for socket data...");
+                $ready = stream_select($read, $write, $except, $timeout, 0);
+
+                if ($ready === false) {
+                    common_log(LOG_ERR, "Error selecting on sockets");
+                } else if ($ready > 0) {
+                    foreach ($read as $socket) {
+                        $index = array_search($socket, $sockets, true);
+                        if ($index !== false) {
+                            $this->logState('queue');
+                            $managers[$index]->handleInput($socket);
+                        } else {
+                            common_log(LOG_ERR, "Saw input on a socket we didn't listen to");
+                        }
+                    }
+                }
+            }
+
+            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...");
+                $this->logState('sleep');
+                sleep($timeout);
+            }
+
+            $this->logState('poll');
+            $this->poll();
+
+            $this->logState('idle');
+            $this->idle();
+
+            $memoryLimit = $this->softMemoryLimit();
+            if ($memoryLimit > 0) {
+                $usage = memory_get_usage();
+                if ($usage > $memoryLimit) {
+                    common_log(LOG_INFO, "Queue thread hit soft memory limit ($usage > $memoryLimit); gracefully restarting.");
+                    break;
+                }
+            }
+        }
+
+        $this->logState('shutdown');
+        $this->finish();
+    }
+
+    /**
+     * Return fully-parsed soft memory limit in bytes.
+     * @return intval 0 or -1 if not set
+     */
+    function softMemoryLimit()
+    {
+        $softLimit = trim(common_config('queue', 'softlimit'));
+        if (substr($softLimit, -1) == '%') {
+            $limit = trim(ini_get('memory_limit'));
+            $limit = $this->parseMemoryLimit($limit);
+            if ($limit > 0) {
+                return intval(substr($softLimit, 0, -1) * $limit / 100);
+            } else {
+                return -1;
+            }
+        } else {
+            return $this->parseMemoryLimit($limit);
+        }
+        return $softLimit;
+    }
+
+    /**
+     * Interpret PHP shorthand for memory_limit and friends.
+     * Why don't they just expose the actual numeric value? :P
+     * @param string $mem
+     * @return int
+     */
+    protected function parseMemoryLimit($mem)
+    {
+        // http://www.php.net/manual/en/faq.using.php#faq.using.shorthandbytes
+        $size = array('k' => 1024,
+                      'm' => 1024*1024,
+                      'g' => 1024*1024*1024);
+        if (empty($mem)) {
+            return 0;
+        } else if (is_numeric($mem)) {
+            return intval($mem);
+        } else {
+            $mult = strtolower(substr($mem, -1));
+            if (isset($size[$mult])) {
+                return substr($mem, 0, -1) * $size[$mult];
+            } else {
+                return intval($mem);
+            }
+        }
+    }
+
+    function start()
+    {
+        foreach ($this->managers as $index => $manager) {
+            $manager->start($this);
+            // @fixme error check
+            if ($manager->pollInterval()) {
+                // We'll want to check for input on the first pass
+                $this->pollTimeouts[$index] = 0;
+                $this->lastPoll[$index] = 0;
+            }
+        }
+    }
+
+    function finish()
+    {
+        foreach ($this->managers as $manager) {
+            $manager->finish();
+            // @fixme error check
+        }
+    }
+
+    /**
+     * Called during the idle portion of the runloop to see which handlers
+     */
+    function poll()
+    {
+        foreach ($this->managers as $index => $manager) {
+            $interval = $manager->pollInterval();
+            if ($interval <= 0) {
+                // Not a polling manager.
+                continue;
+            }
+
+            if (isset($this->pollTimeouts[$index])) {
+                $timeout = $this->pollTimeouts[$index];
+                if (time() - $this->lastPoll[$index] < $timeout) {
+                    // Not time to poll yet.
+                    continue;
+                }
+            } else {
+                $timeout = 0;
+            }
+            $hit = $manager->poll();
+
+            $this->lastPoll[$index] = time();
+            if ($hit) {
+                // Do the next poll quickly, there may be more input!
+                $this->pollTimeouts[$index] = 0;
+            } else {
+                // Empty queue. Exponential backoff up to the maximum poll interval.
+                if ($timeout > 0) {
+                    $timeout = min($timeout * 2, $interval);
+                } else {
+                    $timeout = 1;
+                }
+                $this->pollTimeouts[$index] = $timeout;
+            }
+        }
+    }
+
+    /**
+     * Called after each handled item or empty polling cycle.
+     * This is a good time to e.g. service your XMPP connection.
+     */
+    function idle()
+    {
+        foreach ($this->managers as $manager) {
+            $manager->idle();
+        }
+    }
+
+    /**
+     * Send thread state update to the monitoring server, if configured.
+     *
+     * @param string $state ('init', 'queue', 'shutdown' etc)
+     * @param string $substate (optional, eg queue name 'omb' 'sms' etc)
+     */
+    protected function logState($state, $substate='')
+    {
+        $this->monitor->logState($this->id, $state, $substate);
+    }
+
+    /**
+     * Send thread stats.
+     * Thread ID will be implicit; other owners can be listed as well
+     * for per-queue and per-site records.
+     *
+     * @param string $key counter name
+     * @param array $owners list of owner keys like 'queue:jabber' or 'site:stat01'
+     */
+    public function stats($key, $owners=array())
+    {
+        $owners[] = "thread:" . $this->id;
+        $this->monitor->stats($key, $owners);
+    }
+}
+
diff --git a/lib/jabber.php b/lib/jabber.php
index a821856a8..1d0bb9423 100644
--- a/lib/jabber.php
+++ b/lib/jabber.php
@@ -86,7 +86,11 @@ class Sharing_XMPP extends XMPPHP_XMPP
 }
 
 /**
- * connect the configured Jabber account to the configured server
+ * Lazy-connect the configured Jabber account to the configured server;
+ * if already opened, the same connection will be returned.
+ *
+ * In a multi-site background process, each site configuration
+ * will get its own connection.
  *
  * @param string $resource Resource to connect (defaults to configured resource)
  *
@@ -95,16 +99,19 @@ class Sharing_XMPP extends XMPPHP_XMPP
 
 function jabber_connect($resource=null)
 {
-    static $conn = null;
-    if (!$conn) {
+    static $connections = array();
+    $site = common_config('site', 'server');
+    if (empty($connections[$site])) {
+        if (empty($resource)) {
+            $resource = common_config('xmpp', 'resource');
+        }
         $conn = new Sharing_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'),
-                                ($resource) ? $resource :
-                                common_config('xmpp', 'resource'),
+                                $resource,
                                 common_config('xmpp', 'server'),
                                 common_config('xmpp', 'debug') ?
                                 true : false,
@@ -115,12 +122,16 @@ function jabber_connect($resource=null)
         if (!$conn) {
             return false;
         }
+        $connections[$site] = $conn;
 
         $conn->autoSubscribe();
         $conn->useEncryption(common_config('xmpp', 'encryption'));
 
         try {
-            $conn->connect(true); // true = persistent connection
+            common_log(LOG_INFO, __METHOD__ . ": connecting " .
+                common_config('xmpp', 'user') . '/' . $resource);
+            //$conn->connect(true); // true = persistent connection
+            $conn->connect(); // persistent connections break multisite
         } catch (XMPPHP_Exception $e) {
             common_log(LOG_ERR, $e->getMessage());
             return false;
@@ -128,7 +139,7 @@ function jabber_connect($resource=null)
 
         $conn->processUntil('session_start');
     }
-    return $conn;
+    return $connections[$site];
 }
 
 /**
diff --git a/lib/jabberqueuehandler.php b/lib/jabberqueuehandler.php
new file mode 100644
index 000000000..b1518866d
--- /dev/null
+++ b/lib/jabberqueuehandler.php
@@ -0,0 +1,47 @@
+<?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);
+}
+
+/**
+ * Queue handler for pushing new notices to Jabber users.
+ * @fixme this exception handling doesn't look very good.
+ */
+class JabberQueueHandler extends QueueHandler
+{
+    var $conn = null;
+
+    function transport()
+    {
+        return 'jabber';
+    }
+
+    function handle_notice($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);
+        }
+    }
+}
diff --git a/lib/liberalstomp.php b/lib/liberalstomp.php
new file mode 100644
index 000000000..c9233843a
--- /dev/null
+++ b/lib/liberalstomp.php
@@ -0,0 +1,133 @@
+<?php
+
+/**
+ * Based on code from Stomp PHP library, working around bugs in the base class.
+ *
+ * Original code is copyright 2005-2006 The Apache Software Foundation
+ * Modifications copyright 2009 StatusNet Inc by Brion Vibber <brion@status.net>
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+class LiberalStomp extends Stomp
+{
+    /**
+     * We need to be able to get the socket so advanced daemons can
+     * do a select() waiting for input both from the queue and from
+     * other sources such as an XMPP connection.
+     *
+     * @return resource
+     */
+    function getSocket()
+    {
+        return $this->_socket;
+    }
+
+    /**
+     * Make socket connection to the server
+     * We also set the stream to non-blocking mode, since we'll be
+     * select'ing to wait for updates. In blocking mode it seems
+     * to get confused sometimes.
+     *
+     * @throws StompException
+     */
+    protected function _makeConnection ()
+    {
+        parent::_makeConnection();
+        stream_set_blocking($this->_socket, 0);
+    }
+
+    /**
+     * Version 1.0.0 of the Stomp library gets confused if messages
+     * come in too fast over the connection. This version will read
+     * out as many frames as are ready to be read from the socket.
+     *
+     * Modified from Stomp::readFrame()
+     *
+     * @return StompFrame False when no frame to read
+     */
+    public function readFrames ()
+    {
+        if (!$this->hasFrameToRead()) {
+            return false;
+        }
+        
+        $rb = 1024;
+        $data = '';
+        $end = false;
+        $frames = array();
+
+        do {
+            // @fixme this sometimes hangs in blocking mode...
+            // shouldn't we have been idle until we found there's more data?
+            $read = fread($this->_socket, $rb);
+            if ($read === false) {
+                $this->_reconnect();
+                // @fixme this will lose prior items
+                return $this->readFrames();
+            }
+            $data .= $read;
+            if (strpos($data, "\x00") !== false) {
+                // Frames are null-delimited, but some servers
+                // may append an extra \n according to old bug reports.
+                $data = str_replace("\x00\n", "\x00", $data);
+                $chunks = explode("\x00", $data);
+
+                $data = array_pop($chunks);
+                $frames = array_merge($frames, $chunks);
+                if ($data == '') {
+                    // We're at the end of a frame; stop reading.
+                    break;
+                } else {
+                    // In the middle of a frame; keep going.
+                }
+            }
+            // @fixme find out why this len < 2 check was there
+            //$len = strlen($data);
+        } while (true);//$len < 2 || $end == false);
+
+        return array_map(array($this, 'parseFrame'), $frames);
+    }
+    
+    /**
+     * Parse a raw Stomp frame into an object.
+     * Extracted from Stomp::readFrame()
+     *
+     * @param string $data
+     * @return StompFrame
+     */
+    function parseFrame($data)
+    {
+        list ($header, $body) = explode("\n\n", $data, 2);
+        $header = explode("\n", $header);
+        $headers = array();
+        $command = null;
+        foreach ($header as $v) {
+            if (isset($command)) {
+                list ($name, $value) = explode(':', $v, 2);
+                $headers[$name] = $value;
+            } else {
+                $command = $v;
+            }
+        }
+        $frame = new StompFrame($command, $headers, trim($body));
+        if (isset($frame->headers['transformation']) && $frame->headers['transformation'] == 'jms-map-json') {
+            require_once 'Stomp/Message/Map.php';
+            return new StompMessageMap($frame);
+        } else {
+            return $frame;
+        }
+        return $frame;
+    }
+}
+
diff --git a/lib/ombqueuehandler.php b/lib/ombqueuehandler.php
new file mode 100644
index 000000000..3ffc1313b
--- /dev/null
+++ b/lib/ombqueuehandler.php
@@ -0,0 +1,56 @@
+<?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);
+}
+
+/**
+ * Queue handler for pushing new notices to OpenMicroBlogging subscribers.
+ */
+class OmbQueueHandler extends QueueHandler
+{
+
+    function transport()
+    {
+        return 'omb';
+    }
+
+    /**
+     * @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)
+    {
+        if ($this->is_remote($notice)) {
+            $this->log(LOG_DEBUG, 'Ignoring remote notice ' . $notice->id);
+            return true;
+        } else {
+            require_once(INSTALLDIR.'/lib/omb.php');
+            omb_broadcast_notice($notice);
+            return true;
+        }
+    }
+
+    function is_remote($notice)
+    {
+        $user = User::staticGet($notice->profile_id);
+        return is_null($user);
+    }
+}
diff --git a/lib/pingqueuehandler.php b/lib/pingqueuehandler.php
new file mode 100644
index 000000000..8bb218078
--- /dev/null
+++ b/lib/pingqueuehandler.php
@@ -0,0 +1,37 @@
+<?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);
+}
+
+/**
+ * Queue handler for pushing new notices to ping servers.
+ */
+class PingQueueHandler extends QueueHandler {
+
+    function transport() {
+        return 'ping';
+    }
+
+    function handle_notice($notice) {
+        require_once INSTALLDIR . '/lib/ping.php';
+        return ping_broadcast_notice($notice);
+    }
+}
diff --git a/lib/pluginqueuehandler.php b/lib/pluginqueuehandler.php
new file mode 100644
index 000000000..24d504699
--- /dev/null
+++ b/lib/pluginqueuehandler.php
@@ -0,0 +1,50 @@
+<?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);
+}
+
+/**
+ * Queue handler for letting plugins handle stuff.
+ *
+ * The plugin queue handler accepts notices over the "plugin" queue
+ * and simply passes them through the "HandleQueuedNotice" event.
+ *
+ * This gives plugins a chance to do background processing without
+ * actually registering their own queue and ensuring that things
+ * are queued into it.
+ *
+ * Fancier plugins may wish to instead hook the 'GetQueueHandlerClass'
+ * event with their own class, in which case they must ensure that
+ * their notices get enqueued when they need them.
+ */
+class PluginQueueHandler extends QueueHandler
+{
+    function transport()
+    {
+        return 'plugin';
+    }
+
+    function handle_notice($notice)
+    {
+        Event::handle('HandleQueuedNotice', array(&$notice));
+        return true;
+    }
+}
diff --git a/lib/publicqueuehandler.php b/lib/publicqueuehandler.php
new file mode 100644
index 000000000..9ea9ee73a
--- /dev/null
+++ b/lib/publicqueuehandler.php
@@ -0,0 +1,47 @@
+<?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);
+}
+
+/**
+ * Queue handler for pushing new notices to public XMPP subscribers.
+ * @fixme correct this exception handling
+ */
+class PublicQueueHandler extends QueueHandler
+{
+
+    function transport()
+    {
+        return 'public';
+    }
+
+    function handle_notice($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 true;
+    }
+}
diff --git a/lib/queuehandler.php b/lib/queuehandler.php
index cd43b1e09..613be6e33 100644
--- a/lib/queuehandler.php
+++ b/lib/queuehandler.php
@@ -19,14 +19,6 @@
 
 if (!defined('STATUSNET') && !defined('LACONICA')) { exit(1); }
 
-require_once(INSTALLDIR.'/lib/daemon.php');
-require_once(INSTALLDIR.'/classes/Queue_item.php');
-require_once(INSTALLDIR.'/classes/Notice.php');
-
-define('CLAIM_TIMEOUT', 1200);
-define('QUEUE_HANDLER_MISS_IDLE', 10);
-define('QUEUE_HANDLER_HIT_IDLE', 0);
-
 /**
  * Base class for queue handlers.
  *
@@ -36,24 +28,20 @@ define('QUEUE_HANDLER_HIT_IDLE', 0);
  *
  * Subclasses must override at least the following methods:
  * - transport
- * - start
- * - finish
  * - handle_notice
- *
- * Some subclasses will also want to override the idle handler:
- * - idle
  */
-class QueueHandler extends Daemon
+#class QueueHandler extends Daemon
+class QueueHandler
 {
 
-    function __construct($id=null, $daemonize=true)
-    {
-        parent::__construct($daemonize);
-
-        if ($id) {
-            $this->set_id($id);
-        }
-    }
+#    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
@@ -61,22 +49,23 @@ class QueueHandler extends Daemon
      *
      * 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 timeout()
+#    {
+#        return 60;
+#    }
 
-    function class_name()
-    {
-        return ucfirst($this->transport()) . 'Handler';
-    }
+#    function class_name()
+#    {
+#        return ucfirst($this->transport()) . 'Handler';
+#    }
 
-    function name()
-    {
-        return strtolower($this->class_name().'.'.$this->get_id());
-    }
+#    function name()
+#    {
+#        return strtolower($this->class_name().'.'.$this->get_id());
+#    }
 
     /**
      * Return transport keyword which identifies items this queue handler
@@ -92,30 +81,6 @@ class QueueHandler extends Daemon
         return null;
     }
 
-    /**
-     * Initialization, run when the queue handler starts.
-     * If this function indicates failure, the handler run will be aborted.
-     *
-     * @fixme run() will abort if this doesn't return true,
-     *        but some subclasses don't bother.
-     * @return boolean true on success, false on failure
-     */
-    function start()
-    {
-    }
-
-    /**
-     * Cleanup, run when the queue handler ends.
-     * If this function indicates failure, a warning will be logged.
-     *
-     * @fixme run() will throw warnings if this doesn't return true,
-     *        but many subclasses don't bother.
-     * @return boolean true on success, false on failure
-     */
-    function finish()
-    {
-    }
-
     /**
      * Here's the meat of your queue handler -- you're handed a Notice
      * object, which you may do as you will with.
@@ -169,29 +134,10 @@ class QueueHandler extends Daemon
         return true;
     }
 
-    /**
-     * Called by QueueHandler after each handled item or empty polling cycle.
-     * This is a good time to e.g. service your XMPP connection.
-     *
-     * Doesn't need to be overridden if there's no maintenance to do.
-     *
-     * @param int $timeout seconds to sleep if there's nothing to do
-     */
-    function idle($timeout=0)
-    {
-        if ($timeout > 0) {
-            sleep($timeout);
-        }
-    }
 
     function log($level, $msg)
     {
         common_log($level, $this->class_name() . ' ('. $this->get_id() .'): '.$msg);
     }
-
-    function getSockets()
-    {
-        return array();
-    }
 }
 
diff --git a/lib/queuemanager.php b/lib/queuemanager.php
index 43105b7a8..a98c0efff 100644
--- a/lib/queuemanager.php
+++ b/lib/queuemanager.php
@@ -2,7 +2,7 @@
 /**
  * StatusNet, the distributed open-source microblogging tool
  *
- * Abstract class for queue managers
+ * Abstract class for i/o managers
  *
  * PHP version 5
  *
@@ -23,16 +23,32 @@
  * @package   StatusNet
  * @author    Evan Prodromou <evan@status.net>
  * @author    Sarven Capadisli <csarven@status.net>
- * @copyright 2009 StatusNet, Inc.
+ * @author    Brion Vibber <brion@status.net>
+ * @copyright 2009-2010 StatusNet, Inc.
  * @license   http://www.fsf.org/licensing/licenses/agpl-3.0.html GNU Affero General Public License version 3.0
  * @link      http://status.net/
  */
 
-class QueueManager
+/**
+ * Completed child classes must implement the enqueue() method.
+ *
+ * For background processing, classes should implement either socket-based
+ * input (handleInput(), getSockets()) or idle-loop polling (idle()).
+ */
+abstract class QueueManager extends IoManager
 {
     static $qm = null;
 
-    static function get()
+    /**
+     * Factory function to pull the appropriate QueueManager object
+     * for this site's configuration. It can then be used to queue
+     * events for later processing or to spawn a processing loop.
+     *
+     * Plugins can add to the built-in types by hooking StartNewQueueManager.
+     *
+     * @return QueueManager
+     */
+    public static function get()
     {
         if (empty(self::$qm)) {
 
@@ -62,13 +78,130 @@ class QueueManager
         return self::$qm;
     }
 
-    function enqueue($object, $queue)
+    /**
+     * @fixme wouldn't necessarily work with other class types.
+     * Better to change the interface...?
+     */
+    public static function multiSite()
+    {
+        if (common_config('queue', 'subsystem') == 'stomp') {
+            return IoManager::INSTANCE_PER_PROCESS;
+        } else {
+            return IoManager::SINGLE_ONLY;
+        }
+    }
+
+    function __construct()
     {
-        throw ServerException("Unimplemented function 'enqueue' called");
+        $this->initialize();
     }
 
-    function service($queue, $handler)
+    /**
+     * Store an object (usually/always a Notice) into the given queue
+     * for later processing. No guarantee is made on when it will be
+     * processed; it could be immediately or at some unspecified point
+     * in the future.
+     *
+     * Must be implemented by any queue manager.
+     *
+     * @param Notice $object
+     * @param string $queue
+     */
+    abstract function enqueue($object, $queue);
+
+    /**
+     * Instantiate the appropriate QueueHandler class for the given queue.
+     *
+     * @param string $queue
+     * @return mixed QueueHandler or null
+     */
+    function getHandler($queue)
     {
-        throw ServerException("Unimplemented function 'service' called");
+        if (isset($this->handlers[$queue])) {
+            $class = $this->handlers[$queue];
+            if (class_exists($class)) {
+                return new $class();
+            } else {
+                common_log(LOG_ERR, "Nonexistent handler class '$class' for queue '$queue'");
+            }
+        } else {
+            common_log(LOG_ERR, "Requested handler for unkown queue '$queue'");
+        }
+        return null;
+    }
+
+    /**
+     * Get a list of all registered queue transport names.
+     *
+     * @return array of strings
+     */
+    function getQueues()
+    {
+        return array_keys($this->handlers);
+    }
+
+    /**
+     * Initialize the list of queue handlers
+     *
+     * @event StartInitializeQueueManager
+     * @event EndInitializeQueueManager
+     */
+    function initialize()
+    {
+        if (Event::handle('StartInitializeQueueManager', array($this))) {
+            $this->connect('plugin', 'PluginQueueHandler');
+            $this->connect('omb', 'OmbQueueHandler');
+            $this->connect('ping', 'PingQueueHandler');
+            if (common_config('sms', 'enabled')) {
+                $this->connect('sms', 'SmsQueueHandler');
+            }
+
+            // XMPP output handlers...
+            if (common_config('xmpp', 'enabled')) {
+                $this->connect('jabber', 'JabberQueueHandler');
+                $this->connect('public', 'PublicQueueHandler');
+                
+                // @fixme this should move up a level or should get an actual queue
+                $this->connect('confirm', 'XmppConfirmHandler');
+            }
+
+            // For compat with old plugins not registering their own handlers.
+            $this->connect('plugin', 'PluginQueueHandler');
+        }
+        Event::handle('EndInitializeQueueManager', array($this));
+    }
+
+    /**
+     * Register a queue transport name and handler class for your plugin.
+     * Only registered transports will be reliably picked up!
+     *
+     * @param string $transport
+     * @param string $class
+     */
+    public function connect($transport, $class)
+    {
+        $this->handlers[$transport] = $class;
+    }
+
+    /**
+     * Send a statistic ping to the queue monitoring system,
+     * optionally with a per-queue id.
+     *
+     * @param string $key
+     * @param string $queue
+     */
+    function stats($key, $queue=false)
+    {
+        $owners = array();
+        if ($queue) {
+            $owners[] = "queue:$queue";
+            $owners[] = "site:" . common_config('site', 'server');
+        }
+        if (isset($this->master)) {
+            $this->master->stats($key, $owners);
+        } else {
+            $monitor = new QueueMonitor();
+            $monitor->stats($key, $owners);
+        }
     }
 }
diff --git a/lib/queuemonitor.php b/lib/queuemonitor.php
new file mode 100644
index 000000000..1c306a629
--- /dev/null
+++ b/lib/queuemonitor.php
@@ -0,0 +1,116 @@
+<?php
+/**
+ * StatusNet, the distributed open-source microblogging tool
+ *
+ * Monitoring output helper for IoMaster and IoManager/QueueManager
+ *
+ * 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  QueueManager
+ * @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/
+ */
+
+class QueueMonitor
+{
+    protected $monSocket = null;
+
+    /**
+     * Increment monitoring statistics for a given counter, if configured.
+     * Only explicitly listed thread/site/queue owners will be incremented.
+     *
+     * @param string $key counter name
+     * @param array $owners list of owner keys like 'queue:jabber' or 'site:stat01'
+     */
+    public function stats($key, $owners=array())
+    {
+        $this->ping(array('counter' => $key,
+                          'owners' => $owners));
+    }
+
+    /**
+     * Send thread state update to the monitoring server, if configured.
+     *
+     * @param string $thread ID (eg 'generic.1')
+     * @param string $state ('init', 'queue', 'shutdown' etc)
+     * @param string $substate (optional, eg queue name 'omb' 'sms' etc)
+     */
+    public function logState($threadId, $state, $substate='')
+    {
+        $this->ping(array('thread_id' => $threadId,
+                          'state' => $state,
+                          'substate' => $substate,
+                          'ts' => microtime(true)));
+    }
+
+    /**
+     * General call to the monitoring server
+     */
+    protected function ping($data)
+    {
+        $target = common_config('queue', 'monitor');
+        if (empty($target)) {
+            return;
+        }
+
+        $data = $this->prepMonitorData($data);
+
+        if (substr($target, 0, 4) == 'udp:') {
+            $this->pingUdp($target, $data);
+        } else if (substr($target, 0, 5) == 'http:') {
+            $this->pingHttp($target, $data);
+        } else {
+            common_log(LOG_ERR, __METHOD__ . ' unknown monitor target type ' . $target);
+        }
+    }
+
+    protected function pingUdp($target, $data)
+    {
+        if (!$this->monSocket) {
+            $this->monSocket = stream_socket_client($target, $errno, $errstr);
+        }
+        if ($this->monSocket) {
+            $post = http_build_query($data, '', '&');
+            stream_socket_sendto($this->monSocket, $post);
+        } else {
+            common_log(LOG_ERR, __METHOD__ . " UDP logging fail: $errstr");
+        }
+    }
+
+    protected function pingHttp($target, $data)
+    {
+        $client = new HTTPClient();
+        $result = $client->post($target, array(), $data);
+        
+        if (!$result->isOk()) {
+            common_log(LOG_ERR, __METHOD__ . ' HTTP ' . $result->getStatus() .
+                                ': ' . $result->getBody());
+        }
+    }
+
+    protected function prepMonitorData($data)
+    {
+        #asort($data);
+        #$macdata = http_build_query($data, '', '&');
+        #$key = 'This is a nice old key';
+        #$data['hmac'] = hash_hmac('sha256', $macdata, $key);
+        return $data;
+    }
+
+}
diff --git a/lib/smsqueuehandler.php b/lib/smsqueuehandler.php
new file mode 100644
index 000000000..48a96409d
--- /dev/null
+++ b/lib/smsqueuehandler.php
@@ -0,0 +1,39 @@
+<?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);
+}
+
+/**
+ * Queue handler for pushing new notices to local subscribers using SMS.
+ */
+class SmsQueueHandler extends QueueHandler
+{
+    function transport()
+    {
+        return 'sms';
+    }
+
+    function handle_notice($notice)
+    {
+    	require_once(INSTALLDIR.'/lib/mail.php');
+        return mail_broadcast_notice_sms($notice);
+    }
+}
diff --git a/lib/statusnet.php b/lib/statusnet.php
new file mode 100644
index 000000000..0c5807d7b
--- /dev/null
+++ b/lib/statusnet.php
@@ -0,0 +1,295 @@
+<?php
+/*
+ * StatusNet - the distributed open-source microblogging tool
+ * Copyright (C) 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); }
+
+global $config, $_server, $_path;
+
+/**
+ * Global configuration setup and management.
+ */
+class StatusNet
+{
+    protected static $have_config;
+
+    /**
+     * Configure and instantiate a plugin into the current configuration.
+     * Class definitions will be loaded from standard paths if necessary.
+     * Note that initialization events won't be fired until later.
+     *
+     * @param string $name class name & plugin file/subdir name
+     * @param array $attrs key/value pairs of public attributes to set on plugin instance
+     *
+     * @throws ServerException if plugin can't be found
+     */
+    public static function addPlugin($name, $attrs = null)
+    {
+        $name = ucfirst($name);
+        $pluginclass = "{$name}Plugin";
+
+        if (!class_exists($pluginclass)) {
+
+            $files = array("local/plugins/{$pluginclass}.php",
+                           "local/plugins/{$name}/{$pluginclass}.php",
+                           "local/{$pluginclass}.php",
+                           "local/{$name}/{$pluginclass}.php",
+                           "plugins/{$pluginclass}.php",
+                           "plugins/{$name}/{$pluginclass}.php");
+
+            foreach ($files as $file) {
+                $fullpath = INSTALLDIR.'/'.$file;
+                if (@file_exists($fullpath)) {
+                    include_once($fullpath);
+                    break;
+                }
+            }
+            if (!class_exists($pluginclass)) {
+                throw new ServerException(500, "Plugin $name not found.");
+            }
+        }
+
+        $inst = new $pluginclass();
+        if (!empty($attrs)) {
+            foreach ($attrs as $aname => $avalue) {
+                $inst->$aname = $avalue;
+            }
+        }
+        return true;
+    }
+
+    /**
+     * Initialize, or re-initialize, StatusNet global configuration
+     * and plugins.
+     *
+     * If switching site configurations during script execution, be
+     * careful when working with leftover objects -- global settings
+     * affect many things and they may not behave as you expected.
+     *
+     * @param $server optional web server hostname for picking config
+     * @param $path optional URL path for picking config
+     * @param $conffile optional configuration file path
+     *
+     * @throws NoConfigException if config file can't be found
+     */
+    public static function init($server=null, $path=null, $conffile=null)
+    {
+        StatusNet::initDefaults($server, $path);
+        StatusNet::loadConfigFile($conffile);
+
+        // Load settings from database; note we need autoload for this
+        Config::loadSettings();
+
+        self::initPlugins();
+    }
+
+    /**
+     * Fire initialization events for all instantiated plugins.
+     */
+    protected static function initPlugins()
+    {
+        // Load default plugins
+        foreach (common_config('plugins', 'default') as $name => $params) {
+            if (is_null($params)) {
+                addPlugin($name);
+            } else if (is_array($params)) {
+                if (count($params) == 0) {
+                    addPlugin($name);
+                } else {
+                    $keys = array_keys($params);
+                    if (is_string($keys[0])) {
+                        addPlugin($name, $params);
+                    } else {
+                        foreach ($params as $paramset) {
+                            addPlugin($name, $paramset);
+                        }
+                    }
+                }
+            }
+        }
+
+        // XXX: if plugins should check the schema at runtime, do that here.
+        if (common_config('db', 'schemacheck') == 'runtime') {
+            Event::handle('CheckSchema');
+        }
+
+        // Give plugins a chance to initialize in a fully-prepared environment
+        Event::handle('InitializePlugin');
+    }
+
+    /**
+     * Quick-check if configuration has been established.
+     * Useful for functions which may get used partway through
+     * initialization to back off from fancier things.
+     *
+     * @return bool
+     */
+    public function haveConfig()
+    {
+        return self::$have_config;
+    }
+
+    /**
+     * Build default configuration array
+     * @return array
+     */
+    protected static function defaultConfig()
+    {
+        global $_server, $_path;
+        require(INSTALLDIR.'/lib/default.php');
+        return $default;
+    }
+
+    /**
+     * Establish default configuration based on given or default server and path
+     * Sets global $_server, $_path, and $config
+     */
+    protected static function initDefaults($server, $path)
+    {
+        global $_server, $_path, $config;
+
+        Event::clearHandlers();
+
+        // try to figure out where we are. $server and $path
+        // can be set by including module, else we guess based
+        // on HTTP info.
+
+        if (isset($server)) {
+            $_server = $server;
+        } else {
+            $_server = array_key_exists('SERVER_NAME', $_SERVER) ?
+              strtolower($_SERVER['SERVER_NAME']) :
+            null;
+        }
+
+        if (isset($path)) {
+            $_path = $path;
+        } else {
+            $_path = (array_key_exists('SERVER_NAME', $_SERVER) && array_key_exists('SCRIPT_NAME', $_SERVER)) ?
+              self::_sn_to_path($_SERVER['SCRIPT_NAME']) :
+            null;
+        }
+
+        // Set config values initially to default values
+        $default = self::defaultConfig();
+        $config = $default;
+
+        // default configuration, overwritten in config.php
+        // Keep DB_DataObject's db config synced to ours...
+
+        $config['db'] = &PEAR::getStaticProperty('DB_DataObject','options');
+
+        $config['db'] = $default['db'];
+
+        // Backward compatibility
+
+        $config['site']['design'] =& $config['design'];
+
+        if (function_exists('date_default_timezone_set')) {
+            /* Work internally in UTC */
+            date_default_timezone_set('UTC');
+        }
+    }
+
+    protected function _sn_to_path($sn)
+    {
+        $past_root = substr($sn, 1);
+        $last_slash = strrpos($past_root, '/');
+        if ($last_slash > 0) {
+            $p = substr($past_root, 0, $last_slash);
+        } else {
+            $p = '';
+        }
+        return $p;
+    }
+
+    /**
+     * Load the default or specified configuration file.
+     * Modifies global $config and may establish plugins.
+     *
+     * @throws NoConfigException
+     */
+    protected function loadConfigFile($conffile=null)
+    {
+        global $_server, $_path, $config;
+        
+        // From most general to most specific:
+        // server-wide, then vhost-wide, then for a path,
+        // finally for a dir (usually only need one of the last two).
+        
+        if (isset($conffile)) {
+            $config_files = array($conffile);
+        } else {
+            $config_files = array('/etc/statusnet/statusnet.php',
+                                        '/etc/statusnet/laconica.php',
+                                        '/etc/laconica/laconica.php',
+                                        '/etc/statusnet/'.$_server.'.php',
+                                        '/etc/laconica/'.$_server.'.php');
+
+            if (strlen($_path) > 0) {
+                $config_files[] = '/etc/statusnet/'.$_server.'_'.$_path.'.php';
+                $config_files[] = '/etc/laconica/'.$_server.'_'.$_path.'.php';
+            }
+
+            $config_files[] = INSTALLDIR.'/config.php';
+        }
+
+        self::$have_config = false;
+
+        foreach ($config_files as $_config_file) {
+            if (@file_exists($_config_file)) {
+                include($_config_file);
+                self::$have_config = true;
+            }
+        }
+
+        if (!self::$have_config) {
+            throw new NoConfigException("No configuration file found.",
+                $config_files);
+        }
+
+        // Fixup for statusnet.ini
+        $_db_name = substr($config['db']['database'], strrpos($config['db']['database'], '/') + 1);
+
+        if ($_db_name != 'statusnet' && !array_key_exists('ini_'.$_db_name, $config['db'])) {
+            $config['db']['ini_'.$_db_name] = INSTALLDIR.'/classes/statusnet.ini';
+        }
+
+        // Backwards compatibility
+
+        if (array_key_exists('memcached', $config)) {
+            if ($config['memcached']['enabled']) {
+                addPlugin('Memcache', array('servers' => $config['memcached']['server']));
+            }
+
+            if (!empty($config['memcached']['base'])) {
+                $config['cache']['base'] = $config['memcached']['base'];
+            }
+        }
+    }
+}
+
+class NoConfigException extends Exception
+{
+    public $config_files;
+
+    function __construct($msg, $config_files) {
+        parent::__construct($msg);
+        $this->config_files = $config_files;
+    }
+}
diff --git a/lib/stompqueuemanager.php b/lib/stompqueuemanager.php
index f059b42f0..3090e0bfb 100644
--- a/lib/stompqueuemanager.php
+++ b/lib/stompqueuemanager.php
@@ -30,46 +30,53 @@
 
 require_once 'Stomp.php';
 
-class LiberalStomp extends Stomp
-{
-    function getSocket()
-    {
-        return $this->_socket;
-    }
-}
 
-class StompQueueManager
+class StompQueueManager extends QueueManager
 {
     var $server = null;
     var $username = null;
     var $password = null;
     var $base = null;
     var $con = null;
+    
+    protected $master = null;
+    protected $sites = array();
 
     function __construct()
     {
+        parent::__construct();
         $this->server   = common_config('queue', 'stomp_server');
         $this->username = common_config('queue', 'stomp_username');
         $this->password = common_config('queue', 'stomp_password');
         $this->base     = common_config('queue', 'queue_basename');
     }
 
-    function _connect()
+    /**
+     * Tell the i/o master we only need a single instance to cover
+     * all sites running in this process.
+     */
+    public static function multiSite()
     {
-        if (empty($this->con)) {
-            $this->_log(LOG_INFO, "Connecting to '$this->server' as '$this->username'...");
-            $this->con = new LiberalStomp($this->server);
+        return IoManager::INSTANCE_PER_PROCESS;
+    }
 
-            if ($this->con->connect($this->username, $this->password)) {
-                $this->_log(LOG_INFO, "Connected.");
-            } else {
-                $this->_log(LOG_ERR, 'Failed to connect to queue server');
-                throw new ServerException('Failed to connect to queue server');
-            }
-        }
+    /**
+     * Record each site we'll be handling input for in this process,
+     * so we can listen to the necessary queues for it.
+     *
+     * @fixme possibly actually do subscription here to save another
+     *        loop over all sites later?
+     */
+    public function addSite($server)
+    {
+        $this->sites[] = $server;
     }
 
-    function enqueue($object, $queue)
+    /**
+     * Saves a notice object reference into the queue item table.
+     * @return boolean true on success
+     */
+    public function enqueue($object, $queue)
     {
         $notice = $object;
 
@@ -77,7 +84,7 @@ class StompQueueManager
 
         // XXX: serialize and send entire notice
 
-        $result = $this->con->send($this->_queueName($queue),
+        $result = $this->con->send($this->queueName($queue),
                                    $notice->id, 		// BODY of the message
                                    array ('created' => $notice->created));
 
@@ -88,78 +95,212 @@ class StompQueueManager
 
         common_log(LOG_DEBUG, 'complete remote queueing notice ID = '
                    . $notice->id . ' for ' . $queue);
+        $this->stats('enqueued', $queue);
     }
 
-    function service($queue, $handler)
+    /**
+     * Send any sockets we're listening on to the IO manager
+     * to wait for input.
+     *
+     * @return array of resources
+     */
+    public function getSockets()
     {
-        $result = null;
-
-        $this->_connect();
+        return array($this->con->getSocket());
+    }
 
-        $this->con->setReadTimeout($handler->timeout());
+    /**
+     * We've got input to handle on our socket!
+     * Read any waiting Stomp frame(s) and process them.
+     *
+     * @param resource $socket
+     * @return boolean ok on success
+     */
+    public function handleInput($socket)
+    {
+        assert($socket === $this->con->getSocket());
+        $ok = true;
+        $frames = $this->con->readFrames();
+        foreach ($frames as $frame) {
+            $ok = $ok && $this->_handleNotice($frame);
+        }
+        return $ok;
+    }
 
-        $this->con->subscribe($this->_queueName($queue));
+    /**
+     * Initialize our connection and subscribe to all the queues
+     * we're going to need to handle...
+     *
+     * Side effects: in multi-site mode, may reset site configuration.
+     *
+     * @param IoMaster $master process/event controller
+     * @return bool return false on failure
+     */
+    public function start($master)
+    {
+        parent::start($master);
+        if ($this->sites) {
+            foreach ($this->sites as $server) {
+                StatusNet::init($server);
+                $this->doSubscribe();
+            }
+        } else {
+            $this->doSubscribe();
+        }
+        return true;
+    }
+    
+    /**
+     * Subscribe to all the queues we're going to need to handle...
+     *
+     * Side effects: in multi-site mode, may reset site configuration.
+     *
+     * @return bool return false on failure
+     */
+    public function finish()
+    {
+        if ($this->sites) {
+            foreach ($this->sites as $server) {
+                StatusNet::init($server);
+                $this->doUnsubscribe();
+            }
+        } else {
+            $this->doUnsubscribe();
+        }
+        return true;
+    }
+    
+    /**
+     * Lazy open connection to Stomp queue server.
+     */
+    protected function _connect()
+    {
+        if (empty($this->con)) {
+            $this->_log(LOG_INFO, "Connecting to '$this->server' as '$this->username'...");
+            $this->con = new LiberalStomp($this->server);
 
-        while (true) {
+            if ($this->con->connect($this->username, $this->password)) {
+                $this->_log(LOG_INFO, "Connected.");
+            } else {
+                $this->_log(LOG_ERR, 'Failed to connect to queue server');
+                throw new ServerException('Failed to connect to queue server');
+            }
+        }
+    }
 
-            // Wait for something on one of our sockets
+    /**
+     * Subscribe to all enabled notice queues for the current site.
+     */
+    protected function doSubscribe()
+    {
+        $this->_connect();
+        foreach ($this->getQueues() as $queue) {
+            $rawqueue = $this->queueName($queue);
+            $this->_log(LOG_INFO, "Subscribing to $rawqueue");
+            $this->con->subscribe($rawqueue);
+        }
+    }
+    
+    /**
+     * Subscribe from all enabled notice queues for the current site.
+     */
+    protected function doUnsubscribe()
+    {
+        $this->_connect();
+        foreach ($this->getQueues() as $queue) {
+            $this->con->unsubscribe($this->queueName($queue));
+        }
+    }
 
-            $stompsock = $this->con->getSocket();
+    /**
+     * Handle and acknowledge a notice 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.
+     *
+     * Side effects: in multi-site mode, may reset site configuration to
+     * match the site that queued the event.
+     *
+     * @param StompFrame $frame
+     * @return bool
+     */
+    protected function _handleNotice($frame)
+    {
+        list($site, $queue) = $this->parseDestination($frame->headers['destination']);
+        if ($site != common_config('site', 'server')) {
+            $this->stats('switch');
+            StatusNet::init($site);
+        }
 
-            $handsocks = $handler->getSockets();
+        $id = intval($frame->body);
+        $info = "notice $id posted at {$frame->headers['created']} in queue $queue";
 
-            $socks = array_merge(array($stompsock), $handsocks);
+        $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;
+        }
 
-            $read = $socks;
-            $write = array();
-            $except = array();
+        $handler = $this->getHandler($queue);
+        if (!$handler) {
+            $this->_log(LOG_ERROR, "Missing handler class; skipping $info");
+            $this->con->ack($frame);
+            $this->stats('badhandler', $queue);
+            return false;
+        }
 
-            $ready = stream_select($read, $write, $except, $handler->timeout(), 0);
+        $ok = $handler->handle_notice($notice);
 
-            if ($ready === false) {
-                $this->_log(LOG_ERR, "Error selecting on sockets");
-            } else if ($ready > 0) {
-                if (in_array($stompsock, $read)) {
-                    $this->_handleNotice($queue, $handler);
-                }
-                $handler->idle(QUEUE_HANDLER_HIT_IDLE);
-            }
+        if (!$ok) {
+            $this->_log(LOG_WARNING, "Failed handling $info");
+            // FIXME we probably shouldn't have to do
+            // this kind of queue management ourselves;
+            // if we don't ack, it should resend...
+            $this->con->ack($frame);
+            $this->enqueue($notice, $queue);
+            $this->stats('requeued', $queue);
+            return false;
         }
 
-        $this->con->unsubscribe($this->_queueName($queue));
+        $this->_log(LOG_INFO, "Successfully handled $info");
+        $this->con->ack($frame);
+        $this->stats('handled', $queue);
+        return true;
     }
 
-    function _handleNotice($queue, $handler)
+    /**
+     * Combines the queue_basename from configuration with the
+     * site server name and queue name to give eg:
+     *
+     * /queue/statusnet/identi.ca/sms
+     *
+     * @param string $queue
+     * @return string
+     */
+    protected function queueName($queue)
     {
-        $frame = $this->con->readFrame();
-
-        if (!empty($frame)) {
-            $notice = Notice::staticGet('id', $frame->body);
-
-            if (empty($notice)) {
-                $this->_log(LOG_WARNING, 'Got ID '. $frame->body .' for non-existent notice in queue '. $queue);
-                $this->con->ack($frame);
-            } else {
-                if ($handler->handle_notice($notice)) {
-                    $this->_log(LOG_INFO, 'Successfully handled notice '. $notice->id .' posted at ' . $frame->headers['created'] . ' in queue '. $queue);
-                    $this->con->ack($frame);
-                } else {
-                    $this->_log(LOG_WARNING, 'Failed handling notice '. $notice->id .' posted at ' . $frame->headers['created']  . ' in queue '. $queue);
-                    // FIXME we probably shouldn't have to do
-                    // this kind of queue management ourselves
-                    $this->con->ack($frame);
-                    $this->enqueue($notice, $queue);
-                }
-                unset($notice);
-            }
-
-            unset($frame);
-        }
+        return common_config('queue', 'queue_basename') .
+            common_config('site', 'server') . '/' . $queue;
     }
 
-    function _queueName($queue)
+    /**
+     * Returns the site and queue name from the server-side queue.
+     *
+     * @param string queue destination (eg '/queue/statusnet/identi.ca/sms')
+     * @return array of site and queue: ('identi.ca','sms') or false if unrecognized
+     */
+    protected function parseDestination($dest)
     {
-        return common_config('queue', 'queue_basename') . $queue;
+        $prefix = common_config('queue', 'queue_basename');
+        if (substr($dest, 0, strlen($prefix)) == $prefix) {
+            $rest = substr($dest, strlen($prefix));
+            return explode("/", $rest, 2);
+        } else {
+            common_log(LOG_ERR, "Got a message from unrecognized stomp queue: $dest");
+            return array(false, false);
+        }
     }
 
     function _log($level, $msg)
@@ -167,3 +308,4 @@ class StompQueueManager
         common_log($level, 'StompQueueManager: '.$msg);
     }
 }
+
diff --git a/lib/unqueuemanager.php b/lib/unqueuemanager.php
index 72dbc4eed..5595eac05 100644
--- a/lib/unqueuemanager.php
+++ b/lib/unqueuemanager.php
@@ -23,57 +23,35 @@
  * @package   StatusNet
  * @author    Evan Prodromou <evan@status.net>
  * @author    Sarven Capadisli <csarven@status.net>
+ * @author    Brion Vibber <brion@status.net>
  * @copyright 2009 StatusNet, Inc.
  * @license   http://www.fsf.org/licensing/licenses/agpl-3.0.html GNU Affero General Public License version 3.0
  * @link      http://status.net/
  */
 
-class UnQueueManager
+class UnQueueManager extends QueueManager
 {
+
+    /**
+     * Dummy queue storage manager: instead of saving events for later,
+     * we just process them immediately. This is only suitable for events
+     * that can be processed quickly and don't need polling or long-running
+     * connections to another server such as XMPP.
+     *
+     * @param Notice $object
+     * @param string $queue
+     */
     function enqueue($object, $queue)
     {
         $notice = $object;
-
-        switch ($queue)
-        {
-         case 'omb':
-            if ($this->_isLocal($notice)) {
-                require_once(INSTALLDIR.'/lib/omb.php');
-                omb_broadcast_notice($notice);
-            }
-            break;
-         case 'public':
-            if ($this->_isLocal($notice)) {
-                require_once(INSTALLDIR.'/lib/jabber.php');
-                jabber_public_notice($notice);
-            }
-            break;
-         case 'ping':
-            if ($this->_isLocal($notice)) {
-                require_once INSTALLDIR . '/lib/ping.php';
-                return ping_broadcast_notice($notice);
-            }
-         case 'sms':
-            require_once(INSTALLDIR.'/lib/mail.php');
-            mail_broadcast_notice_sms($notice);
-            break;
-         case 'jabber':
-            require_once(INSTALLDIR.'/lib/jabber.php');
-            jabber_broadcast_notice($notice);
-            break;
-         case 'plugin':
-            Event::handle('HandleQueuedNotice', array(&$notice));
-            break;
-         default:
+        
+        $handler = $this->getHandler($queue);
+        if ($handler) {
+            $handler->handle_notice($notice);
+        } else {
             if (Event::handle('UnqueueHandleNotice', array(&$notice, $queue))) {
                 throw new ServerException("UnQueueManager: Unknown queue: $queue");
             }
         }
     }
-
-    function _isLocal($notice)
-    {
-        return ($notice->is_local == Notice::LOCAL_PUBLIC ||
-                $notice->is_local == Notice::LOCAL_NONPUBLIC);
-    }
-}
\ No newline at end of file
+}
diff --git a/lib/util.php b/lib/util.php
index 1237d718b..f161794e3 100644
--- a/lib/util.php
+++ b/lib/util.php
@@ -1132,8 +1132,9 @@ function common_log_line($priority, $msg)
 function common_request_id()
 {
     $pid = getmypid();
+    $server = common_config('site', 'server');
     if (php_sapi_name() == 'cli') {
-        return $pid;
+        return "$server:$pid";
     } else {
         static $req_id = null;
         if (!isset($req_id)) {
@@ -1143,7 +1144,7 @@ function common_request_id()
             $url = $_SERVER['REQUEST_URI'];
         }
         $method = $_SERVER['REQUEST_METHOD'];
-        return "$pid.$req_id $method $url";
+        return "$server:$pid.$req_id $method $url";
     }
 }
 
diff --git a/lib/xmppconfirmmanager.php b/lib/xmppconfirmmanager.php
new file mode 100644
index 000000000..ee4e294fd
--- /dev/null
+++ b/lib/xmppconfirmmanager.php
@@ -0,0 +1,168 @@
+<?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
new file mode 100644
index 000000000..9662e97d1
--- /dev/null
+++ b/lib/xmppmanager.php
@@ -0,0 +1,273 @@
+<?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); }
+
+/**
+ * XMPP background connection manager for XMPP-using queue handlers,
+ * allowing them to send outgoing messages on the right connection.
+ *
+ * Input is handled during socket select loop, keepalive pings during idle.
+ * Any incoming messages will be forwarded to the main XmppDaemon process,
+ * which handles direct user interaction.
+ *
+ * In a multi-site queuedaemon.php run, one connection will be instantiated
+ * for each site being handled by the current process that has XMPP enabled.
+ */
+
+class XmppManager extends IoManager
+{
+    protected $site = null;
+    protected $pingid = 0;
+    protected $lastping = null;
+
+    static protected $singletons = array();
+    
+    const PING_INTERVAL = 120;
+
+    /**
+     * Fetch the singleton XmppManager for the current site.
+     * @return mixed XmppManager, or false if unneeded
+     */
+    public static function get()
+    {
+        if (common_config('xmpp', 'enabled')) {
+            $site = common_config('site', 'server');
+            if (empty(self::$singletons[$site])) {
+                self::$singletons[$site] = new XmppManager();
+            }
+            return self::$singletons[$site];
+        } 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');
+    }
+
+    /**
+     * Initialize connection to server.
+     * @return boolean true on success
+     */
+    public function start($master)
+    {
+        parent::start($master);
+        $this->switchSite();
+
+        require_once "lib/jabber.php";
+
+        # Low priority; we don't want to receive messages
+
+        common_log(LOG_INFO, "INITIALIZE");
+        $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->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);
+
+        return !is_null($this->conn);
+    }
+
+    /**
+     * Message pump is triggered on socket input, so we only need an idle()
+     * call often enough to trigger our outgoing pings.
+     */
+    function timeout()
+    {
+        return self::PING_INTERVAL;
+    }
+
+    /**
+     * Lists the XMPP connection socket to allow i/o master to wake
+     * when input comes in here as well as from the queue source.
+     *
+     * @return array of resources
+     */
+    public function getSockets()
+    {
+        return array($this->conn->getSocket());
+    }
+
+    /**
+     * Process XMPP events that have come in over the wire.
+     * Side effects: may switch site configuration
+     * @fixme may kill process on XMPP error
+     * @param resource $socket
+     */
+    public function handleInput($socket)
+    {
+        $this->switchSite();
+
+        # Process the queue for as long as needed
+        try {
+            if ($this->conn) {
+                assert($socket === $this->conn->getSocket());
+                
+                common_log(LOG_DEBUG, "Servicing the XMPP queue.");
+                $this->stats('xmpp_process');
+                $this->conn->processTime(0);
+            }
+        } catch (XMPPHP_Exception $e) {
+            common_log(LOG_ERR, "Got an XMPPHP_Exception: " . $e->getMessage());
+            die($e->getMessage());
+        }
+    }
+
+    /**
+     * Idle processing for io manager's execution loop.
+     * Send keepalive pings to server.
+     *
+     * Side effect: kills process on exception from XMPP library.
+     *
+     * @fixme non-dying error handling
+     */
+    public function idle($timeout=0)
+    {
+        if ($this->conn) {
+            $now = time();
+            if (empty($this->lastping) || $now - $this->lastping > self::PING_INTERVAL) {
+                $this->switchSite();
+                try {
+                    $this->sendPing();
+                    $this->lastping = $now;
+                } catch (XMPPHP_Exception $e) {
+                    common_log(LOG_ERR, "Got an XMPPHP_Exception: " . $e->getMessage());
+                    die($e->getMessage());
+                }
+            }
+        }
+    }
+
+    /**
+     * Send a keepalive ping to the XMPP server.
+     */
+    protected function sendPing()
+    {
+        $jid = jabber_daemon_address().'/'.$this->resource();
+        $server = common_config('xmpp', 'server');
+
+        if (!isset($this->pingid)) {
+            $this->pingid = 0;
+        } else {
+            $this->pingid++;
+        }
+
+        common_log(LOG_DEBUG, "Sending ping #{$this->pingid}");
+
+        $this->conn->send("<iq from='{$jid}' to='{$server}' id='ping_{$this->pingid}' type='get'><ping xmlns='urn:xmpp:ping'/></iq>");
+    }
+
+    /**
+     * Callback for Jabber reconnect event
+     * @param $pl
+     */
+    function handle_reconnect(&$pl)
+    {
+        common_log(LOG_NOTICE, 'XMPP reconnected');
+
+        $this->conn->processUntil('session_start');
+        $this->conn->presence(null, 'available', null, 'available', -1);
+    }
+
+    /**
+     * 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
+     */
+    function forward_message(&$pl)
+    {
+        if ($pl['type'] != 'chat') {
+            common_log(LOG_DEBUG, 'Ignoring message of type ' . $pl['type'] . ' from ' . $pl['from']);
+            return;
+        }
+        $listener = $this->listener();
+        if (strtolower($listener) == strtolower($pl['from'])) {
+            common_log(LOG_WARNING, 'Ignoring loop message.');
+            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']));
+    }
+
+    /**
+     * 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)
+    {
+        $address = "<addresses xmlns='http://jabber.org/protocol/address'>\n";
+        $address .= "<address type='ofrom' jid='$from' />\n";
+        $address .= "</addresses>\n";
+        return $address;
+    }
+
+    /**
+     * Build the complete JID of the XmppDaemon process which
+     * handles primary XMPP input for this site.
+     *
+     * @return string Jabber ID
+     */
+    protected function listener()
+    {
+        if (common_config('xmpp', 'listener')) {
+            return common_config('xmpp', 'listener');
+        } else {
+            return jabber_daemon_address() . '/' . common_config('xmpp','resource') . 'daemon';
+        }
+    }
+
+    protected function resource()
+    {
+        return 'queue' . posix_getpid(); // @fixme PIDs won't be host-unique
+    }
+
+    /**
+     * 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/xmppqueuehandler.php b/lib/xmppqueuehandler.php
deleted file mode 100644
index f28fc9088..000000000
--- a/lib/xmppqueuehandler.php
+++ /dev/null
@@ -1,142 +0,0 @@
-<?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); }
-
-require_once(INSTALLDIR.'/lib/queuehandler.php');
-
-define('PING_INTERVAL', 120);
-
-/**
- * Common superclass for all XMPP-using queue handlers. They all need to
- * service their message queues on idle, and forward any incoming messages
- * to the XMPP listener connection. So, we abstract out common code to a
- * superclass.
- */
-
-class XmppQueueHandler extends QueueHandler
-{
-    var $pingid = 0;
-    var $lastping = null;
-
-    function start()
-    {
-        # Low priority; we don't want to receive messages
-
-        $this->log(LOG_INFO, "INITIALIZE");
-        $this->conn = jabber_connect($this->_id.$this->transport());
-
-        if (empty($this->conn)) {
-            $this->log(LOG_ERR, "Couldn't connect to server.");
-            return false;
-        }
-
-        $this->conn->addEventHandler('message', 'forward_message', $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);
-
-        return !is_null($this->conn);
-    }
-
-    function timeout()
-    {
-        return 10;
-    }
-
-    function handle_reconnect(&$pl)
-    {
-        $this->log(LOG_NOTICE, 'reconnected');
-
-        $this->conn->processUntil('session_start');
-        $this->conn->presence(null, 'available', null, 'available', -1);
-    }
-
-    function idle($timeout=0)
-    {
-        # Process the queue for as long as needed
-        try {
-            if ($this->conn) {
-                $this->log(LOG_DEBUG, "Servicing the XMPP queue.");
-                $this->conn->processTime($timeout);
-                $now = time();
-                if (empty($this->lastping) || $now - $this->lastping > PING_INTERVAL) {
-                    $this->sendPing();
-                    $this->lastping = $now;
-                }
-            }
-        } catch (XMPPHP_Exception $e) {
-            $this->log(LOG_ERR, "Got an XMPPHP_Exception: " . $e->getMessage());
-            die($e->getMessage());
-        }
-    }
-
-    function sendPing()
-    {
-        $jid = jabber_daemon_address().'/'.$this->_id.$this->transport();
-        $server = common_config('xmpp', 'server');
-
-        if (!isset($this->pingid)) {
-            $this->pingid = 0;
-        } else {
-            $this->pingid++;
-        }
-
-        $this->log(LOG_DEBUG, "Sending ping #{$this->pingid}");
-
-		$this->conn->send("<iq from='{$jid}' to='{$server}' id='ping_{$this->pingid}' type='get'><ping xmlns='urn:xmpp:ping'/></iq>");
-    }
-
-    function forward_message(&$pl)
-    {
-        if ($pl['type'] != 'chat') {
-            $this->log(LOG_DEBUG, 'Ignoring message of type ' . $pl['type'] . ' from ' . $pl['from']);
-            return;
-        }
-        $listener = $this->listener();
-        if (strtolower($listener) == strtolower($pl['from'])) {
-            $this->log(LOG_WARNING, 'Ignoring loop message.');
-            return;
-        }
-        $this->log(LOG_INFO, 'Forwarding message from ' . $pl['from'] . ' to ' . $listener);
-        $this->conn->message($this->listener(), $pl['body'], 'chat', null, $this->ofrom($pl['from']));
-    }
-
-    function ofrom($from)
-    {
-        $address = "<addresses xmlns='http://jabber.org/protocol/address'>\n";
-        $address .= "<address type='ofrom' jid='$from' />\n";
-        $address .= "</addresses>\n";
-        return $address;
-    }
-
-    function listener()
-    {
-        if (common_config('xmpp', 'listener')) {
-            return common_config('xmpp', 'listener');
-        } else {
-            return jabber_daemon_address() . '/' . common_config('xmpp','resource') . 'daemon';
-        }
-    }
-
-    function getSockets()
-    {
-        return array($this->conn->getSocket());
-    }
-}
diff --git a/plugins/Enjit/README b/plugins/Enjit/README
new file mode 100644
index 000000000..03f989490
--- /dev/null
+++ b/plugins/Enjit/README
@@ -0,0 +1,5 @@
+This doesn't seem to have been functional for a while; can't find other references
+to the enjit configuration or transport enqueuing. Keeping it in case someone
+wants to bring it up to date.
+
+-- brion vibber <brion@status.net> 2009-12-03
diff --git a/plugins/Enjit/enjitqueuehandler.php b/plugins/Enjit/enjitqueuehandler.php
new file mode 100644
index 000000000..f0e706b92
--- /dev/null
+++ b/plugins/Enjit/enjitqueuehandler.php
@@ -0,0 +1,91 @@
+<?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);
+}
+
+/**
+ * Queue handler for watching new notices and posting to enjit.
+ * @fixme is this actually being used/functional atm?
+ */
+class EnjitQueueHandler extends QueueHandler
+{
+    function transport()
+    {
+        return 'enjit';
+    }
+
+    function start()
+    {
+        $this->log(LOG_INFO, "Starting EnjitQueueHandler");
+        $this->log(LOG_INFO, "Broadcasting to ".common_config('enjit', 'apiurl'));
+        return true;
+    }
+
+    function handle_notice($notice)
+    {
+
+        $profile = Profile::staticGet($notice->profile_id);
+
+        $this->log(LOG_INFO, "Posting Notice ".$notice->id." from ".$profile->nickname);
+
+        if ( ! $notice->is_local ) {
+            $this->log(LOG_INFO, "Skipping remote notice");
+            return "skipped";
+        }
+
+        #
+        # Build an Atom message from the notice
+        #
+        $noticeurl = common_local_url('shownotice', array('notice' => $notice->id));
+        $msg = $profile->nickname . ': ' . $notice->content;
+
+        $atom  = "<entry xmlns='http://www.w3.org/2005/Atom'>\n";
+        $atom .= "<apisource>".common_config('enjit','source')."</apisource>\n";
+        $atom .= "<source>\n";
+        $atom .= "<title>" . $profile->nickname . " - " . common_config('site', 'name') . "</title>\n";
+        $atom .= "<link href='" . $profile->profileurl . "'/>\n";
+        $atom .= "<link rel='self' type='application/rss+xml' href='" . common_local_url('userrss', array('nickname' => $profile->nickname)) . "'/>\n";
+        $atom .= "<author><name>" . $profile->nickname . "</name></author>\n";
+        $atom .= "<icon>" . $profile->avatarUrl(AVATAR_PROFILE_SIZE) . "</icon>\n";
+        $atom .= "</source>\n";
+        $atom .= "<title>" . htmlspecialchars($msg) . "</title>\n";
+        $atom .= "<summary>" . htmlspecialchars($msg) . "</summary>\n";
+        $atom .= "<link rel='alternate' href='" . $noticeurl . "' />\n";
+        $atom .= "<id>". $notice->uri . "</id>\n";
+        $atom .= "<published>".common_date_w3dtf($notice->created)."</published>\n";
+        $atom .= "<updated>".common_date_w3dtf($notice->modified)."</updated>\n";
+        $atom .= "</entry>\n";
+
+        $url  = common_config('enjit', 'apiurl') . "/submit/". common_config('enjit','apikey');
+        $data = array(
+            'msg' => $atom,
+        );
+
+        #
+        # POST the message to $config['enjit']['apiurl']
+        #
+        $request = HTTPClient::start();
+        $response = $request->post($url, null, $data);
+
+        return $response->isOk();
+    }
+
+}
diff --git a/plugins/Facebook/FacebookPlugin.php b/plugins/Facebook/FacebookPlugin.php
index de91bf24a..4266b886d 100644
--- a/plugins/Facebook/FacebookPlugin.php
+++ b/plugins/Facebook/FacebookPlugin.php
@@ -114,6 +114,9 @@ class FacebookPlugin extends Plugin
         case 'FBCSettingsNav':
             include_once INSTALLDIR . '/plugins/Facebook/FBCSettingsNav.php';
             return false;
+        case 'FacebookQueueHandler':
+            include_once INSTALLDIR . '/plugins/Facebook/facebookqueuehandler.php';
+            return false;
         default:
             return true;
         }
@@ -508,50 +511,15 @@ class FacebookPlugin extends Plugin
     }
 
     /**
-     * broadcast the message when not using queuehandler
+     * Register Facebook notice queue handler
      *
-     * @param Notice &$notice the notice
-     * @param array  $queue   destination queue
+     * @param QueueManager $manager
      *
      * @return boolean hook return
      */
-
-    function onUnqueueHandleNotice(&$notice, $queue)
-    {
-        if (($queue == 'facebook') && ($this->_isLocal($notice))) {
-            facebookBroadcastNotice($notice);
-            return false;
-        }
-        return true;
-    }
-
-    /**
-     * Determine whether the notice was locally created
-     *
-     * @param Notice $notice the notice
-     *
-     * @return boolean locality
-     */
-
-    function _isLocal($notice)
-    {
-        return ($notice->is_local == Notice::LOCAL_PUBLIC ||
-                $notice->is_local == Notice::LOCAL_NONPUBLIC);
-    }
-
-    /**
-     * Add Facebook queuehandler to the list of daemons to start
-     *
-     * @param array $daemons the list fo daemons to run
-     *
-     * @return boolean hook return
-     *
-     */
-
-    function onGetValidDaemons($daemons)
+    function onEndInitializeQueueManager($manager)
     {
-        array_push($daemons, INSTALLDIR .
-                   '/plugins/Facebook/facebookqueuehandler.php');
+        $manager->connect('facebook', 'FacebookQueueHandler');
         return true;
     }
 
diff --git a/plugins/Facebook/facebookqueuehandler.php b/plugins/Facebook/facebookqueuehandler.php
old mode 100755
new mode 100644
index e4ae7d4ee..1778690e5
--- a/plugins/Facebook/facebookqueuehandler.php
+++ b/plugins/Facebook/facebookqueuehandler.php
@@ -1,4 +1,3 @@
-#!/usr/bin/env php
 <?php
 /*
  * StatusNet - the distributed open-source microblogging tool
@@ -18,21 +17,9 @@
  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  */
 
-define('INSTALLDIR', realpath(dirname(__FILE__) . '/../..'));
+if (!defined('STATUSNET') && !defined('LACONICA')) { exit(1); }
 
-$shortoptions = 'i::';
-$longoptions = array('id::');
-
-$helptext = <<<END_OF_FACEBOOK_HELP
-Daemon script for pushing new notices to Facebook.
-
-    -i --id           Identity (default none)
-
-END_OF_FACEBOOK_HELP;
-
-require_once INSTALLDIR . '/scripts/commandline.inc';
 require_once INSTALLDIR . '/plugins/Facebook/facebookutil.php';
-require_once INSTALLDIR . '/lib/queuehandler.php';
 
 class FacebookQueueHandler extends QueueHandler
 {
@@ -41,33 +28,24 @@ class FacebookQueueHandler extends QueueHandler
         return 'facebook';
     }
 
-    function start()
-    {
-        $this->log(LOG_INFO, "INITIALIZE");
-        return true;
-    }
-
     function handle_notice($notice)
     {
-        return facebookBroadcastNotice($notice);
+        if ($this->_isLocal($notice)) {
+            return facebookBroadcastNotice($notice);
+        }
+        return true;
     }
 
-    function finish()
+    /**
+     * Determine whether the notice was locally created
+     *
+     * @param Notice $notice the notice
+     *
+     * @return boolean locality
+     */
+    function _isLocal($notice)
     {
+        return ($notice->is_local == Notice::LOCAL_PUBLIC ||
+                $notice->is_local == Notice::LOCAL_NONPUBLIC);
     }
-
 }
-
-if (have_option('i')) {
-    $id = get_option_value('i');
-} else if (have_option('--id')) {
-    $id = get_option_value('--id');
-} else if (count($args) > 0) {
-    $id = $args[0];
-} else {
-    $id = null;
-}
-
-$handler = new FacebookQueueHandler($id);
-
-$handler->runOnce();
diff --git a/plugins/MemcachePlugin.php b/plugins/MemcachePlugin.php
index 5f93e9a83..fbc2802f7 100644
--- a/plugins/MemcachePlugin.php
+++ b/plugins/MemcachePlugin.php
@@ -133,6 +133,23 @@ class MemcachePlugin extends Plugin
         return false;
     }
 
+    function onStartCacheReconnect(&$success)
+    {
+        if (empty($this->_conn)) {
+            // nothing to do
+            return true;
+        }
+        if ($this->persistent) {
+            common_log(LOG_ERR, "Cannot close persistent memcached connection");
+            $success = false;
+        } else {
+            common_log(LOG_INFO, "Closing memcached connection");
+            $success = $this->_conn->close();
+            $this->_conn = null;
+        }
+        return false;
+    }
+
     /**
      * Ensure that a connection exists
      *
diff --git a/plugins/TwitterBridge/TwitterBridgePlugin.php b/plugins/TwitterBridge/TwitterBridgePlugin.php
index a87ee2894..57b3c1c99 100644
--- a/plugins/TwitterBridge/TwitterBridgePlugin.php
+++ b/plugins/TwitterBridge/TwitterBridgePlugin.php
@@ -112,7 +112,9 @@ class TwitterBridgePlugin extends Plugin
               strtolower(mb_substr($cls, 0, -6)) . '.php';
             return false;
         case 'TwitterOAuthClient':
-            include_once INSTALLDIR . '/plugins/TwitterBridge/twitteroauthclient.php';
+        case 'TwitterQueueHandler':
+            include_once INSTALLDIR . '/plugins/TwitterBridge/' .
+              strtolower($cls) . '.php';
             return false;
         default:
             return true;
@@ -138,48 +140,15 @@ class TwitterBridgePlugin extends Plugin
         return true;
     }
 
-    /**
-     * broadcast the message when not using queuehandler
-     *
-     * @param Notice &$notice the notice
-     * @param array  $queue   destination queue
-     *
-     * @return boolean hook return
-     */
-    function onUnqueueHandleNotice(&$notice, $queue)
-    {
-        if (($queue == 'twitter') && ($this->_isLocal($notice))) {
-            broadcast_twitter($notice);
-            return false;
-        }
-        return true;
-    }
-
-    /**
-     * Determine whether the notice was locally created
-     *
-     * @param Notice $notice
-     *
-     * @return boolean locality
-     */
-    function _isLocal($notice)
-    {
-        return ($notice->is_local == Notice::LOCAL_PUBLIC ||
-                $notice->is_local == Notice::LOCAL_NONPUBLIC);
-    }
-
     /**
      * Add Twitter bridge daemons to the list of daemons to start
      *
      * @param array $daemons the list fo daemons to run
      *
      * @return boolean hook return
-     *
      */
     function onGetValidDaemons($daemons)
     {
-        array_push($daemons, INSTALLDIR .
-                   '/plugins/TwitterBridge/daemons/twitterqueuehandler.php');
         array_push($daemons, INSTALLDIR .
                    '/plugins/TwitterBridge/daemons/synctwitterfriends.php');
 
@@ -191,6 +160,19 @@ class TwitterBridgePlugin extends Plugin
         return true;
     }
 
+    /**
+     * Register Twitter notice queue handler
+     *
+     * @param QueueManager $manager
+     *
+     * @return boolean hook return
+     */
+    function onEndInitializeQueueManager($manager)
+    {
+        $manager->connect('twitter', 'TwitterQueueHandler');
+        return true;
+    }
+
     function onPluginVersion(&$versions)
     {
         $versions[] = array('name' => 'TwitterBridge',
diff --git a/plugins/TwitterBridge/daemons/twitterqueuehandler.php b/plugins/TwitterBridge/daemons/twitterqueuehandler.php
deleted file mode 100755
index f0e76bb74..000000000
--- a/plugins/TwitterBridge/daemons/twitterqueuehandler.php
+++ /dev/null
@@ -1,73 +0,0 @@
-#!/usr/bin/env php
-<?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/>.
- */
-
-define('INSTALLDIR', realpath(dirname(__FILE__) . '/../../..'));
-
-$shortoptions = 'i::';
-$longoptions = array('id::');
-
-$helptext = <<<END_OF_ENJIT_HELP
-Daemon script for pushing new notices to Twitter.
-
-    -i --id           Identity (default none)
-
-END_OF_ENJIT_HELP;
-
-require_once INSTALLDIR . '/scripts/commandline.inc';
-require_once INSTALLDIR . '/lib/queuehandler.php';
-require_once INSTALLDIR . '/plugins/TwitterBridge/twitter.php';
-
-class TwitterQueueHandler extends QueueHandler
-{
-    function transport()
-    {
-        return 'twitter';
-    }
-
-    function start()
-    {
-        $this->log(LOG_INFO, "INITIALIZE");
-        return true;
-    }
-
-    function handle_notice($notice)
-    {
-        return broadcast_twitter($notice);
-    }
-
-    function finish()
-    {
-    }
-
-}
-
-if (have_option('i')) {
-    $id = get_option_value('i');
-} else if (have_option('--id')) {
-    $id = get_option_value('--id');
-} else if (count($args) > 0) {
-    $id = $args[0];
-} else {
-    $id = null;
-}
-
-$handler = new TwitterQueueHandler($id);
-
-$handler->runOnce();
diff --git a/plugins/TwitterBridge/twitterqueuehandler.php b/plugins/TwitterBridge/twitterqueuehandler.php
new file mode 100644
index 000000000..5089ca7b7
--- /dev/null
+++ b/plugins/TwitterBridge/twitterqueuehandler.php
@@ -0,0 +1,35 @@
+<?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); }
+
+require_once INSTALLDIR . '/plugins/TwitterBridge/twitter.php';
+
+class TwitterQueueHandler extends QueueHandler
+{
+    function transport()
+    {
+        return 'twitter';
+    }
+
+    function handle_notice($notice)
+    {
+        return broadcast_twitter($notice);
+    }
+}
diff --git a/scripts/enjitqueuehandler.php b/scripts/enjitqueuehandler.php
deleted file mode 100755
index afcac539a..000000000
--- a/scripts/enjitqueuehandler.php
+++ /dev/null
@@ -1,121 +0,0 @@
-#!/usr/bin/env php
-<?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/>.
- */
-
-define('INSTALLDIR', realpath(dirname(__FILE__) . '/..'));
-
-$shortoptions = 'i::';
-$longoptions = array('id::');
-
-$helptext = <<<END_OF_ENJIT_HELP
-Daemon script for watching new notices and posting to enjit.
-
-    -i --id           Identity (default none)
-
-END_OF_ENJIT_HELP;
-
-require_once INSTALLDIR.'/scripts/commandline.inc';
-
-require_once INSTALLDIR . '/lib/mail.php';
-require_once INSTALLDIR . '/lib/queuehandler.php';
-
-set_error_handler('common_error_handler');
-
-class EnjitQueueHandler extends QueueHandler
-{
-    function transport()
-    {
-        return 'enjit';
-    }
-
-    function start()
-    {
-        $this->log(LOG_INFO, "Starting EnjitQueueHandler");
-        $this->log(LOG_INFO, "Broadcasting to ".common_config('enjit', 'apiurl'));
-        return true;
-    }
-
-    function handle_notice($notice)
-    {
-
-        $profile = Profile::staticGet($notice->profile_id);
-
-        $this->log(LOG_INFO, "Posting Notice ".$notice->id." from ".$profile->nickname);
-
-        if ( ! $notice->is_local ) {
-            $this->log(LOG_INFO, "Skipping remote notice");
-            return "skipped";
-        }
-
-        #
-        # Build an Atom message from the notice
-        #
-        $noticeurl = common_local_url('shownotice', array('notice' => $notice->id));
-        $msg = $profile->nickname . ': ' . $notice->content;
-
-        $atom  = "<entry xmlns='http://www.w3.org/2005/Atom'>\n";
-        $atom .= "<apisource>".common_config('enjit','source')."</apisource>\n";
-        $atom .= "<source>\n";
-        $atom .= "<title>" . $profile->nickname . " - " . common_config('site', 'name') . "</title>\n";
-        $atom .= "<link href='" . $profile->profileurl . "'/>\n";
-        $atom .= "<link rel='self' type='application/rss+xml' href='" . common_local_url('userrss', array('nickname' => $profile->nickname)) . "'/>\n";
-        $atom .= "<author><name>" . $profile->nickname . "</name></author>\n";
-        $atom .= "<icon>" . $profile->avatarUrl(AVATAR_PROFILE_SIZE) . "</icon>\n";
-        $atom .= "</source>\n";
-        $atom .= "<title>" . htmlspecialchars($msg) . "</title>\n";
-        $atom .= "<summary>" . htmlspecialchars($msg) . "</summary>\n";
-        $atom .= "<link rel='alternate' href='" . $noticeurl . "' />\n";
-        $atom .= "<id>". $notice->uri . "</id>\n";
-        $atom .= "<published>".common_date_w3dtf($notice->created)."</published>\n";
-        $atom .= "<updated>".common_date_w3dtf($notice->modified)."</updated>\n";
-        $atom .= "</entry>\n";
-
-        $url  = common_config('enjit', 'apiurl') . "/submit/". common_config('enjit','apikey');
-        $data = array(
-            'msg' => $atom,
-        );
-
-        #
-        # POST the message to $config['enjit']['apiurl']
-        #
-        $request = HTTPClient::start();
-        $response = $request->post($url, null, $data);
-
-        return $response->isOk();
-    }
-
-}
-
-if (have_option('-i')) {
-    $id = get_option_value('-i');
-} else if (have_option('--id')) {
-    $id = get_option_value('--id');
-} else if (count($args) > 0) {
-    $id = $args[0];
-} else {
-    $id = null;
-}
-
-$handler = new EnjitQueueHandler($id);
-
-if ($handler->start()) {
-    $handler->handle_queue();
-}
-
-$handler->finish();
diff --git a/scripts/getvaliddaemons.php b/scripts/getvaliddaemons.php
index 99ad41b37..a332e06b5 100755
--- a/scripts/getvaliddaemons.php
+++ b/scripts/getvaliddaemons.php
@@ -37,19 +37,10 @@ require_once INSTALLDIR.'/scripts/commandline.inc';
 
 $daemons = array();
 
-$daemons[] = INSTALLDIR.'/scripts/pluginqueuehandler.php';
-$daemons[] = INSTALLDIR.'/scripts/ombqueuehandler.php';
-$daemons[] = INSTALLDIR.'/scripts/pingqueuehandler.php';
+$daemons[] = INSTALLDIR.'/scripts/queuedaemon.php';
 
 if(common_config('xmpp','enabled')) {
     $daemons[] = INSTALLDIR.'/scripts/xmppdaemon.php';
-    $daemons[] = INSTALLDIR.'/scripts/jabberqueuehandler.php';
-    $daemons[] = INSTALLDIR.'/scripts/publicqueuehandler.php';
-    $daemons[] = INSTALLDIR.'/scripts/xmppconfirmhandler.php';
-}
-
-if (common_config('sms', 'enabled')) {
-    $daemons[] = INSTALLDIR.'/scripts/smsqueuehandler.php';
 }
 
 if (Event::handle('GetValidDaemons', array(&$daemons))) {
diff --git a/scripts/handlequeued.php b/scripts/handlequeued.php
new file mode 100755
index 000000000..9031437aa
--- /dev/null
+++ b/scripts/handlequeued.php
@@ -0,0 +1,56 @@
+#!/usr/bin/env php
+<?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/>.
+ */
+
+define('INSTALLDIR', realpath(dirname(__FILE__) . '/..'));
+
+$helptext = <<<END_OF_QUEUE_HELP
+USAGE: handlequeued.php <queue> <notice id>
+Run a single queued notice through background processing
+as if it were being run through the queue.
+
+
+END_OF_QUEUE_HELP;
+
+require_once INSTALLDIR.'/scripts/commandline.inc';
+
+if (count($args) != 2) {
+    show_help();
+}
+
+$queue = trim($args[0]);
+$noticeId = intval($args[1]);
+
+$qm = QueueManager::get();
+$handler = $qm->getHandler($queue);
+if (!$handler) {
+    print "No handler for queue '$queue'.\n";
+    exit(1);
+}
+
+$notice = Notice::staticGet('id', $noticeId);
+if (empty($notice)) {
+    print "Invalid notice id $noticeId\n";
+    exit(1);
+}
+
+if (!$handler->handle_notice($notice)) {
+    print "Failed to handle notice id $noticeId on queue '$queue'.\n";
+    exit(1);
+}
diff --git a/scripts/jabberqueuehandler.php b/scripts/jabberqueuehandler.php
deleted file mode 100755
index 8f3a56944..000000000
--- a/scripts/jabberqueuehandler.php
+++ /dev/null
@@ -1,78 +0,0 @@
-#!/usr/bin/env php
-<?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/>.
- */
-
-define('INSTALLDIR', realpath(dirname(__FILE__) . '/..'));
-
-$shortoptions = 'i::';
-$longoptions = array('id::');
-
-$helptext = <<<END_OF_JABBER_HELP
-Daemon script for pushing new notices to Jabber users.
-
-    -i --id           Identity (default none)
-
-END_OF_JABBER_HELP;
-
-require_once INSTALLDIR.'/scripts/commandline.inc';
-
-require_once INSTALLDIR . '/lib/common.php';
-require_once INSTALLDIR . '/lib/jabber.php';
-require_once INSTALLDIR . '/lib/xmppqueuehandler.php';
-
-class JabberQueueHandler extends XmppQueueHandler
-{
-    var $conn = null;
-
-    function transport()
-    {
-        return 'jabber';
-    }
-
-    function handle_notice($notice)
-    {
-        try {
-            return jabber_broadcast_notice($notice);
-        } catch (XMPPHP_Exception $e) {
-            $this->log(LOG_ERR, "Got an XMPPHP_Exception: " . $e->getMessage());
-            exit(1);
-        }
-    }
-}
-
-// Abort immediately if xmpp is not enabled, otherwise the daemon chews up
-// lots of CPU trying to connect to unconfigured servers
-if (common_config('xmpp','enabled')==false) {
-    print "Aborting daemon - xmpp is disabled\n";
-    exit();
-}
-
-if (have_option('i')) {
-    $id = get_option_value('i');
-} else if (have_option('--id')) {
-    $id = get_option_value('--id');
-} else if (count($args) > 0) {
-    $id = $args[0];
-} else {
-    $id = null;
-}
-
-$handler = new JabberQueueHandler($id);
-
-$handler->runOnce();
diff --git a/scripts/ombqueuehandler.php b/scripts/ombqueuehandler.php
deleted file mode 100755
index be33b9821..000000000
--- a/scripts/ombqueuehandler.php
+++ /dev/null
@@ -1,87 +0,0 @@
-#!/usr/bin/env php
-<?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/>.
- */
-
-define('INSTALLDIR', realpath(dirname(__FILE__) . '/..'));
-
-$shortoptions = 'i::';
-$longoptions = array('id::');
-
-$helptext = <<<END_OF_OMB_HELP
-Daemon script for pushing new notices to OpenMicroBlogging subscribers.
-
-    -i --id           Identity (default none)
-
-END_OF_OMB_HELP;
-
-require_once INSTALLDIR.'/scripts/commandline.inc';
-
-require_once INSTALLDIR . '/lib/omb.php';
-require_once INSTALLDIR . '/lib/queuehandler.php';
-
-set_error_handler('common_error_handler');
-
-class OmbQueueHandler extends QueueHandler
-{
-
-    function transport()
-    {
-        return 'omb';
-    }
-
-    function start()
-    {
-        $this->log(LOG_INFO, "INITIALIZE");
-        return true;
-    }
-
-    function handle_notice($notice)
-    {
-        if ($this->is_remote($notice)) {
-            $this->log(LOG_DEBUG, 'Ignoring remote notice ' . $notice->id);
-            return true;
-        } else {
-            return omb_broadcast_notice($notice);
-        }
-    }
-
-    function finish()
-    {
-    }
-
-    function is_remote($notice)
-    {
-        $user = User::staticGet($notice->profile_id);
-        return is_null($user);
-    }
-}
-
-if (have_option('i')) {
-    $id = get_option_value('i');
-} else if (have_option('--id')) {
-    $id = get_option_value('--id');
-} else if (count($args) > 0) {
-    $id = $args[0];
-} else {
-    $id = null;
-}
-
-$handler = new OmbQueueHandler($id);
-
-$handler->runOnce();
diff --git a/scripts/pingqueuehandler.php b/scripts/pingqueuehandler.php
deleted file mode 100755
index c92337e36..000000000
--- a/scripts/pingqueuehandler.php
+++ /dev/null
@@ -1,69 +0,0 @@
-#!/usr/bin/env php
-<?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/>.
- */
-
-define('INSTALLDIR', realpath(dirname(__FILE__) . '/..'));
-
-$shortoptions = 'i::';
-$longoptions = array('id::');
-
-$helptext = <<<END_OF_PING_HELP
-Daemon script for pushing new notices to ping servers.
-
-    -i --id           Identity (default none)
-
-END_OF_PING_HELP;
-
-require_once INSTALLDIR.'/scripts/commandline.inc';
-
-require_once INSTALLDIR . '/lib/ping.php';
-require_once INSTALLDIR . '/lib/queuehandler.php';
-
-class PingQueueHandler extends QueueHandler {
-
-	function transport() {
-		return 'ping';
-	}
-
-	function start() {
-		$this->log(LOG_INFO, "INITIALIZE");
-		return true;
-	}
-
-	function handle_notice($notice) {
-		return ping_broadcast_notice($notice);
-	}
-
-	function finish() {
-	}
-}
-
-if (have_option('i')) {
-    $id = get_option_value('i');
-} else if (have_option('--id')) {
-    $id = get_option_value('--id');
-} else if (count($args) > 0) {
-    $id = $args[0];
-} else {
-    $id = null;
-}
-
-$handler = new PingQueueHandler($id);
-
-$handler->runOnce();
diff --git a/scripts/pluginqueuehandler.php b/scripts/pluginqueuehandler.php
deleted file mode 100755
index fa39bdda6..000000000
--- a/scripts/pluginqueuehandler.php
+++ /dev/null
@@ -1,64 +0,0 @@
-#!/usr/bin/env php
-<?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/>.
- */
-
-define('INSTALLDIR', realpath(dirname(__FILE__) . '/..'));
-
-$shortoptions = 'i::';
-$longoptions = array('id::');
-
-$helptext = <<<END_OF_OMB_HELP
-Daemon script for letting plugins handle stuff at queue time
-
-    -i --id           Identity (default none)
-
-END_OF_OMB_HELP;
-
-require_once INSTALLDIR.'/scripts/commandline.inc';
-require_once INSTALLDIR . '/lib/queuehandler.php';
-
-class PluginQueueHandler extends QueueHandler
-{
-
-    function transport()
-    {
-        return 'plugin';
-    }
-
-    function start()
-    {
-        $this->log(LOG_INFO, "INITIALIZE");
-        return true;
-    }
-
-    function handle_notice($notice)
-    {
-        Event::handle('HandleQueuedNotice', array(&$notice));
-        return true;
-    }
-}
-
-if (have_option('i', 'id')) {
-    $id = get_option_value('i', 'id');
-} else {
-    $id = null;
-}
-
-$handler = new PluginQueueHandler($id);
-$handler->runOnce();
diff --git a/scripts/publicqueuehandler.php b/scripts/publicqueuehandler.php
deleted file mode 100755
index 50a11bcba..000000000
--- a/scripts/publicqueuehandler.php
+++ /dev/null
@@ -1,76 +0,0 @@
-#!/usr/bin/env php
-<?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/>.
- */
-
-define('INSTALLDIR', realpath(dirname(__FILE__) . '/..'));
-
-$shortoptions = 'i::';
-$longoptions = array('id::');
-
-$helptext = <<<END_OF_PUBLIC_HELP
-Daemon script for pushing new notices to public XMPP subscribers.
-
-    -i --id           Identity (default none)
-
-END_OF_PUBLIC_HELP;
-
-require_once INSTALLDIR.'/scripts/commandline.inc';
-
-require_once INSTALLDIR . '/lib/jabber.php';
-require_once INSTALLDIR . '/lib/xmppqueuehandler.php';
-
-class PublicQueueHandler extends XmppQueueHandler
-{
-
-    function transport()
-    {
-        return 'public';
-    }
-
-    function handle_notice($notice)
-    {
-        try {
-            return jabber_public_notice($notice);
-        } catch (XMPPHP_Exception $e) {
-            $this->log(LOG_ERR, "Got an XMPPHP_Exception: " . $e->getMessage());
-            die($e->getMessage());
-        }
-    }
-}
-
-// Abort immediately if xmpp is not enabled, otherwise the daemon chews up
-// lots of CPU trying to connect to unconfigured servers
-if (common_config('xmpp','enabled')==false) {
-    print "Aborting daemon - xmpp is disabled\n";
-    exit();
-}
-
-if (have_option('i')) {
-    $id = get_option_value('i');
-} else if (have_option('--id')) {
-    $id = get_option_value('--id');
-} else if (count($args) > 0) {
-    $id = $args[0];
-} else {
-    $id = null;
-}
-
-$handler = new PublicQueueHandler($id);
-
-$handler->runOnce();
diff --git a/scripts/queuedaemon.php b/scripts/queuedaemon.php
new file mode 100755
index 000000000..8ef364fe7
--- /dev/null
+++ b/scripts/queuedaemon.php
@@ -0,0 +1,265 @@
+#!/usr/bin/env php
+<?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/>.
+ */
+
+define('INSTALLDIR', realpath(dirname(__FILE__) . '/..'));
+
+$shortoptions = 'fi:at:';
+$longoptions = array('id=', 'foreground', 'all', 'threads=');
+
+/**
+ * Attempts to get a count of the processors available on the current system
+ * to fan out multiple threads.
+ *
+ * Recognizes Linux and Mac OS X; others will return default of 1.
+ *
+ * @return intval
+ */
+function getProcessorCount()
+{
+    $cpus = 0;
+    switch (PHP_OS) {
+    case 'Linux':
+        $cpuinfo = file('/proc/cpuinfo');
+        foreach (file('/proc/cpuinfo') as $line) {
+            if (preg_match('/^processor\s+:\s+(\d+)\s?$/', $line)) {
+                $cpus++;
+            }
+        }
+        break;
+    case 'Darwin':
+        $cpus = intval(shell_exec("/usr/sbin/sysctl -n hw.ncpu 2>/dev/null"));
+        break;
+    }
+    if ($cpus) {
+        return $cpus;
+    }
+    return 1;
+}
+
+$threads = getProcessorCount();
+$helptext = <<<END_OF_QUEUE_HELP
+Daemon script for running queued items.
+
+    -i --id           Identity (default none)
+    -f --foreground   Stay in the foreground (default background)
+    -a --all          Handle queues for all local sites
+                      (requires Stomp queue handler, status_network setup)
+    -t --threads=<n>  Spawn <n> processing threads (default $threads)
+
+
+END_OF_QUEUE_HELP;
+
+require_once INSTALLDIR.'/scripts/commandline.inc';
+
+require_once(INSTALLDIR.'/lib/daemon.php');
+require_once(INSTALLDIR.'/classes/Queue_item.php');
+require_once(INSTALLDIR.'/classes/Notice.php');
+
+define('CLAIM_TIMEOUT', 1200);
+
+/**
+ * Queue handling daemon...
+ *
+ * The queue daemon by default launches in the background, at which point
+ * it'll pass control to the configured QueueManager class to poll for updates.
+ *
+ * We can then pass individual items through the QueueHandler subclasses
+ * they belong to.
+ */
+class QueueDaemon extends Daemon
+{
+    protected $allsites;
+    protected $threads=1;
+
+    function __construct($id=null, $daemonize=true, $threads=1, $allsites=false)
+    {
+        parent::__construct($daemonize);
+
+        if ($id) {
+            $this->set_id($id);
+        }
+        $this->all = $allsites;
+        $this->threads = $threads;
+    }
+
+    /**
+     * 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.
+     *
+     * @return int timeout in seconds
+     */
+    function timeout()
+    {
+        return 60;
+    }
+
+    function name()
+    {
+        return strtolower(get_class($this).'.'.$this->get_id());
+    }
+
+    function run()
+    {
+        if ($this->threads > 1) {
+            return $this->runThreads();
+        } else {
+            return $this->runLoop();
+        }
+    }
+    
+    function runThreads()
+    {
+        $children = array();
+        for ($i = 1; $i <= $this->threads; $i++) {
+            $pid = pcntl_fork();
+            if ($pid < 0) {
+                print "Couldn't fork for thread $i; aborting\n";
+                exit(1);
+            } else if ($pid == 0) {
+                $this->runChild($i);
+                exit(0);
+            } 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) {
+                    print "Couldn't fork to respawn thread $i; aborting thread.\n";
+                } else if ($pid == 0) {
+                    $this->runChild($i);
+                    exit(0);
+                } else {
+                    $this->log(LOG_INFO, "Respawned thread $i as pid $pid");
+                    $children[$i] = $pid;
+                }
+            }
+        }
+        $this->log(LOG_INFO, "All child processes complete.");
+        return true;
+    }
+
+    function runChild($thread)
+    {
+        $this->set_id($this->get_id() . "." . $thread);
+        $this->resetDb();
+        $this->runLoop();
+    }
+
+    /**
+     * Reconnect to the database for each child process,
+     * or they'll get very confused trying to use the
+     * same socket.
+     */
+    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;
+        }
+    }
+
+    /**
+     * 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 on to the QueueHandler's 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.
+     *
+     * @return boolean true on success, false on failure
+     */
+    function runLoop()
+    {
+        $this->log(LOG_INFO, 'checking for queued notices');
+
+        $master = new IoMaster($this->get_id());
+        $master->init($this->all);
+        $master->service();
+
+        $this->log(LOG_INFO, 'finished servicing the queue');
+
+        $this->log(LOG_INFO, 'terminating normally');
+
+        return true;
+    }
+
+    function log($level, $msg)
+    {
+        common_log($level, get_class($this) . ' ('. $this->get_id() .'): '.$msg);
+    }
+}
+
+if (have_option('i')) {
+    $id = get_option_value('i');
+} else if (have_option('--id')) {
+    $id = get_option_value('--id');
+} else if (count($args) > 0) {
+    $id = $args[0];
+} else {
+    $id = null;
+}
+
+if (have_option('t')) {
+    $threads = intval(get_option_value('t'));
+} else if (have_option('--threads')) {
+    $threads = intval(get_option_value('--threads'));
+} else {
+    $threads = 0;
+}
+if (!$threads) {
+    $threads = getProcessorCount();
+}
+
+$daemonize = !(have_option('f') || have_option('--foreground'));
+$all = have_option('a') || have_option('--all');
+
+$daemon = new QueueDaemon($id, $daemonize, $threads, $all);
+$daemon->runOnce();
+
diff --git a/scripts/smsqueuehandler.php b/scripts/smsqueuehandler.php
deleted file mode 100755
index 6583a77da..000000000
--- a/scripts/smsqueuehandler.php
+++ /dev/null
@@ -1,73 +0,0 @@
-#!/usr/bin/env php
-<?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/>.
- */
-
-define('INSTALLDIR', realpath(dirname(__FILE__) . '/..'));
-
-$shortoptions = 'i::';
-$longoptions = array('id::');
-
-$helptext = <<<END_OF_SMS_HELP
-Daemon script for pushing new notices to local subscribers using SMS.
-
-    -i --id           Identity (default none)
-
-END_OF_SMS_HELP;
-
-require_once INSTALLDIR.'/scripts/commandline.inc';
-
-require_once INSTALLDIR . '/lib/mail.php';
-require_once INSTALLDIR . '/lib/queuehandler.php';
-
-class SmsQueueHandler extends QueueHandler
-{
-    function transport()
-    {
-        return 'sms';
-    }
-
-    function start()
-    {
-        $this->log(LOG_INFO, "INITIALIZE");
-        return true;
-    }
-
-    function handle_notice($notice)
-    {
-        return mail_broadcast_notice_sms($notice);
-    }
-
-    function finish()
-    {
-    }
-}
-
-if (have_option('i')) {
-    $id = get_option_value('i');
-} else if (have_option('--id')) {
-    $id = get_option_value('--id');
-} else if (count($args) > 0) {
-    $id = $args[0];
-} else {
-    $id = null;
-}
-
-$handler = new SmsQueueHandler($id);
-
-$handler->runOnce();
diff --git a/scripts/xmppconfirmhandler.php b/scripts/xmppconfirmhandler.php
deleted file mode 100755
index 2e3974136..000000000
--- a/scripts/xmppconfirmhandler.php
+++ /dev/null
@@ -1,161 +0,0 @@
-#!/usr/bin/env php
-<?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/>.
- */
-
-define('INSTALLDIR', realpath(dirname(__FILE__) . '/..'));
-
-$shortoptions = 'i::';
-$longoptions = array('id::');
-
-$helptext = <<<END_OF_JABBER_HELP
-Daemon script for pushing new confirmations to Jabber users.
-
-    -i --id           Identity (default none)
-
-END_OF_JABBER_HELP;
-
-require_once INSTALLDIR.'/scripts/commandline.inc';
-require_once INSTALLDIR . '/lib/jabber.php';
-require_once INSTALLDIR . '/lib/xmppqueuehandler.php';
-
-class XmppConfirmHandler extends XmppQueueHandler
-{
-    var $_id = 'confirm';
-
-    function class_name()
-    {
-        return 'XmppConfirmHandler';
-    }
-
-    function run()
-    {
-        if (!$this->start()) {
-            return false;
-        }
-        $this->log(LOG_INFO, 'checking for queued confirmations');
-        do {
-            $confirm = $this->next_confirm();
-            if ($confirm) {
-                $this->log(LOG_INFO, 'Sending confirmation for ' . $confirm->address);
-                $user = User::staticGet($confirm->user_id);
-                if (!$user) {
-                    $this->log(LOG_WARNING, 'Confirmation for unknown user ' . $confirm->user_id);
-                    continue;
-                }
-                $success = jabber_confirm_address($confirm->code,
-                                                  $user->nickname,
-                                                  $confirm->address);
-                if (!$success) {
-                    $this->log(LOG_ERR, 'Confirmation failed for ' . $confirm->address);
-                    # Just let the claim age out; hopefully things work then
-                    continue;
-                } else {
-                    $this->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__);
-                        continue;
-                    }
-                    $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
-                        continue;
-                    }
-                    $dupe->free();
-                    unset($dupe);
-                }
-                $user->free();
-                unset($user);
-                $confirm->free();
-                unset($confirm);
-                $this->idle(0);
-            } else {
-#                $this->clear_old_confirm_claims();
-                $this->idle(10);
-            }
-        } while (true);
-        if (!$this->finish()) {
-            return false;
-        }
-        return true;
-    }
-
-    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)) {
-            $this->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) {
-                $this->log(LOG_INFO, 'Succeeded in claim! '. $result);
-                return $confirm;
-            } else {
-                $this->log(LOG_INFO, 'Failed in claim!');
-                return false;
-            }
-        }
-        return null;
-    }
-
-    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);
-    }
-}
-
-// Abort immediately if xmpp is not enabled, otherwise the daemon chews up
-// lots of CPU trying to connect to unconfigured servers
-if (common_config('xmpp','enabled')==false) {
-    print "Aborting daemon - xmpp is disabled\n";
-    exit();
-}
-
-if (have_option('i')) {
-    $id = get_option_value('i');
-} else if (have_option('--id')) {
-    $id = get_option_value('--id');
-} else if (count($args) > 0) {
-    $id = $args[0];
-} else {
-    $id = null;
-}
-
-$handler = new XmppConfirmHandler($id);
-
-$handler->runOnce();
-
-- 
cgit v1.2.3-54-g00ecf


From fb0d837ddc572993f8c6d1db3de2c96d70101841 Mon Sep 17 00:00:00 2001
From: Evan Prodromou <evan@status.net>
Date: Tue, 12 Jan 2010 23:41:33 -0800
Subject: remove transaction for Notice save; causes deadlocks

---
 classes/Notice.php | 4 ----
 1 file changed, 4 deletions(-)

(limited to 'classes')

diff --git a/classes/Notice.php b/classes/Notice.php
index 6284b8ca5..306956422 100644
--- a/classes/Notice.php
+++ b/classes/Notice.php
@@ -299,8 +299,6 @@ class Notice extends Memcached_DataObject
 
             // XXX: some of these functions write to the DB
 
-            $notice->query('BEGIN');
-
             $id = $notice->insert();
 
             if (!$id) {
@@ -342,8 +340,6 @@ class Notice extends Memcached_DataObject
 
             $notice->saveUrls();
 
-            $notice->query('COMMIT');
-
             Event::handle('EndNoticeSave', array($notice));
         }
 
-- 
cgit v1.2.3-54-g00ecf


From 3d579fc58007ccb20657bc3959aad220e72e8f30 Mon Sep 17 00:00:00 2001
From: Evan Prodromou <evan@status.net>
Date: Wed, 13 Jan 2010 00:16:54 -0800
Subject: memoize Notice::whoGets()

---
 classes/Notice.php | 14 ++++++++++++++
 1 file changed, 14 insertions(+)

(limited to 'classes')

diff --git a/classes/Notice.php b/classes/Notice.php
index 306956422..c8edc98e1 100644
--- a/classes/Notice.php
+++ b/classes/Notice.php
@@ -828,6 +828,15 @@ class Notice extends Memcached_DataObject
 
     function whoGets()
     {
+        $c = self::memcache();
+
+        if (!empty($c)) {
+            $ni = $c->get(common_cache_key('notice:who_gets:'.$this->id));
+            if ($ni !== false) {
+                return $ni;
+            }
+        }
+
         $users = $this->getSubscribedUsers();
 
         // FIXME: kind of ignoring 'transitional'...
@@ -867,6 +876,11 @@ class Notice extends Memcached_DataObject
             }
         }
 
+        if (!empty($c)) {
+            // XXX: pack this data better
+            $c->set(common_cache_key('notice:who_gets:'.$this->id), $ni);
+        }
+
         return $ni;
     }
 
-- 
cgit v1.2.3-54-g00ecf


From 1bbd19169f22b9405b4ee1964da7e94a228a8dd5 Mon Sep 17 00:00:00 2001
From: Evan Prodromou <evan@status.net>
Date: Wed, 13 Jan 2010 00:44:29 -0800
Subject: unimplement repeatedToMe() until we figure out how

---
 classes/User.php | 51 +--------------------------------------------------
 1 file changed, 1 insertion(+), 50 deletions(-)

(limited to 'classes')

diff --git a/classes/User.php b/classes/User.php
index bde3f71b9..d6b52be01 100644
--- a/classes/User.php
+++ b/classes/User.php
@@ -899,56 +899,7 @@ class User extends Memcached_DataObject
 
     function repeatedToMe($offset=0, $limit=20, $since_id=null, $max_id=null)
     {
-        $ids = Notice::stream(array($this, '_repeatedToMeDirect'),
-                              array(),
-                              'user:repeated_to_me:'.$this->id,
-                              $offset, $limit, $since_id, $max_id, null);
-
-        return Notice::getStreamByIds($ids);
-    }
-
-    function _repeatedToMeDirect($offset, $limit, $since_id, $max_id, $since)
-    {
-        $qry =
-          'SELECT notice.id AS id ' .
-          'FROM notice JOIN notice_inbox ON notice.id = notice_inbox.notice_id ' .
-          'WHERE notice_inbox.user_id = ' . $this->id . ' ' .
-          'AND notice.repeat_of IS NOT NULL ';
-
-        if ($since_id != 0) {
-            $qry .= 'AND notice.id > ' . $since_id . ' ';
-        }
-
-        if ($max_id != 0) {
-            $qry .= 'AND notice.id <= ' . $max_id . ' ';
-        }
-
-        if (!is_null($since)) {
-            $qry .= 'AND notice.modified > \'' . date('Y-m-d H:i:s', $since) . '\' ';
-        }
-
-        // NOTE: we sort by fave time, not by notice time!
-
-        $qry .= 'ORDER BY notice.id DESC ';
-
-        if (!is_null($offset)) {
-            $qry .= "LIMIT $limit OFFSET $offset";
-        }
-
-        $ids = array();
-
-        $notice = new Notice();
-
-        $notice->query($qry);
-
-        while ($notice->fetch()) {
-            $ids[] = $notice->id;
-        }
-
-        $notice->free();
-        $notice = NULL;
-
-        return $ids;
+        throw new Exception("Not implemented since inbox change.");
     }
 
     function shareLocation()
-- 
cgit v1.2.3-54-g00ecf


From 5397249e505b6e82c814c2a7ce5bc1bfaad2adf3 Mon Sep 17 00:00:00 2001
From: Evan Prodromou <evan@status.net>
Date: Wed, 13 Jan 2010 00:45:12 -0800
Subject: remove vestiges of Notice_inbox from cache clearing code in Notice

---
 classes/Notice.php | 3 ---
 1 file changed, 3 deletions(-)

(limited to 'classes')

diff --git a/classes/Notice.php b/classes/Notice.php
index c8edc98e1..d651ef80e 100644
--- a/classes/Notice.php
+++ b/classes/Notice.php
@@ -498,9 +498,6 @@ class Notice extends Memcached_DataObject
                     $original->free();
                     unset($original);
                 }
-
-                $ni->free();
-                unset($ni);
             }
         }
     }
-- 
cgit v1.2.3-54-g00ecf


From 8bfa7fdeaf85091907e5c0c9f181655a9a5dcdf5 Mon Sep 17 00:00:00 2001
From: Evan Prodromou <evan@status.net>
Date: Wed, 13 Jan 2010 00:47:12 -0800
Subject: repeats don't get posted to groups

---
 classes/Notice.php | 6 ++++++
 1 file changed, 6 insertions(+)

(limited to 'classes')

diff --git a/classes/Notice.php b/classes/Notice.php
index d651ef80e..f48e77c1d 100644
--- a/classes/Notice.php
+++ b/classes/Notice.php
@@ -919,6 +919,12 @@ class Notice extends Memcached_DataObject
 
     function saveGroups()
     {
+        // Don't save groups for repeats
+
+        if (!empty($this->repeat_of)) {
+            return array();
+        }
+
         $groups = array();
 
         /* extract all !group */
-- 
cgit v1.2.3-54-g00ecf


From 43ff54218934663910f11762ee2a1c1083b119c7 Mon Sep 17 00:00:00 2001
From: Evan Prodromou <evan@status.net>
Date: Wed, 13 Jan 2010 01:13:06 -0800
Subject: Don't save reply info for repeats

---
 classes/Notice.php | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

(limited to 'classes')

diff --git a/classes/Notice.php b/classes/Notice.php
index f48e77c1d..02cd20391 100644
--- a/classes/Notice.php
+++ b/classes/Notice.php
@@ -275,7 +275,6 @@ class Notice extends Memcached_DataObject
 
         if (isset($repeat_of)) {
             $notice->repeat_of = $repeat_of;
-            $notice->reply_to = $repeat_of;
         } else {
             $notice->reply_to = self::getReplyTo($reply_to, $profile_id, $source, $final);
         }
@@ -995,6 +994,12 @@ class Notice extends Memcached_DataObject
      */
     function saveReplies()
     {
+        // Don't save reply data for repeats
+
+        if (!empty($this->repeat_of)) {
+            return array();
+        }
+
         // Alternative reply format
         $tname = false;
         if (preg_match('/^T ([A-Z0-9]{1,64}) /', $this->content, $match)) {
-- 
cgit v1.2.3-54-g00ecf


From 430bd69312a27f9f97bda78566a78c9f7eec1f14 Mon Sep 17 00:00:00 2001
From: Evan Prodromou <evan@status.net>
Date: Wed, 13 Jan 2010 02:16:13 -0800
Subject: add events for subscribing to people and joining groups

---
 EVENTS.txt               | 32 ++++++++++++++++++
 actions/joingroup.php    | 16 ++++-----
 actions/leavegroup.php   | 21 ++++--------
 classes/Group_member.php | 37 ++++++++++++++++++++
 lib/command.php          | 45 ++++++++++---------------
 lib/subs.php             | 88 +++++++++++++++++++++++++++++-------------------
 6 files changed, 153 insertions(+), 86 deletions(-)

(limited to 'classes')

diff --git a/EVENTS.txt b/EVENTS.txt
index 64e345b69..e6400244e 100644
--- a/EVENTS.txt
+++ b/EVENTS.txt
@@ -655,3 +655,35 @@ StartUnblockProfile: when we're about to unblock
 EndUnblockProfile: when an unblock has succeeded
 - $user: the person doing the unblock
 - $profile: the person unblocked, can be remote
+
+StartSubscribe: when a subscription is starting
+- $user: the person subscribing
+- $other: the person being subscribed to
+
+EndSubscribe: when a subscription is finished
+- $user: the person subscribing
+- $other: the person being subscribed to
+
+StartUnsubscribe: when an unsubscribe is starting
+- $user: the person unsubscribing
+- $other: the person being unsubscribed from
+
+EndUnsubscribe: when an unsubscribe is done
+- $user: the person unsubscribing
+- $other: the person being unsubscribed to
+
+StartJoinGroup: when a user is joining a group
+- $group: the group being joined
+- $user: the user joining
+
+EndJoinGroup: when a user finishes joining a group
+- $group: the group being joined
+- $user: the user joining
+
+StartLeaveGroup: when a user is leaving a group
+- $group: the group being left
+- $user: the user leaving
+
+EndLeaveGroup: when a user has left a group
+- $group: the group being left
+- $user: the user leaving
diff --git a/actions/joingroup.php b/actions/joingroup.php
index 05e33e7cb..235e5ab4c 100644
--- a/actions/joingroup.php
+++ b/actions/joingroup.php
@@ -115,16 +115,12 @@ class JoingroupAction extends Action
 
         $cur = common_current_user();
 
-        $member = new Group_member();
-
-        $member->group_id   = $this->group->id;
-        $member->profile_id = $cur->id;
-        $member->created    = common_sql_now();
-
-        $result = $member->insert();
-
-        if (!$result) {
-            common_log_db_error($member, 'INSERT', __FILE__);
+        try {
+            if (Event::handle('StartJoinGroup', array($this->group, $cur))) {
+                Group_member::join($this->group->id, $cur->id);
+                Event::handle('EndJoinGroup', array($this->group, $cur));
+            }
+        } catch (Exception $e) {
             $this->serverError(sprintf(_('Could not join user %1$s to group %2$s.'),
                                        $cur->nickname, $this->group->nickname));
         }
diff --git a/actions/leavegroup.php b/actions/leavegroup.php
index b0f973e1a..9b9d83b6c 100644
--- a/actions/leavegroup.php
+++ b/actions/leavegroup.php
@@ -110,22 +110,15 @@ class LeavegroupAction extends Action
 
         $cur = common_current_user();
 
-        $member = new Group_member();
-
-        $member->group_id   = $this->group->id;
-        $member->profile_id = $cur->id;
-
-        if (!$member->find(true)) {
-            $this->serverError(_('Could not find membership record.'));
-            return;
-        }
-
-        $result = $member->delete();
-
-        if (!$result) {
-            common_log_db_error($member, 'DELETE', __FILE__);
+        try {
+            if (Event::handle('StartLeaveGroup', array($this->group, $cur))) {
+                Group_member::leave($this->group->id, $cur->id);
+                Event::handle('EndLeaveGroup', array($this->group, $cur));
+            }
+        } catch (Exception $e) {
             $this->serverError(sprintf(_('Could not remove user %1$s from group %2$s.'),
                                        $cur->nickname, $this->group->nickname));
+            return;
         }
 
         if ($this->boolean('ajax')) {
diff --git a/classes/Group_member.php b/classes/Group_member.php
index 069b2c7a1..7b1760f76 100644
--- a/classes/Group_member.php
+++ b/classes/Group_member.php
@@ -25,4 +25,41 @@ class Group_member extends Memcached_DataObject
     {
         return Memcached_DataObject::pkeyGet('Group_member', $kv);
     }
+
+    static function join($group_id, $profile_id)
+    {
+        $member = new Group_member();
+
+        $member->group_id   = $group_id;
+        $member->profile_id = $profile_id;
+        $member->created    = common_sql_now();
+
+        $result = $member->insert();
+
+        if (!$result) {
+            common_log_db_error($member, 'INSERT', __FILE__);
+            throw new Exception(_("Group join failed."));
+        }
+
+        return true;
+    }
+
+    static function leave($group_id, $profile_id)
+    {
+        $member = Group_member::pkeyGet(array('group_id' => $group_id,
+                                              'profile_id' => $profile_id));
+
+        if (empty($member)) {
+            throw new Exception(_("Not part of group."));
+        }
+
+        $result = $member->delete();
+
+        if (!$result) {
+            common_log_db_error($member, 'INSERT', __FILE__);
+            throw new Exception(_("Group leave failed."));
+        }
+
+        return true;
+    }
 }
diff --git a/lib/command.php b/lib/command.php
index f846fb823..c0a32e1b1 100644
--- a/lib/command.php
+++ b/lib/command.php
@@ -222,18 +222,15 @@ class JoinCommand extends Command
             return;
         }
 
-        $member = new Group_member();
-
-        $member->group_id   = $group->id;
-        $member->profile_id = $cur->id;
-        $member->created    = common_sql_now();
-
-        $result = $member->insert();
-        if (!$result) {
-          common_log_db_error($member, 'INSERT', __FILE__);
-          $channel->error($cur, sprintf(_('Could not join user %s to group %s'),
-                                       $cur->nickname, $group->nickname));
-          return;
+        try {
+            if (Event::handle('StartJoinGroup', array($group, $cur))) {
+                Group_member::join($group->id, $cur->id);
+                Event::handle('EndJoinGroup', array($group, $cur));
+            }
+        } catch (Exception $e) {
+            $channel->error($cur, sprintf(_('Could not join user %s to group %s'),
+                                          $cur->nickname, $group->nickname));
+            return;
         }
 
         $channel->output($cur, sprintf(_('%s joined group %s'),
@@ -269,21 +266,15 @@ class DropCommand extends Command
             return;
         }
 
-        $member = new Group_member();
-
-        $member->group_id   = $group->id;
-        $member->profile_id = $cur->id;
-
-        if (!$member->find(true)) {
-          $channel->error($cur,_('Could not find membership record.'));
-          return;
-        }
-        $result = $member->delete();
-        if (!$result) {
-          common_log_db_error($member, 'INSERT', __FILE__);
-          $channel->error($cur, sprintf(_('Could not remove user %s to group %s'),
-                                       $cur->nickname, $group->nickname));
-          return;
+        try {
+            if (Event::handle('StartLeaveGroup', array($group, $cur))) {
+                Group_member::leave($group->id, $cur->id);
+                Event::handle('EndLeaveGroup', array($group, $cur));
+            }
+        } catch (Exception $e) {
+            $channel->error($cur, sprintf(_('Could not remove user %s to group %s'),
+                                          $cur->nickname, $group->nickname));
+            return;
         }
 
         $channel->output($cur, sprintf(_('%s left group %s'),
diff --git a/lib/subs.php b/lib/subs.php
index 4b6b03967..5ac1a75a5 100644
--- a/lib/subs.php
+++ b/lib/subs.php
@@ -56,35 +56,44 @@ function subs_subscribe_to($user, $other)
         return _('User has blocked you.');
     }
 
-    if (!$user->subscribeTo($other)) {
-        return _('Could not subscribe.');
-        return;
-    }
+    try {
+        if (Event::handle('StartSubscribe', array($user, $other))) {
 
-    subs_notify($other, $user);
+            if (!$user->subscribeTo($other)) {
+                return _('Could not subscribe.');
+                return;
+            }
 
-    $cache = common_memcache();
+            subs_notify($other, $user);
 
-    if ($cache) {
-        $cache->delete(common_cache_key('user:notices_with_friends:' . $user->id));
-	}
+            $cache = common_memcache();
 
-    $profile = $user->getProfile();
+            if ($cache) {
+                $cache->delete(common_cache_key('user:notices_with_friends:' . $user->id));
+            }
 
-    $profile->blowSubscriptionsCount();
-    $other->blowSubscribersCount();
+            $profile = $user->getProfile();
 
-    if ($other->autosubscribe && !$other->isSubscribed($user) && !$user->hasBlocked($other)) {
-        if (!$other->subscribeTo($user)) {
-            return _('Could not subscribe other to you.');
-        }
-        $cache = common_memcache();
+            $profile->blowSubscriptionsCount();
+            $other->blowSubscribersCount();
+
+            if ($other->autosubscribe && !$other->isSubscribed($user) && !$user->hasBlocked($other)) {
+                if (!$other->subscribeTo($user)) {
+                    return _('Could not subscribe other to you.');
+                }
+                $cache = common_memcache();
 
-        if ($cache) {
-            $cache->delete(common_cache_key('user:notices_with_friends:' . $other->id));
-		}
+                if ($cache) {
+                    $cache->delete(common_cache_key('user:notices_with_friends:' . $other->id));
+                }
 
-        subs_notify($user, $other);
+                subs_notify($user, $other);
+            }
+
+            Event::handle('EndSubscribe', array($user, $other));
+        }
+    } catch (Exception $e) {
+        return $e->getMessage();
     }
 
     return true;
@@ -133,28 +142,37 @@ function subs_unsubscribe_to($user, $other)
         return _('Couldn\'t delete self-subscription.');
     }
 
-    $sub = DB_DataObject::factory('subscription');
+    try {
+        if (Event::handle('StartUnsubscribe', array($user, $other))) {
 
-    $sub->subscriber = $user->id;
-    $sub->subscribed = $other->id;
+            $sub = DB_DataObject::factory('subscription');
 
-    $sub->find(true);
+            $sub->subscriber = $user->id;
+            $sub->subscribed = $other->id;
 
-    // note we checked for existence above
+            $sub->find(true);
 
-    if (!$sub->delete())
-        return _('Couldn\'t delete subscription.');
+            // note we checked for existence above
 
-    $cache = common_memcache();
+            if (!$sub->delete())
+              return _('Couldn\'t delete subscription.');
 
-    if ($cache) {
-        $cache->delete(common_cache_key('user:notices_with_friends:' . $user->id));
-	}
+            $cache = common_memcache();
 
-    $profile = $user->getProfile();
+            if ($cache) {
+                $cache->delete(common_cache_key('user:notices_with_friends:' . $user->id));
+            }
 
-    $profile->blowSubscriptionsCount();
-    $other->blowSubscribersCount();
+            $profile = $user->getProfile();
+
+            $profile->blowSubscriptionsCount();
+            $other->blowSubscribersCount();
+
+            Event::handle('EndUnsubscribe', array($user, $other));
+        }
+    } catch (Exception $e) {
+        return $e->getMessage();
+    }
 
     return true;
 }
-- 
cgit v1.2.3-54-g00ecf


From 0e1f2d4b47e5e340679c4245b62e1d64c6b9c9b9 Mon Sep 17 00:00:00 2001
From: Brion Vibber <brion@pobox.com>
Date: Wed, 13 Jan 2010 18:17:36 +0000
Subject: Fix regressions in XMPP output during inbox refactoring -
 NOTICE_INBOX_SOURCE_* constants moved to common.php since Notice_inbox.php
 not always loaded - fixed typo in User::staticGet() call which caused user #1
 to receive messages once for each subscriber instead of for him/herself -
 'continue' -> 'continue 2' inside switch() statement to fix loop escape (PHP
 considers switch() a looping construct for break & continue)

---
 classes/Notice_inbox.php | 6 ------
 lib/common.php           | 6 ++++++
 lib/jabber.php           | 8 ++++----
 3 files changed, 10 insertions(+), 10 deletions(-)

(limited to 'classes')

diff --git a/classes/Notice_inbox.php b/classes/Notice_inbox.php
index 6c328e685..c27dcdfd6 100644
--- a/classes/Notice_inbox.php
+++ b/classes/Notice_inbox.php
@@ -31,12 +31,6 @@ define('NOTICE_INBOX_GC_MAX', 12800);
 define('NOTICE_INBOX_LIMIT', 1000);
 define('NOTICE_INBOX_SOFT_LIMIT', 1000);
 
-define('NOTICE_INBOX_SOURCE_SUB', 1);
-define('NOTICE_INBOX_SOURCE_GROUP', 2);
-define('NOTICE_INBOX_SOURCE_REPLY', 3);
-define('NOTICE_INBOX_SOURCE_FORWARD', 4);
-define('NOTICE_INBOX_SOURCE_GATEWAY', -1);
-
 class Notice_inbox extends Memcached_DataObject
 {
     ###START_AUTOCODE
diff --git a/lib/common.php b/lib/common.php
index 61decebb7..00e80373e 100644
--- a/lib/common.php
+++ b/lib/common.php
@@ -41,6 +41,12 @@ define('FOREIGN_NOTICE_SEND_REPLY', 4);
 define('FOREIGN_FRIEND_SEND', 1);
 define('FOREIGN_FRIEND_RECV', 2);
 
+define('NOTICE_INBOX_SOURCE_SUB', 1);
+define('NOTICE_INBOX_SOURCE_GROUP', 2);
+define('NOTICE_INBOX_SOURCE_REPLY', 3);
+define('NOTICE_INBOX_SOURCE_FORWARD', 4);
+define('NOTICE_INBOX_SOURCE_GATEWAY', -1);
+
 # append our extlib dir as the last-resort place to find libs
 
 set_include_path(get_include_path() . PATH_SEPARATOR . INSTALLDIR . '/extlib/');
diff --git a/lib/jabber.php b/lib/jabber.php
index 69f0d3570..4cdfa6746 100644
--- a/lib/jabber.php
+++ b/lib/jabber.php
@@ -359,7 +359,7 @@ function jabber_broadcast_notice($notice)
     $ni = $notice->whoGets();
 
     foreach ($ni as $user_id => $reason) {
-        $user = User::staticGet('user_id', $user_id);
+        $user = User::staticGet($user_id);
         if (empty($user) ||
             empty($user->jabber) ||
             !$user->jabbernotify) {
@@ -369,20 +369,20 @@ function jabber_broadcast_notice($notice)
         switch ($reason) {
         case NOTICE_INBOX_SOURCE_REPLY:
             if (!$user->jabberreplies) {
-                continue;
+                continue 2;
             }
             break;
         case NOTICE_INBOX_SOURCE_SUB:
             $sub = Subscription::pkeyGet(array('subscriber' => $user->id,
                                                'subscribed' => $notice->profile_id));
             if (empty($sub) || !$sub->jabber) {
-                continue;
+                continue 2;
             }
             break;
         case NOTICE_INBOX_SOURCE_GROUP:
             break;
         default:
-            throw new Exception(_("Unknown inbox source."));
+            throw new Exception(sprintf(_("Unknown inbox source %d."), $reason));
         }
 
         common_log(LOG_INFO,
-- 
cgit v1.2.3-54-g00ecf