1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192 |
- <?php
- if (!defined('STATUSNET')) {
- exit(1);
- }
- class HubPrepQueueHandler extends QueueHandler
- {
-
-
-
-
-
-
- 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));
-
-
-
-
-
-
-
- try {
- $n = 0;
- while (count($pushCallbacks) && $n < self::ROLLING_BATCH) {
- $n++;
- $callback = array_shift($pushCallbacks);
- $sub = HubSub::getByHashkey($topic, $callback);
- if (!$sub) {
- common_log(LOG_ERR, "Skipping WebSub delivery for deleted(?) consumer $callback on $topic");
- continue;
- }
- $sub->distribute($atom);
- }
- } catch (Exception $e) {
- common_log(LOG_ERR, "Exception during WebSub batch out: " .
- $e->getMessage() .
- " prepping $topic to $callback");
- }
-
- if (count($pushCallbacks) > 0) {
- $sub = new HubSub();
- $sub->topic = $topic;
- $sub->bulkDistribute($atom, $pushCallbacks);
- }
- return true;
- }
- }
|