X-Git-Url: https://git.mxchange.org/?a=blobdiff_plain;f=xmppdaemon.php;h=9aaa953fd0c3016e027c579b889bf164e3942247;hb=38b215805e17e3df93c4b93c323ed7e72b927820;hp=b42e33566a1fb285aa96e81b222810df88847934;hpb=8b89e1cfb074d9255ab3eb66aa16ee0a01a0db7a;p=quix0rs-gnu-social.git diff --git a/xmppdaemon.php b/xmppdaemon.php old mode 100644 new mode 100755 index b42e33566a..9aaa953fd0 --- a/xmppdaemon.php +++ b/xmppdaemon.php @@ -1,3 +1,4 @@ +#!/usr/bin/env php $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->server, $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 handle_session($pl) { + # XXX what to do here? + return true; + } + + 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'])) { return; + } else if ($this->is_autoreply($pl['body'])) { + $this->log(LOG_INFO, 'Ignoring auto reply from ' . $from); + return; } else { $this->add_notice($user, $pl); } } + function is_autoreply($txt) { + if (preg_match('/[\[\(]?[Aa]uto-?[Rr]eply[\]\)]/', $txt)) { + return true; + } else { + return false; + } + } + + 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; @@ -119,6 +197,7 @@ class XMPPDaemon { $notice = new Notice(); $notice->profile_id = $user->id; $notice->content = trim(substr($pl['body'], 0, 140)); + $notice->rendered = common_render_content($notice->content, $notice); $notice->created = DB_DataObject_Cast::dateTime(); $notice->query('BEGIN'); $id = $notice->insert(); @@ -141,34 +220,206 @@ class XMPPDaemon { ': ' . $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 handle_session(&$pl) { - $this->conn->presence($status="Send me a message to post a notice"); + 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 log($level, $msg) { - common_log($level, 'XMPPDaemon('.$this->resource.'): '.$msg); + 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 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 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(); } + ?>