fixed thread for deployer

deployer/__main__.py

@@ -10,12 +10,17 @@ import logging
 import pika
 import yaml
 import sys
+from time import sleep
+from concurrent.futures import thread
+from threading import Thread
 
 from service import tosca, k8s_service
 from service import ansible_service
 
 logger = logging.getLogger(__name__)
 
+done = False
+
 
 # if not getattr(logger, 'handler_set', None):
 # logger.setLevel(logging.INFO)
@@ -37,7 +42,7 @@ def init_chanel(args):
     connection = pika.BlockingConnection(pika.ConnectionParameters(host=rabbitmq_host))
     channel = connection.channel()
     channel.queue_declare(queue=queue_name)
-    return channel
+    return channel, connection
 
 
 def start(this_channel):
@@ -106,6 +111,12 @@ def handle_delivery(message):
     return json.dumps(response)
 
 
+def threaded_function(args):
+    while not done:
+        connection.process_data_events()
+        sleep(5)
+
+
 if __name__ == "__main__":
     logging.basicConfig(level=logging.INFO)
     if sys.argv[1] == "test_local":
@@ -124,7 +135,16 @@ if __name__ == "__main__":
 
     else:
         logger.info("Input args: " + sys.argv[0] + ' ' + sys.argv[1] + ' ' + sys.argv[2])
-        channel = init_chanel(sys.argv)
-        global queue_name
+        global channel, queue_name, connection
+        channel, connection = init_chanel(sys.argv)
         queue_name = sys.argv[2]
+        logger.info("Awaiting RPC requests")
+        try:
+            thread = Thread(target=threaded_function, args=(1,))
+            thread.start()
             start(channel)
+        except:
+            done = True
+            e = sys.exc_info()[0]
+            logger.info("Error: " + str(e))
+            exit(-1)

planner/__main__.py

@@ -68,12 +68,13 @@ def on_request(ch, method, props, body):
     ch.basic_ack(delivery_tag=method.delivery_tag)
 
 
-def handle_delivery(message):
+def handle_delivery(message, sys=None):
     logger.info("Got: " + str(message))
     try:
         message = message.decode()
     except (UnicodeDecodeError, AttributeError):
-        pass
+        e = sys.exc_info()[0]
+        logger.info("Parsing Error: " + str(e))
     parsed_json_message = json.loads(message)
     owner = parsed_json_message['owner']
     tosca_file_name = 'tosca_template'
@@ -143,22 +144,20 @@ if __name__ == "__main__":
         test_response = {'toscaTemplate': template_dict}
         logger.info("Output message:" + json.dumps(test_response))
     else:
-        print("Input args: " + sys.argv[0] + ' ' + sys.argv[1] + ' ' + sys.argv[2])
         logger.info("Input args: " + sys.argv[0] + ' ' + sys.argv[1] + ' ' + sys.argv[2])
         global channel
         global connection
         channel, connection = init_chanel(sys.argv)
         global queue_name
         queue_name = sys.argv[2]
-        # start(channel)
-        thread = Thread(target=threaded_function, args=(1,))
-        thread.start()
+
+        # thread = Thread(target=threaded_function, args=(1,))
+        # thread.start()
 
         logger.info("Awaiting RPC requests")
         try:
-            channel.start_consuming()
-        except KeyboardInterrupt:
-            # thread.stop()
-            done = True
-            thread.join()
-            logger.info("Threads successfully closed")
+            start(channel)
+        except:
+            e = sys.exc_info()[0]
+            logger.info("Error: " + str(e))
+            exit(-1)