diff options
| -rw-r--r-- | plugins/OStatus/OStatusPlugin.php | 28 | ||||
| -rw-r--r-- | plugins/OStatus/actions/groupsalmon.php | 9 | ||||
| -rw-r--r-- | plugins/OStatus/classes/Ostatus_profile.php | 15 | ||||
| -rw-r--r-- | plugins/OStatus/lib/ostatusqueuehandler.php (renamed from plugins/OStatus/lib/hubdistribqueuehandler.php) | 95 | ||||
| -rw-r--r-- | plugins/OStatus/lib/salmonoutqueuehandler.php | 44 |
5 files changed, 140 insertions, 51 deletions
diff --git a/plugins/OStatus/OStatusPlugin.php b/plugins/OStatus/OStatusPlugin.php index 35f952935..9376c048d 100644 --- a/plugins/OStatus/OStatusPlugin.php +++ b/plugins/OStatus/OStatusPlugin.php @@ -78,11 +78,16 @@ class OStatusPlugin extends Plugin */ function onEndInitializeQueueManager(QueueManager $qm) { + // Prepare outgoing distributions after notice save. + $qm->connect('ostatus', 'OStatusQueueHandler'); + // Outgoing from our internal PuSH hub $qm->connect('hubverify', 'HubVerifyQueueHandler'); - $qm->connect('hubdistrib', 'HubDistribQueueHandler'); $qm->connect('hubout', 'HubOutQueueHandler'); + // Outgoing Salmon replies (when we don't need a return value) + $qm->connect('salmonout', 'SalmonOutQueueHandler'); + // Incoming from a foreign PuSH hub $qm->connect('pushinput', 'PushInputQueueHandler'); return true; @@ -93,7 +98,7 @@ class OStatusPlugin extends Plugin */ function onStartEnqueueNotice($notice, &$transports) { - $transports[] = 'hubdistrib'; + $transports[] = 'ostatus'; return true; } @@ -199,25 +204,6 @@ class OStatusPlugin extends Plugin function onEndNoticeSave($notice) { - $mentioned = $notice->getReplies(); - - foreach ($mentioned as $profile_id) { - - $oprofile = Ostatus_profile::staticGet('profile_id', $profile_id); - - if (!empty($oprofile) && !empty($oprofile->salmonuri)) { - - common_log(LOG_INFO, "Sending notice '{$notice->uri}' to remote profile '{$oprofile->uri}'."); - - // FIXME: this needs to go out in a queue handler - - $xml = '<?xml version="1.0" encoding="UTF-8" ?' . '>'; - $xml .= $notice->asAtomEntry(true, true); - - $salmon = new Salmon(); - $salmon->post($oprofile->salmonuri, $xml); - } - } } /** diff --git a/plugins/OStatus/actions/groupsalmon.php b/plugins/OStatus/actions/groupsalmon.php index 2e4fe9443..29377b5fa 100644 --- a/plugins/OStatus/actions/groupsalmon.php +++ b/plugins/OStatus/actions/groupsalmon.php @@ -46,6 +46,11 @@ class GroupsalmonAction extends SalmonAction $this->clientError(_('No such group.')); } + $oprofile = Ostatus_profile::staticGet('group_id', $id); + if ($oprofile) { + $this->clientError(_m("Can't accept remote posts for a remote group.")); + } + return true; } @@ -74,13 +79,13 @@ class GroupsalmonAction extends SalmonAction throw new ClientException("Not to the attention of anyone."); } else { $uri = common_local_url('groupbyid', array('id' => $this->group->id)); - if (!in_array($context->attention, $uri)) { + if (!in_array($uri, $context->attention)) { throw new ClientException("Not to the attention of this group."); } } $profile = $this->ensureProfile(); - // @fixme save the post + $this->saveNotice(); } /** diff --git a/plugins/OStatus/classes/Ostatus_profile.php b/plugins/OStatus/classes/Ostatus_profile.php index 6beaf0f5d..82dbf773d 100644 --- a/plugins/OStatus/classes/Ostatus_profile.php +++ b/plugins/OStatus/classes/Ostatus_profile.php @@ -650,6 +650,7 @@ class Ostatus_profile extends Memcached_DataObject */ protected function filterReplies($sender, &$attention_uris) { + common_log(LOG_DEBUG, "Original reply recipients: " . implode(', ', $attention_uris)); $groups = array(); $replies = array(); foreach ($attention_uris as $recipient) { @@ -668,6 +669,8 @@ class Ostatus_profile extends Memcached_DataObject // Deliver to local members of this remote group. // @fixme sender verification? $groups[] = $oprofile->group_id; + } else { + common_log(LOG_DEBUG, "Skipping reply to remote profile $recipient"); } continue; } @@ -683,14 +686,24 @@ class Ostatus_profile extends Memcached_DataObject $group = User_group::staticGet('id', $id); if ($group) { // Deliver to all members of this local group if allowed. - if ($sender->localProfile()->isMember($group)) { + $profile = $sender->localProfile(); + if ($profile->isMember($group)) { $groups[] = $group->id; + } else { + common_log(LOG_DEBUG, "Skipping reply to local group $group->nickname as sender $profile->id is not a member"); } continue; + } else { + common_log(LOG_DEBUG, "Skipping reply to bogus group $recipient"); } } + + common_log(LOG_DEBUG, "Skipping reply to unrecognized profile $recipient"); + } $attention_uris = $replies; + common_log(LOG_DEBUG, "Local reply recipients: " . implode(', ', $replies)); + common_log(LOG_DEBUG, "Local group recipients: " . implode(', ', $groups)); return $groups; } diff --git a/plugins/OStatus/lib/hubdistribqueuehandler.php b/plugins/OStatus/lib/ostatusqueuehandler.php index c2bd630f9..c1e50bffa 100644 --- a/plugins/OStatus/lib/hubdistribqueuehandler.php +++ b/plugins/OStatus/lib/ostatusqueuehandler.php @@ -18,46 +18,89 @@ */ /** - * Send a PuSH subscription verification from our internal hub. - * Queue up final distribution for - * @package Hub + * Prepare PuSH and Salmon distributions for an outgoing message. + * + * @package OStatusPlugin * @author Brion Vibber <brion@status.net> */ -class HubDistribQueueHandler extends QueueHandler +class OStatusQueueHandler extends QueueHandler { function transport() { - return 'hubdistrib'; + return 'ostatus'; } function handle($notice) { assert($notice instanceof Notice); - $this->pushUser($notice); + $this->notice = $notice; + $this->user = User::staticGet($notice->profile_id); + + $this->pushUser(); + foreach ($notice->getGroups() as $group) { - $this->pushGroup($notice, $group->id); + $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($notice) + + function pushUser() { - // 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); + 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($notice, $group_id) + 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, $notice); + $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 version="1.0" encoding="UTF-8" ?' . '>'; + $xml .= $this->notice->asAtomEntry(true, true); + + $data = array('salmonuri' => $oprofile->salmonuri, + 'entry' => $xml); + + $qm = QueueManager::get(); + $qm->enqueue($data, 'salmonout'); + } + } } /** @@ -122,7 +165,6 @@ class HubDistribQueueHandler extends QueueHandler function pushFeedInternal($atom, $sub) { common_log(LOG_INFO, "Preparing $sub->N PuSH distribution(s) for $sub->topic"); - $qm = QueueManager::get(); while ($sub->fetch()) { $sub->distribute($atom); } @@ -130,20 +172,19 @@ class HubDistribQueueHandler extends QueueHandler /** * Build a single-item version of the sending user's Atom feed. - * @param Notice $notice * @return string */ - function userFeedForNotice($notice) + 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' => $notice->profile_id, + $api->prepare(array('id' => $this->notice->profile_id, 'format' => 'atom', - 'max_id' => $notice->id, - 'since_id' => $notice->id - 1)); + 'max_id' => $this->notice->id, + 'since_id' => $this->notice->id - 1)); $api->showTimeline(); $feed = ob_get_clean(); @@ -155,7 +196,7 @@ class HubDistribQueueHandler extends QueueHandler return $feed; } - function groupFeedForNotice($group_id, $notice) + function groupFeedForNotice($group_id) { // @fixme this feels VERY hacky... // should probably be a cleaner way to do it @@ -164,8 +205,8 @@ class HubDistribQueueHandler extends QueueHandler $api = new ApiTimelineGroupAction(); $args = array('id' => $group_id, 'format' => 'atom', - 'max_id' => $notice->id, - 'since_id' => $notice->id - 1); + 'max_id' => $this->notice->id, + 'since_id' => $this->notice->id - 1); $api->prepare($args); $api->handle($args); $feed = ob_get_clean(); diff --git a/plugins/OStatus/lib/salmonoutqueuehandler.php b/plugins/OStatus/lib/salmonoutqueuehandler.php new file mode 100644 index 000000000..536ff94af --- /dev/null +++ b/plugins/OStatus/lib/salmonoutqueuehandler.php @@ -0,0 +1,44 @@ +<?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/>. + */ + +/** + * Send a Salmon notification in the background. + * @package OStatusPlugin + * @author Brion Vibber <brion@status.net> + */ +class SalmonOutQueueHandler extends QueueHandler +{ + function transport() + { + return 'salmonout'; + } + + function handle($data) + { + assert(is_array($data)); + assert(is_string($data['salmonuri'])); + assert(is_string($data['entry'])); + + $salmon = new Salmon(); + $salmon->post($data['salmonuri'], $data['entry']); + + // @fixme detect failure and attempt to resend + return true; + } +} |