1 files changed, 107 insertions, 9 deletions

diff --git a/plugins/OStatus/lib/hubdistribqueuehandler.php b/plugins/OStatus/lib/hubdistribqueuehandler.php
index 126f1355f..245a57f72 100644
--- a/plugins/OStatus/lib/hubdistribqueuehandler.php
+++ b/plugins/OStatus/lib/hubdistribqueuehandler.php
@@ -34,27 +34,101 @@ class HubDistribQueueHandler extends QueueHandler
     {
         assert($notice instanceof Notice);
 
+        $this->pushUser($notice);
+        foreach ($notice->getGroups() as $group) {
+            $this->pushGroup($notice, $group->group_id);
+        }
+        return true;
+    }
+    
+    function pushUser($notice)
+    {
         // See if there's any PuSH subscriptions, including OStatus clients.
         // @fixme handle group subscriptions as well
         // http://identi.ca/api/statuses/user_timeline/1.atom
         $feed = common_local_url('ApiTimelineUser',
                                  array('id' => $notice->profile_id,
                                        'format' => 'atom'));
+        $this->pushFeed($feed, array($this, 'userFeedForNotice'), $notice);
+    }
+
+    function pushGroup($notice, $group_id)
+    {
+        $feed = common_local_url('ApiTimelineGroup',
+                                 array('id' => $group_id,
+                                       'format' => 'atom'));
+        $this->pushFeed($feed, array($this, 'groupFeedForNotice'), $group_id, $notice);
+    }
+
+    /**
+     * @param string $feed URI to the feed
+     * @param callable $callback function to generate Atom feed update if needed
+     *        any additional params are passed to the callback.
+     */
+    function pushFeed($feed, $callback)
+    {
+        $hub = common_config('ostatus', 'hub');
+        if ($hub) {
+            $this->pushFeedExternal($feed, $hub);
+        }
+
         $sub = new HubSub();
         $sub->topic = $feed;
         if ($sub->find()) {
-            common_log(LOG_INFO, "Preparing $sub->N PuSH distribution(s) for $feed");
-            $qm = QueueManager::get();
-            $atom = $this->userFeedForNotice($notice);
-            while ($sub->fetch()) {
-                common_log(LOG_INFO, "Prepping PuSH distribution to $sub->callback for $feed");
-                $data = array('sub' => clone($sub),
-                              'atom' => $atom);
-                $qm->enqueue($data, 'hubout');
-            }
+            $args = array_slice(func_get_args(), 2);
+            $atom = call_user_func_array($callback, $args);
+            $this->pushFeedInternal($atom, $sub);
         } else {
             common_log(LOG_INFO, "No PuSH subscribers for $feed");
         }
+        return true;
+    }
+
+    /**
+     * Ping external hub about this update.
+     * The hub will pull the feed and check for new items later.
+     * Not guaranteed safe in an environment with database replication.
+     *
+     * @param string $feed feed topic URI
+     * @param string $hub PuSH hub URI
+     * @fixme can consolidate pings for user & group posts
+     */
+    function pushFeedExternal($feed, $hub)
+    {
+        $client = new HTTPClient();
+        try {
+            $data = array('hub.mode' => 'publish',
+                          'hub.url' => $feed);
+            $response = $client->post($hub, array(), $data);
+            if ($response->getStatus() == 204) {
+                common_log(LOG_INFO, "PuSH ping to hub $hub for $feed ok");
+                return true;
+            } else {
+                common_log(LOG_ERR, "PuSH ping to hub $hub for $feed failed with HTTP " .
+                                    $response->getStatus() . ': ' .
+                                    $response->getBody());
+            }
+        } catch (Exception $e) {
+            common_log(LOG_ERR, "PuSH ping to hub $hub for $feed failed: " . $e->getMessage());
+            return false;
+        }
+    }
+
+    /**
+     * Queue up direct feed update pushes to subscribers on our internal hub.
+     * @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");
+        $qm = QueueManager::get();
+        while ($sub->fetch()) {
+            common_log(LOG_INFO, "Prepping PuSH distribution to $sub->callback for $sub->topic");
+            $data = array('sub' => clone($sub),
+                          'atom' => $atom);
+            $qm->enqueue($data, 'hubout');
+        }
     }
 
     /**
@@ -83,5 +157,29 @@ class HubDistribQueueHandler extends QueueHandler
         common_log(LOG_DEBUG, $feed);
         return $feed;
     }
+
+    function groupFeedForNotice($group_id, $notice)
+    {
+        // @fixme this feels VERY hacky...
+        // should probably be a cleaner way to do it
+
+        ob_start();
+        $api = new ApiTimelineGroupAction();
+        $args = array('id' => $group_id,
+                      'format' => 'atom',
+                      'max_id' => $notice->id,
+                      'since_id' => $notice->id - 1);
+        $api->prepare($args);
+        $api->handle($args);
+        $feed = ob_get_clean();
+        
+        // ...and override the content-type back to something normal... eww!
+        // hope there's no other headers that got set while we weren't looking.
+        header('Content-Type: text/html; charset=utf-8');
+
+        common_log(LOG_DEBUG, $feed);
+        return $feed;
+    }
+
 }