commit 102d0b9a4584b78ec7d10992d1b2b8e034e76b80
author Shad Ansari <shad@opennetworking.org> Mon Apr 22 15:31:53 2019 -0700
committer Shad Ansari <shad@opennetworking.org> Fri May 17 06:07:11 2019 +0000
tree 7ff51793926db3362e63180b8947642962170dfd
parent 33b3a24c886f934838adeca6d95ce615fc513f4d

OpenoltKafkaProxy

Change-Id: I10f46ffd417f5c6814d30c56b1656a492b265528

voltha/adapters/openolt/openolt_grpc.py

diff --git a/voltha/adapters/openolt/openolt_grpc.py b/voltha/adapters/openolt/openolt_grpc.py
index 0db15a0..2dcc4c5 100644
--- a/voltha/adapters/openolt/openolt_grpc.py
+++ b/voltha/adapters/openolt/openolt_grpc.py
@@ -19,8 +19,10 @@
 import grpc
 import threading
 
+from voltha.registry import registry
 from voltha.adapters.openolt.openolt_kafka_producer import kafka_send_pb
 from voltha.adapters.openolt.protos import openolt_pb2_grpc, openolt_pb2
+from voltha.adapters.openolt.openolt_kafka_proxy import OpenoltKafkaProxy
 
 log = structlog.get_logger()
 
@@ -79,4 +81,9 @@
     broker = sys.argv[1]
     host = sys.argv[2]
 
-    process_indications(broker, host)
+    kafka_proxy = registry.register(
+        'openolt_kafka_proxy',
+        OpenoltKafkaProxy(broker)
+    ).start()
+
+    process_indications(host)

voltha/adapters/openolt/openolt_kafka_producer.py

diff --git a/voltha/adapters/openolt/openolt_kafka_producer.py b/voltha/adapters/openolt/openolt_kafka_producer.py
index 58b3613..567933b 100644
--- a/voltha/adapters/openolt/openolt_kafka_producer.py
+++ b/voltha/adapters/openolt/openolt_kafka_producer.py
@@ -25,7 +25,7 @@
 def kafka_send_pb(topic, msg):
     try:
         log.debug('send protobuf to kafka', topic=topic, msg=msg)
-        kafka_proxy = registry('kafka_proxy')
+        kafka_proxy = registry('openolt_kafka_proxy')
         if kafka_proxy and not kafka_proxy.is_faulty():
             log.debug('kafka-proxy-available')
             kafka_proxy.send_message(

voltha/main.py

diff --git a/voltha/main.py b/voltha/main.py
index 253b643..243bffa 100755
--- a/voltha/main.py
+++ b/voltha/main.py
@@ -42,6 +42,7 @@
 from voltha.northbound.diagnostics import Diagnostics
 from voltha.northbound.grpc.grpc_server import VolthaGrpcServer
 from voltha.northbound.kafka.kafka_proxy import KafkaProxy
+from voltha.adapters.openolt.openolt_kafka_proxy import OpenoltKafkaProxy
 from voltha.northbound.rest.health_check import init_rest_service
 from voltha.protos.common_pb2 import LogLevel
 from voltha.registry import registry, IComponent
@@ -438,6 +439,11 @@
             ).start()
 
             yield registry.register(
+                'openolt_kafka_proxy',
+                OpenoltKafkaProxy(self.args.kafka)
+            ).start()
+
+            yield registry.register(
                 'frameio',
                 FrameIOManager()
             ).start()