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     var $frames = array();
  41
  42     function __construct()
  43     {
  44         $this->server   = common_config('queue', 'stomp_server');
  45         $this->username = common_config('queue', 'stomp_username');
  46         $this->password = common_config('queue', 'stomp_password');
  47         $this->base     = common_config('queue', 'queue_basename');
  48     }
  49
  50     function _connect()
  51     {
  52         if (empty($this->con)) {
  53             $this->_log(LOG_INFO, "Connecting to '$this->server' as '$this->username'...");
  54             $this->con = new Stomp($this->server);
  55
  56             if ($this->con->connect($this->username, $this->password)) {
  57                 $this->_log(LOG_INFO, "Connected.");
  58             } else {
  59                 $this->_log(LOG_ERR, 'Failed to connect to queue server');
  60                 throw new ServerException('Failed to connect to queue server');
  61             }
  62         }
  63     }
  64
  65     function enqueue($object, $queue)
  66     {
  67         $notice = $object;
  68
  69         $this->_connect();
  70
  71         // XXX: serialize and send entire notice
  72
  73         $result = $this->con->send($this->_queueName($queue),
  74                                    $notice->id,                 // BODY of the message
  75                                    array ('created' => $notice->created));
  76
  77         if (!$result) {
  78             common_log(LOG_ERR, 'Error sending to '.$queue.' queue');
  79             return false;
  80         }
  81
  82         common_log(LOG_DEBUG, 'complete remote queueing notice ID = '
  83                    . $notice->id . ' for ' . $queue);
  84     }
  85
  86     function service($queue, $handler)
  87     {
  88         $result = null;
  89
  90         $this->_connect();
  91
  92         $this->con->setReadTimeout($handler->timeout());
  93
  94         $this->con->subscribe($this->_queueName($queue));
  95
  96         while (true) {
  97
  98             $frame = $this->con->readFrame();
  99
 100             if ($frame) {
 101                 $notice = Notice::staticGet($frame->body);
 102
 103                 if ($handler->handle_notice($notice)) {
 104                     $this->_log(LOG_INFO, 'Successfully handled notice '. $notice->id .' posted at ' . $frame->headers['created']);
 105                     $this->con->ack($frame);
 106                 }
 107             }
 108
 109             $handler->idle(0);
 110         }
 111
 112         $this->con->unsubscribe($this->_queueName($queue));
 113     }
 114
 115     function _queueName($queue)
 116     {
 117         return common_config('queue', 'queue_basename') . $queue;
 118     }
 119
 120     function _log($level, $msg)
 121     {
 122         common_log($level, 'StompQueueManager: '.$msg);
 123     }
 124 }