From 2e58802cc9959763f28e2f43c8e0cd0dbe7bcd8e Mon Sep 17 00:00:00 2001 From: Brion Vibber Date: Wed, 24 Feb 2010 02:19:13 +0000 Subject: OStatus: fix group delivery, send reply/group Salmon pings from background. --- plugins/OStatus/lib/ostatusqueuehandler.php | 223 ++++++++++++++++++++++++++++ 1 file changed, 223 insertions(+) create mode 100644 plugins/OStatus/lib/ostatusqueuehandler.php (limited to 'plugins/OStatus/lib/ostatusqueuehandler.php') diff --git a/plugins/OStatus/lib/ostatusqueuehandler.php b/plugins/OStatus/lib/ostatusqueuehandler.php new file mode 100644 index 000000000..c1e50bffa --- /dev/null +++ b/plugins/OStatus/lib/ostatusqueuehandler.php @@ -0,0 +1,223 @@ +. + */ + +/** + * Prepare PuSH and Salmon distributions for an outgoing message. + * + * @package OStatusPlugin + * @author Brion Vibber + */ +class OStatusQueueHandler extends QueueHandler +{ + function transport() + { + return 'ostatus'; + } + + function handle($notice) + { + assert($notice instanceof Notice); + + $this->notice = $notice; + $this->user = User::staticGet($notice->profile_id); + + $this->pushUser(); + + foreach ($notice->getGroups() as $group) { + $oprofile = Ostatus_profile::staticGet('group_id', $group->id); + if ($oprofile) { + $this->pingReply($oprofile); + } else { + $this->pushGroup($group->id); + } + } + + foreach ($notice->getReplies() as $profile_id) { + $oprofile = Ostatus_profile::staticGet('profile_id', $profile_id); + if ($oprofile) { + $this->pingReply($oprofile); + } + } + + return true; + } + + function pushUser() + { + if ($this->user) { + // For local posts, ping the PuSH hub to update their feed. + // http://identi.ca/api/statuses/user_timeline/1.atom + $feed = common_local_url('ApiTimelineUser', + array('id' => $this->user->id, + 'format' => 'atom')); + $this->pushFeed($feed, array($this, 'userFeedForNotice')); + } + } + + function pushGroup($group_id) + { + // For a local group, ping the PuSH hub to update its feed. + // Updates may come from either a local or a remote user. + $feed = common_local_url('ApiTimelineGroup', + array('id' => $group_id, + 'format' => 'atom')); + $this->pushFeed($feed, array($this, 'groupFeedForNotice'), $group_id); + } + + function pingReply($oprofile) + { + if ($this->user) { + if (!empty($oprofile->salmonuri)) { + // For local posts, send a Salmon ping to the mentioned + // remote user or group. + // @fixme as an optimization we can skip this if the + // remote profile is subscribed to the author. + + common_log(LOG_INFO, "Prepping to send notice '{$this->notice->uri}' to remote profile '{$oprofile->uri}'."); + + $xml = ''; + $xml .= $this->notice->asAtomEntry(true, true); + + $data = array('salmonuri' => $oprofile->salmonuri, + 'entry' => $xml); + + $qm = QueueManager::get(); + $qm->enqueue($data, 'salmonout'); + } + } + } + + /** + * @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()) { + $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"); + while ($sub->fetch()) { + $sub->distribute($atom); + } + } + + /** + * Build a single-item version of the sending user's Atom feed. + * @return string + */ + function userFeedForNotice() + { + // @fixme this feels VERY hacky... + // should probably be a cleaner way to do it + + ob_start(); + $api = new ApiTimelineUserAction(); + $api->prepare(array('id' => $this->notice->profile_id, + 'format' => 'atom', + 'max_id' => $this->notice->id, + 'since_id' => $this->notice->id - 1)); + $api->showTimeline(); + $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; + } + + function groupFeedForNotice($group_id) + { + // @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' => $this->notice->id, + 'since_id' => $this->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; + } + +} + -- cgit v1.2.3-54-g00ecf