diff --git a/src/context/client/EventsCollector.py b/src/context/client/EventsCollector.py
index 9715098bd3cd979d78a83b4839e40613d3997d1e..f5fc3fbc735c2f62b39223b9ed20aa3730ecd11d 100644
--- a/src/context/client/EventsCollector.py
+++ b/src/context/client/EventsCollector.py
@@ -12,7 +12,8 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-import grpc, logging, queue, threading
+from typing import Callable
+import grpc, logging, queue, threading, time
from common.proto.context_pb2 import Empty
from common.tools.grpc.Tools import grpc_message_to_json_string
from context.client.ContextClient import ContextClient
@@ -20,6 +21,41 @@ from context.client.ContextClient import ContextClient
LOGGER = logging.getLogger(__name__)
LOGGER.setLevel(logging.DEBUG)
+class _Collector(threading.Thread):
+ def __init__(
+ self, subscription_func : Callable, events_queue = queue.Queue,
+ terminate = threading.Event, log_events_received: bool = False
+ ) -> None:
+ super().__init__(daemon=False)
+ self._subscription_func = subscription_func
+ self._events_queue = events_queue
+ self._terminate = terminate
+ self._log_events_received = log_events_received
+ self._stream = None
+
+ def cancel(self) -> None:
+ if self._stream is None: return
+ self._stream.cancel()
+
+ def run(self) -> None:
+ while not self._terminate.is_set():
+ self._stream = self._subscription_func()
+ try:
+ for event in self._stream:
+ if self._log_events_received:
+ str_event = grpc_message_to_json_string(event)
+ LOGGER.info('[_collect] event: {:s}'.format(str_event))
+ self._events_queue.put_nowait(event)
+ except grpc.RpcError as e:
+ if e.code() == grpc.StatusCode.UNAVAILABLE:
+ LOGGER.info('[_collect] UNAVAILABLE... retrying...')
+ time.sleep(0.5)
+ continue
+ elif e.code() == grpc.StatusCode.CANCELLED:
+ break
+ else:
+ raise # pragma: no cover
+
class EventsCollector:
def __init__(
self, context_client : ContextClient,
@@ -31,60 +67,49 @@ class EventsCollector:
activate_service_collector : bool = True,
activate_slice_collector : bool = True,
activate_connection_collector : bool = True,
-
) -> None:
self._events_queue = queue.Queue()
+ self._terminate = threading.Event()
self._log_events_received = log_events_received
- self._context_stream, self._context_thread = None, None
- if activate_context_collector:
- self._context_stream = context_client.GetContextEvents(Empty())
- self._context_thread = self._create_collector_thread(self._context_stream)
-
- self._topology_stream, self._topology_thread = None, None
- if activate_topology_collector:
- self._topology_stream = context_client.GetTopologyEvents(Empty())
- self._topology_thread = self._create_collector_thread(self._topology_stream)
-
- self._device_stream, self._device_thread = None, None
- if activate_device_collector:
- self._device_stream = context_client.GetDeviceEvents(Empty())
- self._device_thread = self._create_collector_thread(self._device_stream)
-
- self._link_stream, self._link_thread = None, None
- if activate_link_collector:
- self._link_stream = context_client.GetLinkEvents(Empty())
- self._link_thread = self._create_collector_thread(self._link_stream)
-
- self._service_stream, self._service_thread = None, None
- if activate_service_collector:
- self._service_stream = context_client.GetServiceEvents(Empty())
- self._service_thread = self._create_collector_thread(self._service_stream)
-
- self._slice_stream, self._slice_thread = None, None
- if activate_slice_collector:
- self._slice_stream = context_client.GetSliceEvents(Empty())
- self._slice_thread = self._create_collector_thread(self._slice_stream)
-
- self._connection_stream, self._connection_thread = None, None
- if activate_connection_collector:
- self._connection_stream = context_client.GetConnectionEvents(Empty())
- self._connection_thread = self._create_collector_thread(self._connection_stream)
-
- def _create_collector_thread(self, stream, as_daemon : bool = False):
- return threading.Thread(target=self._collect, args=(stream,), daemon=as_daemon)
-
- def _collect(self, events_stream) -> None:
- try:
- for event in events_stream:
- if self._log_events_received:
- LOGGER.info('[_collect] event: {:s}'.format(grpc_message_to_json_string(event)))
- self._events_queue.put_nowait(event)
- except grpc.RpcError as e:
- if e.code() != grpc.StatusCode.CANCELLED: # pylint: disable=no-member
- raise # pragma: no cover
+ self._context_thread = _Collector(
+ lambda: context_client.GetContextEvents(Empty()),
+ self._events_queue, self._terminate, self._log_events_received
+ ) if activate_context_collector else None
+
+ self._topology_thread = _Collector(
+ lambda: context_client.GetTopologyEvents(Empty()),
+ self._events_queue, self._terminate, self._log_events_received
+ ) if activate_topology_collector else None
+
+ self._device_thread = _Collector(
+ lambda: context_client.GetDeviceEvents(Empty()),
+ self._events_queue, self._terminate, self._log_events_received
+ ) if activate_device_collector else None
+
+ self._link_thread = _Collector(
+ lambda: context_client.GetLinkEvents(Empty()),
+ self._events_queue, self._terminate, self._log_events_received
+ ) if activate_link_collector else None
+
+ self._service_thread = _Collector(
+ lambda: context_client.GetServiceEvents(Empty()),
+ self._events_queue, self._terminate, self._log_events_received
+ ) if activate_service_collector else None
+
+ self._slice_thread = _Collector(
+ lambda: context_client.GetSliceEvents(Empty()),
+ self._events_queue, self._terminate, self._log_events_received
+ ) if activate_slice_collector else None
+
+ self._connection_thread = _Collector(
+ lambda: context_client.GetConnectionEvents(Empty()),
+ self._events_queue, self._terminate, self._log_events_received
+ ) if activate_connection_collector else None
def start(self):
+ self._terminate.clear()
+
if self._context_thread is not None: self._context_thread.start()
if self._topology_thread is not None: self._topology_thread.start()
if self._device_thread is not None: self._device_thread.start()
@@ -102,25 +127,28 @@ class EventsCollector:
def get_events(self, block : bool = True, timeout : float = 0.1, count : int = None):
events = []
if count is None:
- while True:
+ while not self._terminate.is_set():
event = self.get_event(block=block, timeout=timeout)
if event is None: break
events.append(event)
else:
for _ in range(count):
+ if self._terminate.is_set(): break
event = self.get_event(block=block, timeout=timeout)
if event is None: continue
events.append(event)
return sorted(events, key=lambda e: e.event.timestamp.timestamp)
def stop(self):
- if self._context_stream is not None: self._context_stream.cancel()
- if self._topology_stream is not None: self._topology_stream.cancel()
- if self._device_stream is not None: self._device_stream.cancel()
- if self._link_stream is not None: self._link_stream.cancel()
- if self._service_stream is not None: self._service_stream.cancel()
- if self._slice_stream is not None: self._slice_stream.cancel()
- if self._connection_stream is not None: self._connection_stream.cancel()
+ self._terminate.set()
+
+ if self._context_thread is not None: self._context_thread.cancel()
+ if self._topology_thread is not None: self._topology_thread.cancel()
+ if self._device_thread is not None: self._device_thread.cancel()
+ if self._link_thread is not None: self._link_thread.cancel()
+ if self._service_thread is not None: self._service_thread.cancel()
+ if self._slice_thread is not None: self._slice_thread.cancel()
+ if self._connection_thread is not None: self._connection_thread.cancel()
if self._context_thread is not None: self._context_thread.join()
if self._topology_thread is not None: self._topology_thread.join()