OSM MON Installation and Packaging Updates

[osm/MON.git] / core / message_bus / producer.py

diff --git a/core/message_bus/producer.py b/core/message_bus/producer.py
index 07742e2..f5abce2 100644 (file)
--- a/core/message_bus/producer.py
+++ b/core/message_bus/producer.py
@@ -29,7 +29,7 @@ __author__ = "Prithiv Mohan"
 __date__   = "06/Sep/2017"
 
 
-from kafka import KafkaProducer
+from kafka import KafkaProducer as kaf
 from kafka.errors import KafkaError
 import logging
 import json
@@ -38,52 +38,53 @@ import os
 from os import listdir
 from jsmin import jsmin
 
+json_path = os.path.join(os.pardir+"/models/")
+
+
 class KafkaProducer(object):
 
     def __init__(self, topic):
 
         self._topic= topic
 
-    if "ZOOKEEPER_URI" in os.environ:
-        broker = os.getenv("ZOOKEEPER_URI")
-    else:
-        broker = "localhost:2181"
+        if "BROKER_URI" in os.environ:
+            broker = os.getenv("BROKER_URI")
+        else:
+            broker = "localhost:9092"
 
         '''
-        If the zookeeper broker URI is not set in the env, by default,
+        If the broker URI is not set in the env, by default,
         localhost container is taken as the host because an instance of
         is already running.
         '''
 
