Merge commit 'origin/0.9.x' into 0.9.x

20 files changed, 457 insertions, 167 deletions

diff --git a/plugins/OStatus/OStatusPlugin.php b/plugins/OStatus/OStatusPlugin.php
index 5b153216e..6fef20d6f 100644
--- a/plugins/OStatus/OStatusPlugin.php
+++ b/plugins/OStatus/OStatusPlugin.php
@@ -28,6 +28,15 @@ set_include_path(get_include_path() . PATH_SEPARATOR . dirname(__FILE__) . '/ext
 
 class FeedSubException extends Exception
 {
+    function __construct($msg=null)
+    {
+        $type = get_class($this);
+        if ($msg) {
+            parent::__construct("$type: $msg");
+        } else {
+            parent::__construct($type);
+        }
+    }
 }
 
 class OStatusPlugin extends Plugin
@@ -87,6 +96,8 @@ class OStatusPlugin extends Plugin
 
         // Outgoing from our internal PuSH hub
         $qm->connect('hubconf', 'HubConfQueueHandler');
+        $qm->connect('hubprep', 'HubPrepQueueHandler');
+
         $qm->connect('hubout', 'HubOutQueueHandler');
 
         // Outgoing Salmon replies (when we don't need a return value)
@@ -102,8 +113,10 @@ class OStatusPlugin extends Plugin
      */
     function onStartEnqueueNotice($notice, &$transports)
     {
-        // put our transport first, in case there's any conflict (like OMB)
-        array_unshift($transports, 'ostatus');
+        if ($notice->isLocal()) {
+            // put our transport first, in case there's any conflict (like OMB)
+            array_unshift($transports, 'ostatus');
+        }
         return true;
     }
 
@@ -154,6 +167,9 @@ class OStatusPlugin extends Plugin
 
             // Also, we'll add in the salmon link
             $salmon = common_local_url($salmonAction, array('id' => $id));
+            $feed->addLink($salmon, array('rel' => Salmon::REL_SALMON));
+
+            // XXX: these are deprecated
             $feed->addLink($salmon, array('rel' => Salmon::NS_REPLIES));
             $feed->addLink($salmon, array('rel' => Salmon::NS_MENTIONS));
         }
