lib/dbqueuemanager.php

   1 <?php
   2 /**
   3  * StatusNet, the distributed open-source microblogging tool
   4  *
   5  * Simple-minded queue manager for storing items in the database
   6  *
   7  * PHP version 5
   8  *
   9  * LICENCE: This program is free software: you can redistribute it and/or modify
  10  * it under the terms of the GNU Affero General Public License as published by
  11  * the Free Software Foundation, either version 3 of the License, or
  12  * (at your option) any later version.
  13  *
  14  * This program is distributed in the hope that it will be useful,
  15  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  16  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  17  * GNU Affero General Public License for more details.
  18  *
  19  * You should have received a copy of the GNU Affero General Public License
  20  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  21  *
  22  * @category  QueueManager
  23  * @package   StatusNet
  24  * @author    Evan Prodromou <evan@status.net>
  25  * @author    Brion Vibber <brion@status.net>
  26  * @copyright 2009-2010 StatusNet, Inc.
  27  * @license   http://www.fsf.org/licensing/licenses/agpl-3.0.html GNU Affero General Public License version 3.0
  28  * @link      http://status.net/
  29  */
  30
  31 class DBQueueManager extends QueueManager
  32 {
  33     /**
  34      * Saves a notice object reference into the queue item table.
  35      * @return boolean true on success
  36      * @throws ServerException on failure
  37      */
  38     public function enqueue($object, $queue)
  39     {
  40         $notice = $object;
  41
  42         $qi = new Queue_item();
  43
  44         $qi->notice_id = $notice->id;
  45         $qi->transport = $queue;
  46         $qi->created   = $notice->created;
  47         $result        = $qi->insert();
  48
  49         if (!$result) {
  50             common_log_db_error($qi, 'INSERT', __FILE__);
  51             throw new ServerException('DB error inserting queue item');
  52         }
  53
  54         $this->stats('enqueued', $queue);
  55
  56         return true;
  57     }
  58
  59     /**
  60      * Poll every minute for new events during idle periods.
  61      * We'll look in more often when there's data available.
  62      *
  63      * @return int seconds
  64      */
  65     public function pollInterval()
  66     {
  67         return 60;
  68     }
  69
  70     /**
  71      * Run a polling cycle during idle processing in the input loop.
  72      * @return boolean true if we had a hit
  73      */
  74     public function poll()
  75     {
  76         $this->_log(LOG_DEBUG, 'Checking for notices...');
  77         $item = $this->_nextItem();
  78         if ($item === false) {
  79             $this->_log(LOG_DEBUG, 'No notices waiting; idling.');
  80             return false;
  81         }
  82         if ($item === true) {
  83             // We dequeued an entry for a deleted or invalid notice.
  84             // Consider it a hit for poll rate purposes.
  85             return true;
  86         }
  87
  88         list($queue, $notice) = $item;
  89         $this->_log(LOG_INFO, 'Got notice '. $notice->id . ' for transport ' . $queue);
  90
  91         // Yay! Got one!
  92         $handler = $this->getHandler($queue);
  93         if ($handler) {
  94             if ($handler->handle_notice($notice)) {
  95                 $this->_log(LOG_INFO, "[$queue:notice $notice->id] Successfully handled notice");
  96                 $this->_done($notice, $queue);
  97             } else {
  98                 $this->_log(LOG_INFO, "[$queue:notice $notice->id] Failed to handle notice");
  99                 $this->_fail($notice, $queue);
 100             }
 101         } else {
 102             $this->_log(LOG_INFO, "[$queue:notice $notice->id] No handler for queue $queue; discarding.");
 103             $this->_done($notice, $queue);
 104         }
 105         return true;
 106     }
 107
 108     /**
 109      * Pop the oldest unclaimed item off the queue set and claim it.
 110      *
 111      * @return mixed false if no items; true if bogus hit; otherwise array(string, Notice)
 112      *               giving the queue transport name.
 113      */
 114     protected function _nextItem()
 115     {
 116         $start = time();
 117         $result = null;
 118
 119         $qi = Queue_item::top();
 120         if (empty($qi)) {
 121             return false;
 122         }
 123
 124         $queue = $qi->transport;
 125         $notice = Notice::staticGet('id', $qi->notice_id);
 126         if (empty($notice)) {
 127             $this->_log(LOG_INFO, "[$queue:notice $notice->id] dequeued non-existent notice");
 128             $qi->delete();
 129             return true;
 130         }
 131
 132         $result = $notice;
 133         return array($queue, $notice);
 134     }
 135
 136     /**
 137      * Delete our claimed item from the queue after successful processing.
 138      *
 139      * @param Notice $object
 140      * @param string $queue
 141      */
 142     protected function _done($object, $queue)
 143     {
 144         // XXX: right now, we only handle notices
 145
 146         $notice = $object;
 147
 148         $qi = Queue_item::pkeyGet(array('notice_id' => $notice->id,
 149                                         'transport' => $queue));
 150
 151         if (empty($qi)) {
 152             $this->_log(LOG_INFO, "[$queue:notice $notice->id] Cannot find queue item");
 153         } else {
 154             if (empty($qi->claimed)) {
 155                 $this->_log(LOG_WARNING, "[$queue:notice $notice->id] Reluctantly releasing unclaimed queue item");
 156             }
 157             $qi->delete();
 158             $qi->free();
 159         }
 160
 161         $this->_log(LOG_INFO, "[$queue:notice $notice->id] done with item");
 162         $this->stats('handled', $queue);
 163
 164         $notice->free();
 165     }
 166
 167     /**
 168      * Free our claimed queue item for later reprocessing in case of
 169      * temporary failure.
 170      *
 171      * @param Notice $object
 172      * @param string $queue
 173      */
 174     protected function _fail($object, $queue)
 175     {
 176         // XXX: right now, we only handle notices
 177
 178         $notice = $object;
 179
 180         $qi = Queue_item::pkeyGet(array('notice_id' => $notice->id,
 181                                         'transport' => $queue));
 182
 183         if (empty($qi)) {
 184             $this->_log(LOG_INFO, "[$queue:notice $notice->id] Cannot find queue item");
 185         } else {
 186             if (empty($qi->claimed)) {
 187                 $this->_log(LOG_WARNING, "[$queue:notice $notice->id] Ignoring failure for unclaimed queue item");
 188             } else {
 189                 $orig = clone($qi);
 190                 $qi->claimed = null;
 191                 $qi->update($orig);
 192                 $qi = null;
 193             }
 194         }
 195
 196         $this->_log(LOG_INFO, "[$queue:notice $notice->id] done with queue item");
 197         $this->stats('error', $queue);
 198
 199         $notice->free();
 200     }
 201
 202     protected function _log($level, $msg)
 203     {
 204         common_log($level, 'DBQueueManager: '.$msg);
 205     }
 206 }