# Copyright (C) 2026 Hector van der Aa # Copyright (C) 2026 Association Exergie # SPDX-License-Identifier: GPL-3.0-or-later import logging from queue import Empty, Full, Queue import threading from threading import RLock, Thread from time import monotonic, monotonic_ns, sleep from uuid import UUID from dynalab_core.constants import HELLO_PACKET from dynalab_core.protocols.errors import ( ConnectorEndpointQueueFullError, ConnectorRegistryAlreadyRegisteredError, ) from dynalab_core.protocols.packets import ProtocolMessage from dynalab_core.protocols.packets.data import ValueBatch, ValueDescriptor from dynalab_core.protocols.packets.handshake import ( ConnectorHello, HandshakeAccepted, HandshakeRejected, ) from dynalab_core.protocols.packets.heartbeat import Heartbeat log = logging.getLogger(__name__) class ConnectorEndpoint: def __init__( self, hello: ConnectorHello, timeout_event: threading.Event, reject_event: threading.Event, core_input_queue: Queue, ) -> None: # external IO queues self._packet_ingress_queue: Queue[ProtocolMessage] = Queue(524288) self._packet_egress_queue: Queue[ProtocolMessage] = Queue(524288) self._core_input_queue: Queue[ProtocolMessage] = core_input_queue # internal IO queues self._heartbeat_ingress_queue: Queue[ProtocolMessage] = Queue(524288) self._heartbeat_egress_queue: Queue[ProtocolMessage] = Queue(524288) self._connector_hello = hello self._timed_out_event = timeout_event self._reject_event = reject_event self._stop_event = threading.Event() self._input_worker_stopped_event = threading.Event() self._input_worker_thread = Thread( target=self._input_worker, name=f"endpoint_input_worker_{self._connector_hello.connector_uuid}", daemon=True, ) self._output_worker_stopped_event = threading.Event() self._output_worker_thread = Thread( target=self._output_worker, name=f"endpoint_output_worker_{self._connector_hello.connector_uuid}", daemon=True, ) self._heartbeat_worker_stopped_event = threading.Event() self._heartbeat_worker_thread = Thread( target=self._heartbeat_worker, name=f"endpoint_heartbeat_worker_{self._connector_hello.connector_uuid}", daemon=True, ) self._input_worker_thread.start() self._output_worker_thread.start() def stop(self) -> None: connector_uuid = str(self.uuid()) log.debug( "Stopping connector endpoint %s", connector_uuid, extra={"event": "endpoint.stopping", "connector_uuid": connector_uuid}, ) self._stop_event.set() if not self._input_worker_thread.is_alive(): self._input_worker_stopped_event.set() if not self._output_worker_thread.is_alive(): self._output_worker_stopped_event.set() if not self._heartbeat_worker_thread.is_alive(): self._heartbeat_worker_stopped_event.set() self._input_worker_stopped_event.wait(1) self._output_worker_stopped_event.wait(1) self._heartbeat_worker_stopped_event.wait(1) if ( not self._input_worker_stopped_event.is_set() or not self._output_worker_stopped_event.is_set() or not self._heartbeat_worker_stopped_event.is_set() ): log.error( "Connector endpoint %s did not stop within 10 seconds", connector_uuid, extra={ "event": "endpoint.stop_timeout", "connector_uuid": connector_uuid, "timeout_s": 10, }, ) else: log.debug( "Connector endpoint %s stopped", connector_uuid, extra={"event": "endpoint.stopped", "connector_uuid": connector_uuid}, ) def uuid(self) -> UUID: return self._connector_hello.connector_uuid def put_ingress_packet(self, packet: ProtocolMessage) -> None: try: self._packet_ingress_queue.put_nowait(packet) except Full: self._log_queue_full("ingress", self._packet_ingress_queue) raise ConnectorEndpointQueueFullError def get_egress_packet(self, timeout: float | None) -> ProtocolMessage: return self._packet_egress_queue.get(block=True, timeout=timeout) def get_egress_packet_no_wait(self) -> ProtocolMessage: return self._packet_egress_queue.get_nowait() def _get_ingress_packet(self, timeout: float | None) -> ProtocolMessage: return self._packet_ingress_queue.get(block=True, timeout=timeout) def _get_ingress_packet_no_wait(self) -> ProtocolMessage: return self._packet_ingress_queue.get_nowait() def _put_egress_packet(self, packet: ProtocolMessage) -> None: try: self._packet_egress_queue.put_nowait(packet) except Full: self._log_queue_full("egress", self._packet_egress_queue) raise ConnectorEndpointQueueFullError def _log_queue_full(self, direction: str, queue: Queue[ProtocolMessage]) -> None: connector_uuid = str(self.uuid()) log.warning( "Connector %s %s queue is full (%d/%d)", connector_uuid, direction, queue.qsize(), queue.maxsize, extra={ "event": "endpoint.queue_full", "connector_uuid": connector_uuid, "queue_direction": direction, "queue_size": queue.qsize(), "queue_capacity": queue.maxsize, }, ) def _heartbeat_worker(self) -> None: last_send: int = 0 last_recieved: int = 0 first_received: bool = False ctr: int = 0 try: while not self._stop_event.is_set(): now = round(monotonic() * 1000) if now > last_send + HELLO_PACKET.heartbeat_interval_ms: try: self._heartbeat_egress_queue.put_nowait( Heartbeat(sequence=ctr, send_timestamp=now) ) last_send = now ctr += 1 log.debug("Sent heartbeat") except Full: sleep(0.01) try: received = self._heartbeat_ingress_queue.get_nowait() except Empty: sleep(0.01) else: if isinstance(received, Heartbeat): first_received = True if received.return_timestamp is not None: last_recieved = received.return_timestamp log.debug("Received heartbeat") if not first_received: if ctr > round( HELLO_PACKET.heartbeat_timeout_ms / HELLO_PACKET.heartbeat_interval_ms ): self._timed_out_event.set() return elif last_send > last_recieved + HELLO_PACKET.heartbeat_timeout_ms: self._timed_out_event.set() return finally: self._heartbeat_worker_stopped_event.set() def _input_worker(self) -> None: connector_uuid = str(self.uuid()) log.debug( "Connector endpoint input worker %s started", self._input_worker_thread.name, extra={ "event": "endpoint.input_worker_started", "connector_uuid": connector_uuid, "thread_name": self._input_worker_thread.name, }, ) last_log = 0 while not self._stop_event.is_set(): now = monotonic_ns() messages: list[ProtocolMessage] = [] try: while not self._packet_ingress_queue.empty(): messages.append(self._get_ingress_packet_no_wait()) except Empty: continue else: for message in messages: if isinstance(message, Heartbeat): self._heartbeat_ingress_queue.put_nowait(message) elif isinstance(message, ValueDescriptor): if message.signal_id in ( signal.id for signal in self._connector_hello.signals ): self._core_input_queue.put_nowait(message) elif isinstance(message, ValueBatch): for value in message.values: if value.signal_id in ( signal.id for signal in self._connector_hello.signals ): self._core_input_queue.put_nowait(value) if now > last_log + 2000 * 1_000_000: log.debug(f"Endpoint queue size: {self._packet_ingress_queue.qsize()}") last_log = now self._input_worker_stopped_event.set() log.debug( "Connector endpoint input worker %s stopped", self._input_worker_thread.name, extra={ "event": "endpoint.input_worker_stopped", "connector_uuid": connector_uuid, "thread_name": self._input_worker_thread.name, }, ) def _output_worker(self) -> None: try: connector_uuid = str(self.uuid()) log.debug( "Connector endpoint output worker %s started", self._input_worker_thread.name, extra={ "event": "endpoint.output_worker_started", "connector_uuid": connector_uuid, "thread_name": self._input_worker_thread.name, }, ) if ( not self._connector_hello.protocol_version == HELLO_PACKET.protocol_version ): self._put_egress_packet( HandshakeRejected(reason="Protocol versions mismatch") ) log.debug(f"Rejected handshake for endpoint {self.uuid()}") self._reject_event.set() return if not self._connector_hello.signals: self._put_egress_packet( HandshakeRejected(reason="No signals available") ) log.debug(f"Rejected handshake for endpoint {self.uuid()}") self._reject_event.set() return self._put_egress_packet(HandshakeAccepted(accepted_signals=[])) log.debug(f"Accepted handshake for endpoint {self.uuid()}") log.debug(f"Accepted signals: {self._connector_hello.signals}") self._heartbeat_worker_thread.start() while not self._stop_event.is_set(): try: message = self._heartbeat_egress_queue.get_nowait() except Empty: self._stop_event.wait(0.01) pass else: self._put_egress_packet(message) finally: self._output_worker_stopped_event.set() log.debug( "Connector endpoint output worker %s stopped", self._input_worker_thread.name, extra={ "event": "endpoint.output_worker_stopped", "connector_uuid": connector_uuid, "thread_name": self._input_worker_thread.name, }, ) class ConnectorRegistry: def __init__(self, core_input_queue: Queue) -> None: self._endpoints: dict[UUID, ConnectorEndpoint] = {} self._lock = RLock() self._core_input_queue = core_input_queue def register( self, hello: ConnectorHello, timeout_event: threading.Event, reject_event: threading.Event, ) -> ConnectorEndpoint: endpoint = ConnectorEndpoint( hello, timeout_event, reject_event, self._core_input_queue ) with self._lock: current = self._endpoints.get(hello.connector_uuid) if current is not None: log.warning( "Connector %s is already registered", hello.connector_uuid, extra={ "event": "connector.registration_rejected", "connector_uuid": str(hello.connector_uuid), "reason": "duplicate_uuid", }, ) raise ConnectorRegistryAlreadyRegisteredError self._endpoints[hello.connector_uuid] = endpoint endpoint_count = len(self._endpoints) log.info( "Registered connector %s (%s)", hello.connector_name, hello.connector_uuid, extra={ "event": "connector.registered", "connector_uuid": str(hello.connector_uuid), "connector_name": hello.connector_name, "connector_version": hello.connector_version, "protocol_version": hello.protocol_version.get_version(), "endpoint_count": endpoint_count, }, ) return endpoint def unregister(self, endpoint: ConnectorEndpoint) -> None: connector_uuid = endpoint.uuid() with self._lock: current = self._endpoints.get(connector_uuid) if current is endpoint: del self._endpoints[connector_uuid] endpoint_count = len(self._endpoints) else: log.debug( "Connector %s was not registered", connector_uuid, extra={ "event": "connector.unregister_noop", "connector_uuid": str(connector_uuid), }, ) return endpoint.stop() log.info( "Unregistered connector %s", connector_uuid, extra={ "event": "connector.unregistered", "connector_uuid": str(connector_uuid), "endpoint_count": endpoint_count, }, ) def get(self, connector_uuid: UUID) -> ConnectorEndpoint | None: with self._lock: return self._endpoints.get(connector_uuid) def stop(self) -> None: with self._lock: endpoints = tuple(self._endpoints.values()) self._endpoints.clear() log.debug( "Stopping connector registry with %d endpoint(s)", len(endpoints), extra={ "event": "connector_registry.stopping", "endpoint_count": len(endpoints), }, ) for endpoint in endpoints: endpoint.stop() log.debug( "Connector registry stopped", extra={ "event": "connector_registry.stopped", "endpoint_count": 0, }, )