@@ -473,6 +489,24 @@ class OStatusPlugin extends Plugin
     }
 
     /**
+     * Tell the FeedSub infrastructure whether we have any active OStatus
+     * usage for the feed; if not it'll be able to garbage-collect the
+     * feed subscription.
+     * 
+     * @param FeedSub $feedsub
+     * @param integer $count in/out
+     * @return mixed hook return code
+     */
+    function onFeedSubSubscriberCount($feedsub, &$count)
+    {
+        $oprofile = Ostatus_profile::staticGet('feeduri', $feedsub->uri);
+        if ($oprofile) {
+            $count += $oprofile->subscriberCount();
+        }
+        return true;
+    }
+
+    /**
      * When about to subscribe to a remote user, start a server-to-server
      * PuSH subscription if needed. If we can't establish that, abort.
      *
@@ -949,4 +983,16 @@ class OStatusPlugin extends Plugin
         }
         return false;
     }
+
+    public function onStartProfileGetAtomFeed($profile, &$feed)
+    {
+        $oprofile = Ostatus_profile::staticGet('profile_id', $profile->id);
+
+        if (empty($oprofile)) {
+            return true;
+        }
+
+        $feed = $oprofile->feeduri;
+        return false;
+    }
 }
diff --git a/plugins/OStatus/README b/plugins/OStatus/README
index 3a98b7b25..ea5dfc055 100644
--- a/plugins/OStatus/README
+++ b/plugins/OStatus/README
@@ -1,18 +1,42 @@
-Plugin to support importing updates from external RSS and Atom feeds into your timeline.
+Plugin to support importing and exporting notices through Atom and RSS feeds.
+The OStatus plugin concentrates on user-to-user cases for federating StatusNet
+and similar social networking / microblogging / blogging sites, but includes
+low-level feed subscription systems which are used by some other plugins.
+
+Uses PubSubHubbub for push feed updates; currently non-PuSH feeds cannot be
+subscribed unless an external PuSH hub proxy is used.
 
-Uses PubSubHubbub for push feed updates; currently non-PuSH feeds cannot be subscribed.
 
 Configuration options available:
 
 $config['ostatus']['hub']
     (default internal hub)
-    Set to URL of an external PuSH hub to use it instead of our internal hub.
+    Set to URL of an external PuSH hub to use it instead of our internal hub
+    for sending outgoing updates in user and group feeds.
 
 $config['ostatus']['hub_retries']
     (default 0)
     Number of times to retry a PuSH send to consumers if using internal hub
 
 
+Settings controlling incoming feed subscription:
+
+$config['feedsub']['fallback_hub']
+    To subscribe to feeds that don't have a hub, an external PuSH proxy hub
+    such as Superfeedr may be used. Any feed without a hub of its own will
+    be subscribed through the specified hub URL instead. If the external hub
+    has usage charges, be aware that there is no restriction placed to how
+    many feeds may be subscribed!
+
+    $config['feedsub']['fallback_hub'] = 'https://superfeedr.com/hubbub';
+
+$config['feedsub']['hub_user']
+$config['feedsub']['hub_password']
+    If using the fallback hub mode, these settings may be used to provide
+    HTTP authentication credentials for contacting the hub. Default hubs
+    specified from feeds are assumed to not require
+
+
 For testing, shouldn't be used in production:
 
 $config['ostatus']['skip_signatures']
@@ -23,12 +47,11 @@ $config['feedsub']['nohub']
     (default require hub)
     Allow low-level feed subscription setup for feeds without hubs.
     Not actually usable at this stage, OStatus will check for hubs too
-    and we have no polling backend.
+    and we have no polling backend. (The fallback hub option can be used
+    with a 3rd-party service to provide such polling.)
 
 
 Todo:
-* fully functional l10n
-* redo non-OStatus feed support
-** rssCloud support?
-** possibly a polling daemon to support non-PuSH feeds?
+* better support for feeds that aren't natively oriented at social networking
 * make use of tags/categories from feeds
+* better repeat handling
diff --git a/plugins/OStatus/actions/groupsalmon.php b/plugins/OStatus/actions/groupsalmon.php
index d60725a71..5094dccf0 100644
--- a/plugins/OStatus/actions/groupsalmon.php
+++ b/plugins/OStatus/actions/groupsalmon.php
@@ -61,7 +61,7 @@ class GroupsalmonAction extends SalmonAction
     function handlePost()
     {
         // @fixme process all objects?
-        switch ($this->act->objects[0]->type) {
+        switch ($this->activity->objects[0]->type) {
         case ActivityObject::ARTICLE:
         case ActivityObject::BLOGENTRY:
         case ActivityObject::NOTE:
@@ -74,7 +74,7 @@ class GroupsalmonAction extends SalmonAction
 
         // Notice must be to the attention of this group
 
-        $context = $this->act->context;
+        $context = $this->activity->context;
 
         if (empty($context->attention)) {
             throw new ClientException("Not to the attention of anyone.");
diff --git a/plugins/OStatus/actions/hostmeta.php b/plugins/OStatus/actions/hostmeta.php
index 6d35ada6c..8ca07f916 100644
--- a/plugins/OStatus/actions/hostmeta.php
+++ b/plugins/OStatus/actions/hostmeta.php
@@ -36,13 +36,12 @@ class HostMetaAction extends Action
         $url.= '?uri={uri}';
 
         $xrd = new XRD();
-        
-        $xrd = new XRD();
         $xrd->host = $domain;
         $xrd->links[] = array('rel' => Discovery::LRDD_REL,
                               'template' => $url,
                               'title' => array('Resource Descriptor'));
 
+        header('Content-type: application/xrd+xml');
         print $xrd->toXML();
     }
 }
diff --git a/plugins/OStatus/actions/ostatusgroup.php b/plugins/OStatus/actions/ostatusgroup.php
index f325ba053..1b368de63 100644
--- a/plugins/OStatus/actions/ostatusgroup.php
+++ b/plugins/OStatus/actions/ostatusgroup.php
@@ -104,7 +104,7 @@ class OStatusGroupAction extends OStatusSubAction
         }
 
         $this->showEntity($group,
-                          $group->getProfileUrl(),
+                          $group->homeUrl(),
                           $group->homepage_logo,
                           $group->description);
         return $ok;
diff --git a/plugins/OStatus/actions/usersalmon.php b/plugins/OStatus/actions/usersalmon.php
index 6c360c49f..641e131ab 100644
--- a/plugins/OStatus/actions/usersalmon.php
+++ b/plugins/OStatus/actions/usersalmon.php
@@ -55,10 +55,10 @@ class UsersalmonAction extends SalmonAction
      */
     function handlePost()
     {
-        common_log(LOG_INFO, "Received post of '{$this->act->objects[0]->id}' from '{$this->act->actor->id}'");
+        common_log(LOG_INFO, "Received post of '{$this->activity->objects[0]->id}' from '{$this->activity->actor->id}'");
 
         // @fixme: process all activity objects?
-        switch ($this->act->objects[0]->type) {
+        switch ($this->activity->objects[0]->type) {
         case ActivityObject::ARTICLE:
         case ActivityObject::BLOGENTRY:
         case ActivityObject::NOTE:
@@ -72,7 +72,7 @@ class UsersalmonAction extends SalmonAction
         // Notice must either be a) in reply to a notice by this user
         // or b) to the attention of this user
 
-        $context = $this->act->context;
+        $context = $this->activity->context;
 
         if (!empty($context->replyToID)) {
             $notice = Notice::staticGet('uri', $context->replyToID);
@@ -92,7 +92,7 @@ class UsersalmonAction extends SalmonAction
             throw new ClientException("Not to anyone in reply to anything!");
         }
 
-        $existing = Notice::staticGet('uri', $this->act->objects[0]->id);
+        $existing = Notice::staticGet('uri', $this->activity->objects[0]->id);
 
         if (!empty($existing)) {
             common_log(LOG_ERR, "Not saving notice '{$existing->uri}'; already exists.");
@@ -143,7 +143,7 @@ class UsersalmonAction extends SalmonAction
 
     function handleFavorite()
     {
-        $notice = $this->getNotice($this->act->objects[0]);
+        $notice = $this->getNotice($this->activity->objects[0]);
         $profile = $this->ensureProfile()->localProfile();
 
         $old = Fave::pkeyGet(array('user_id' => $profile->id,
@@ -164,7 +164,7 @@ class UsersalmonAction extends SalmonAction
      */
     function handleUnfavorite()
     {
-        $notice = $this->getNotice($this->act->objects[0]);
+        $notice = $this->getNotice($this->activity->objects[0]);
         $profile = $this->ensureProfile()->localProfile();
 
         $fave = Fave::pkeyGet(array('user_id' => $profile->id,
diff --git a/plugins/OStatus/classes/FeedSub.php b/plugins/OStatus/classes/FeedSub.php
index b10509dae..dd1968db1 100644
--- a/plugins/OStatus/classes/FeedSub.php
+++ b/plugins/OStatus/classes/FeedSub.php
@@ -207,8 +207,8 @@ class FeedSub extends Memcached_DataObject
         $discover = new FeedDiscovery();
         $discover->discoverFromFeedURL($feeduri);
 
-        $huburi = $discover->getAtomLink('hub');
-        if (!$huburi) {
+        $huburi = $discover->getHubLink();
+        if (!$huburi && !common_config('feedsub', 'fallback_hub')) {
             throw new FeedSubNoHubException();
         }
 
@@ -241,8 +241,12 @@ class FeedSub extends Memcached_DataObject
             common_log(LOG_WARNING, "Attempting to (re)start PuSH subscription to $this->uri in unexpected state $this->sub_state");
         }
         if (empty($this->huburi)) {
-            if (common_config('feedsub', 'nohub')) {
+            if (common_config('feedsub', 'fallback_hub')) {
+                // No native hub on this feed?
+                // Use our fallback hub, which handles polling on our behalf.
+            } else if (common_config('feedsub', 'nohub')) {
                 // Fake it! We're just testing remote feeds w/o hubs.
+                // We'll never actually get updates in this mode.
                 return true;
             } else {
                 throw new ServerException("Attempting to start PuSH subscription for feed with no hub");
@@ -255,6 +259,9 @@ class FeedSub extends Memcached_DataObject
     /**
      * Send a PuSH unsubscription request to the hub for this feed.
      * The hub will later send us a confirmation POST to /main/push/callback.
+     * Warning: this will cancel the subscription even if someone else in
+     * the system is using it. Most callers will want garbageCollect() instead,
+     * which confirms there's no uses left.
      *
      * @return bool true on success, false on failure
      * @throws ServerException if feed state is not valid
@@ -264,8 +271,12 @@ class FeedSub extends Memcached_DataObject
             common_log(LOG_WARNING, "Attempting to (re)end PuSH subscription to $this->uri in unexpected state $this->sub_state");
         }
         if (empty($this->huburi)) {
-            if (common_config('feedsub', 'nohub')) {
+            if (common_config('feedsub', 'fallback_hub')) {
+                // No native hub on this feed?
+                // Use our fallback hub, which handles polling on our behalf.
+            } else if (common_config('feedsub', 'nohub')) {
                 // Fake it! We're just testing remote feeds w/o hubs.
+                // We'll never actually get updates in this mode.
                 return true;
             } else {
                 throw new ServerException("Attempting to end PuSH subscription for feed with no hub");
@@ -275,6 +286,33 @@ class FeedSub extends Memcached_DataObject
         return $this->doSubscribe('unsubscribe');
     }
 
+    /**
+     * Check if there are any active local uses of this feed, and if not then
+     * make sure it's inactive, unsubscribing if necessary.
+     *
+     * @return boolean true if the subscription is now inactive, false if still active.
+     */
+    public function garbageCollect()
+    {
+        if ($this->sub_state == '' || $this->sub_state == 'inactive') {
+            // No active PuSH subscription, we can just leave it be.
+            return true;
+        } else {
+            // PuSH subscription is either active or in an indeterminate state.
+            // Check if we're out of subscribers, and if so send an unsubscribe.
+            $count = 0;
+            Event::handle('FeedSubSubscriberCount', array($this, &$count));
+
+            if ($count) {
+                common_log(LOG_INFO, __METHOD__ . ': ok, ' . $count . ' user(s) left for ' . $this->uri);
+                return false;
+            } else {
+                common_log(LOG_INFO, __METHOD__ . ': unsubscribing, no users left for ' . $this->uri);
+                return $this->unsubscribe();
+            }
+        }
+    }
+
     protected function doSubscribe($mode)
     {
         $orig = clone($this);
@@ -296,7 +334,21 @@ class FeedSub extends Memcached_DataObject
                           'hub.secret' => $this->secret,
                           'hub.topic' => $this->uri);
             $client = new HTTPClient();
-            $response = $client->post($this->huburi, $headers, $post);
+            if ($this->huburi) {
+                $hub = $this->huburi;
+            } else {
+                if (common_config('feedsub', 'fallback_hub')) {
+                    $hub = common_config('feedsub', 'fallback_hub');
+                    if (common_config('feedsub', 'hub_user')) {
+                        $u = common_config('feedsub', 'hub_user');
+                        $p = common_config('feedsub', 'hub_pass');
+                        $client->setAuth($u, $p);
+                    }
+                } else {
+                    throw new FeedSubException('WTF?');
+                }
+            }
+            $response = $client->post($hub, $headers, $post);
             $status = $response->getStatus();
             if ($status == 202) {
                 common_log(LOG_INFO, __METHOD__ . ': sub req ok, awaiting verification callback');
diff --git a/plugins/OStatus/classes/HubSub.php b/plugins/OStatus/classes/HubSub.php
index cdace3c1f..7db528a4e 100644
--- a/plugins/OStatus/classes/HubSub.php
+++ b/plugins/OStatus/classes/HubSub.php
@@ -260,6 +260,37 @@ class HubSub extends Memcached_DataObject
             $retries = intval(common_config('ostatus', 'hub_retries'));
         }
 
+        if (common_config('ostatus', 'local_push_bypass')) {
+            // If target is a local site, bypass the web server and drop the
+            // item directly into the target's input queue.
+            $url = parse_url($this->callback);
+            $wildcard = common_config('ostatus', 'local_wildcard');
+            $site = Status_network::getFromHostname($url['host'], $wildcard);
+
+            if ($site) {
+                if ($this->secret) {
+                    $hmac = 'sha1=' . hash_hmac('sha1', $atom, $this->secret);
+                } else {
+                    $hmac = '';
+                }
+
+                // Hack: at the moment we stick the subscription ID in the callback
+                // URL so we don't have to look inside the Atom to route the subscription.
+                // For now this means we need to extract that from the target URL
+                // so we can include it in the data.
+                $parts = explode('/', $url['path']);
+                $subId = intval(array_pop($parts));
+
+                $data = array('feedsub_id' => $subId,
+                              'post' => $atom,
+                              'hmac' => $hmac);
+                common_log(LOG_DEBUG, "Cross-site PuSH bypass enqueueing straight to $site->nickname feed $subId");
+                $qm = QueueManager::get();
+                $qm->enqueue($data, 'pushin', $site->nickname);
+                return;
+            }
+        }
+
         // We dare not clone() as when the clone is discarded it'll
         // destroy the result data for the parent query.
         // @fixme use clone() again when it's safe to copy an
@@ -274,6 +305,26 @@ class HubSub extends Memcached_DataObject
     }
 
     /**
+     * Queue up a large batch of pushes to multiple subscribers
+     * for this same topic update.
+     * 
+     * If queues are disabled, this will run immediately.
+     * 
+     * @param string $atom well-formed Atom feed
+     * @param array $pushCallbacks list of callback URLs
+     */
+    function bulkDistribute($atom, $pushCallbacks)
+    {
+        $data = array('atom' => $atom,
+                      'topic' => $this->topic,
+                      'pushCallbacks' => $pushCallbacks);
+        common_log(LOG_INFO, "Queuing PuSH batch: $this->topic to " .
+                             count($pushCallbacks) . " sites");
+        $qm = QueueManager::get();
+        $qm->enqueue($data, 'hubprep');
+    }
+
+    /**
      * Send a 'fat ping' to the subscriber's callback endpoint
      * containing the given Atom feed chunk.
      *
diff --git a/plugins/OStatus/classes/Ostatus_profile.php b/plugins/OStatus/classes/Ostatus_profile.php
index 5d3f37cd0..cc4307b14 100644
--- a/plugins/OStatus/classes/Ostatus_profile.php
+++ b/plugins/OStatus/classes/Ostatus_profile.php
@@ -215,22 +215,13 @@ class Ostatus_profile extends Memcached_DataObject
     }
 
     /**
-     * Send a PuSH unsubscription request to the hub for this feed.
-     * The hub will later send us a confirmation POST to /main/push/callback.
+     * Check if this remote profile has any active local subscriptions, and
+     * if not drop the PuSH subscription feed.
      *
      * @return bool true on success, false on failure
-     * @throws ServerException if feed state is not valid
      */
     public function unsubscribe() {
-        $feedsub = FeedSub::staticGet('uri', $this->feeduri);
-        if (!$feedsub || $feedsub->sub_state == '' || $feedsub->sub_state == 'inactive') {
-            // No active PuSH subscription, we can just leave it be.
-            return true;
-        } else {
-            // PuSH subscription is either active or in an indeterminate state.
-            // Send an unsubscribe.
-            return $feedsub->unsubscribe();
-        }
+        $this->garbageCollect();
     }
 
     /**
@@ -241,19 +232,35 @@ class Ostatus_profile extends Memcached_DataObject
      */
     public function garbageCollect()
     {
+        $feedsub = FeedSub::staticGet('uri', $this->feeduri);
+        return $feedsub->garbageCollect();
+    }
+
+    /**
+     * Check if this remote profile has any active local subscriptions, so the
+     * PuSH subscription layer can decide if it can drop the feed.
+     *
+     * This gets called via the FeedSubSubscriberCount event when running
+     * FeedSub::garbageCollect().
+     *
+     * @return int
+     */
+    public function subscriberCount()
+    {
         if ($this->isGroup()) {
             $members = $this->localGroup()->getMembers(0, 1);
             $count = $members->N;
         } else {
             $count = $this->localProfile()->subscriberCount();
         }
-        if ($count == 0) {
-            common_log(LOG_INFO, "Unsubscribing from now-unused remote feed $this->feeduri");
-            $this->unsubscribe();
-            return true;
-        } else {
-            return false;
-        }
+        common_log(LOG_INFO, __METHOD__ . " SUB COUNT BEFORE: $count");
+
+        // Other plugins may be piggybacking on OStatus without having
+        // an active group or user-to-user subscription we know about.
+        Event::handle('Ostatus_profileSubscriberCount', array($this, &$count));
+        common_log(LOG_INFO, __METHOD__ . " SUB COUNT AFTER: $count");
+
+        return $count;
     }
 
     /**
@@ -438,26 +445,32 @@ class Ostatus_profile extends Memcached_DataObject
      * @param DOMElement $feed for context
      * @param string $source identifier ("push" or "salmon")
      */
+
     public function processEntry($entry, $feed, $source)
     {
         $activity = new Activity($entry, $feed);
 
-        // @todo process all activity objects
-        switch ($activity->objects[0]->type) {
-        case ActivityObject::ARTICLE:
-        case ActivityObject::BLOGENTRY:
-        case ActivityObject::NOTE:
-        case ActivityObject::STATUS:
-        case ActivityObject::COMMENT:
-            break;
-        default:
-            throw new ClientException("Can't handle that kind of post.");
-        }
+        if (Event::handle('StartHandleFeedEntry', array($activity))) {
+
+            // @todo process all activity objects
+            switch ($activity->objects[0]->type) {
+            case ActivityObject::ARTICLE:
+            case ActivityObject::BLOGENTRY:
+            case ActivityObject::NOTE:
+            case ActivityObject::STATUS:
+            case ActivityObject::COMMENT:
+			case null:
+                if ($activity->verb == ActivityVerb::POST) {
+                    $this->processPost($activity, $source);
+                } else {
+                    common_log(LOG_INFO, "Ignoring activity with unrecognized verb $activity->verb");
+                }
+                break;
+            default:
+                throw new ClientException("Can't handle that kind of post.");
+            }
 
-        if ($activity->verb == ActivityVerb::POST) {
-            $this->processPost($activity, $source);
-        } else {
-            common_log(LOG_INFO, "Ignoring activity with unrecognized verb $activity->verb");
+            Event::handle('EndHandleFeedEntry', array($activity));
         }
     }
 
@@ -486,8 +499,17 @@ class Ostatus_profile extends Memcached_DataObject
                 // OK here! assume the default
             } else if ($actor->id == $this->uri || $actor->link == $this->uri) {
                 $this->updateFromActivityObject($actor);
+            } else if ($actor->id) {
+                // We have an ActivityStreams actor with an explicit ID that doesn't match the feed owner.
+                // This isn't what we expect from mainline OStatus person feeds!
+                // Group feeds go down another path, with different validation...
+                // Most likely this is a plain ol' blog feed of some kind which
+                // doesn't match our expectations. We'll take the entry, but ignore
+                // the <author> info.
+                common_log(LOG_WARNING, "Got an actor '{$actor->title}' ({$actor->id}) on single-user feed for {$this->uri}");
             } else {
-                throw new Exception("Got an actor '{$actor->title}' ({$actor->id}) on single-user feed for {$this->uri}");
+                // Plain <author> without ActivityStreams actor info.
+                // We'll just ignore this info for now and save the update under the feed's identity.
             }
 
             $oprofile = $this;
@@ -668,7 +690,7 @@ class Ostatus_profile extends Memcached_DataObject
         common_log(LOG_DEBUG, "Original reply recipients: " . implode(', ', $attention_uris));
         $groups = array();
         $replies = array();
-        foreach ($attention_uris as $recipient) {
+        foreach (array_unique($attention_uris) as $recipient) {
             // Is the recipient a local user?
             $user = User::staticGet('uri', $recipient);
             if ($user) {
@@ -862,12 +884,12 @@ class Ostatus_profile extends Memcached_DataObject
         $feeduri = $discover->discoverFromFeedURL($feed_url);
         $hints['feedurl'] = $feeduri;
 
-        $huburi = $discover->getAtomLink('hub');
+        $huburi = $discover->getHubLink();
         $hints['hub'] = $huburi;
         $salmonuri = $discover->getAtomLink(Salmon::NS_REPLIES);
         $hints['salmon'] = $salmonuri;
 
-        if (!$huburi) {
+        if (!$huburi && !common_config('feedsub', 'fallback_hub')) {
             // We can only deal with folks with a PuSH hub
             throw new FeedSubNoHubException();
         }
@@ -1263,10 +1285,10 @@ class Ostatus_profile extends Memcached_DataObject
                 $discover = new FeedDiscovery();
                 $discover->discoverFromFeedURL($hints['feedurl']);
             }
-            $huburi = $discover->getAtomLink('hub');
+            $huburi = $discover->getHubLink();
         }
 
-        if (!$huburi) {
+        if (!$huburi && !common_config('feedsub', 'fallback_hub')) {
             // We can only deal with folks with a PuSH hub
             throw new FeedSubNoHubException();
         }
diff --git a/plugins/OStatus/lib/discoveryhints.php b/plugins/OStatus/lib/discoveryhints.php
index 34c9be277..fa2ead732 100644
--- a/plugins/OStatus/lib/discoveryhints.php
+++ b/plugins/OStatus/lib/discoveryhints.php
@@ -114,9 +114,10 @@ class DiscoveryHints {
 
     static function _hcard($body, $url)
     {
-        // DOMDocument::loadHTML may throw warnings on unrecognized elements.
+        // DOMDocument::loadHTML may throw warnings on unrecognized elements,
+        // and notices on unrecognized namespaces.
 
-        $old = error_reporting(error_reporting() & ~E_WARNING);
+        $old = error_reporting(error_reporting() & ~(E_WARNING | E_NOTICE));
 
         $doc = new DOMDocument();
         $doc->loadHTML($body);
diff --git a/plugins/OStatus/lib/feeddiscovery.php b/plugins/OStatus/lib/feeddiscovery.php
index 4ac243832..8a166a0be 100644
--- a/plugins/OStatus/lib/feeddiscovery.php
+++ b/plugins/OStatus/lib/feeddiscovery.php
@@ -88,6 +88,16 @@ class FeedDiscovery
     }
 
     /**
+     * Get the referenced PuSH hub link from an Atom feed.
+     *
+     * @return mixed string or false
+     */
+    public function getHubLink()
+    {
+        return $this->getAtomLink('hub');
+    }
+
+    /**
      * @param string $url
      * @param bool $htmlOk pass false here if you don't want to follow web pages.
      * @return string with validated URL
@@ -186,8 +196,9 @@ class FeedDiscovery
      */
     function discoverFromHTML($url, $body)
     {
-        // DOMDocument::loadHTML may throw warnings on unrecognized elements.
-        $old = error_reporting(error_reporting() & ~E_WARNING);
+        // DOMDocument::loadHTML may throw warnings on unrecognized elements,
+        // and notices on unrecognized namespaces.
+        $old = error_reporting(error_reporting() & ~(E_WARNING | E_NOTICE));
         $dom = new DOMDocument();
         $ok = $dom->loadHTML($body);
         error_reporting($old);
diff --git a/plugins/OStatus/lib/hubprepqueuehandler.php b/plugins/OStatus/lib/hubprepqueuehandler.php
new file mode 100644
index 000000000..0d585938f
--- /dev/null
+++ b/plugins/OStatus/lib/hubprepqueuehandler.php
@@ -0,0 +1,87 @@
+<?php
+/*
+ * StatusNet - the distributed open-source microblogging tool
+ * Copyright (C) 2010, StatusNet, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+/**
+ * When we have a large batch of PuSH consumers, we break the data set
+ * into smaller chunks. Enqueue final destinations...
+ *
+ * @package Hub
+ * @author Brion Vibber <brion@status.net>
+ */
+class HubPrepQueueHandler extends QueueHandler
+{
+    // Enqueue this many low-level distributions before re-queueing the rest
+    // of the batch to be processed later. Helps to keep latency down for other
+    // things happening during a particularly long OStatus delivery session.
+    //
+    // [Could probably ditch this if we had working message delivery priorities
+    // for queueing, but this isn't supported in ActiveMQ 5.3.]
+    const ROLLING_BATCH = 20;
+
+    function transport()
+    {
+        return 'hubprep';
+    }
+
+    function handle($data)
+    {
+        $topic = $data['topic'];
+        $atom = $data['atom'];
+        $pushCallbacks = $data['pushCallbacks'];
+
+        assert(is_string($atom));
+        assert(is_string($topic));
+        assert(is_array($pushCallbacks));
+
+        // Set up distribution for the first n subscribing sites...
+        // If we encounter an uncatchable error, queue handling should
+        // automatically re-run the batch, which could lead to some dupe
+        // distributions.
+        //
+        // Worst case is if one of these hubprep entries dies too many
+        // times and gets dropped; the rest of the batch won't get processed.
+        try {
+            $n = 0;
+            while (count($pushCallbacks) && $n < self::ROLLING_BATCH) {
+                $n++;
+                $callback = array_shift($pushCallbacks);
+                $sub = HubSub::staticGet($topic, $callback);
+                if (!$sub) {
+                    common_log(LOG_ERR, "Skipping PuSH delivery for deleted(?) consumer $callback on $topic");
+                    continue;
+                }
+
+                $sub->distribute($atom);
+            }
+        } catch (Exception $e) {
+            common_log(LOG_ERR, "Exception during PuSH batch out: " .
+                                $e->getMessage() .
+                                " prepping $topic to $callback");
+        }
+
+        // And re-queue the rest of the batch!
+        if (count($pushCallbacks) > 0) {
+            $sub = new HubSub();
+            $sub->topic = $topic;
+            $sub->bulkDistribute($atom, $pushCallbacks);
+        }
+
+        return true;
+    }
+}
diff --git a/plugins/OStatus/lib/magicenvelope.php b/plugins/OStatus/lib/magicenvelope.php
index f39686b71..967e5f6d1 100644
--- a/plugins/OStatus/lib/magicenvelope.php
+++ b/plugins/OStatus/lib/magicenvelope.php
@@ -97,24 +97,18 @@ class MagicEnvelope
     }
 
     public function toXML($env) {
-        $dom = new DOMDocument();
-
-        $envelope = $dom->createElementNS(MagicEnvelope::NS, 'me:env');
-        $envelope->setAttribute('xmlns:me', MagicEnvelope::NS);
-        $data = $dom->createElementNS(MagicEnvelope::NS, 'me:data', $env['data']);
-        $data->setAttribute('type', $env['data_type']);
-        $envelope->appendChild($data);
-        $enc = $dom->createElementNS(MagicEnvelope::NS, 'me:encoding', $env['encoding']);
-        $envelope->appendChild($enc);
-        $alg = $dom->createElementNS(MagicEnvelope::NS, 'me:alg', $env['alg']);
-        $envelope->appendChild($alg);
-        $sig = $dom->createElementNS(MagicEnvelope::NS, 'me:sig', $env['sig']);
-        $envelope->appendChild($sig);
-
-        $dom->appendChild($envelope);
-        
+        $xs = new XMLStringer();
+        $xs->startXML();
+        $xs->elementStart('me:env', array('xmlns:me' => MagicEnvelope::NS));
+        $xs->element('me:data', array('type' => $env['data_type']), $env['data']);
+        $xs->element('me:encoding', null, $env['encoding']);
+        $xs->element('me:alg', null, $env['alg']);
+        $xs->element('me:sig', null, $env['sig']);
+        $xs->elementEnd('me:env');
         
-        return $dom->saveXML();
+        $string =  $xs->getString();
+        common_debug($string);
+        return $string;
     }
 
     
@@ -216,13 +210,13 @@ class MagicEnvelope
         }
 
         $data_element = $env_element->getElementsByTagNameNS(MagicEnvelope::NS, 'data')->item(0);
-        
+        $sig_element = $env_element->getElementsByTagNameNS(MagicEnvelope::NS, 'sig')->item(0);
         return array(
-            'data' => trim($data_element->nodeValue),
+            'data' => preg_replace('/\s/', '', $data_element->nodeValue),
             'data_type' => $data_element->getAttribute('type'),
             'encoding' => $env_element->getElementsByTagNameNS(MagicEnvelope::NS, 'encoding')->item(0)->nodeValue,
             'alg' => $env_element->getElementsByTagNameNS(MagicEnvelope::NS, 'alg')->item(0)->nodeValue,
-            'sig' => $env_element->getElementsByTagNameNS(MagicEnvelope::NS, 'sig')->item(0)->nodeValue,
+            'sig' => preg_replace('/\s/', '', $sig_element->nodeValue),
         );
     }
 
diff --git a/plugins/OStatus/lib/ostatusqueuehandler.php b/plugins/OStatus/lib/ostatusqueuehandler.php
index d1e58f1d6..8905d2e21 100644
--- a/plugins/OStatus/lib/ostatusqueuehandler.php
+++ b/plugins/OStatus/lib/ostatusqueuehandler.php
@@ -25,6 +25,18 @@
  */
 class OStatusQueueHandler extends QueueHandler
 {
+    // If we have more than this many subscribing sites on a single feed,
+    // break up the PuSH distribution into smaller batches which will be
+    // rolled into the queue progressively. This reduces disruption to
+    // other, shorter activities being enqueued while we work.
+    const MAX_UNBATCHED = 50;
+
+    // Each batch (a 'hubprep' entry) will have this many items.
+    // Selected to provide a balance between queue packet size
+    // and number of batches that will end up getting processed.
+    // For 20,000 target sites, 1000 should work acceptably.
+    const BATCH_SIZE = 1000;
+
     function transport()
     {
         return 'ostatus';
@@ -147,14 +159,31 @@ class OStatusQueueHandler extends QueueHandler
 
     /**
      * Queue up direct feed update pushes to subscribers on our internal hub.
+     * If there are a large number of subscriber sites, intermediate bulk
+     * distribution triggers may be queued.
+     * 
      * @param string $atom update feed, containing only new/changed items
      * @param HubSub $sub open query of subscribers
      */
     function pushFeedInternal($atom, $sub)
     {
         common_log(LOG_INFO, "Preparing $sub->N PuSH distribution(s) for $sub->topic");
+        $n = 0;
+        $batch = array();
         while ($sub->fetch()) {
-            $sub->distribute($atom);
+            $n++;
+            if ($n < self::MAX_UNBATCHED) {
+                $sub->distribute($atom);
+            } else {
+                $batch[] = $sub->callback;
+                if (count($batch) >= self::BATCH_SIZE) {
+                    $sub->bulkDistribute($atom, $batch);
+                    $batch = array();
+                }
+            }
+        }
+        if (count($batch) >= 0) {
+            $sub->bulkDistribute($atom, $batch);
         }
     }
 
diff --git a/plugins/OStatus/lib/salmon.php b/plugins/OStatus/lib/salmon.php
index 3d3341bc6..ef7719a40 100644
--- a/plugins/OStatus/lib/salmon.php
+++ b/plugins/OStatus/lib/salmon.php
@@ -28,9 +28,11 @@
  */
 class Salmon
 {
+    const REL_SALMON = 'salmon';
+    const REL_MENTIONED = 'mentioned';
 
+    // XXX: these are deprecated 
     const NS_REPLIES = "http://salmon-protocol.org/ns/salmon-replies";
-
     const NS_MENTIONS = "http://salmon-protocol.org/ns/salmon-mention";
     
     /**
diff --git a/plugins/OStatus/lib/salmonaction.php b/plugins/OStatus/lib/salmonaction.php
index fa9dc3b1d..9d6c6b269 100644
--- a/plugins/OStatus/lib/salmonaction.php
+++ b/plugins/OStatus/lib/salmonaction.php
@@ -47,7 +47,6 @@ class SalmonAction extends Action
 
         $xml = file_get_contents('php://input');
 
-
         // Check the signature
         $salmon = new Salmon;
         if (!$salmon->verifyMagicEnv($xml)) {
@@ -58,7 +57,6 @@ class SalmonAction extends Action
             $env = $magic_env->parse($xml);
             $xml = $magic_env->unfold($env);
         }
-        
 
         $dom = DOMDocument::loadXML($xml);
         if ($dom->documentElement->namespaceURI != Activity::ATOM ||
@@ -67,7 +65,7 @@ class SalmonAction extends Action
             $this->clientError(_m('Salmon post must be an Atom entry.'));
         }
 
-        $this->act = new Activity($dom->documentElement);
+        $this->activity = new Activity($dom->documentElement);
         return true;
     }
 
@@ -79,9 +77,9 @@ class SalmonAction extends Action
     {
         StatusNet::setApi(true); // Send smaller error pages
 
-        common_log(LOG_DEBUG, "Got a " . $this->act->verb);
+        common_log(LOG_DEBUG, "Got a " . $this->activity->verb);
         if (Event::handle('StartHandleSalmon', array($this->activity))) {
-            switch ($this->act->verb)
+            switch ($this->activity->verb)
             {
             case ActivityVerb::POST:
                 $this->handlePost();
@@ -164,12 +162,12 @@ class SalmonAction extends Action
      */
     function handleUpdateProfile()
     {
-        $oprofile = Ostatus_profile::getActorProfile($this->act);
+        $oprofile = Ostatus_profile::getActorProfile($this->activity);
         if ($oprofile) {
             common_log(LOG_INFO, "Got a profile-update ping from $oprofile->uri");
-            $oprofile->updateFromActivityObject($this->act->actor);
+            $oprofile->updateFromActivityObject($this->activity->actor);
         } else {
-            common_log(LOG_INFO, "Ignoring profile-update ping from unknown " . $this->act->actor->id);
+            common_log(LOG_INFO, "Ignoring profile-update ping from unknown " . $this->activity->actor->id);
         }
     }
 
@@ -178,10 +176,10 @@ class SalmonAction extends Action
      */
     function ensureProfile()
     {
-        $actor = $this->act->actor;
+        $actor = $this->activity->actor;
         if (empty($actor->id)) {
             common_log(LOG_ERR, "broken actor: " . var_export($actor, true));
-            common_log(LOG_ERR, "activity with no actor: " . var_export($this->act, true));
+            common_log(LOG_ERR, "activity with no actor: " . var_export($this->activity, true));
             throw new Exception("Received a salmon slap from unidentified actor.");
         }
 
@@ -191,6 +189,6 @@ class SalmonAction extends Action
     function saveNotice()
     {
         $oprofile = $this->ensureProfile();
-        return $oprofile->processPost($this->act, 'salmon');
+        return $oprofile->processPost($this->activity, 'salmon');
     }
 }
diff --git a/plugins/OStatus/lib/xrd.php b/plugins/OStatus/lib/xrd.php
index 34b28790b..a10b9f427 100644
--- a/plugins/OStatus/lib/xrd.php
+++ b/plugins/OStatus/lib/xrd.php
@@ -106,44 +106,43 @@ class XRD
 
     public function toXML()
     {
-        $dom = new DOMDocument('1.0', 'UTF-8');
-        $dom->formatOutput = true;
-        
-        $xrd_dom = $dom->createElementNS(XRD::XRD_NS, 'XRD');
-        $dom->appendChild($xrd_dom);
+        $xs = new XMLStringer();
+
+        $xs->startXML();
+        $xs->elementStart('XRD', array('xmlns' => XRD::XRD_NS));
 
         if ($this->host) {
-            $host_dom = $dom->createElement('hm:Host', $this->host);
-            $xrd_dom->setAttributeNS(XRD::XML_NS, 'xmlns:hm', XRD::HOST_META_NS);
-            $xrd_dom->appendChild($host_dom);
+            $xs->element('hm:Host', array('xmlns:hm' => XRD::HOST_META_NS), $this->host);
         }
         
-		if ($this->expires) {
-			$expires_dom = $dom->createElement('Expires', $this->expires);
-			$xrd_dom->appendChild($expires_dom);
-		}
-
-		if ($this->subject) {
-			$subject_dom = $dom->createElement('Subject', $this->subject);
-			$xrd_dom->appendChild($subject_dom);
-		}
-
-		foreach ($this->alias as $alias) {
-			$alias_dom = $dom->createElement('Alias', $alias);
-			$xrd_dom->appendChild($alias_dom);
-		}
-
-		foreach ($this->types as $type) {
-			$type_dom = $dom->createElement('Type', $type);
-			$xrd_dom->appendChild($type_dom);
-		}
-
-		foreach ($this->links as $link) {
-			$link_dom = $this->saveLink($dom, $link);
-			$xrd_dom->appendChild($link_dom);
-		}
-
-        return $dom->saveXML();
+        if ($this->expires) {
+            $xs->element('Expires', null, $this->expires);
+        }
+
+        if ($this->subject) {
+            $xs->element('Subject', null, $this->subject);
+        }
+
+        foreach ($this->alias as $alias) {
+            $xs->element('Alias', null, $alias);
+        }
+
+        foreach ($this->links as $link) {
+            $titles = array();
+            if (isset($link['title'])) {
+                $titles = $link['title'];
+                unset($link['title']);
+            }
+            $xs->elementStart('Link', $link);
+            foreach ($titles as $title) {
+                $xs->element('Title', null, $title);
+            }
+            $xs->elementEnd('Link');
+        }
+        
+        $xs->elementEnd('XRD');
+
+        return $xs->getString();
     }
 
     function parseType($element)
@@ -169,32 +168,5 @@ class XRD
 
         return $link;
     }
-
-    function saveLink($doc, $link)
-    {
-        $link_element = $doc->createElement('Link');
-        if (!empty($link['rel'])) {
-            $link_element->setAttribute('rel', $link['rel']);
-        }
-        if (!empty($link['type'])) {
-            $link_element->setAttribute('type', $link['type']);
-        }
-        if (!empty($link['href'])) {
-            $link_element->setAttribute('href', $link['href']);
-        }
-        if (!empty($link['template'])) {
-            $link_element->setAttribute('template', $link['template']);
-        }
-
-        if (!empty($link['title']) && is_array($link['title'])) {
-            foreach($link['title'] as $title) {
-                $title = $doc->createElement('Title', $title);
-                $link_element->appendChild($title);
-            }
-        }
-
-        
-        return $link_element;
-    }
 }
 
diff --git a/plugins/OStatus/lib/xrdaction.php b/plugins/OStatus/lib/xrdaction.php
index f1a56e0a8..d8cf648d6 100644
--- a/plugins/OStatus/lib/xrdaction.php
+++ b/plugins/OStatus/lib/xrdaction.php
@@ -76,6 +76,9 @@ class XrdAction extends Action
         $salmon_url = common_local_url('usersalmon',
                                        array('id' => $this->user->id));
 
+        $xrd->links[] = array('rel' => Salmon::REL_SALMON,
+                              'href' => $salmon_url);
+        // XXX : Deprecated - to be removed.
         $xrd->links[] = array('rel' => Salmon::NS_REPLIES,
                               'href' => $salmon_url);
 
@@ -98,7 +101,7 @@ class XrdAction extends Action
         $xrd->links[] = array('rel' => 'http://ostatus.org/schema/1.0/subscribe',
                               'template' => $url );
 
-        header('Content-type: text/xml');
+        header('Content-type: application/xrd+xml');
         print $xrd->toXML();
     }
 
diff --git a/plugins/OStatus/scripts/update-profile.php b/plugins/OStatus/scripts/update-profile.php
index d06de4f90..64afa0f35 100644
--- a/plugins/OStatus/scripts/update-profile.php
+++ b/plugins/OStatus/scripts/update-profile.php
@@ -55,7 +55,7 @@ print "Re-running feed discovery for profile URL $oprofile->uri\n";
 // @fixme will bork where the URI isn't the profile URL for now
 $discover = new FeedDiscovery();
 $feedurl = $discover->discoverFromURL($oprofile->uri);
-$huburi = $discover->getAtomLink('hub');
+$huburi = $discover->getHubLink();
 $salmonuri = $discover->getAtomLink(Salmon::NS_REPLIES);
 
 print "  Feed URL: $feedurl\n";
diff --git a/plugins/OStatus/tests/FeedDiscoveryTest.php b/plugins/OStatus/tests/FeedDiscoveryTest.php
index 1c5249701..0e6354a86 100644
--- a/plugins/OStatus/tests/FeedDiscoveryTest.php
+++ b/plugins/OStatus/tests/FeedDiscoveryTest.php
@@ -10,7 +10,7 @@ define('STATUSNET', true);
 define('LACONICA', true);
 
 require_once INSTALLDIR . '/lib/common.php';
-require_once INSTALLDIR . '/plugins/FeedSub/feedsub.php';
+require_once INSTALLDIR . '/plugins/OStatus/lib/feeddiscovery.php';
 
 class FeedDiscoveryTest extends PHPUnit_Framework_TestCase
 {