1 files changed, 69 insertions, 11 deletions

diff --git a/lib/util.php b/lib/util.php
index b17a44bd8..78b5a36f8 100644
--- a/lib/util.php
+++ b/lib/util.php
@@ -874,18 +874,76 @@ function common_broadcast_notice($notice, $remote=false)
 
 function common_enqueue_notice($notice)
 {
-    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;
+    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;
+    }
+    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_log(LOG_DEBUG, 'complete queueing notice ID = ' . $notice->id . ' for ' . $transport);
     }
     return $result;
 }