lib/xmppmanager.php

   1 <?php
   2 /*
   3  * StatusNet - the distributed open-source microblogging tool
   4  * Copyright (C) 2008, 2009, 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') && !defined('LACONICA')) { exit(1); }
  21
  22 /**
  23  * XMPP background connection manager for XMPP-using queue handlers,
  24  * allowing them to send outgoing messages on the right connection.
  25  *
  26  * Input is handled during socket select loop, keepalive pings during idle.
  27  * Any incoming messages will be forwarded to the main XmppDaemon process,
  28  * which handles direct user interaction.
  29  *
  30  * In a multi-site queuedaemon.php run, one connection will be instantiated
  31  * for each site being handled by the current process that has XMPP enabled.
  32  */
  33
  34 class XmppManager extends IoManager
  35 {
  36     protected $site = null;
  37     protected $pingid = 0;
  38     protected $lastping = null;
  39
  40     static protected $singletons = array();
  41
  42     const PING_INTERVAL = 120;
  43
  44     /**
  45      * Fetch the singleton XmppManager for the current site.
  46      * @return mixed XmppManager, or false if unneeded
  47      */
  48     public static function get()
  49     {
  50         if (common_config('xmpp', 'enabled')) {
  51             $site = common_config('site', 'server');
  52             if (empty(self::$singletons[$site])) {
  53                 self::$singletons[$site] = new XmppManager();
  54             }
  55             return self::$singletons[$site];
  56         } else {
  57             return false;
  58         }
  59     }
  60
  61     /**
  62      * Tell the i/o master we need one instance for each supporting site
  63      * being handled in this process.
  64      */
  65     public static function multiSite()
  66     {
  67         return IoManager::INSTANCE_PER_SITE;
  68     }
  69
  70     function __construct()
  71     {
  72         $this->site = common_config('site', 'server');
  73     }
  74
  75     /**
  76      * Initialize connection to server.
  77      * @return boolean true on success
  78      */
  79     public function start($master)
  80     {
  81         parent::start($master);
  82         $this->switchSite();
  83
  84         require_once "lib/jabber.php";
  85
  86         # Low priority; we don't want to receive messages
  87
  88         common_log(LOG_INFO, "INITIALIZE");
  89         $this->conn = jabber_connect($this->resource());
  90
  91         if (empty($this->conn)) {
  92             common_log(LOG_ERR, "Couldn't connect to server.");
  93             return false;
  94         }
  95
  96         $this->conn->addEventHandler('message', 'forward_message', $this);
  97         $this->conn->addEventHandler('reconnect', 'handle_reconnect', $this);
  98         $this->conn->setReconnectTimeout(600);
  99         jabber_send_presence("Send me a message to post a notice", 'available', null, 'available', -1);
 100
 101         return !is_null($this->conn);
 102     }
 103
 104     /**
 105      * Message pump is triggered on socket input, so we only need an idle()
 106      * call often enough to trigger our outgoing pings.
 107      */
 108     function timeout()
 109     {
 110         return self::PING_INTERVAL;
 111     }
 112
 113     /**
 114      * Lists the XMPP connection socket to allow i/o master to wake
 115      * when input comes in here as well as from the queue source.
 116      *
 117      * @return array of resources
 118      */
 119     public function getSockets()
 120     {
 121         return array($this->conn->getSocket());
 122     }
 123
 124     /**
 125      * Process XMPP events that have come in over the wire.
 126      * Side effects: may switch site configuration
 127      * @fixme may kill process on XMPP error
 128      * @param resource $socket
 129      */
 130     public function handleInput($socket)
 131     {
 132         $this->switchSite();
 133
 134         # Process the queue for as long as needed
 135         try {
 136             if ($this->conn) {
 137                 assert($socket === $this->conn->getSocket());
 138
 139                 common_log(LOG_DEBUG, "Servicing the XMPP queue.");
 140                 $this->stats('xmpp_process');
 141                 $this->conn->processTime(0);
 142             }
 143         } catch (XMPPHP_Exception $e) {
 144             common_log(LOG_ERR, "Got an XMPPHP_Exception: " . $e->getMessage());
 145             die($e->getMessage());
 146         }
 147     }
 148
 149     /**
 150      * Idle processing for io manager's execution loop.
 151      * Send keepalive pings to server.
 152      *
 153      * Side effect: kills process on exception from XMPP library.
 154      *
 155      * @fixme non-dying error handling
 156      */
 157     public function idle($timeout=0)
 158     {
 159         if ($this->conn) {
 160             $now = time();
 161             if (empty($this->lastping) || $now - $this->lastping > self::PING_INTERVAL) {
 162                 $this->switchSite();
 163                 try {
 164                     $this->sendPing();
 165                     $this->lastping = $now;
 166                 } catch (XMPPHP_Exception $e) {
 167                     common_log(LOG_ERR, "Got an XMPPHP_Exception: " . $e->getMessage());
 168                     die($e->getMessage());
 169                 }
 170             }
 171         }
 172     }
 173
 174     /**
 175      * Send a keepalive ping to the XMPP server.
 176      */
 177     protected function sendPing()
 178     {
 179         $jid = jabber_daemon_address().'/'.$this->resource();
 180         $server = common_config('xmpp', 'server');
 181
 182         if (!isset($this->pingid)) {
 183             $this->pingid = 0;
 184         } else {
 185             $this->pingid++;
 186         }
 187
 188         common_log(LOG_DEBUG, "Sending ping #{$this->pingid}");
 189
 190         $this->conn->send("<iq from='{$jid}' to='{$server}' id='ping_{$this->pingid}' type='get'><ping xmlns='urn:xmpp:ping'/></iq>");
 191     }
 192
 193     /**
 194      * Callback for Jabber reconnect event
 195      * @param $pl
 196      */
 197     function handle_reconnect(&$pl)
 198     {
 199         common_log(LOG_NOTICE, 'XMPP reconnected');
 200
 201         $this->conn->processUntil('session_start');
 202         $this->conn->presence(null, 'available', null, 'available', -1);
 203     }
 204
 205     /**
 206      * Callback for Jabber message event.
 207      *
 208      * This connection handles output; if we get a message straight to us,
 209      * forward it on to our XmppDaemon listener for processing.
 210      *
 211      * @param $pl
 212      */
 213     function forward_message(&$pl)
 214     {
 215         if ($pl['type'] != 'chat') {
 216             common_log(LOG_DEBUG, 'Ignoring message of type ' . $pl['type'] . ' from ' . $pl['from']);
 217             return;
 218         }
 219         $listener = $this->listener();
 220         if (strtolower($listener) == strtolower($pl['from'])) {
 221             common_log(LOG_WARNING, 'Ignoring loop message.');
 222             return;
 223         }
 224         common_log(LOG_INFO, 'Forwarding message from ' . $pl['from'] . ' to ' . $listener);
 225         $this->conn->message($this->listener(), $pl['body'], 'chat', null, $this->ofrom($pl['from']));
 226     }
 227
 228     /**
 229      * Build an <addresses> block with an ofrom entry for forwarded messages
 230      *
 231      * @param string $from Jabber ID of original sender
 232      * @return string XML fragment
 233      */
 234     protected function ofrom($from)
 235     {
 236         $address = "<addresses xmlns='http://jabber.org/protocol/address'>\n";
 237         $address .= "<address type='ofrom' jid='$from' />\n";
 238         $address .= "</addresses>\n";
 239         return $address;
 240     }
 241
 242     /**
 243      * Build the complete JID of the XmppDaemon process which
 244      * handles primary XMPP input for this site.
 245      *
 246      * @return string Jabber ID
 247      */
 248     protected function listener()
 249     {
 250         if (common_config('xmpp', 'listener')) {
 251             return common_config('xmpp', 'listener');
 252         } else {
 253             return jabber_daemon_address() . '/' . common_config('xmpp','resource') . 'daemon';
 254         }
 255     }
 256
 257     protected function resource()
 258     {
 259         return 'queue' . posix_getpid(); // @fixme PIDs won't be host-unique
 260     }
 261
 262     /**
 263      * Make sure we're on the right site configuration
 264      */
 265     protected function switchSite()
 266     {
 267         if ($this->site != common_config('site', 'server')) {
 268             common_log(LOG_DEBUG, __METHOD__ . ": switching to site $this->site");
 269             $this->stats('switch');
 270             StatusNet::init($this->site);
 271         }
 272     }
 273 }