send rabbit message
@@ -8,6 +8,7 @@ import logging
import time
import humanize
+import networkml
import pika
@@ -19,6 +20,25 @@ class NetworkML():
self.main(raw_args=raw_args)
+ @staticmethod
+ def connect_rabbit(host='messenger', port=5672, queue='task_queue'):
+ params = pika.ConnectionParameters(host=host, port=port)
+ connection = pika.BlockingConnection(params)
+ channel = connection.channel()
+ channel.queue_declare(queue=queue, durable=True)
+ return channel
+
+
+ @staticmethod
+ def send_rabbit_msg(msg, channel, exchange='', routing_key='task_queue'):
+ channel.basic_publish(exchange=exchange,
+ routing_key=routing_key,
+ body=json.dumps(msg),
+ properties=pika.BasicProperties(delivery_mode=2))
+ print(" [X] %s UTC %r %r" % (str(datetime.datetime.utcnow()),
+ str(msg['id']), str(msg['file_path'])))
+
+
@staticmethod
def parse_args(raw_args=None):
parser = argparse.ArgumentParser()
@@ -72,6 +92,15 @@ class NetworkML():
if self.final_stage == 'algorithm' and self.operation == 'predict':
self.logger.info(f'Prediction: {result}')
+ uid = os.getenv('id', 'None')
+ file_path = os.getenv('file_path', 'None')
+ try:
+ channel = NetworkML.connect_rabbit()
+ body = {'id': uid, 'type': 'metadata', 'file_path': file_path, 'data': result, 'results': {'tool': 'networkml', 'version': version}}
+ NetworkML.send_rabbit_msg(body, channel)
+ body = {'id': uid, 'type': 'metadata', 'file_path': file_path, 'data': '', 'results': {'tool': 'networkml', 'version': networkml.__version__}}
+ except Exception as e: # pragma: no cover
+ self.logger.error(f'Failed to send Rabbit message because: {e}')
def main(self, raw_args=None):
