lib/queuehandler.php

   1 <?php
   2 /*
   3  * Laconica - a distributed open-source microblogging tool
   4  * Copyright (C) 2008, 2009, Control Yourself, 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('LACONICA')) { exit(1); }
  21
  22 require_once(INSTALLDIR.'/lib/daemon.php');
  23 require_once(INSTALLDIR.'/classes/Queue_item.php');
  24 require_once(INSTALLDIR.'/classes/Notice.php');
  25
  26 define('CLAIM_TIMEOUT', 1200);
  27 define('QUEUE_HANDLER_MISS_IDLE', 10);
  28 define('QUEUE_HANDLER_HIT_IDLE', 10);
  29
  30 class QueueHandler extends Daemon
  31 {
  32     var $_id = 'generic';
  33
  34     function QueueHandler($id=null)
  35     {
  36         if ($id) {
  37             $this->set_id($id);
  38         }
  39     }
  40
  41     function timeout()
  42     {
  43         return null;
  44     }
  45
  46     function class_name()
  47     {
  48         return ucfirst($this->transport()) . 'Handler';
  49     }
  50
  51     function name()
  52     {
  53         return strtolower($this->class_name().'.'.$this->get_id());
  54     }
  55
  56     function get_id()
  57     {
  58         return $this->_id;
  59     }
  60
  61     function set_id($id)
  62     {
  63         $this->_id = $id;
  64     }
  65
  66     function transport()
  67     {
  68         return null;
  69     }
  70
  71     function start()
  72     {
  73     }
  74
  75     function finish()
  76     {
  77     }
  78
  79     function handle_notice($notice)
  80     {
  81         return true;
  82     }
  83
  84     function run()
  85     {
  86         if (!$this->start()) {
  87             return false;
  88         }
  89
  90         $this->log(LOG_INFO, 'checking for queued notices');
  91
  92         $queue   = $this->transport();
  93         $timeout = $this->timeout();
  94
  95         $qm = QueueManager::get();
  96
  97         while (true) {
  98             $this->log(LOG_DEBUG, 'Checking for notices...');
  99             $notice = $qm->nextItem($queue, $timeout);
 100             if (empty($notice)) {
 101                 $this->log(LOG_DEBUG, 'No notices waiting; idling.');
 102                 // Nothing in the queue. Do you
 103                 // have other tasks, like servicing your
 104                 // XMPP connection, to do?
 105                 $this->idle(QUEUE_HANDLER_MISS_IDLE);
 106             } else {
 107                 $this->log(LOG_INFO, 'Got notice '. $notice->id);
 108                 // Yay! Got one!
 109                 if ($this->handle_notice($notice)) {
 110                     $this->log(LOG_INFO, 'Successfully handled notice '. $notice->id);
 111                     $qm->done($notice, $queue);
 112                 } else {
 113                     $this->log(LOG_INFO, 'Failed to handle notice '. $notice->id);
 114                     $qm->fail($notice, $queue);
 115                 }
 116                 // Chance to e.g. service your XMPP connection
 117                 $this->log(LOG_DEBUG, 'Idling after success.');
 118                 $this->idle(QUEUE_HANDLER_HIT_IDLE);
 119             }
 120             // XXX: when do we give up?
 121         }
 122
 123         if (!$this->finish()) {
 124             return false;
 125         }
 126         return true;
 127     }
 128
 129     function idle($timeout=0)
 130     {
 131         if ($timeout > 0) {
 132             sleep($timeout);
 133         }
 134     }
 135
 136     function log($level, $msg)
 137     {
 138         common_log($level, $this->class_name() . ' ('. $this->get_id() .'): '.$msg);
 139     }
 140 }
 141