public queue handler and xmpp confirm handler service their message queues

darcs-hash:20080830172217-84dde-b5501719421b84ea1ee8346cf8a5ad170fd1db88.gz

diff --git a/scripts/publicqueuehandler.php b/scripts/publicqueuehandler.php
index 081ce1e38b624c09b88aac8c39b5e7b0e84b8df2..eb18d2c347b754ec4a957b1b6bc0b06dd727bf3f 100755 (executable)
--- a/scripts/publicqueuehandler.php
+++ b/scripts/publicqueuehandler.php
@@ -42,6 +42,7 @@ class PublicQueueHandler extends QueueHandler {
        function start() {
                # Low priority; we don't want to receive messages
                $this->conn = jabber_connect($this->_id, NULL, -1);
+               $this->conn->addEventHandler('message', 'forward_message', $this);
                return !is_null($this->conn);
        }
 
@@ -49,7 +50,45 @@ class PublicQueueHandler extends QueueHandler {
                return jabber_public_notice($notice);
        }
        
-       function finish() {
+       function idle() {
+           $this->log(LOG_DEBUG, 'Checking the incoming message queue.');
+               # Process the queue for a second
+               if ($this->conn->readyToProcess()) {
+                       $this->log(LOG_DEBUG, 'Something in the incoming message queue; processing it.');
+                       $this->conn->processTime(1);
+                       $this->log(LOG_DEBUG, 'Done processing incoming message queue.');
+               } else {
+                       $this->log(LOG_DEBUG, 'Nothing in the incoming message queue; skipping it.');
+               }
+       }
+
+       function forward_message(&$pl) {
+               if ($pl['type'] != 'chat') {
+                   $this->log(LOG_DEBUG, 'Ignoring message of type ' . $pl['type'] . ' from ' . $pl['from']);
+                       return;
+               }
+               $listener = $this->listener();
+               if (strtolower($listener) == strtolower($pl['from'])) {
+                       $this->log(LOG_WARNING, 'Ignoring loop message.');
+                       return;
+               }
+               $this->log(LOG_INFO, 'Forwarding message from ' . $pl['from'] . ' to ' . $listener);
+               $this->conn->message($this->listener(), $pl['body'], 'chat', NULL, $this->ofrom($pl['from']));
+       }
+
+       function ofrom($from) {
+               $address = "<addresses xmlns='http://jabber.org/protocol/address'>\n";
+               $address .= "<address type='ofrom' jid='$from' />\n";
+               $address .= "</addresses>\n";
+               return $address;
+       }
+
+       function listener() {
+               if (common_config('xmpp', 'listener')) {
+                       return common_config('xmpp', 'listener');
+               } else {
+                       return jabber_daemon_address() . '/' . common_config('xmpp','resource') . '-listener';
+               }
        }
 }
 

diff --git a/scripts/xmppconfirmhandler.php b/scripts/xmppconfirmhandler.php
index b6c1ee7a420b91697d62c0b0ce90c5a8406db784..7a369a33f25c8060f3aef1805e7ac3852bf40f61 100755 (executable)
--- a/scripts/xmppconfirmhandler.php
+++ b/scripts/xmppconfirmhandler.php
@@ -48,6 +48,7 @@ class XmppConfirmHandler {
        function start() {
                # Low priority; we don't want to receive messages
                $this->conn = jabber_connect($this->_id, NULL, -1);
+               $this->conn->addEventHandler('message', 'forward_message', $this);
                return !is_null($this->conn);
        }
        
@@ -81,8 +82,15 @@ class XmppConfirmHandler {
                                                continue;
                                        }
                                }
+                               $this->idle();
                        } else {
 #                              $this->clear_old_confirm_claims();
+                               $start = microtime();
+                               $this->idle();
+                               $used = microtime() - $start;
+                               if ($used < 10000000) {
+                                       usleep(10000000 - $used);
+                               }
                                sleep(10);
                        }
                } while (true);
@@ -124,6 +132,47 @@ class XmppConfirmHandler {
        function log($level, $msg) {
                common_log($level, 'XmppConfirmHandler ('. $this->_id .'): '.$msg);
        }
+       
+       function idle() {
+           $this->log(LOG_DEBUG, 'Checking the incoming message queue.');
+               # Process the queue for a second
+               if ($this->conn->readyToProcess()) {
+                       $this->log(LOG_DEBUG, 'Something in the incoming message queue; processing it.');
+                       $this->conn->processTime(1);
+                       $this->log(LOG_DEBUG, 'Done processing incoming message queue.');
+               } else {
+                       $this->log(LOG_DEBUG, 'Nothing in the incoming message queue; skipping it.');
+               }
+       }
+
+       function forward_message(&$pl) {
+               if ($pl['type'] != 'chat') {
+                   $this->log(LOG_DEBUG, 'Ignoring message of type ' . $pl['type'] . ' from ' . $pl['from']);
+                       return;
+               }
+               $listener = $this->listener();
+               if (strtolower($listener) == strtolower($pl['from'])) {
+                       $this->log(LOG_WARNING, 'Ignoring loop message.');
+                       return;
+               }
+               $this->log(LOG_INFO, 'Forwarding message from ' . $pl['from'] . ' to ' . $listener);
+               $this->conn->message($this->listener(), $pl['body'], 'chat', NULL, $this->ofrom($pl['from']));
+       }
+
+       function ofrom($from) {
+               $address = "<addresses xmlns='http://jabber.org/protocol/address'>\n";
+               $address .= "<address type='ofrom' jid='$from' />\n";
+               $address .= "</addresses>\n";
+               return $address;
+       }
+
+       function listener() {
+               if (common_config('xmpp', 'listener')) {
+                       return common_config('xmpp', 'listener');
+               } else {
+                       return jabber_daemon_address() . '/' . common_config('xmpp','resource') . '-listener';
+               }
+       }
 }
 
 mb_internal_encoding('UTF-8');