plugins/OStatus/lib/hubdistribqueuehandler.php

   1 <?php
   2 /*
   3  * StatusNet - the distributed open-source microblogging tool
   4  * Copyright (C) 2010, StatusNet, Inc.
   5  *
   6  * This program is free software: you can redistribute it and/or modify
   7  * it under the terms of the GNU Affero General Public License as published by
   8  * the Free Software Foundation, either version 3 of the License, or
   9  * (at your option) any later version.
  10  *
  11  * This program is distributed in the hope that it will be useful,
  12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14  * GNU Affero General Public License for more details.
  15  *
  16  * You should have received a copy of the GNU Affero General Public License
  17  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  18  */
  19
  20 /**
  21  * Send a PuSH subscription verification from our internal hub.
  22  * Queue up final distribution for
  23  * @package Hub
  24  * @author Brion Vibber <brion@status.net>
  25  */
  26 class HubDistribQueueHandler extends QueueHandler
  27 {
  28     function transport()
  29     {
  30         return 'hubdistrib';
  31     }
  32
  33     function handle($notice)
  34     {
  35         assert($notice instanceof Notice);
  36
  37         $this->pushUser($notice);
  38         foreach ($notice->getGroups() as $group) {
  39             $this->pushGroup($notice, $group->group_id);
  40         }
  41         return true;
  42     }
  43
  44     function pushUser($notice)
  45     {
  46         // See if there's any PuSH subscriptions, including OStatus clients.
  47         // @fixme handle group subscriptions as well
  48         // http://identi.ca/api/statuses/user_timeline/1.atom
  49         $feed = common_local_url('ApiTimelineUser',
  50                                  array('id' => $notice->profile_id,
  51                                        'format' => 'atom'));
  52         $this->pushFeed($feed, array($this, 'userFeedForNotice'), $notice);
  53     }
  54
  55     function pushGroup($notice, $group_id)
  56     {
  57         $feed = common_local_url('ApiTimelineGroup',
  58                                  array('id' => $group_id,
  59                                        'format' => 'atom'));
  60         $this->pushFeed($feed, array($this, 'groupFeedForNotice'), $group_id, $notice);
  61     }
  62
  63     /**
  64      * @param string $feed URI to the feed
  65      * @param callable $callback function to generate Atom feed update if needed
  66      *        any additional params are passed to the callback.
  67      */
  68     function pushFeed($feed, $callback)
  69     {
  70         $hub = common_config('ostatus', 'hub');
  71         if ($hub) {
  72             $this->pushFeedExternal($feed, $hub);
  73         }
  74
  75         $sub = new HubSub();
  76         $sub->topic = $feed;
  77         if ($sub->find()) {
  78             $args = array_slice(func_get_args(), 2);
  79             $atom = call_user_func_array($callback, $args);
  80             $this->pushFeedInternal($atom, $sub);
  81         } else {
  82             common_log(LOG_INFO, "No PuSH subscribers for $feed");
  83         }
  84         return true;
  85     }
  86
  87     /**
  88      * Ping external hub about this update.
  89      * The hub will pull the feed and check for new items later.
  90      * Not guaranteed safe in an environment with database replication.
  91      *
  92      * @param string $feed feed topic URI
  93      * @param string $hub PuSH hub URI
  94      * @fixme can consolidate pings for user & group posts
  95      */
  96     function pushFeedExternal($feed, $hub)
  97     {
  98         $client = new HTTPClient();
  99         try {
 100             $data = array('hub.mode' => 'publish',
 101                           'hub.url' => $feed);
 102             $response = $client->post($hub, array(), $data);
 103             if ($response->getStatus() == 204) {
 104                 common_log(LOG_INFO, "PuSH ping to hub $hub for $feed ok");
 105                 return true;
 106             } else {
 107                 common_log(LOG_ERR, "PuSH ping to hub $hub for $feed failed with HTTP " .
 108                                     $response->getStatus() . ': ' .
 109                                     $response->getBody());
 110             }
 111         } catch (Exception $e) {
 112             common_log(LOG_ERR, "PuSH ping to hub $hub for $feed failed: " . $e->getMessage());
 113             return false;
 114         }
 115     }
 116
 117     /**
 118      * Queue up direct feed update pushes to subscribers on our internal hub.
 119      * @param string $atom update feed, containing only new/changed items
 120      * @param HubSub $sub open query of subscribers
 121      */
 122     function pushFeedInternal($atom, $sub)
 123     {
 124         common_log(LOG_INFO, "Preparing $sub->N PuSH distribution(s) for $sub->topic");
 125         $qm = QueueManager::get();
 126         while ($sub->fetch()) {
 127             common_log(LOG_INFO, "Prepping PuSH distribution to $sub->callback for $sub->topic");
 128             $data = array('sub' => clone($sub),
 129                           'atom' => $atom);
 130             $qm->enqueue($data, 'hubout');
 131         }
 132     }
 133
 134     /**
 135      * Build a single-item version of the sending user's Atom feed.
 136      * @param Notice $notice
 137      * @return string
 138      */
 139     function userFeedForNotice($notice)
 140     {
 141         // @fixme this feels VERY hacky...
 142         // should probably be a cleaner way to do it
 143
 144         ob_start();
 145         $api = new ApiTimelineUserAction();
 146         $api->prepare(array('id' => $notice->profile_id,
 147                             'format' => 'atom',
 148                             'max_id' => $notice->id,
 149                             'since_id' => $notice->id - 1));
 150         $api->showTimeline();
 151         $feed = ob_get_clean();
 152
 153         // ...and override the content-type back to something normal... eww!
 154         // hope there's no other headers that got set while we weren't looking.
 155         header('Content-Type: text/html; charset=utf-8');
 156
 157         common_log(LOG_DEBUG, $feed);
 158         return $feed;
 159     }
 160
 161     function groupFeedForNotice($group_id, $notice)
 162     {
 163         // @fixme this feels VERY hacky...
 164         // should probably be a cleaner way to do it
 165
 166         ob_start();
 167         $api = new ApiTimelineGroupAction();
 168         $args = array('id' => $group_id,
 169                       'format' => 'atom',
 170                       'max_id' => $notice->id,
 171                       'since_id' => $notice->id - 1);
 172         $api->prepare($args);
 173         $api->handle($args);
 174         $feed = ob_get_clean();
 175
 176         // ...and override the content-type back to something normal... eww!
 177         // hope there's no other headers that got set while we weren't looking.
 178         header('Content-Type: text/html; charset=utf-8');
 179
 180         common_log(LOG_DEBUG, $feed);
 181         return $feed;
 182     }
 183
 184 }
 185