add extractor_key param

Author: max-zilla

pyclowder/connectors.py

@@ -66,7 +66,7 @@ class Connector(object):
     registered_clowder = list()
 
     def __init__(self, extractor_name, extractor_info, check_message=None, process_message=None, ssl_verify=True,
-                 mounted_paths=None, clowder_url=None, max_retry=10):
+                 mounted_paths=None, clowder_url=None, max_retry=10, extractor_key=None):
         self.extractor_name = extractor_name
         self.extractor_info = extractor_info
         self.check_message = check_message
@@ -77,6 +77,7 @@ def __init__(self, extractor_name, extractor_info, check_message=None, process_m
         else:
             self.mounted_paths = mounted_paths
         self.clowder_url = clowder_url
+        self.extractor_key = extractor_key
         self.max_retry = max_retry
 
         filename = 'notifications.json'
@@ -391,10 +392,16 @@ def _process_message(self, body):
             return
 
         # register extractor
-        url = "%sapi/extractors" % source_host
-        if url not in Connector.registered_clowder:
-            Connector.registered_clowder.append(url)
-            self.register_extractor("%s?key=%s" % (url, secret_key))
+        if self.extractor_key is None:
+            url = "%sapi/extractors" % source_host
+            if url not in Connector.registered_clowder:
+                Connector.registered_clowder.append(url)
+                self.register_extractor("%s?key=%s" % (url, secret_key))
+        else:
+            url = "%sapi/extractors/private/%s" % (source_host, self.extractor_key)
+            if url not in Connector.registered_clowder:
+                Connector.registered_clowder.append(url)
+                self.register_extractor("%s?key=%s" % (url, secret_key))
 
         # tell everybody we are starting to process the file
         self.status_update(pyclowder.utils.StatusMessage.start, resource, "Started processing.")
@@ -630,7 +637,7 @@ class RabbitMQConnector(Connector):
     def __init__(self, extractor_name, extractor_info,
                  rabbitmq_uri, rabbitmq_exchange=None, rabbitmq_key=None, rabbitmq_queue=None,
                  check_message=None, process_message=None, ssl_verify=True, mounted_paths=None,
-                 heartbeat=5*60, clowder_url=None, max_retry=10):
+                 heartbeat=5*60, clowder_url=None, max_retry=10, extractor_key=None):
         super(RabbitMQConnector, self).__init__(extractor_name, extractor_info, check_message, process_message,
                                                 ssl_verify, mounted_paths, clowder_url, max_retry)
         self.rabbitmq_uri = rabbitmq_uri
@@ -640,6 +647,9 @@ def __init__(self, extractor_name, extractor_info,
             self.rabbitmq_queue = extractor_info['name']
         else:
             self.rabbitmq_queue = rabbitmq_queue
+        self.extractor_key = extractor_key
+        if extractor_key is not None:
+            self.rabbitmq_queue += extractor_key
         self.channel = None
         self.connection = None
         self.consumer_tag = None

pyclowder/extractors.py

@@ -66,6 +66,7 @@ def __init__(self):
         rabbitmq_exchange = os.getenv('RABBITMQ_EXCHANGE', "")
         clowder_url = os.getenv("CLOWDER_URL", "")
         registration_endpoints = os.getenv('REGISTRATION_ENDPOINTS', "")
+        extractor_key = os.getenv("EXTRACTOR_KEY", "")
         logging_config = os.getenv("LOGGING")
         mounted_paths = os.getenv("MOUNTED_PATHS", "{}")
         input_file_path = os.getenv("INPUT_FILE_PATH")
@@ -90,6 +91,9 @@ def __init__(self):
         self.parser.add_argument('--register', '-r', nargs='?', dest="registration_endpoints",
                                  default=registration_endpoints,
                                  help='Clowder registration URL (default=%s)' % registration_endpoints)
+        self.parser.add_argument('--key', '-k', dest="extractor_key",
+                                 default=extractor_key,
+                                 help='Unique key to use for extractor queue ID (sets extractor to private)')
         self.parser.add_argument('--rabbitmqURI', nargs='?', dest='rabbitmq_uri', default=rabbitmq_uri,
                                  help='rabbitMQ URI (default=%s)' % rabbitmq_uri.replace("%", "%%"))
         self.parser.add_argument('--rabbitmqQUEUE', nargs='?', dest='rabbitmq_queuename',
@@ -155,6 +159,7 @@ def start(self):
                                 else:
                                     rabbitmq_key.append("*.%s.%s" % (key, mt.replace("/", ".")))
 
+                logger.info('Creating connector with key '+self.args.extractor_key)
                 connector = RabbitMQConnector(self.args.rabbitmq_queuename,
                                               self.extractor_info,
                                               check_message=self.check_message,
@@ -165,8 +170,10 @@ def start(self):
                                               rabbitmq_queue=self.args.rabbitmq_queuename,
                                               mounted_paths=json.loads(self.args.mounted_paths),
                                               clowder_url=self.args.clowder_url,
-                                              max_retry=self.args.max_retry)
+                                              max_retry=self.args.max_retry,
+                                              extractor_key=self.args.extractor_key)
                 connector.connect()
+                logger.info("new version check OK")
                 connector.register_extractor(self.args.registration_endpoints)
                 threading.Thread(target=connector.listen, name="RabbitMQConnector").start()