Merge branch '0.8.x' of git@gitorious.org:laconica/dev into 0.8.x

4 files changed, 97 insertions, 81 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') {