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::getByHashkey($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 }