summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
-rw-r--r--plugins/OStatus/OStatusPlugin.php4
-rw-r--r--plugins/OStatus/actions/pushcallback.php9
-rw-r--r--plugins/OStatus/classes/FeedSub.php10
-rw-r--r--plugins/OStatus/lib/pushinputqueuehandler.php49
4 files changed, 70 insertions, 2 deletions
diff --git a/plugins/OStatus/OStatusPlugin.php b/plugins/OStatus/OStatusPlugin.php
index b966661db..c5a2db3d8 100644
--- a/plugins/OStatus/OStatusPlugin.php
+++ b/plugins/OStatus/OStatusPlugin.php
@@ -78,9 +78,13 @@ class OStatusPlugin extends Plugin
*/
function onEndInitializeQueueManager(QueueManager $qm)
{
+ // Outgoing from our internal PuSH hub
$qm->connect('hubverify', 'HubVerifyQueueHandler');
$qm->connect('hubdistrib', 'HubDistribQueueHandler');
$qm->connect('hubout', 'HubOutQueueHandler');
+
+ // Incoming from a foreign PuSH hub
+ $qm->connect('pushinput', 'PushInputQueueHandler');
return true;
}
diff --git a/plugins/OStatus/actions/pushcallback.php b/plugins/OStatus/actions/pushcallback.php
index 7e1227a66..9e976a80d 100644
--- a/plugins/OStatus/actions/pushcallback.php
+++ b/plugins/OStatus/actions/pushcallback.php
@@ -60,9 +60,14 @@ class PushCallbackAction extends Action
$post = file_get_contents('php://input');
- // @fixme Queue this to a background process; we should return
+ // Queue this to a background process; we should return
// as quickly as possible from a distribution POST.
- $feedsub->receive($post, $hmac);
+ // If queues are disabled this'll process immediately.
+ $data = array('feedsub_id' => $feedsub->id,
+ 'post' => $post,
+ 'hmac' => $hmac);
+ $qm = QueueManager::get();
+ $qm->enqueue($data, 'pushinput');
}
/**
diff --git a/plugins/OStatus/classes/FeedSub.php b/plugins/OStatus/classes/FeedSub.php
index bf9d063fa..31241d3de 100644
--- a/plugins/OStatus/classes/FeedSub.php
+++ b/plugins/OStatus/classes/FeedSub.php
@@ -372,6 +372,12 @@ class FeedSub extends Memcached_DataObject
* feed (as a DOMDocument) will be passed to the StartFeedSubHandleFeed
* and EndFeedSubHandleFeed events for processing.
*
+ * Not guaranteed to be running in an immediate POST context; may be run
+ * from a queue handler.
+ *
+ * Side effects: the feedsub record's lastupdate field will be updated
+ * to the current time (not published time) if we got a legit update.
+ *
* @param string $post source of Atom or RSS feed
* @param string $hmac X-Hub-Signature header, if present
*/
@@ -402,6 +408,10 @@ class FeedSub extends Memcached_DataObject
return;
}
+ $orig = clone($this);
+ $this->last_update = common_sql_now();
+ $this->update($orig);
+
Event::handle('StartFeedSubReceive', array($this, $feed));
Event::handle('EndFeedSubReceive', array($this, $feed));
}
diff --git a/plugins/OStatus/lib/pushinputqueuehandler.php b/plugins/OStatus/lib/pushinputqueuehandler.php
new file mode 100644
index 000000000..cbd9139b5
--- /dev/null
+++ b/plugins/OStatus/lib/pushinputqueuehandler.php
@@ -0,0 +1,49 @@
+<?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/>.
+ */
+
+/**
+ * Process a feed distribution POST from a PuSH hub.
+ * @package FeedSub
+ * @author Brion Vibber <brion@status.net>
+ */
+
+class PushInputQueueHandler extends QueueHandler
+{
+ function transport()
+ {
+ return 'pushinput';
+ }
+
+ function handle($data)
+ {
+ assert(is_array($data));
+
+ $feedsub_id = $data['feedsub_id'];
+ $post = $data['post'];
+ $hmac = $data['hmac'];
+
+ $feedsub = FeedSub::staticGet('id', $feedsub_id);
+ if ($feedsub) {
+ $feedsub->receive($post, $hmac);
+ } else {
+ common_log(LOG_ERR, "Discarding POST to unknown feed subscription id $feedsub_id");
+ }
+ return true;
+ }
+}