fix heartbeat

Author: robkooper

CHANGELOG.md

@@ -6,23 +6,33 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 ## [Unreleased]
 
+### Changed
+- Heartbeat is now every five minutes, was every 5 seconds
+- the python3 images were actually python2 images
+
 ## 2.2.0 - 2019-04-03
 
 ### Fixed
 - Code is now compatible with both Python 2 and Python 3.
 - RabbitMQ queue name can be different from extractor name.
 - Updated dependencies.
-- A race condition existed where an ACK could be lost, resulting in an extractor not processing more messages [CATSPYC-1](https://opensource.ncsa.illinois.edu/jira/browse/CATSPYC-1)
-- Error decoding messages where filename contains non-ascii characters[CATSPYC-18](https://opensource.ncsa.illinois.edu/jira/browse/CATSPYC-18)
+- A race condition existed where an ACK could be lost, resulting in an extractor not processing more messages
+  [CATSPYC-1](https://opensource.ncsa.illinois.edu/jira/browse/CATSPYC-1)
+- Error decoding messages where filename contains non-ascii characters
+  [CATSPYC-18](https://opensource.ncsa.illinois.edu/jira/browse/CATSPYC-18)
 
 ### Changed
-- files.upload_preview and collections.upload_preview does not require section_id for metadata [CATS-935](https://opensource.ncsa.illinois.edu/jira/browse/CATS-935)
-- Extractors will not register by default to clowder [CATSPYC-1](https://opensource.ncsa.illinois.edu/jira/browse/CATSPYC-1)
+- files.upload_preview and collections.upload_preview does not require section_id for metadata
+  [CATS-935](https://opensource.ncsa.illinois.edu/jira/browse/CATS-935)
+- Extractors will not register by default to clowder
+  [CATSPYC-1](https://opensource.ncsa.illinois.edu/jira/browse/CATSPYC-1)
 
 ### Added
 - Simple R extractor. Allows wrapping R code as a clowder extractor, see sample-extractors/wordcount-simple-r-extractor for an example.
-- Email notification when extraction is done [CATSPYC-17](https://opensource.ncsa.illinois.edu/jira/browse/CATSPYC-17)
-- Docker compose file for starting up the Clowder stack [BD-2226](https://opensource.ncsa.illinois.edu/jira/browse/BD-2226)
+- Email notification when extraction is done
+  [CATSPYC-17](https://opensource.ncsa.illinois.edu/jira/browse/CATSPYC-17)
+- Docker compose file for starting up the Clowder stack
+  [BD-2226](https://opensource.ncsa.illinois.edu/jira/browse/BD-2226)
 - PyClowder will now send heartbeats on extractors exchange, processes can listen for broadcast to get notified when new extractors come online.
 - Monitor application to leverage new heartbeat send out by extractors
 - Extractors now send version and name to clowder as part of the agent information.

pyclowder/connectors.py

@@ -621,7 +621,8 @@ class RabbitMQConnector(Connector):
     # pylint: disable=too-many-arguments
     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):
+                 check_message=None, process_message=None, ssl_verify=True, mounted_paths=None,
+                 heartbeat=5*60):
         super(RabbitMQConnector, self).__init__(extractor_name, extractor_info, check_message, process_message,
                                                 ssl_verify, mounted_paths)
         self.rabbitmq_uri = rabbitmq_uri
@@ -636,6 +637,7 @@ def __init__(self, extractor_name, extractor_info,
         self.consumer_tag = None
         self.worker = None
         self.announcer = None
+        self.heartbeat = 5*60
 
     def connect(self):
         """connect to rabbitmq using URL parameters"""
@@ -678,7 +680,7 @@ def connect(self):
                                     routing_key="extractors." + self.extractor_name)
 
         # start the extractor announcer
-        self.announcer = RabbitMQBroadcast(self.rabbitmq_uri, self.extractor_info, self.rabbitmq_queue, 5)
+        self.announcer = RabbitMQBroadcast(self.rabbitmq_uri, self.extractor_info, self.rabbitmq_queue, self.heartbeat)
         self.announcer.start_thread()
 
     def listen(self):
@@ -808,8 +810,8 @@ def send_heartbeat(self):
         }
         while self.thread:
             try:
-                time.sleep(self.heartbeat)
                 self.channel.basic_publish(exchange='extractors', routing_key='', body=json.dumps(message))
+                time.sleep(self.heartbeat)
             except SystemExit:
                 raise
             except KeyboardInterrupt: