summaryrefslogtreecommitdiff
path: root/plugins
diff options
context:
space:
mode:
Diffstat (limited to 'plugins')
-rw-r--r--plugins/OStatus/OStatusPlugin.php2
-rw-r--r--plugins/OStatus/classes/HubSub.php20
-rw-r--r--plugins/OStatus/lib/ostatusqueuehandler.php31
3 files changed, 52 insertions, 1 deletions
diff --git a/plugins/OStatus/OStatusPlugin.php b/plugins/OStatus/OStatusPlugin.php
index 5a657c83d..c61e2cc5f 100644
--- a/plugins/OStatus/OStatusPlugin.php
+++ b/plugins/OStatus/OStatusPlugin.php
@@ -87,6 +87,8 @@ class OStatusPlugin extends Plugin
// Outgoing from our internal PuSH hub
$qm->connect('hubconf', 'HubConfQueueHandler');
+ $qm->connect('hubprep', 'HubPrepQueueHandler');
+
$qm->connect('hubout', 'HubOutQueueHandler');
// Outgoing Salmon replies (when we don't need a return value)
diff --git a/plugins/OStatus/classes/HubSub.php b/plugins/OStatus/classes/HubSub.php
index 9748b4a56..7db528a4e 100644
--- a/plugins/OStatus/classes/HubSub.php
+++ b/plugins/OStatus/classes/HubSub.php
@@ -305,6 +305,26 @@ class HubSub extends Memcached_DataObject
}
/**
+ * Queue up a large batch of pushes to multiple subscribers
+ * for this same topic update.
+ *
+ * If queues are disabled, this will run immediately.
+ *
+ * @param string $atom well-formed Atom feed
+ * @param array $pushCallbacks list of callback URLs
+ */
+ function bulkDistribute($atom, $pushCallbacks)
+ {
+ $data = array('atom' => $atom,
+ 'topic' => $this->topic,
+ 'pushCallbacks' => $pushCallbacks);
+ common_log(LOG_INFO, "Queuing PuSH batch: $this->topic to " .
+ count($pushCallbacks) . " sites");
+ $qm = QueueManager::get();
+ $qm->enqueue($data, 'hubprep');
+ }
+
+ /**
* Send a 'fat ping' to the subscriber's callback endpoint
* containing the given Atom feed chunk.
*
diff --git a/plugins/OStatus/lib/ostatusqueuehandler.php b/plugins/OStatus/lib/ostatusqueuehandler.php
index d1e58f1d6..8905d2e21 100644
--- a/plugins/OStatus/lib/ostatusqueuehandler.php
+++ b/plugins/OStatus/lib/ostatusqueuehandler.php
@@ -25,6 +25,18 @@
*/
class OStatusQueueHandler extends QueueHandler
{
+ // If we have more than this many subscribing sites on a single feed,
+ // break up the PuSH distribution into smaller batches which will be
+ // rolled into the queue progressively. This reduces disruption to
+ // other, shorter activities being enqueued while we work.
+ const MAX_UNBATCHED = 50;
+
+ // Each batch (a 'hubprep' entry) will have this many items.
+ // Selected to provide a balance between queue packet size
+ // and number of batches that will end up getting processed.
+ // For 20,000 target sites, 1000 should work acceptably.
+ const BATCH_SIZE = 1000;
+
function transport()
{
return 'ostatus';
@@ -147,14 +159,31 @@ class OStatusQueueHandler extends QueueHandler
/**
* Queue up direct feed update pushes to subscribers on our internal hub.
+ * If there are a large number of subscriber sites, intermediate bulk
+ * distribution triggers may be queued.
+ *
* @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");
+ $n = 0;
+ $batch = array();
while ($sub->fetch()) {
- $sub->distribute($atom);
+ $n++;
+ if ($n < self::MAX_UNBATCHED) {
+ $sub->distribute($atom);
+ } else {
+ $batch[] = $sub->callback;
+ if (count($batch) >= self::BATCH_SIZE) {
+ $sub->bulkDistribute($atom, $batch);
+ $batch = array();
+ }
+ }
+ }
+ if (count($batch) >= 0) {
+ $sub->bulkDistribute($atom, $batch);
}
}