]> git.mxchange.org Git - quix0rs-gnu-social.git/blob - plugins/OStatus/lib/hubprepqueuehandler.php
tweak RSSCloud plugin to use new schema layout (but still the old event); the timesta...
[quix0rs-gnu-social.git] / plugins / OStatus / lib / hubprepqueuehandler.php
1 <?php
2 /*
3  * StatusNet - the distributed open-source microblogging tool
4  * Copyright (C) 2010, StatusNet, Inc.
5  *
6  * This program is free software: you can redistribute it and/or modify
7  * it under the terms of the GNU Affero General Public License as published by
8  * the Free Software Foundation, either version 3 of the License, or
9  * (at your option) any later version.
10  *
11  * This program is distributed in the hope that it will be useful,
12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
14  * GNU Affero General Public License for more details.
15  *
16  * You should have received a copy of the GNU Affero General Public License
17  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
18  */
19
20 if (!defined('STATUSNET')) {
21     exit(1);
22 }
23
24 /**
25  * When we have a large batch of PuSH consumers, we break the data set
26  * into smaller chunks. Enqueue final destinations...
27  *
28  * @package Hub
29  * @author Brion Vibber <brion@status.net>
30  */
31 class HubPrepQueueHandler extends QueueHandler
32 {
33     // Enqueue this many low-level distributions before re-queueing the rest
34     // of the batch to be processed later. Helps to keep latency down for other
35     // things happening during a particularly long OStatus delivery session.
36     //
37     // [Could probably ditch this if we had working message delivery priorities
38     // for queueing, but this isn't supported in ActiveMQ 5.3.]
39     const ROLLING_BATCH = 20;
40
41     function transport()
42     {
43         return 'hubprep';
44     }
45
46     function handle($data)
47     {
48         $topic = $data['topic'];
49         $atom = $data['atom'];
50         $pushCallbacks = $data['pushCallbacks'];
51
52         assert(is_string($atom));
53         assert(is_string($topic));
54         assert(is_array($pushCallbacks));
55
56         // Set up distribution for the first n subscribing sites...
57         // If we encounter an uncatchable error, queue handling should
58         // automatically re-run the batch, which could lead to some dupe
59         // distributions.
60         //
61         // Worst case is if one of these hubprep entries dies too many
62         // times and gets dropped; the rest of the batch won't get processed.
63         try {
64             $n = 0;
65             while (count($pushCallbacks) && $n < self::ROLLING_BATCH) {
66                 $n++;
67                 $callback = array_shift($pushCallbacks);
68                 $sub = HubSub::staticGet($topic, $callback);
69                 if (!$sub) {
70                     common_log(LOG_ERR, "Skipping PuSH delivery for deleted(?) consumer $callback on $topic");
71                     continue;
72                 }
73
74                 $sub->distribute($atom);
75             }
76         } catch (Exception $e) {
77             common_log(LOG_ERR, "Exception during PuSH batch out: " .
78                                 $e->getMessage() .
79                                 " prepping $topic to $callback");
80         }
81
82         // And re-queue the rest of the batch!
83         if (count($pushCallbacks) > 0) {
84             $sub = new HubSub();
85             $sub->topic = $topic;
86             $sub->bulkDistribute($atom, $pushCallbacks);
87         }
88
89         return true;
90     }
91 }