3 * StatusNet - the distributed open-source microblogging tool
4 * Copyright (C) 2010, StatusNet, Inc.
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.
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.
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/>.
21 * Prepare PuSH and Salmon distributions for an outgoing message.
23 * @package OStatusPlugin
24 * @author Brion Vibber <brion@status.net>
26 class OStatusQueueHandler extends QueueHandler
33 function handle($notice)
35 assert($notice instanceof Notice);
37 $this->notice = $notice;
38 $this->user = User::staticGet($notice->profile_id);
42 foreach ($notice->getGroups() as $group) {
43 $oprofile = Ostatus_profile::staticGet('group_id', $group->id);
45 $this->pingReply($oprofile);
47 $this->pushGroup($group->id);
51 foreach ($notice->getReplies() as $profile_id) {
52 $oprofile = Ostatus_profile::staticGet('profile_id', $profile_id);
54 $this->pingReply($oprofile);
64 // For local posts, ping the PuSH hub to update their feed.
65 // http://identi.ca/api/statuses/user_timeline/1.atom
66 $feed = common_local_url('ApiTimelineUser',
67 array('id' => $this->user->id,
69 $this->pushFeed($feed, array($this, 'userFeedForNotice'));
73 function pushGroup($group_id)
75 // For a local group, ping the PuSH hub to update its feed.
76 // Updates may come from either a local or a remote user.
77 $feed = common_local_url('ApiTimelineGroup',
78 array('id' => $group_id,
80 $this->pushFeed($feed, array($this, 'groupFeedForNotice'), $group_id);
83 function pingReply($oprofile)
86 if (!empty($oprofile->salmonuri)) {
87 // For local posts, send a Salmon ping to the mentioned
88 // remote user or group.
89 // @fixme as an optimization we can skip this if the
90 // remote profile is subscribed to the author.
92 common_log(LOG_INFO, "Prepping to send notice '{$this->notice->uri}' to remote profile '{$oprofile->uri}'.");
94 $xml = '<?xml version="1.0" encoding="UTF-8" ?' . '>';
95 $xml .= $this->notice->asAtomEntry(true, true);
97 $data = array('salmonuri' => $oprofile->salmonuri,
100 $qm = QueueManager::get();
101 $qm->enqueue($data, 'salmonout');
107 * @param string $feed URI to the feed
108 * @param callable $callback function to generate Atom feed update if needed
109 * any additional params are passed to the callback.
111 function pushFeed($feed, $callback)
113 $hub = common_config('ostatus', 'hub');
115 $this->pushFeedExternal($feed, $hub);
121 $args = array_slice(func_get_args(), 2);
122 $atom = call_user_func_array($callback, $args);
123 $this->pushFeedInternal($atom, $sub);
125 common_log(LOG_INFO, "No PuSH subscribers for $feed");
131 * Ping external hub about this update.
132 * The hub will pull the feed and check for new items later.
133 * Not guaranteed safe in an environment with database replication.
135 * @param string $feed feed topic URI
136 * @param string $hub PuSH hub URI
137 * @fixme can consolidate pings for user & group posts
139 function pushFeedExternal($feed, $hub)
141 $client = new HTTPClient();
143 $data = array('hub.mode' => 'publish',
145 $response = $client->post($hub, array(), $data);
146 if ($response->getStatus() == 204) {
147 common_log(LOG_INFO, "PuSH ping to hub $hub for $feed ok");
150 common_log(LOG_ERR, "PuSH ping to hub $hub for $feed failed with HTTP " .
151 $response->getStatus() . ': ' .
152 $response->getBody());
154 } catch (Exception $e) {
155 common_log(LOG_ERR, "PuSH ping to hub $hub for $feed failed: " . $e->getMessage());
161 * Queue up direct feed update pushes to subscribers on our internal hub.
162 * @param string $atom update feed, containing only new/changed items
163 * @param HubSub $sub open query of subscribers
165 function pushFeedInternal($atom, $sub)
167 common_log(LOG_INFO, "Preparing $sub->N PuSH distribution(s) for $sub->topic");
168 while ($sub->fetch()) {
169 $sub->distribute($atom);
174 * Build a single-item version of the sending user's Atom feed.
177 function userFeedForNotice()
179 // @fixme this feels VERY hacky...
180 // should probably be a cleaner way to do it
183 $api = new ApiTimelineUserAction();
184 $api->prepare(array('id' => $this->notice->profile_id,
186 'max_id' => $this->notice->id,
187 'since_id' => $this->notice->id - 1));
188 $api->showTimeline();
189 $feed = ob_get_clean();
191 // ...and override the content-type back to something normal... eww!
192 // hope there's no other headers that got set while we weren't looking.
193 header('Content-Type: text/html; charset=utf-8');
195 common_log(LOG_DEBUG, $feed);
199 function groupFeedForNotice($group_id)
201 // @fixme this feels VERY hacky...
202 // should probably be a cleaner way to do it
205 $api = new ApiTimelineGroupAction();
206 $args = array('id' => $group_id,
208 'max_id' => $this->notice->id,
209 'since_id' => $this->notice->id - 1);
210 $api->prepare($args);
212 $feed = ob_get_clean();
214 // ...and override the content-type back to something normal... eww!
215 // hope there's no other headers that got set while we weren't looking.
216 header('Content-Type: text/html; charset=utf-8');
218 common_log(LOG_DEBUG, $feed);