trackrdrd: allow configuration of multiple MQ URIs

src/activemq/amq.cpp

@@ -57,26 +57,30 @@ using namespace cms;
 using namespace decaf::lang;
 using namespace decaf::lang::exceptions;
 
-ActiveMQConnectionFactory* AMQ_Worker::factory = NULL;
-
-void
-AMQ_Worker::initConnectionFactory(const std::string& brokerURI) {
-    factory = new ActiveMQConnectionFactory(brokerURI);
-}
+std::vector<ActiveMQConnectionFactory*> AMQ_Worker::factories;
 
 void
 AMQ_Worker::shutdown() {
-    delete factory;
-    factory = NULL;
+    for (unsigned i = 0; i < factories.size(); i++)
+        delete factories[i];
+    factories.resize(0);
 }
     
-AMQ_Worker::AMQ_Worker(std::string& qName,
+AMQ_Worker::AMQ_Worker(std::string& brokerURI, std::string& qName,
     Session::AcknowledgeMode ackMode = Session::AUTO_ACKNOWLEDGE,
     int deliveryMode = DeliveryMode::NON_PERSISTENT) {
-    
-    if (factory == NULL)
-        throw cms::IllegalStateException("Connection factory not initialized");
-    
+
+    ActiveMQConnectionFactory* factory = NULL;
+    for (unsigned i = 0; i < factories.size(); i++)
+        if (factories[i]->getBrokerURI().toString().compare(brokerURI) == 0) {
+            factory = factories[i];
+            break;
+        }
+    if (factory == NULL) {
+        factory = new ActiveMQConnectionFactory(brokerURI);
+        factories.push_back(factory);
+    }
+       
     connection = factory->createConnection();
     connection->start();
     session = connection->createSession(ackMode);
@@ -100,15 +104,12 @@ AMQ_Worker::~AMQ_Worker() {
 	session = NULL;
     }
     if (connection != NULL) {
-#if 0	    
 	connection->close();
-#endif	    
 	delete connection;
 	connection = NULL;
     }
 }
 
-/* XXX: Timeout */
 void
 AMQ_Worker::send(std::string& text) {
     if (msg == NULL || producer == NULL)
@@ -126,23 +127,22 @@ AMQ_Worker::getVersion() {
 }
 
 const char *
-AMQ_GlobalInit(char *uri)
+AMQ_GlobalInit(void)
 {
-    activemq::library::ActiveMQCPP::initializeLibrary();
     try {
-        string brokerURI (uri);
-        AMQ_Worker::initConnectionFactory(brokerURI);
+        activemq::library::ActiveMQCPP::initializeLibrary();
         return NULL;
     }
     CATCHALL
 }
 
 const char *
-AMQ_WorkerInit(AMQ_Worker **worker, char *qName)
+AMQ_WorkerInit(AMQ_Worker **worker, char *uri, char *qName)
 {
     try {
+        string brokerURI (uri);
         string queueName (qName);
-        std::auto_ptr<AMQ_Worker> w (new AMQ_Worker(queueName));
+        std::auto_ptr<AMQ_Worker> w (new AMQ_Worker(brokerURI, queueName));
         *worker = w.release();
         return NULL;
     }

src/activemq/amq.h

@@ -34,6 +34,8 @@
 
 #ifdef __cplusplus
 
+#include <vector>
+
 #include <activemq/core/ActiveMQConnectionFactory.h>
 #include <cms/Connection.h>
 #include <cms/Session.h>
@@ -45,7 +47,7 @@ using namespace cms;
 
 class AMQ_Worker {
 private:
-    static ActiveMQConnectionFactory* factory;
+    static std::vector<ActiveMQConnectionFactory*> factories;
     Connection* connection;
     Session* session;
     Queue* queue;
@@ -54,11 +56,10 @@ private:
     AMQ_Worker() {};
 
 public:
-    static void initConnectionFactory(const std::string& brokerURI);
     static void shutdown();
     
-    AMQ_Worker(std::string& qName, Session::AcknowledgeMode ackMode,
-        int deliveryMode);
+    AMQ_Worker(std::string& brokerURI, std::string& qName,
+        Session::AcknowledgeMode ackMode, int deliveryMode);
     virtual ~AMQ_Worker();
     void send(std::string& text);
     std::string getVersion();
@@ -71,8 +72,8 @@ typedef struct AMQ_Worker AMQ_Worker;
 extern "C" {
 #endif
 
-    const char *AMQ_GlobalInit(char *uri);
-    const char *AMQ_WorkerInit(AMQ_Worker **worker, char *qName);
+    const char *AMQ_GlobalInit(void);
+    const char *AMQ_WorkerInit(AMQ_Worker **worker, char *uri, char *qName);
     const char *AMQ_Send(AMQ_Worker *worker, const char *data, unsigned len);
     const char *AMQ_Version(AMQ_Worker *worker, char *version);
     const char *AMQ_WorkerShutdown(AMQ_Worker **worker);

src/config.c

@@ -124,7 +124,6 @@ CONF_Add(const char *lval, const char *rval)
     confString("log.file", log_file);
     confString("varnish.bindump", varnish_bindump);
     confString("processor.log", processor_log);
-    confString("mq.uri", mq_uri);
     confString("mq.qname", mq_qname);
 
     confUnsignedMinVal("maxopen.scale", maxopen_scale, MIN_MAXOPEN_SCALE);
@@ -178,6 +177,19 @@ CONF_Add(const char *lval, const char *rval)
         return(EINVAL);
     }
 
+    if (strcmp(lval, "mq.uri") == 0) {
+        int n = config.n_mq_uris++;
+        config.mq_uri = (char **) realloc(config.mq_uri,
+            config.n_mq_uris * sizeof(char **));
+        if (config.mq_uri == NULL)
+            return(errno);
+        config.mq_uri[n] = (char *) malloc(strlen(rval) + 1);
+        if (config.mq_uri[n] == NULL)
+            return(errno);
+        strcpy(config.mq_uri[n], rval);
+        return(0);
+    }
+
     return EINVAL;
 }
 
@@ -226,7 +238,9 @@ CONF_Init(void)
     config.hash_ttl = DEF_HASH_TTL;
     config.hash_mlt = DEF_HASH_MTL;
 
-    config.mq_uri[0] = '\0';
+    config.n_mq_uris = 0;
+    config.mq_uri = (char **) malloc (sizeof(char **));
+    AN(config.mq_uri);
     config.mq_qname[0] = '\0';
     config.nworkers = 1;
     config.restarts = 1;
@@ -326,8 +340,12 @@ CONF_Dump(void)
     confdump("hash_ttl = %u", config.hash_ttl);
     confdump("hash_mlt = %u", config.hash_mlt);
 
-
-    confdump("mq.uri = %s", config.mq_uri);
+    if (config.n_mq_uris > 0)
+        for (int i = 0; i < config.n_mq_uris; i++)
+            confdump("mq.uri = %s", config.mq_uri[i]);
+    else
+        LOG_Log0(LOG_DEBUG, "config: mq.uri = ");
+    
     confdump("mq.qname = %s", config.mq_qname);
     confdump("nworkers = %u", config.nworkers);
     confdump("restarts = %u", config.restarts);

src/mq.c

@@ -35,13 +35,13 @@
 const char *
 MQ_GlobalInit(void)
 {
-    return AMQ_GlobalInit(config.mq_uri);
+    return AMQ_GlobalInit();
 }
 
 const char *
-MQ_WorkerInit(void **priv)
+MQ_WorkerInit(void **priv, char *uri)
 {
-    return AMQ_WorkerInit((AMQ_Worker **) priv, config.mq_qname);
+    return AMQ_WorkerInit((AMQ_Worker **) priv, uri, config.mq_qname);
 }
 
 const char *

src/test/test_mq.c

@@ -50,7 +50,6 @@ static char
 
     printf("... testing MQ global initialization\n");
 
-    strcpy(config.mq_uri, "tcp://localhost:61616");
     err = MQ_GlobalInit();
     sprintf(errmsg, "MQ_GlobalInit: %s", err);
     mu_assert(errmsg, err == NULL);
@@ -62,11 +61,13 @@ static const char
 *test_worker_init(void)
 {
     const char *err;
+    char uri[sizeof("tcp://localhost:61616")];
 
     printf("... test worker init (including connect to ActiveMQ)\n");
 
+    strcpy(uri, "tcp://localhost:61616");
     strcpy(config.mq_qname, "lhoste/tracking/test");
-    err = MQ_WorkerInit(&worker);
+    err = MQ_WorkerInit(&worker, uri);
     if (err != NULL && strstr(err, "Connection refused") != NULL) {
         printf("Connection refused, ActiveMQ assumed not running\n");
         exit(EXIT_SKIPPED);
@@ -118,7 +119,7 @@ static const char
 
     printf("... testing worker shutdown\n");
 
-    mu_assert("MQ_WorkerShhutdown: worker is NULL before call", worker != NULL);
+    mu_assert("MQ_WorkerShutdown: worker is NULL before call", worker != NULL);
     err = MQ_WorkerShutdown(&worker);
     sprintf(errmsg, "MQ_WorkerShutdown: %s", err);
     mu_assert(errmsg, err == NULL);

src/test/test_worker.c

@@ -63,8 +63,19 @@ static char
     config.maxopen_scale = 10;
     config.maxdone_scale = 10;
     config.nworkers = NWORKERS;
-    strcpy(config.mq_uri, "tcp://localhost:61616");
     strcpy(config.mq_qname, "lhoste/tracking/test");
+
+    config.n_mq_uris = 2;
+    config.mq_uri = (char **) malloc(2 * sizeof(char**));
+    AN(config.mq_uri);
+    config.mq_uri[0] = (char *) malloc(strlen("tcp://localhost:61616") + 1);
+    AN(config.mq_uri[0]);
+    strcpy(config.mq_uri[0], "tcp://localhost:61616");
+    config.mq_uri[1] = (char *)
+        malloc(strlen("tcp://localhost:61616?connection.sendTimeout=1000") + 1);
+    AN(config.mq_uri[1]);
+    strcpy(config.mq_uri[1],
+        "tcp://localhost:61616?connection.sendTimeout=1000");
     
     error = MQ_GlobalInit();
     sprintf(errmsg, "MQ_GlobalInit failed: %s", error);

src/trackrdrd.h

@@ -151,7 +151,7 @@ int		spmcq_datawaiter;
 
 /* mq.c */
 const char *MQ_GlobalInit(void);
-const char *MQ_WorkerInit(void **priv);
+const char *MQ_WorkerInit(void **priv, char *uri);
 const char *MQ_Send(void *priv, const char *data, unsigned len);
 const char *MQ_Version(void *priv, char *version);
 const char *MQ_WorkerShutdown(void **priv);
@@ -305,7 +305,8 @@ struct config {
     unsigned	hash_mlt;
 #define DEF_HASH_MTL 5
 
-    char	mq_uri[BUFSIZ];
+    unsigned	n_mq_uris;
+    char	**mq_uri;
     char	mq_qname[BUFSIZ];
     unsigned	nworkers;
     unsigned	restarts;

src/worker.c

@@ -137,14 +137,15 @@ static void
     void *amq_worker;
     dataentry *entry;
     const char *err;
-    char version[VERSION_LEN];
+    char version[VERSION_LEN], *uri;
 
     LOG_Log(LOG_INFO, "Worker %d: starting", wrk->id);
     CHECK_OBJ_NOTNULL(wrk, WORKER_DATA_MAGIC);
     wrk->state = WRK_INITIALIZING;
     wrk->tid = pthread_self();
+    uri = config.mq_uri[wrk->id % config.n_mq_uris];
 
-    err = MQ_WorkerInit(&amq_worker);
+    err = MQ_WorkerInit(&amq_worker, uri);
     if (err != NULL) {
         LOG_Log(LOG_ALERT, "Worker %d: Cannot initialize queue connection: %s",
             wrk->id, err);
@@ -167,7 +168,7 @@ static void
     running++;
     AZ(pthread_mutex_unlock(&running_lock));
 
-    LOG_Log(LOG_INFO, "Worker %d: running (%s)", wrk->id, version);
+    LOG_Log(LOG_INFO, "Worker %d: running (%s, uri=%s)", wrk->id, version, uri);
     
     while (run) {
 	entry = (dataentry *) SPMCQ_Deq();