Refactor openolt_grpc
Change-Id: I4e8b7671f05d527bc3a25a79ff26ba1044efedf6
diff --git a/voltha/adapters/openolt/openolt_grpc.py b/voltha/adapters/openolt/openolt_grpc.py
index 52f5f06..26b9547 100644
--- a/voltha/adapters/openolt/openolt_grpc.py
+++ b/voltha/adapters/openolt/openolt_grpc.py
@@ -14,61 +14,71 @@
# limitations under the License.
#
+import sys
import structlog
import grpc
import threading
+from structlog import get_logger
+
from voltha.northbound.kafka.kafka_proxy import kafka_send_pb
from voltha.adapters.openolt.protos import openolt_pb2_grpc, openolt_pb2
+log = get_logger()
+
class OpenoltGrpc(object):
def __init__(self, host_and_port, device):
super(OpenoltGrpc, self).__init__()
self.log = structlog.get_logger()
- self.log.debug('openolt grpc init')
+ log.debug('openolt grpc init')
self.device = device
self.host_and_port = host_and_port
self.channel = grpc.insecure_channel(self.host_and_port)
- self.channel_ready_future = grpc.channel_ready_future(self.channel)
self.stub = openolt_pb2_grpc.OpenoltStub(self.channel)
def start(self):
try:
# Start indications thread
- self.log.debug('openolt grpc starting')
+ log.debug('openolt grpc starting')
self.indications_thread_handle = threading.Thread(
- target=self.indications_thread)
- # Old getter/setter API for daemon; use it directly as a
- # property instead. The Jinkins error will happon on the reason of
- # Exception in thread Thread-1 (most likely raised # during
- # interpreter shutdown)
+ target=process_indications,
+ args=(self.host_and_port,))
self.indications_thread_handle.setDaemon(True)
self.indications_thread_handle.start()
except Exception as e:
- self.log.exception('indication start failed', e=e)
+ log.exception('indication start failed', e=e)
else:
- self.log.debug('openolt grpc started')
+ log.debug('openolt grpc started')
- def indications_thread(self):
- self.indications = self.stub.EnableIndication(openolt_pb2.Empty())
+def process_indications(host_and_port):
+ channel = grpc.insecure_channel(host_and_port)
+ stub = openolt_pb2_grpc.OpenoltStub(channel)
+ stream = stub.EnableIndication(openolt_pb2.Empty())
- topic = 'openolt.ind-{}'.format(
- self.device.host_and_port.split(':')[0])
+ topic = 'openolt.ind-{}'.format(host_and_port.split(':')[0])
- while True:
- try:
- # get the next indication from olt
- ind = next(self.indications)
- except Exception as e:
- self.log.warn('openolt grpc connection lost', error=e)
- ind = openolt_pb2.Indication()
- ind.olt_ind.oper_state = 'down'
- kafka_send_pb(topic, ind)
- break
- else:
- self.log.debug("openolt grpc rx indication", indication=ind)
+ while True:
+ try:
+ # get the next indication from olt
+ ind = next(stream)
+ except Exception as e:
+ log.warn('openolt grpc connection lost', error=e)
+ ind = openolt_pb2.Indication()
+ ind.olt_ind.oper_state = 'down'
+ kafka_send_pb(topic, ind)
+ break
+ else:
+ log.debug("openolt grpc rx indication", indication=ind)
+ kafka_send_pb(topic, ind)
- topic = 'openolt.ind-{}'.format(
- self.device.host_and_port.split(':')[0])
- kafka_send_pb(topic, ind)
+
+if __name__ == '__main__':
+ if len(sys.argv) < 2:
+ sys.stderr.write('Usage: %s <olt hostname or ip>\n\n' % sys.argv[0])
+ sys.exit(1)
+
+ broker = sys.argv[1]
+ host = sys.argv[2]
+
+ process_indications(broker, host)