*/
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;
*/
function onStartEnqueueNotice($notice, &$transports)
{
- $transports[] = 'hubdistrib';
+ $transports[] = 'ostatus';
return true;
}
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);
- }
- }
}
/**
$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;
}
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();
}
/**
*/
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) {
// 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;
}
$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;
}
+++ /dev/null
-<?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 PuSH subscription verification from our internal hub.
- * Queue up final distribution for
- * @package Hub
- * @author Brion Vibber <brion@status.net>
- */
-class HubDistribQueueHandler extends QueueHandler
-{
- function transport()
- {
- return 'hubdistrib';
- }
-
- function handle($notice)
- {
- assert($notice instanceof Notice);
-
- $this->pushUser($notice);
- foreach ($notice->getGroups() as $group) {
- $this->pushGroup($notice, $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()) {
- $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()) {
- $sub->distribute($atom);
- }
- }
-
- /**
- * Build a single-item version of the sending user's Atom feed.
- * @param Notice $notice
- * @return string
- */
- function userFeedForNotice($notice)
- {
- // @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,
- 'format' => 'atom',
- 'max_id' => $notice->id,
- 'since_id' => $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, $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;
- }
-
-}
-
--- /dev/null
+<?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/>.
+ */
+
+/**
+ * Prepare PuSH and Salmon distributions for an outgoing message.
+ *
+ * @package OStatusPlugin
+ * @author Brion Vibber <brion@status.net>
+ */
+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 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');
+ }
+ }
+ }
+
+ /**
+ * @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;
+ }
+
+}
+
--- /dev/null
+<?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;
+ }
+}
projects / quix0rs-gnu-social.git / commitdiff