-        producer = KafkaProducer(key_serializer=str.encode,
+        self.producer = kaf(key_serializer=str.encode,
                    value_serializer=lambda v: json.dumps(v).encode('ascii'),
                    bootstrap_servers=broker, api_version=(0,10))
 
 
-    def publish(self, key, message, topic=None):
+    def publish(self, key, value, topic=None):
         try:
-            future = producer.send('key', 'payload',group_id='osm_mon')
-            producer.flush()
+            future = self.producer.send(key, value, topic)
+            self.producer.flush()
         except Exception:
-            log.exception("Error publishing to {} topic." .format(topic))
+            logging.exception("Error publishing to {} topic." .format(topic))
             raise
         try:
             record_metadata = future.get(timeout=10)
-            self._log.debug("TOPIC:", record_metadata.topic)
-            self._log.debug("PARTITION:", record_metadata.partition)
-            self._log.debug("OFFSET:", record_metadata.offset)
+            logging.debug("TOPIC:", record_metadata.topic)
+            logging.debug("PARTITION:", record_metadata.partition)
+            logging.debug("OFFSET:", record_metadata.offset)
         except KafkaError:
             pass
 
-    json_path = os.path.join(os.pardir+"/models/")
-
     def create_alarm_request(self, key, message, topic):
 
        #External to MON
 
         payload_create_alarm = jsmin(open(os.path.join(json_path,
                                          'create_alarm.json')).read())
-        publish(key,
+        self.publish(key,
                 value=json.dumps(payload_create_alarm),
                 topic='alarm_request')
 
@@ -94,8 +95,8 @@ class KafkaProducer(object):
         payload_create_alarm_resp = jsmin(open(os.path.join(json_path,
                                          'create_alarm_resp.json')).read())
 
-        publish(key,
-                value = json.dumps(payload_create_alarm_resp),
+        self.publish(key,
+                value = message,
                 topic = 'alarm_response')
 
     def acknowledge_alarm(self, key, message, topic):
@@ -105,7 +106,7 @@ class KafkaProducer(object):
         payload_acknowledge_alarm = jsmin(open(os.path.join(json_path,
                                          'acknowledge_alarm.json')).read())
 
-        publish(key,
+        self.publish(key,
                 value = json.dumps(payload_acknowledge_alarm),
                 topic = 'alarm_request')
 
@@ -116,7 +117,7 @@ class KafkaProducer(object):
         payload_alarm_list_req = jsmin(open(os.path.join(json_path,
                                       'list_alarm_req.json')).read())
 
-        publish(key,
+        self.publish(key,
                 value=json.dumps(payload_alarm_list_req),
                 topic='alarm_request')
 
@@ -125,8 +126,8 @@ class KafkaProducer(object):
         payload_notify_alarm = jsmin(open(os.path.join(json_path,
                                           'notify_alarm.json')).read())
 
-        publish(key,
-                value=json.dumps(payload_notify_alarm),
+        self.publish(key,
+                value=message,
                 topic='alarm_response')
 
     def list_alarm_response(self, key, message, topic):
@@ -134,8 +135,8 @@ class KafkaProducer(object):
         payload_list_alarm_resp = jsmin(open(os.path.join(json_path,
                                              'list_alarm_resp.json')).read())
 
-        publish(key,
-                value=json.dumps(payload_list_alarm_resp),
+        self.publish(key,
+                value=message,
                 topic='alarm_response')
 
 
@@ -146,7 +147,7 @@ class KafkaProducer(object):
         payload_update_alarm_req = jsmin(open(os.path.join(json_path,
                                         'update_alarm_req.json')).read())
 
-        publish(key,
+        self.publish(key,
                 value=json.dumps(payload_update_alarm_req),
                 topic='alarm_request')
 
@@ -158,8 +159,8 @@ class KafkaProducer(object):
         payload_update_alarm_resp = jsmin(open(os.path.join(json_path,
                                         'update_alarm_resp.json')).read())
 
-        publish(key,
-                value=json.dumps(payload_update_alarm_resp),
+        self.publish(key,
+                value=message,
                 topic='alarm_response')
 
 
@@ -170,7 +171,7 @@ class KafkaProducer(object):
         payload_delete_alarm_req = jsmin(open(os.path.join(json_path,
                                         'delete_alarm_req.json')).read())
 
-        publish(key,
+        self.publish(key,
                 value=json.dumps(payload_delete_alarm_req),
                 topic='alarm_request')
 
@@ -181,8 +182,8 @@ class KafkaProducer(object):
         payload_delete_alarm_resp = jsmin(open(os.path.join(json_path,
                                                'delete_alarm_resp.json')).read())
 
-        publish(key,
-                value=json.dumps(payload_delete_alarm_resp),
+        self.publish(key,
+                value=message,
                 topic='alarm_response')
 
 
@@ -194,7 +195,7 @@ class KafkaProducer(object):
         payload_create_metrics_req = jsmin(open(os.path.join(json_path,
                                                 'create_metric_req.json')).read())
 
-        publish(key,
+        self.publish(key,
                 value=json.dumps(payload_create_metrics_req),
                 topic='metric_request')
 
@@ -206,8 +207,8 @@ class KafkaProducer(object):
         payload_create_metrics_resp = jsmin(open(os.path.join(json_path,
                                                  'create_metric_resp.json')).read())
 
-        publish(key,
-                value=json.dumps(payload_create_metrics_resp),
+        self.publish(key,
+                value=message,
                 topic='metric_response')
 
 
@@ -218,7 +219,7 @@ class KafkaProducer(object):
         payload_read_metric_data_request = jsmin(open(os.path.join(json_path,
                                                       'read_metric_data_req.json')).read())
 
-        publish(key,
+        self.publish(key,
                 value=json.dumps(payload_read_metric_data_request),
                 topic='metric_request')
 
@@ -230,8 +231,8 @@ class KafkaProducer(object):
         payload_metric_data_response = jsmin(open(os.path.join(json_path,
                                                   'read_metric_data_resp.json')).read())
 
-        publish(key,
-                value=json.dumps(payload_metric_data_response),
+        self.publish(key,
+                value=message,
                 topic='metric_response')
 
 
@@ -242,7 +243,7 @@ class KafkaProducer(object):
         payload_metric_list_req = jsmin(open(os.path.join(json_path,
                                              'list_metric_req.json')).read())
 
-        publish(key,
+        self.publish(key,
                 value=json.dumps(payload_metric_list_req),
                 topic='metric_request')
 
@@ -253,8 +254,8 @@ class KafkaProducer(object):
         payload_metric_list_resp = jsmin(open(os.path.join(json_path,
                                               'list_metrics_resp.json')).read())
 
-        publish(key,
-                value=json.dumps(payload_metric_list_resp),
+        self.publish(key,
+                value=message,
                 topic='metric_response')
 
 
@@ -265,7 +266,7 @@ class KafkaProducer(object):
         payload_delete_metric_req = jsmin(open(os.path.join(json_path,
                                                'delete_metric_req.json')).read())
 
-        publish(key,
+        self.publish(key,
                 value=json.dumps(payload_delete_metric_req),
                 topic='metric_request')
 
@@ -277,8 +278,8 @@ class KafkaProducer(object):
         payload_delete_metric_resp = jsmin(open(os.path.join(json_path,
                                                 'delete_metric_resp.json')).read())
 
-        publish(key,
-                value=json.dumps(payload_delete_metric_resp),
+        self.publish(key,
+                value=message,
                 topic='metric_response')
 
 
@@ -289,7 +290,7 @@ class KafkaProducer(object):
         payload_update_metric_req = jsmin(open(os.path.join(json_path,
                                                'update_metric_req.json')).read())
 
-        publish(key,
+        self.publish(key,
                 value=json.dumps(payload_update_metric_req),
                 topic='metric_request')
 
@@ -301,8 +302,8 @@ class KafkaProducer(object):
         payload_update_metric_resp = jsmin(open(os.path.join(json_path,
                                                 'update_metric_resp.json')).read())
 
-        publish(key,
-                value=json.dumps(payload_update_metric_resp),
+        self.publish(key,
+                value=message,
                 topic='metric_response')
 
     def access_credentials(self, key, message, topic):
@@ -310,6 +311,6 @@ class KafkaProducer(object):
         payload_access_credentials = jsmin(open(os.path.join(json_path,
                                                 'access_credentials.json')).read())
 
-        publish(key,
+        self.publish(key,
                 value=json.dumps(payload_access_credentials),
                 topic='access_credentials')