trackrdrd: shutdown & restart AMQ workers & connections more robustly

trackrdrd/src/activemq/amq.cpp

@@ -71,15 +71,16 @@ AMQ_Worker::AMQ_Worker(Connection* cn, std::string& qName,
 }
 
 AMQ_Worker::~AMQ_Worker() {
-    if (producer != NULL) {
-	delete producer;
-	producer = NULL;
-    }
     if (queue != NULL) {
 	delete queue;
 	queue = NULL;
     }
+    if (producer != NULL) {
+	delete producer;
+	producer = NULL;
+    }
     if (session != NULL) {
+        session->stop();
 	delete session;
 	session = NULL;
     }

trackrdrd/src/activemq/amq_connection.cpp

@@ -30,6 +30,7 @@
  */
 
 #include "amq_connection.h"
+#include <decaf/lang/exceptions/NullPointerException.h>
 
 #define CATCHALL                \
 catch (CMSException& cex) {     \
@@ -49,14 +50,27 @@ using namespace std;
 using namespace activemq::core;
 using namespace cms;
 using namespace decaf::lang;
+using namespace decaf::lang::exceptions;
 
 ActiveMQConnectionFactory* AMQ_Connection::factory = NULL;
 
 AMQ_Connection::AMQ_Connection(std::string& brokerURI) {
 
+    if (brokerURI.length() == 0)
+        throw IllegalArgumentException(__FILE__, __LINE__,
+                                       "Broker URI is empty");
+    
     factory = new ActiveMQConnectionFactory(brokerURI);
+    if (factory == NULL)
+        throw NullPointerException(__FILE__, __LINE__,
+                                   "Factory created for %s is NULL",
+                                   brokerURI.c_str());
        
     connection = factory->createConnection();
+    if (connection == NULL)
+        throw NullPointerException(__FILE__, __LINE__,
+                                   "Connection created for %s is NULL",
+                                   brokerURI.c_str());
     connection->start();
 }
 
@@ -67,6 +81,8 @@ AMQ_Connection::getConnection() {
 
 AMQ_Connection::~AMQ_Connection() {
     if (connection != NULL) {
+        connection->stop();
+        connection->close();
 	delete connection;
 	connection = NULL;
     }