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 INSTALLDIR . "/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         if ($this->conn) {
 122             return array($this->conn->getSocket());
 123         } else {
 124             return array();
 125         }
 126     }
 127
 128     /**
 129      * Process XMPP events that have come in over the wire.
 130      * Side effects: may switch site configuration
 131      * @fixme may kill process on XMPP error
 132      * @param resource $socket
 133      */
 134     public function handleInput($socket)
 135     {
 136         $this->switchSite();
 137
 138         # Process the queue for as long as needed
 139         try {
 140             if ($this->conn) {
 141                 assert($socket === $this->conn->getSocket());
 142
 143                 common_log(LOG_DEBUG, "Servicing the XMPP queue.");
 144                 $this->stats('xmpp_process');
 145                 $this->conn->processTime(0);
 146             }
 147         } catch (XMPPHP_Exception $e) {
 148             common_log(LOG_ERR, "Got an XMPPHP_Exception: " . $e->getMessage());
 149             die($e->getMessage());
 150         }
 151     }
 152
 153     /**
 154      * Idle processing for io manager's execution loop.
 155      * Send keepalive pings to server.
 156      *
 157      * Side effect: kills process on exception from XMPP library.
 158      *
 159      * @fixme non-dying error handling
 160      */
 161     public function idle($timeout=0)
 162     {
 163         if ($this->conn) {
 164             $now = time();
 165             if (empty($this->lastping) || $now - $this->lastping > self::PING_INTERVAL) {
 166                 $this->switchSite();
 167                 try {
 168                     $this->sendPing();
 169                     $this->lastping = $now;
 170                 } catch (XMPPHP_Exception $e) {
 171                     common_log(LOG_ERR, "Got an XMPPHP_Exception: " . $e->getMessage());
 172                     die($e->getMessage());
 173                 }
 174             }
 175         }
 176     }
 177
 178     /**
 179      * Send a keepalive ping to the XMPP server.
 180      */
 181     protected function sendPing()
 182     {
 183         $jid = jabber_daemon_address().'/'.$this->resource();
 184         $server = common_config('xmpp', 'server');
 185
 186         if (!isset($this->pingid)) {
 187             $this->pingid = 0;
 188         } else {
 189             $this->pingid++;
 190         }
 191
 192         common_log(LOG_DEBUG, "Sending ping #{$this->pingid}");
 193
 194         $this->conn->send("<iq from='{$jid}' to='{$server}' id='ping_{$this->pingid}' type='get'><ping xmlns='urn:xmpp:ping'/></iq>");
 195     }
 196
 197     /**
 198      * Callback for Jabber reconnect event
 199      * @param $pl
 200      */
 201     function handle_reconnect(&$pl)
 202     {
 203         common_log(LOG_NOTICE, 'XMPP reconnected');
 204
 205         $this->conn->processUntil('session_start');
 206         $this->conn->presence(null, 'available', null, 'available', -1);
 207     }
 208
 209     /**
 210      * Callback for Jabber message event.
 211      *
 212      * This connection handles output; if we get a message straight to us,
 213      * forward it on to our XmppDaemon listener for processing.
 214      *
 215      * @param $pl
 216      */
 217     function forward_message(&$pl)
 218     {
 219         if ($pl['type'] != 'chat') {
 220             common_log(LOG_DEBUG, 'Ignoring message of type ' . $pl['type'] . ' from ' . $pl['from']);
 221             return;
 222         }
 223         $listener = $this->listener();
 224         if (strtolower($listener) == strtolower($pl['from'])) {
 225             common_log(LOG_WARNING, 'Ignoring loop message.');
 226             return;
 227         }
 228         common_log(LOG_INFO, 'Forwarding message from ' . $pl['from'] . ' to ' . $listener);
 229         $this->conn->message($this->listener(), $pl['body'], 'chat', null, $this->ofrom($pl['from']));
 230     }
 231
 232     /**
 233      * Build an <addresses> block with an ofrom entry for forwarded messages
 234      *
 235      * @param string $from Jabber ID of original sender
 236      * @return string XML fragment
 237      */
 238     protected function ofrom($from)
 239     {
 240         $address = "<addresses xmlns='http://jabber.org/protocol/address'>\n";
 241         $address .= "<address type='ofrom' jid='$from' />\n";
 242         $address .= "</addresses>\n";
 243         return $address;
 244     }
 245
 246     /**
 247      * Build the complete JID of the XmppDaemon process which
 248      * handles primary XMPP input for this site.
 249      *
 250      * @return string Jabber ID
 251      */
 252     protected function listener()
 253     {
 254         if (common_config('xmpp', 'listener')) {
 255             return common_config('xmpp', 'listener');
 256         } else {
 257             return jabber_daemon_address() . '/' . common_config('xmpp','resource') . 'daemon';
 258         }
 259     }
 260
 261     protected function resource()
 262     {
 263         return 'queue' . posix_getpid(); // @fixme PIDs won't be host-unique
 264     }
 265
 266     /**
 267      * Make sure we're on the right site configuration
 268      */
 269     protected function switchSite()
 270     {
 271         if ($this->site != common_config('site', 'server')) {
 272             common_log(LOG_DEBUG, __METHOD__ . ": switching to site $this->site");
 273             $this->stats('switch');
 274             StatusNet::init($this->site);
 275         }
 276     }
 277 }