X-Git-Url: https://git.mxchange.org/?a=blobdiff_plain;f=xmppdaemon.php;h=6509ed1faf4d51f57510baf853aed472d3f6d469;hb=4b0abe0ce7fd8c9e2cc8ddf52f76cd4c93a5f279;hp=f1a817e3f108bf3fc2af854c99c517a62bc94e46;hpb=cf96f138f9de1e45a1e1a5527a77ffcae0761b86;p=quix0rs-gnu-social.git diff --git a/xmppdaemon.php b/xmppdaemon.php old mode 100644 new mode 100755 index f1a817e3f1..6509ed1faf --- a/xmppdaemon.php +++ b/xmppdaemon.php @@ -1,3 +1,4 @@ +#!/usr/bin/env php . */ +# Abort if called from a web server +if (isset($_SERVER) && array_key_exists('REQUEST_METHOD', $_SERVER)) { + print "This script must be run from the command line\n"; + exit(); +} + define('INSTALLDIR', dirname(__FILE__)); define('LACONICA', true); require_once(INSTALLDIR . '/lib/common.php'); -require_once('xmpp.php'); +require_once(INSTALLDIR . '/lib/jabber.php'); + +# This is kind of clunky; we create a class to call the global functions +# in jabber.php, which create a new XMPP class. A more elegant (?) solution +# might be to use make this a subclass of XMPP. class XMPPDaemon { - - function XMPPDaemon() { - foreach (array('server', 'port', 'user', 'password', 'resource') as $attr) { + + function XMPPDaemon($resource=NULL) { + static $attrs = array('server', 'port', 'user', 'password', 'host'); + + foreach ($attrs as $attr) + { $this->$attr = common_config('xmpp', $attr); } + + if ($resource) { + $this->resource = $resource; + } else { + $this->resource = common_config('xmpp', 'resource') . 'daemon'; + } + + $this->log(LOG_INFO, "{$this->user}@{$this->server}/{$this->resource}"); } function connect() { - $this->conn = new XMPP($this->host, $this->port, $this->user, - $this->password, $this->resource); + + $connect_to = ($this->host) ? $this->host : $this->server; + + $this->log(LOG_INFO, "Connecting to $connect_to on port $this->port"); + + $this->conn = jabber_connect($this->resource); + if (!$this->conn) { return false; } - $this->conn->connect(); - return !$this->conn->disconnected; + + return !$this->conn->isDisconnected(); } - + function handle() { - while(!$this->conn->disconnected) { - $payloads = $this->conn->processUntil(array('message', 'presence', - 'end_stream', 'session_start')); - foreach($payloads as $event) { - $pl = $event[1]; - switch($event[0]) { - case 'message': - $this->handle_message($pl); - break; - case 'presence': - $this->handle_presence($pl); - break; - case 'session_start': - $this->handle_session($pl); - break; + + static $parts = array('message', 'presence', + 'end_stream', 'session_start'); + + while(!$this->conn->isDisconnected()) { + + $payloads = $this->conn->processUntil($parts, 10); + + if ($payloads) { + foreach($payloads as $event) { + $pl = $event[1]; + switch($event[0]) { + case 'message': + $this->handle_message($pl); + break; + case 'presence': + $this->handle_presence($pl); + break; + case 'session_start': + $this->handle_session($pl); + break; + } } } + + $this->broadcast_queue(); + $this->confirmation_queue(); + } + } + + function get_user($from) { + $user = User::staticGet('jabber', jabber_normalize_jid($from)); + return $user; + } + + function get_confirmation($from) { + $confirm = new Confirm_address(); + $confirm->address = $from; + $confirm->address_type = 'jabber'; + if ($confirm->find(TRUE)) { + return $confirm; + } else { + return NULL; } } function handle_message(&$pl) { - $user = User::staticGet('jabber', $pl['from']); + if ($pl['type'] != 'chat') { + return; + } + if (strlen($pl['body']) == 0) { + return; + } + + $from = jabber_normalize_jid($pl['from']); + $user = $this->get_user($from); + if (!$user) { - $this->log(LOG_WARNING, 'Message from unknown user ' . $pl['from']); + $this->from_site($from, 'Unknown user; go to ' . + common_local_url('imsettings') . + ' to add your address to your account'); + $this->log(LOG_WARNING, 'Message from unknown user ' . $from); return; } if ($this->handle_command($user, $pl['body'])) { @@ -75,14 +140,21 @@ class XMPPDaemon { } } + function from_site($address, $msg) { + $text = '['.common_config('site', 'name') . '] ' . $msg; + jabber_send_message($address, $text); + } + function handle_command($user, $body) { # XXX: localise switch(trim($body)) { case 'on': $this->set_notify($user, true); + $this->from_site($user->jabber, 'notifications on'); return true; case 'off': $this->set_notify($user, false); + $this->from_site($user->jabber, 'notifications off'); return true; default: return false; @@ -95,16 +167,16 @@ class XMPPDaemon { $result = $user->update($orig); if (!$id) { $last_error = &PEAR::getStaticProperty('DB_DataObject','lastError'); - $this->log(LOG_ERROR, + $this->log(LOG_ERROR, 'Could not set notify flag to ' . $notify . - ' for user ' . common_log_objstring($user) . + ' for user ' . common_log_objstring($user) . ': ' . $last_error->message); } else { $this->log(LOG_INFO, 'User ' . $user->nickname . ' set notify flag to ' . $notify); } } - + function add_notice(&$user, &$pl) { $notice = new Notice(); $notice->profile_id = $user->id; @@ -114,9 +186,9 @@ class XMPPDaemon { $id = $notice->insert(); if (!$id) { $last_error = &PEAR::getStaticProperty('DB_DataObject','lastError'); - $this->log(LOG_ERROR, - 'Could not insert ' . common_log_objstring($notice) . - ' for user ' . common_log_objstring($user) . + $this->log(LOG_ERROR, + 'Could not insert ' . common_log_objstring($notice) . + ' for user ' . common_log_objstring($user) . ': ' . $last_error->message); return; } @@ -125,40 +197,212 @@ class XMPPDaemon { $result = $notice->update($orig); if (!$result) { $last_error = &PEAR::getStaticProperty('DB_DataObject','lastError'); - $this->log(LOG_ERROR, - 'Could not add URI to ' . common_log_objstring($notice) . - ' for user ' . common_log_objstring($user) . + $this->log(LOG_ERROR, + 'Could not add URI to ' . common_log_objstring($notice) . + ' for user ' . common_log_objstring($user) . ': ' . $last_error->message); return; } - common_broadcast_notice($notice); + $notice->query('COMMIT'); + common_save_replies($notice); + common_real_broadcast($notice); $this->log(LOG_INFO, 'Added notice ' . $notice->id . ' from user ' . $user->nickname); } - + function handle_presence(&$pl) { - $user = User::staticGet('jabber', $pl['from']); - if (!$user) { - $this->log(LOG_WARNING, 'Message from unknown user ' . $pl['from']); - return; + $from = jabber_normalize_jid($pl['from']); + switch ($pl['type']) { + case 'subscribe': + # We let anyone subscribe + $this->subscribed($from); + $this->log(LOG_INFO, + 'Accepted subscription from ' . $from); + break; + case 'subscribed': + case 'unsubscribed': + case 'unsubscribe': + $this->log(LOG_INFO, + 'Ignoring "' . $pl['type'] . '" from ' . $from); + break; + default: + if (!$pl['type']) { + $user = User::staticGet('jabber', $from); + if (!$user) { + $this->log(LOG_WARNING, 'Message from unknown user ' . $from); + return; + } + if ($user->updatefrompresence) { + $this->log(LOG_INFO, 'Updating ' . $user->nickname . + ' status from presence.'); + $this->add_notice($user, $pl); + } + } + break; } - if ($user->updatefrompresence) { - $this->add_notice($user, $pl); + } + + function log($level, $msg) { + common_log($level, 'XMPPDaemon('.$this->resource.'): '.$msg); + } + + function subscribed($to) { + jabber_special_presence('subscribed', $to); + } + + function set_status($status) { + $this->log(LOG_INFO, 'Setting status to "' . $status . '"'); + jabber_send_presence($status); + } + + function top_queue_item() { + + $qi = new Queue_item(); + $qi->orderBy('created'); + $qi->whereAdd('claimed is NULL'); + + $qi->limit(1); + + $cnt = $qi->find(TRUE); + + if ($cnt) { + # XXX: potential race condition + # can we force it to only update if claimed is still NULL + # (or old)? + $this->log(LOG_INFO, 'claiming queue item = ' . $qi->notice_id); + $orig = clone($qi); + $qi->claimed = DB_DataObject_Cast::dateTime(); + $result = $qi->update($orig); + if ($result) { + $this->log(LOG_INFO, 'claim succeeded.'); + return $qi; + } else { + $this->log(LOG_INFO, 'claim failed.'); + } } + $qi = NULL; + return NULL; + } + + function broadcast_queue() { + $this->clear_old_claims(); + $this->log(LOG_INFO, 'checking for queued notices'); + do { + $qi = $this->top_queue_item(); + if ($qi) { + $this->log(LOG_INFO, 'Got item enqueued '.common_exact_date($qi->created)); + $notice = Notice::staticGet($qi->notice_id); + if ($notice) { + $this->log(LOG_INFO, 'broadcasting notice ID = ' . $notice->id); + # XXX: what to do if broadcast fails? + $result = common_real_broadcast($notice, $this->is_remote($notice)); + if (!$result) { + $this->log(LOG_WARNING, 'Failed broadcast for notice ID = ' . $notice->id); + $orig = $qi; + $qi->claimed = NULL; + $qi->update($orig); + $this->log(LOG_WARNING, 'Abandoned claim for notice ID = ' . $notice->id); + continue; + } + $this->log(LOG_INFO, 'finished broadcasting notice ID = ' . $notice->id); + $notice = NULL; + } else { + $this->log(LOG_WARNING, 'queue item for notice that does not exist'); + } + $qi->delete(); + } + } while ($qi); + } + + function clear_old_claims() { + $qi = new Queue_item(); + $qi->claimed = NULL; + $qi->whereAdd('now() - claimed > '.CLAIM_TIMEOUT); + $qi->update(DB_DATAOBJECT_WHEREADD_ONLY); + } + + function is_remote($notice) { + $user = User::staticGet($notice->profile_id); + return !$user; + } + + function confirmation_queue() { + # $this->clear_old_confirm_claims(); + $this->log(LOG_INFO, 'checking for queued confirmations'); + do { + $confirm = $this->next_confirm(); + if ($confirm) { + $this->log(LOG_INFO, 'Sending confirmation for ' . $confirm->address); + $user = User::staticGet($confirm->user_id); + if (!$user) { + $this->log(LOG_WARNING, 'Confirmation for unknown user ' . $confirm->user_id); + continue; + } + $success = jabber_confirm_address($confirm->code, + $user->nickname, + $confirm->address); + if (!$success) { + $this->log(LOG_ERROR, 'Confirmation failed for ' . $confirm->address); + # Just let the claim age out; hopefully things work then + continue; + } else { + $this->log(LOG_INFO, 'Confirmation sent for ' . $confirm->address); + # Mark confirmation sent + $original = clone($confirm); + $confirm->sent = $confirm->claimed; + $result = $confirm->update($original); + if (!$result) { + $this->log(LOG_ERROR, 'Cannot mark sent for ' . $confirm->address); + # Just let the claim age out; hopefully things work then + continue; + } + } + } + } while ($confirm); } - function handle_session(&$pl) { - $conn->presence($status="Send me a message to post a notice"); + function next_confirm() { + $confirm = new Confirm_address(); + $confirm->whereAdd('claimed IS NULL'); + $confirm->whereAdd('sent IS NULL'); + # XXX: eventually we could do other confirmations in the queue, too + $confirm->address_type = 'jabber'; + $confirm->orderBy('modified DESC'); + $confirm->limit(1); + if ($confirm->find(TRUE)) { + $this->log(LOG_INFO, 'Claiming confirmation for ' . $confirm->address); + # working around some weird DB_DataObject behaviour + $confirm->whereAdd(''); # clears where stuff + $original = clone($confirm); + $confirm->claimed = DB_DataObject_Cast::dateTime(); + $result = $confirm->update($original); + if ($result) { + $this->log(LOG_INFO, 'Succeeded in claim! '. $result); + return $confirm; + } else { + $this->log(LOG_INFO, 'Failed in claim!'); + return false; + } + } + return NULL; } - function log($level, $msg) { - common_log($level, 'XMPPDaemon('.$this->resource.'): '.$msg); + function clear_old_confirm_claims() { + $confirm = new Confirm(); + $confirm->claimed = NULL; + $confirm->whereAdd('now() - claimed > '.CLAIM_TIMEOUT); + $confirm->update(DB_DATAOBJECT_WHEREADD_ONLY); } + } -$daemon = new XMPPDaemon(); +$resource = ($argc > 1) ? $argv[1] : NULL; + +$daemon = new XMPPDaemon($resource); if ($daemon->connect()) { + $daemon->set_status("Send me a message to post a notice"); $daemon->handle(); } + ?>