Make Voltha/OFAgent Ctrl-C terminatable
Change-Id: I8d64b126d8d8d6f368d6cc236b2293fbcd108416
diff --git a/ofagent/grpc_client.py b/ofagent/grpc_client.py
index d1b78bd..75c48ce 100644
--- a/ofagent/grpc_client.py
+++ b/ofagent/grpc_client.py
@@ -85,13 +85,19 @@
def receive_packet_in_stream():
streaming_rpc_method = self.local_stub.ReceivePacketsIn
- iterator = streaming_rpc_method(empty_pb2.Empty())
- for packet_in in iterator:
- reactor.callFromThread(self.packet_in_queue.put,
- packet_in)
- log.debug('enqued-packet-in',
- packet_in=packet_in,
- queue_len=len(self.packet_in_queue.pending))
+ iterator = streaming_rpc_method(empty_pb2.Empty(), timeout=1.0)
+ while not self.stopped:
+ try:
+ for packet_in in iterator:
+ reactor.callFromThread(self.packet_in_queue.put,
+ packet_in)
+ log.debug('enqued-packet-in',
+ packet_in=packet_in,
+ queue_len=len(self.packet_in_queue.pending))
+ except _Rendezvous, e:
+ if e.code() == StatusCode.DEADLINE_EXCEEDED:
+ continue
+ raise
reactor.callInThread(receive_packet_in_stream)
@@ -99,12 +105,18 @@
def receive_change_events():
streaming_rpc_method = self.local_stub.ReceiveChangeEvents
- iterator = streaming_rpc_method(empty_pb2.Empty())
- for event in iterator:
- reactor.callFromThread(self.change_event_queue.put, event)
- log.debug('enqued-change-event',
- change_event=event,
- queue_len=len(self.change_event_queue.pending))
+ iterator = streaming_rpc_method(empty_pb2.Empty(), timeout=1.0)
+ while not self.stopped:
+ try:
+ for event in iterator:
+ reactor.callFromThread(self.change_event_queue.put, event)
+ log.debug('enqued-change-event',
+ change_event=event,
+ queue_len=len(self.change_event_queue.pending))
+ except _Rendezvous, e:
+ if e.code() == StatusCode.DEADLINE_EXCEEDED:
+ continue
+ raise
reactor.callInThread(receive_change_events)