diff options
-rw-r--r-- | lib/common.php | 8 | ||||
-rw-r--r-- | lib/queuehandler.php | 15 | ||||
-rw-r--r-- | lib/util.php | 146 | ||||
-rw-r--r-- | lib/xmppqueuehandler.php | 9 | ||||
-rwxr-xr-x | scripts/getvaliddaemons.php | 3 | ||||
-rwxr-xr-x | scripts/jabberqueuehandler.php | 20 | ||||
-rwxr-xr-x | scripts/publicqueuehandler.php | 20 | ||||
-rwxr-xr-x | scripts/startdaemons.sh | 14 | ||||
-rwxr-xr-x | scripts/xmppconfirmhandler.php | 20 | ||||
-rwxr-xr-x | scripts/xmppdaemon.php | 22 |
10 files changed, 147 insertions, 130 deletions
diff --git a/lib/common.php b/lib/common.php index 8eb464d7d..bb1a4255d 100644 --- a/lib/common.php +++ b/lib/common.php @@ -125,7 +125,13 @@ $config = array('appname' => 'laconica', # for syslog 'priority' => 'debug'), # XXX: currently ignored 'queue' => - array('enabled' => false), + array('enabled' => false, + 'subsystem' => 'db', # default to database, or 'stomp' + 'stomp_server' => null, + 'queue_basename' => 'laconica', + 'stomp_username' => null, + 'stomp_password' => null, + ), 'license' => array('url' => 'http://creativecommons.org/licenses/by/3.0/', 'title' => 'Creative Commons Attribution 3.0', diff --git a/lib/queuehandler.php b/lib/queuehandler.php index d5e0150d9..ae403c65e 100644 --- a/lib/queuehandler.php +++ b/lib/queuehandler.php @@ -112,12 +112,21 @@ class QueueHandler extends Daemon } function stomp_dispatch() { - require("Stomp.php"); - $con = new Stomp(common_config('queue','stomp_server')); - if (!$con->connect()) { + + // use an external message queue system via STOMP + require_once("Stomp.php"); + + $server = common_config('queue','stomp_server'); + $username = common_config('queue', 'stomp_username'); + $password = common_config('queue', 'stomp_password'); + + $con = new Stomp($server); + + if (!$con->connect($username, $password)) { $this->log(LOG_ERR, 'Failed to connect to queue server'); return false; } + $queue_basename = common_config('queue','queue_basename'); // subscribe to the relevant queue (format: basename-transport) $con->subscribe('/queue/'.$queue_basename.'-'.$this->transport()); diff --git a/lib/util.php b/lib/util.php index 1af462516..30b767c3e 100644 --- a/lib/util.php +++ b/lib/util.php @@ -826,89 +826,91 @@ function common_broadcast_notice($notice, $remote=false) function common_enqueue_notice($notice) { + $transports = array('omb', 'sms', 'public', 'twitter', 'facebook', 'ping'); + + if (common_config('xmpp', 'enabled')) + { + $transports[] = 'jabber'; + } + if (common_config('queue','subsystem') == 'stomp') { - // use an external message queue system via STOMP - require_once("Stomp.php"); - $con = new Stomp(common_config('queue','stomp_server')); - if (!$con->connect()) { - common_log(LOG_ERR, 'Failed to connect to queue server'); - return false; - } - $queue_basename = common_config('queue','queue_basename'); - foreach (array('jabber', 'omb', 'sms', 'public', 'twitter', 'facebook', 'ping') as $transport) { - if (!$con->send( - '/queue/'.$queue_basename.'-'.$transport, // QUEUE - $notice->id, // BODY of the message - array ( // HEADERS of the msg - 'created' => $notice->created - ))) { - common_log(LOG_ERR, 'Error sending to '.$transport.' queue'); - return false; - } - common_log(LOG_DEBUG, 'complete remote queueing notice ID = ' . $notice->id . ' for ' . $transport); - } - - //send tags as headers, so they can be used as JMS selectors - common_log(LOG_DEBUG, 'searching for tags ' . $notice->id); - $tags = array(); - $tag = new Notice_tag(); - $tag->notice_id = $notice->id; - if ($tag->find()) { - while ($tag->fetch()) { - common_log(LOG_DEBUG, 'tag found = ' . $tag->tag); - array_push($tags,$tag->tag); - } - } - $tag->free(); - - $con->send('/topic/laconica.'.$notice->profile_id, - $notice->content, - array( - 'profile_id' => $notice->profile_id, - 'created' => $notice->created, - 'tags' => implode($tags,' - ') - ) - ); - common_log(LOG_DEBUG, 'sent to personal topic ' . $notice->id); - $con->send('/topic/laconica.allusers', - $notice->content, - array( - 'profile_id' => $notice->profile_id, - 'created' => $notice->created, - 'tags' => implode($tags,' - ') - ) - ); - common_log(LOG_DEBUG, 'sent to catch-all topic ' . $notice->id); - $result = true; + common_enqueue_notice_stomp($notice, $transports); } else { - // in any other case, 'internal' - foreach (array('jabber', 'omb', 'sms', 'public', 'twitter', 'facebook', 'ping') as $transport) { - $qi = new Queue_item(); - $qi->notice_id = $notice->id; - $qi->transport = $transport; - $qi->created = $notice->created; - $result = $qi->insert(); - if (!$result) { - $last_error = &PEAR::getStaticProperty('DB_DataObject','lastError'); - common_log(LOG_ERR, 'DB error inserting queue item: ' . $last_error->message); - return false; - } - common_log(LOG_DEBUG, 'complete queueing notice ID = ' . $notice->id . ' for ' . $transport); - } + common_enqueue_notice_db($notice, $transports); } return $result; } -function common_post_inbox_transports() +function common_enqueue_notice_stomp($notice, $transports) { - $transports = array('omb', 'sms'); + // use an external message queue system via STOMP + require_once("Stomp.php"); + + $server = common_config('queue','stomp_server'); + $username = common_config('queue', 'stomp_username'); + $password = common_config('queue', 'stomp_password'); - if (common_config('xmpp', 'enabled')) { - $transports = array_merge($transports, array('jabber', 'public')); + $con = new Stomp($server); + + if (!$con->connect($username, $password)) { + common_log(LOG_ERR, 'Failed to connect to queue server'); + return false; } - return $transports; + $queue_basename = common_config('queue','queue_basename'); + + foreach ($transports as $transport) { + $result = $con->send('/queue/'.$queue_basename.'-'.$transport, // QUEUE + $notice->id, // BODY of the message + array ('created' => $notice->created)); + if (!$result) { + common_log(LOG_ERR, 'Error sending to '.$transport.' queue'); + return false; + } + common_log(LOG_DEBUG, 'complete remote queueing notice ID = ' . $notice->id . ' for ' . $transport); + } + + //send tags as headers, so they can be used as JMS selectors + common_log(LOG_DEBUG, 'searching for tags ' . $notice->id); + $tags = array(); + $tag = new Notice_tag(); + $tag->notice_id = $notice->id; + if ($tag->find()) { + while ($tag->fetch()) { + common_log(LOG_DEBUG, 'tag found = ' . $tag->tag); + array_push($tags,$tag->tag); + } + } + $tag->free(); + + $con->send('/topic/laconica.'.$notice->profile_id, + $notice->content, + array( + 'profile_id' => $notice->profile_id, + 'created' => $notice->created, + 'tags' => implode($tags,' - ') + ) + ); + common_log(LOG_DEBUG, 'sent to personal topic ' . $notice->id); + $con->send('/topic/laconica.allusers', + $notice->content, + array( + 'profile_id' => $notice->profile_id, + 'created' => $notice->created, + 'tags' => implode($tags,' - ') + ) + ); + common_log(LOG_DEBUG, 'sent to catch-all topic ' . $notice->id); + $result = true; +} + +function common_enqueue_notice_db($notice, $transports) +{ + // in any other case, 'internal' + foreach ($transports as $transport) { + common_enqueue_notice_transport($notice, $transport); + } } function common_enqueue_notice_transport($notice, $transport) diff --git a/lib/xmppqueuehandler.php b/lib/xmppqueuehandler.php index a078cd9f7..986e09c25 100644 --- a/lib/xmppqueuehandler.php +++ b/lib/xmppqueuehandler.php @@ -22,7 +22,7 @@ if (!defined('LACONICA')) { exit(1); } require_once(INSTALLDIR.'/lib/queuehandler.php'); /** - * Common superclass for all XMPP-using queue handlers. They all need to + * Common superclass for all XMPP-using queue handlers. They all need to * service their message queues on idle, and forward any incoming messages * to the XMPP listener connection. So, we abstract out common code to a * superclass. @@ -30,12 +30,11 @@ require_once(INSTALLDIR.'/lib/queuehandler.php'); class XmppQueueHandler extends QueueHandler { - function start() { # Low priority; we don't want to receive messages $this->log(LOG_INFO, "INITIALIZE"); - $this->conn = jabber_connect($this->_id); + $this->conn = jabber_connect($this->_id.$this->transport()); if ($this->conn) { $this->conn->addEventHandler('message', 'forward_message', $this); $this->conn->addEventHandler('reconnect', 'handle_reconnect', $this); @@ -44,7 +43,7 @@ class XmppQueueHandler extends QueueHandler } return !is_null($this->conn); } - + function handle_reconnect(&$pl) { $this->conn->processUntil('session_start'); @@ -63,7 +62,7 @@ class XmppQueueHandler extends QueueHandler die($e->getMessage()); } } - + function forward_message(&$pl) { if ($pl['type'] != 'chat') { diff --git a/scripts/getvaliddaemons.php b/scripts/getvaliddaemons.php index 198ea8fb9..97c230784 100755 --- a/scripts/getvaliddaemons.php +++ b/scripts/getvaliddaemons.php @@ -38,9 +38,6 @@ if(common_config('xmpp','enabled')) { echo "xmppdaemon.php jabberqueuehandler.php publicqueuehandler.php "; echo "xmppconfirmhandler.php "; } -if(common_config('memcached','enabled')) { - echo "memcachedqueuehandler.php "; -} if(common_config('twitterbridge','enabled')) { echo "twitterstatusfetcher.php "; } diff --git a/scripts/jabberqueuehandler.php b/scripts/jabberqueuehandler.php index a44993236..5b581629d 100755 --- a/scripts/jabberqueuehandler.php +++ b/scripts/jabberqueuehandler.php @@ -20,13 +20,13 @@ define('INSTALLDIR', realpath(dirname(__FILE__) . '/..')); -$shortoptions = 'r::'; -$longoptions = array('resource::'); +$shortoptions = 'i::'; +$longoptions = array('id::'); $helptext = <<<END_OF_JABBER_HELP Daemon script for pushing new notices to Jabber users. - -r --resource Jabber Resource ID (default to config) + -i --id Identity (default none) END_OF_JABBER_HELP; @@ -63,16 +63,16 @@ if (common_config('xmpp','enabled')==false) { exit(); } -if (have_option('r')) { - $resource = get_option_value('r'); -} else if (have_option('--resource')) { - $resource = get_option_value('--resource'); +if (have_option('i')) { + $id = get_option_value('i'); +} else if (have_option('--id')) { + $id = get_option_value('--id'); } else if (count($args) > 0) { - $resource = $args[0]; + $id = $args[0]; } else { - $resource = null; + $id = null; } -$handler = new JabberQueueHandler($resource); +$handler = new JabberQueueHandler($id); $handler->runOnce(); diff --git a/scripts/publicqueuehandler.php b/scripts/publicqueuehandler.php index 58ecc1745..701d50e01 100755 --- a/scripts/publicqueuehandler.php +++ b/scripts/publicqueuehandler.php @@ -20,13 +20,13 @@ define('INSTALLDIR', realpath(dirname(__FILE__) . '/..')); -$shortoptions = 'r::'; -$longoptions = array('resource::'); +$shortoptions = 'i::'; +$longoptions = array('id::'); $helptext = <<<END_OF_PUBLIC_HELP Daemon script for pushing new notices to public XMPP subscribers. - -r --resource Jabber Resource ID + -i --id Identity (default none) END_OF_PUBLIC_HELP; @@ -61,16 +61,16 @@ if (common_config('xmpp','enabled')==false) { exit(); } -if (have_option('r')) { - $resource = get_option_value('r'); -} else if (have_option('--resource')) { - $resource = get_option_value('--resource'); +if (have_option('i')) { + $id = get_option_value('i'); +} else if (have_option('--id')) { + $id = get_option_value('--id'); } else if (count($args) > 0) { - $resource = $args[0]; + $id = $args[0]; } else { - $resource = null; + $id = null; } -$handler = new PublicQueueHandler($resource); +$handler = new PublicQueueHandler($id); $handler->runOnce(); diff --git a/scripts/startdaemons.sh b/scripts/startdaemons.sh index a44362b57..9ead20acd 100755 --- a/scripts/startdaemons.sh +++ b/scripts/startdaemons.sh @@ -20,23 +20,27 @@ # This program tries to start the daemons for Laconica. # Note that the 'maildaemon' needs to run as a mail filter. -ARGS= +ARGSG= +ARGSD= if [ $# -gt 0 ]; then - ARGS="$ARGS -s$1" + ARGSG="$ARGSG -s$1" + ID=`echo $1 | sed s/\\\\./_/g` + ARGSD="$ARGSD -s$1 -i$ID" fi if [ $# -gt 1 ]; then - ARGS="$ARGS -p$2" + ARGSD="$ARGSD -p$2" + ARGSG="$ARGSG -p$2" fi DIR=`dirname $0` -DAEMONS=`php $DIR/getvaliddaemons.php $ARGS` +DAEMONS=`php $DIR/getvaliddaemons.php $ARGSG` for f in $DAEMONS; do printf "Starting $f..."; - php $DIR/$f $ARGS + php $DIR/$f $ARGSD printf "DONE.\n" done diff --git a/scripts/xmppconfirmhandler.php b/scripts/xmppconfirmhandler.php index 883934fd6..d6821ddef 100755 --- a/scripts/xmppconfirmhandler.php +++ b/scripts/xmppconfirmhandler.php @@ -20,13 +20,13 @@ define('INSTALLDIR', realpath(dirname(__FILE__) . '/..')); -$shortoptions = 'r::'; -$longoptions = array('resource::'); +$shortoptions = 'i::'; +$longoptions = array('id::'); $helptext = <<<END_OF_JABBER_HELP Daemon script for pushing new confirmations to Jabber users. - -r --resource Jabber Resource ID (default to config) + -i --id Identity (default none) END_OF_JABBER_HELP; @@ -147,17 +147,17 @@ if (common_config('xmpp','enabled')==false) { exit(); } -if (have_option('r')) { - $resource = get_option_value('r'); -} else if (have_option('--resource')) { - $resource = get_option_value('--resource'); +if (have_option('i')) { + $id = get_option_value('i'); +} else if (have_option('--id')) { + $id = get_option_value('--id'); } else if (count($args) > 0) { - $resource = $args[0]; + $id = $args[0]; } else { - $resource = null; + $id = null; } -$handler = new XmppConfirmHandler($resource); +$handler = new XmppConfirmHandler($id); $handler->runOnce(); diff --git a/scripts/xmppdaemon.php b/scripts/xmppdaemon.php index 661631937..3eecfec29 100755 --- a/scripts/xmppdaemon.php +++ b/scripts/xmppdaemon.php @@ -20,13 +20,13 @@ define('INSTALLDIR', realpath(dirname(__FILE__) . '/..')); -$shortoptions = 'r::'; -$longoptions = array('resource::'); +$shortoptions = 'i::'; +$longoptions = array('id::'); $helptext = <<<END_OF_XMPP_HELP Daemon script for receiving new notices from Jabber users. - -r --resource Jabber Resource ID (default to config) + -i --id Identity (default none) END_OF_XMPP_HELP; @@ -52,7 +52,7 @@ class XMPPDaemon extends Daemon } if ($resource) { - $this->resource = $resource; + $this->resource = $resource . 'daemon'; } else { $this->resource = common_config('xmpp', 'resource') . 'daemon'; } @@ -323,16 +323,16 @@ if (common_config('xmpp','enabled')==false) { exit(); } -if (have_option('r')) { - $resource = get_option_value('r'); -} else if (have_option('--resource')) { - $resource = get_option_value('--resource'); +if (have_option('i')) { + $id = get_option_value('i'); +} else if (have_option('--id')) { + $id = get_option_value('--id'); } else if (count($args) > 0) { - $resource = $args[0]; + $id = $args[0]; } else { - $resource = null; + $id = null; } -$daemon = new XMPPDaemon($resource); +$daemon = new XMPPDaemon($id); $daemon->runOnce(); |