diff options
| -rw-r--r-- | plugins/OStatus/OStatusPlugin.php | 11 | ||||
| -rw-r--r-- | plugins/OStatus/classes/Feedinfo.php | 4 | ||||
| -rw-r--r-- | plugins/OStatus/classes/HubSub.php | 2 | ||||
| -rw-r--r-- | plugins/OStatus/lib/hubdistribqueuehandler.php | 70 |
4 files changed, 67 insertions, 20 deletions
diff --git a/plugins/OStatus/OStatusPlugin.php b/plugins/OStatus/OStatusPlugin.php index 62ecaf631..4b9b4d2c3 100644 --- a/plugins/OStatus/OStatusPlugin.php +++ b/plugins/OStatus/OStatusPlugin.php @@ -114,10 +114,15 @@ class OStatusPlugin extends Plugin if ($action instanceof ApiTimelineUserAction || $action instanceof ApiTimelineGroupAction) { $id = $action->arg('id'); if (strval(intval($id)) === strval($id)) { - // Canonical form of id in URL? - // Updates will be handled for our internal PuSH hub. + // Canonical form of id in URL? These are used for OStatus syndication. + + $hub = common_config('ostatus', 'hub'); + if (empty($hub)) { + // Updates will be handled through our internal PuSH hub. + $hub = common_local_url('pushhub'); + } $action->element('link', array('rel' => 'hub', - 'href' => common_local_url('pushhub'))); + 'href' => $hub)); // Also, we'll add in the salmon link $action->element('link', array('rel' => 'salmon', diff --git a/plugins/OStatus/classes/Feedinfo.php b/plugins/OStatus/classes/Feedinfo.php index b4e55c364..2344a4a0e 100644 --- a/plugins/OStatus/classes/Feedinfo.php +++ b/plugins/OStatus/classes/Feedinfo.php @@ -160,7 +160,7 @@ class Feedinfo extends Memcached_DataObject function keyTypes() { - return array('id' => 'K'); // @fixme we'll need a profile_id key at least + return array('id' => 'K', 'feeduri' => 'U'); // @fixme we'll need a profile_id key at least } function sequenceKey() @@ -323,7 +323,7 @@ class Feedinfo extends Memcached_DataObject if ($this->secret) { if (preg_match('/^sha1=([0-9a-fA-F]{40})$/', $hmac, $matches)) { $their_hmac = strtolower($matches[1]); - $our_hmac = sha1($xml . $this->secret); + $our_hmac = hash_hmac('sha1', $xml, $this->secret); if ($their_hmac !== $our_hmac) { common_log(LOG_ERR, __METHOD__ . ": ignoring PuSH with bad SHA-1 HMAC: got $their_hmac, expected $our_hmac"); return; diff --git a/plugins/OStatus/classes/HubSub.php b/plugins/OStatus/classes/HubSub.php index 1769f6c94..7071ee5b4 100644 --- a/plugins/OStatus/classes/HubSub.php +++ b/plugins/OStatus/classes/HubSub.php @@ -242,7 +242,7 @@ class HubSub extends Memcached_DataObject { $headers = array('Content-Type: application/atom+xml'); if ($this->secret) { - $hmac = sha1($atom . $this->secret); + $hmac = hash_hmac('sha1', $atom, $this->secret); $headers[] = "X-Hub-Signature: sha1=$hmac"; } else { $hmac = '(none)'; diff --git a/plugins/OStatus/lib/hubdistribqueuehandler.php b/plugins/OStatus/lib/hubdistribqueuehandler.php index a35b8874c..245a57f72 100644 --- a/plugins/OStatus/lib/hubdistribqueuehandler.php +++ b/plugins/OStatus/lib/hubdistribqueuehandler.php @@ -49,15 +49,7 @@ class HubDistribQueueHandler extends QueueHandler $feed = common_local_url('ApiTimelineUser', array('id' => $notice->profile_id, 'format' => 'atom')); - $sub = new HubSub(); - $sub->topic = $feed; - if ($sub->find()) { - $atom = $this->userFeedForNotice($notice); - $this->pushFeeds($atom, $sub); - } else { - common_log(LOG_INFO, "No PuSH subscribers for $feed"); - } - return true; + $this->pushFeed($feed, array($this, 'userFeedForNotice'), $notice); } function pushGroup($notice, $group_id) @@ -65,19 +57,69 @@ class HubDistribQueueHandler extends QueueHandler $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()) { - common_log(LOG_INFO, "Building PuSH feed for $feed"); - $atom = $this->groupFeedForNotice($group_id, $notice); - $this->pushFeeds($atom, $sub); + $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; } - - function pushFeeds($atom, $sub) + /** + * 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(); |