From 2ac684f53df3f76aacb4d1be79d7db64567f1964 Mon Sep 17 00:00:00 2001
From: Federico Marani <federico.marani@ymail.com>
Date: Fri, 20 Feb 2009 18:57:20 +0000
Subject: * implemented STOMP support for external message queueing systems *
 (ticket-724) * included PHP STOMP libraries
 (http://code.google.com/p/stompcli/)

---
 lib/util.php | 48 +++++++++++++++++++++++++++++++++++++-----------
 1 file changed, 37 insertions(+), 11 deletions(-)

(limited to 'lib/util.php')

diff --git a/lib/util.php b/lib/util.php
index c5a092f63..8ef19f579 100644
--- a/lib/util.php
+++ b/lib/util.php
@@ -1138,18 +1138,44 @@ function common_twitter_broadcast($notice, $flink)
 
 function common_enqueue_notice($notice)
 {
-    foreach (array('jabber', 'omb', 'sms', 'public') 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') 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);
+	}
+	$result = true;
+    }
+    else {
+	// in any other case, 'internal'
+    	foreach (array('jabber', 'omb', 'sms', 'public') 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;
 }
-- 
cgit v1.2.3-54-g00ecf