make sure to listen to right queuename

this will not overwrite the extractor_info['name'] which was done in the past.
bump version number

Author: robkooper

CHANGELOG.md

@@ -4,10 +4,17 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](http://keepachangelog.com/) 
 and this project adheres to [Semantic Versioning](http://semver.org/).
 
+## 2.1.1 - 2018-07-12
+
+### Fixed
+- RABBITMQ_QUEUE variable/flag was ignored when set and would connect
+  to default queue.
+
 ## 2.1.0 - 2018-07-06
 
 ### Added
-- [Simple extractor](https://opensource.ncsa.illinois.edu/confluence/display/CATS/Simple+Extractor+wrapper+for+basic+functions), now can create a extractor from a single function.
+- [Simple extractor](https://opensource.ncsa.illinois.edu/confluence/display/CATS/Simple+Extractor+wrapper+for+basic+functions),
+  now can create a extractor from a single function.
 
 ### Fixed
 - Acks were not always send due to racing condition

docs/source/conf.py

@@ -59,7 +59,7 @@
 # The short X.Y version.
 version = u'2.1'
 # The full version, including alpha/beta/rc tags.
-release = u'2.1.0'
+release = u'2.1.1'
 
 # The language for content autogenerated by Sphinx. Refer to documentation
 # for a list of supported languages.

pyclowder/connectors.py

@@ -58,7 +58,9 @@ class Connector(object):
 
     registered_clowder = list()
 
-    def __init__(self, extractor_info, check_message=None, process_message=None, ssl_verify=True, mounted_paths=None):
+    def __init__(self, extractor_name, extractor_info, check_message=None, process_message=None, ssl_verify=True,
+                 mounted_paths=None):
+        self.extractor_name = extractor_name
         self.extractor_info = extractor_info
         self.check_message = check_message
         self.process_message = process_message
@@ -135,13 +137,14 @@ def _build_resource(self, body, host, secret_key):
             resource_type = "file"
         elif message_type.find("metadata.added") > -1:
             resource_type = "metadata"
-        elif message_type == "extractors."+self.extractor_info['name']:
+        elif message_type == "extractors." + self.extractor_name \
+                or message_type == "extractors." + self.extractor_info['name']:
             # This was a manually submitted extraction
             if datasetid == fileid:
                 resource_type = "dataset"
             else:
                 resource_type = "file"
-        elif message_type.endswith(self.extractor_info['name']):
+        elif message_type.endswith(self.extractor_info['name']) or message_type.endswith(self.extractor_name):
             # This was migrated from another queue (e.g. error queue) so use extractor default
             for key, value in self.extractor_info['process'].iteritems():
                 if key == "dataset":
@@ -551,9 +554,10 @@ class RabbitMQConnector(Connector):
     """
 
     # pylint: disable=too-many-arguments
-    def __init__(self, extractor_info, rabbitmq_uri, rabbitmq_exchange=None, rabbitmq_key=None,
+    def __init__(self, extractor_name, extractor_info, rabbitmq_uri, rabbitmq_exchange=None, rabbitmq_key=None,
                  check_message=None, process_message=None, ssl_verify=True, mounted_paths=None):
-        Connector.__init__(self, extractor_info, check_message, process_message, ssl_verify, mounted_paths)
+        super(RabbitMQConnector, self).__init__(extractor_name, extractor_info, check_message, process_message,
+                                                ssl_verify, mounted_paths)
         self.rabbitmq_uri = rabbitmq_uri
         self.rabbitmq_exchange = rabbitmq_exchange
         self.rabbitmq_key = rabbitmq_key
@@ -576,8 +580,8 @@ def connect(self):
         self.channel.basic_qos(prefetch_count=1)
 
         # declare the queue in case it does not exist
-        self.channel.queue_declare(queue=self.extractor_info['name'], durable=True)
-        self.channel.queue_declare(queue='error.'+self.extractor_info['name'], durable=True)
+        self.channel.queue_declare(queue=self.extractor_name, durable=True)
+        self.channel.queue_declare(queue='error.'+self.extractor_name, durable=True)
 
         # register with an exchange
         if self.rabbitmq_exchange:
@@ -588,18 +592,18 @@ def connect(self):
             # connect queue and exchange
             if self.rabbitmq_key:
                 if isinstance(self.rabbitmq_key, str):
-                    self.channel.queue_bind(queue=self.extractor_info['name'],
+                    self.channel.queue_bind(queue=self.extractor_name,
                                             exchange=self.rabbitmq_exchange,
                                             routing_key=self.rabbitmq_key)
                 else:
                     for key in self.rabbitmq_key:
-                        self.channel.queue_bind(queue=self.extractor_info['name'],
+                        self.channel.queue_bind(queue=self.extractor_name,
                                                 exchange=self.rabbitmq_exchange,
                                                 routing_key=key)
 
-            self.channel.queue_bind(queue=self.extractor_info['name'],
+            self.channel.queue_bind(queue=self.extractor_name,
                                     exchange=self.rabbitmq_exchange,
-                                    routing_key="extractors." + self.extractor_info['name'])
+                                    routing_key="extractors." + self.extractor_name)
 
     def listen(self):
         """Listen for messages coming from RabbitMQ"""
@@ -609,8 +613,7 @@ def listen(self):
             self.connect()
 
         # create listener
-        self.consumer_tag = self.channel.basic_consume(self.on_message, queue=self.extractor_info['name'],
-                                                       no_ack=False)
+        self.consumer_tag = self.channel.basic_consume(self.on_message, queue=self.extractor_name, no_ack=False)
 
         # start listening
         logging.getLogger(__name__).info("Starting to listen for messages.")
@@ -665,8 +668,8 @@ def on_message(self, channel, method, header, body):
         if 'routing_key' not in json_body and method.routing_key:
             json_body['routing_key'] = method.routing_key
 
-        self.worker = RabbitMQHandler(self.extractor_info, self.check_message, self.process_message,
-                                      self.ssl_verify, self.mounted_paths, method, header, body)
+        self.worker = RabbitMQHandler(self.extractor_name, self.extractor_info, self.check_message,
+                                      self.process_message, self.ssl_verify, self.mounted_paths, method, header, body)
         self.worker.start_thread(json_body)
 
 
@@ -677,9 +680,10 @@ class RabbitMQHandler(Connector):
     a queue of messages that the super- loop can access and send later.
     """
 
-    def __init__(self, extractor_info, check_message=None, process_message=None, ssl_verify=True,
+    def __init__(self, extractor_name, extractor_info, check_message=None, process_message=None, ssl_verify=True,
                  mounted_paths=None, method=None, header=None, body=None):
-        Connector.__init__(self, extractor_info, check_message, process_message, ssl_verify, mounted_paths)
+        super(RabbitMQHandler, self).__init__(extractor_name, extractor_info, check_message, process_message,
+                                              ssl_verify, mounted_paths)
         self.method = method
         self.header = header
         self.body = body
@@ -728,7 +732,7 @@ def process_messages(self, channel):
             elif msg["type"] == 'error':
                 properties = pika.BasicProperties(delivery_mode=2, reply_to=self.header.reply_to)
                 channel.basic_publish(exchange='',
-                                      routing_key='error.' + self.extractor_info['name'],
+                                      routing_key='error.' + self.extractor_name,
                                       properties=properties,
                                       body=self.body)
                 channel.basic_ack(self.method.delivery_tag)
@@ -737,7 +741,7 @@ def process_messages(self, channel):
 
             elif msg["type"] == 'resubmit':
                 retry_count = msg['retry_count']
-                queue = self.extractor_info['name']
+                queue = self.extractor_name
                 properties = pika.BasicProperties(delivery_mode=2, reply_to=self.header.reply_to)
                 jbody = json.loads(self.body)
                 jbody['retry_count'] = retry_count
@@ -790,9 +794,10 @@ class HPCConnector(Connector):
     """Takes pickle files and processes them."""
 
     # pylint: disable=too-many-arguments
-    def __init__(self, extractor_info, picklefile,
+    def __init__(self, extractor_name, extractor_info, picklefile,
                  check_message=None, process_message=None, ssl_verify=True, mounted_paths=None):
-        Connector.__init__(self, extractor_info, check_message, process_message, ssl_verify, mounted_paths)
+        super(HPCConnector, self).__init__(extractor_name, extractor_info, check_message, process_message,
+                                           ssl_verify, mounted_paths)
         self.picklefile = picklefile
         self.logfile = None
 
@@ -847,8 +852,8 @@ class LocalConnector(Connector):
 
     """
 
-    def __init__(self, extractor_info, input_file_path, process_message=None, output_file_path=None):
-        super(LocalConnector, self).__init__(extractor_info, process_message=process_message)
+    def __init__(self, extractor_name, extractor_info, input_file_path, process_message=None, output_file_path=None):
+        super(LocalConnector, self).__init__(extractor_name, extractor_info, process_message=process_message)
         self.input_file_path = input_file_path
         self.output_file_path = output_file_path
         self.completed_processing = False

pyclowder/extractors.py

@@ -149,7 +149,8 @@ def start(self):
                                     else:
                                         rabbitmq_key.append("*.%s.%s" % (key, mt.replace("/", ".")))
 
-                    rconn = RabbitMQConnector(self.extractor_info,
+                    rconn = RabbitMQConnector(self.args.rabbitmq_queuename,
+                                              self.extractor_info,
                                               check_message=self.check_message,
                                               process_message=self.process_message,
                                               rabbitmq_uri=self.args.rabbitmq_uri,
@@ -164,7 +165,8 @@ def start(self):
                 if 'hpc_picklefile' not in self.args:
                     logger.error("Missing hpc_picklefile for HPCExtractor")
                 else:
-                    hconn = HPCConnector(self.extractor_info,
+                    hconn = HPCConnector(self.extractor_info['name'],
+                                         self.extractor_info,
                                          check_message=self.check_message,
                                          process_message=self.process_message,
                                          picklefile=self.args.hpc_picklefile,
@@ -180,7 +182,9 @@ def start(self):
                 elif not os.path.isfile(self.args.input_file_path):
                     logger.error("Local input file is not a regular file. Please check the path.")
                 else:
-                    local_connector = LocalConnector(self.extractor_info, self.args.input_file_path,
+                    local_connector = LocalConnector(self.extractor_info['name'],
+                                                     self.extractor_info,
+                                                     self.args.input_file_path,
                                                      process_message=self.process_message,
                                                      output_file_path=self.args.output_file_path)
                     connectors.append(local_connector)

release.sh

@@ -20,7 +20,7 @@ $(dirname $0)/docker.sh
 
 # check branch and set version
 if [ "${BRANCH}" = "master" ]; then
-    VERSION=${VERSION:-"2.1.0 2.1 2 latest"}
+    VERSION=${VERSION:-"2.1.1 2.1 2 latest"}
 elif [ "${BRANCH}" = "develop" ]; then
     VERSION="develop"
 else

setup.py

@@ -9,7 +9,7 @@ def description():
 
 
 setup(name='pyclowder',
-      version='2.1.0',
+      version='2.1.1',
       packages=find_packages(),
       description='Python SDK for the Clowder Data Management System',
       long_description=description(),