lib/stompqueuemanager.php

   1 <?php
   2 /**
   3  * Laconica, the distributed open-source microblogging tool
   4  *
   5  * Abstract class for queue managers
   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   Laconica
  24  * @author    Evan Prodromou <evan@controlyourself.ca>
  25  * @author    Sarven Capadisli <csarven@controlyourself.ca>
  26  * @copyright 2009 Control Yourself, Inc.
  27  * @license   http://www.fsf.org/licensing/licenses/agpl-3.0.html GNU Affero General Public License version 3.0
  28  * @link      http://laconi.ca/
  29  */
  30
  31 require_once 'Stomp.php';
  32
  33 class StompQueueManager
  34 {
  35     var $server = null;
  36     var $username = null;
  37     var $password = null;
  38     var $base = null;
  39     var $con = null;
  40
  41     function __construct()
  42     {
  43         $this->server   = common_config('queue', 'stomp_server');
  44         $this->username = common_config('queue', 'stomp_username');
  45         $this->password = common_config('queue', 'stomp_password');
  46         $this->base     = common_config('queue', 'queue_basename');
  47     }
  48
  49     function _connect()
  50     {
  51         if (empty($this->con)) {
  52             $this->_log(LOG_INFO, "Connecting to '$this->server' as '$this->username'...");
  53             $this->con = new Stomp($this->server);
  54
  55             if ($this->con->connect($this->username, $this->password)) {
  56                 $this->_log(LOG_INFO, "Connected.");
  57             } else {
  58                 $this->_log(LOG_ERR, 'Failed to connect to queue server');
  59                 throw new ServerException('Failed to connect to queue server');
  60             }
  61         }
  62     }
  63
  64     function enqueue($object, $queue)
  65     {
  66         $notice = $object;
  67
  68         $this->_connect();
  69
  70         // XXX: serialize and send entire notice
  71
  72         $result = $this->con->send($this->_queueName($queue),
  73                                    $notice->id,                 // BODY of the message
  74                                    array ('created' => $notice->created));
  75
  76         if (!$result) {
  77             common_log(LOG_ERR, 'Error sending to '.$queue.' queue');
  78             return false;
  79         }
  80
  81         common_log(LOG_DEBUG, 'complete remote queueing notice ID = '
  82                    . $notice->id . ' for ' . $queue);
  83     }
  84
  85     function service($queue, $handler)
  86     {
  87         $result = null;
  88
  89         $this->_connect();
  90
  91         $this->con->setReadTimeout($handler->timeout());
  92
  93         $this->con->subscribe($this->_queueName($queue));
  94
  95         while (true) {
  96
  97             $frame = $this->con->readFrame();
  98
  99             if (!empty($frame)) {
 100                 $notice = Notice::staticGet('id', $frame->body);
 101
 102                 if (empty($notice)) {
 103                     $this->_log(LOG_WARNING, 'Got ID '. $frame->body .' for non-existent notice in queue '. $queue);
 104                     $this->con->ack($frame);
 105                 } else {
 106                     if ($handler->handle_notice($notice)) {
 107                         $this->_log(LOG_INFO, 'Successfully handled notice '. $notice->id .' posted at ' . $frame->headers['created'] . ' in queue '. $queue);
 108                         $this->con->ack($frame);
 109                     } else {
 110                         $this->_log(LOG_WARNING, 'Failed handling notice '. $notice->id .' posted at ' . $frame->headers['created']  . ' in queue '. $queue);
 111                         // FIXME we probably shouldn't have to do
 112                         // this kind of queue management ourselves
 113                         $this->con->ack($frame);
 114                         $this->enqueue($notice, $queue);
 115                     }
 116                     unset($notice);
 117                 }
 118
 119                 unset($frame);
 120
 121                 $handler->idle(QUEUE_HANDLER_HIT_IDLE);
 122
 123             } else {
 124                 $handler->idle(QUEUE_HANDLER_MISS_IDLE);
 125             }
 126         }
 127
 128         $this->con->unsubscribe($this->_queueName($queue));
 129     }
 130
 131     function _queueName($queue)
 132     {
 133         return common_config('queue', 'queue_basename') . $queue;
 134     }
 135
 136     function _log($level, $msg)
 137     {
 138         common_log($level, 'StompQueueManager: '.$msg);
 139     }
 140 }