diff --git a/plugins/OStatus/lib/hubprepqueuehandler.php b/plugins/OStatus/lib/hubprepqueuehandler.php new file mode 100644 index 0000000000..0d585938f4 --- /dev/null +++ b/plugins/OStatus/lib/hubprepqueuehandler.php @@ -0,0 +1,87 @@ +. + */ + +/** + * When we have a large batch of PuSH consumers, we break the data set + * into smaller chunks. Enqueue final destinations... + * + * @package Hub + * @author Brion Vibber + */ +class HubPrepQueueHandler extends QueueHandler +{ + // Enqueue this many low-level distributions before re-queueing the rest + // of the batch to be processed later. Helps to keep latency down for other + // things happening during a particularly long OStatus delivery session. + // + // [Could probably ditch this if we had working message delivery priorities + // for queueing, but this isn't supported in ActiveMQ 5.3.] + const ROLLING_BATCH = 20; + + function transport() + { + return 'hubprep'; + } + + function handle($data) + { + $topic = $data['topic']; + $atom = $data['atom']; + $pushCallbacks = $data['pushCallbacks']; + + assert(is_string($atom)); + assert(is_string($topic)); + assert(is_array($pushCallbacks)); + + // Set up distribution for the first n subscribing sites... + // If we encounter an uncatchable error, queue handling should + // automatically re-run the batch, which could lead to some dupe + // distributions. + // + // Worst case is if one of these hubprep entries dies too many + // times and gets dropped; the rest of the batch won't get processed. + try { + $n = 0; + while (count($pushCallbacks) && $n < self::ROLLING_BATCH) { + $n++; + $callback = array_shift($pushCallbacks); + $sub = HubSub::staticGet($topic, $callback); + if (!$sub) { + common_log(LOG_ERR, "Skipping PuSH delivery for deleted(?) consumer $callback on $topic"); + continue; + } + + $sub->distribute($atom); + } + } catch (Exception $e) { + common_log(LOG_ERR, "Exception during PuSH batch out: " . + $e->getMessage() . + " prepping $topic to $callback"); + } + + // And re-queue the rest of the batch! + if (count($pushCallbacks) > 0) { + $sub = new HubSub(); + $sub->topic = $topic; + $sub->bulkDistribute($atom, $pushCallbacks); + } + + return true; + } +}