Openolt kafka proxy - add file missing from prev commit
Change-Id: I9001e3c32b6288d34f50642fe70e198ab399f484
diff --git a/voltha/adapters/openolt/openolt_grpc.py b/voltha/adapters/openolt/openolt_grpc.py
index 2dcc4c5..a5c3a92 100644
--- a/voltha/adapters/openolt/openolt_grpc.py
+++ b/voltha/adapters/openolt/openolt_grpc.py
@@ -20,9 +20,9 @@
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
+from voltha.adapters.openolt.openolt_kafka_proxy import OpenoltKafkaProxy, \
+ kafka_send_pb
log = structlog.get_logger()
diff --git a/voltha/adapters/openolt/openolt_kafka_producer.py b/voltha/adapters/openolt/openolt_kafka_producer.py
deleted file mode 100644
index 567933b..0000000
--- a/voltha/adapters/openolt/openolt_kafka_producer.py
+++ /dev/null
@@ -1,39 +0,0 @@
-#
-# Copyright 2017 the original author or authors.
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-from structlog import get_logger
-from simplejson import dumps
-from google.protobuf.json_format import MessageToJson
-from voltha.registry import registry
-
-log = get_logger()
-
-
-def kafka_send_pb(topic, msg):
- try:
- log.debug('send protobuf to kafka', topic=topic, msg=msg)
- kafka_proxy = registry('openolt_kafka_proxy')
- if kafka_proxy and not kafka_proxy.is_faulty():
- log.debug('kafka-proxy-available')
- kafka_proxy.send_message(
- topic,
- dumps(MessageToJson(
- msg,
- including_default_value_fields=True)))
- else:
- log.error('kafka-proxy-unavailable')
- except Exception, e:
- log.exception('failed-sending-protobuf', e=e)
diff --git a/voltha/adapters/openolt/openolt_kafka_proxy.py b/voltha/adapters/openolt/openolt_kafka_proxy.py
new file mode 100644
index 0000000..1541bd9
--- /dev/null
+++ b/voltha/adapters/openolt/openolt_kafka_proxy.py
@@ -0,0 +1,87 @@
+#
+# Copyright 2017 the original author or authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+
+from structlog import get_logger
+from simplejson import dumps
+from google.protobuf.json_format import MessageToJson
+from zope.interface import implementer
+from confluent_kafka import Producer
+
+from voltha.registry import registry, IComponent
+
+log = get_logger()
+
+
+@implementer(IComponent)
+class OpenoltKafkaProxy(object):
+ """
+ This is a singleton proxy kafka class to hide the kafka client details.
+ This proxy uses confluent-kafka-python as the kafka client.
+ """
+ _kafka_instance = None
+
+ def __init__(self, kafka_endpoint='localhost:9092'):
+
+ # return an exception if the object already exist
+ if OpenoltKafkaProxy._kafka_instance:
+ raise Exception('Singleton exist for :{}'.format(
+ OpenoltKafkaProxy))
+
+ log.debug('initializing', endpoint=kafka_endpoint)
+ self.kafka_endpoint = kafka_endpoint
+ self.kproducer = None
+ self.stopping = False
+ self.faulty = False
+ log.debug('initialized', endpoint=kafka_endpoint)
+
+ def start(self):
+ log.debug('starting')
+ self._get_kafka_producer()
+ OpenoltKafkaProxy._kafka_instance = self
+ log.info('started')
+ self.stopping = False
+ return self
+
+ def stop(self):
+ pass
+
+ def _get_kafka_producer(self):
+ conf = {'bootstrap.servers': self.kafka_endpoint}
+ self.kproducer = Producer(**conf)
+
+ def send_message(self, topic, msg, key=None):
+ self.kproducer.produce(topic, msg)
+
+ def is_faulty(self):
+ return self.faulty
+
+
+def kafka_send_pb(topic, msg):
+ try:
+ log.debug('send protobuf to kafka', topic=topic, msg=msg)
+ kafka_proxy = registry('openolt_kafka_proxy')
+ if kafka_proxy and not kafka_proxy.is_faulty():
+ log.debug('kafka-proxy-available')
+ kafka_proxy.send_message(
+ topic,
+ dumps(MessageToJson(
+ msg,
+ including_default_value_fields=True)))
+ else:
+ log.error('kafka-proxy-unavailable')
+ except Exception, e:
+ log.exception('failed-sending-protobuf', e=e)
diff --git a/voltha/core/local_handler.py b/voltha/core/local_handler.py
index 3aea9f1..87600c9 100644
--- a/voltha/core/local_handler.py
+++ b/voltha/core/local_handler.py
@@ -41,7 +41,7 @@
from voltha.protos.omci_alarm_db_pb2 import AlarmDeviceData
from requests.api import request
from common.utils.asleep import asleep
-from voltha.adapters.openolt.openolt_kafka_producer import kafka_send_pb
+from voltha.adapters.openolt.openolt_kafka_proxy import kafka_send_pb
log = structlog.get_logger()