Implemented handshake and heartbeat

Implemented handshake in json server and handoff to endpoint

Endpoint handles connector handshake accept/decline then launches both
its IO threads alongside its heartbeat thread which contains simple
timeout logic, timeout calls connection handles to close, subsequently
killing the endpoint

src/dynalab_core/protocols/endpoint.py

@@ -2,36 +2,106 @@
 # Copyright (C) 2026 Association Exergie <association.exergie@gmail.com>
 # SPDX-License-Identifier: GPL-3.0-or-later
 
+import logging
 from queue import Empty, Full, Queue
-from threading import RLock, Thread
 import threading
+from threading import RLock, Thread
+from time import monotonic, 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.handshake import ConnectorHello
+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) -> None:
+    def __init__(
+        self,
+        hello: ConnectorHello,
+        timeout_event: threading.Event | None = None,
+    ) -> None:
+        # external IO queues
         self._packet_ingress_queue: Queue[ProtocolMessage] = Queue(524288)
         self._packet_egress_queue: Queue[ProtocolMessage] = Queue(524288)
+        # internal IO queues
+        self._heartbeat_ingress_queue: Queue[ProtocolMessage] = Queue(524288)
+        self._heartbeat_egress_queue: Queue[ProtocolMessage] = Queue(524288)
+
         self._connector_hello = hello
-        self._worker_thread = Thread(
-            target=self._worker,
-            name=f"endpoint_worker_{self._connector_hello.connector_uuid}",
+        self._timed_out_event = timeout_event or threading.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._stop_event = threading.Event()
-        self._stopped_event = threading.Event()
+        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._worker_thread.is_alive():
-            self._stopped_event.set()
-        self._stopped_event.wait(10)
+
+        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
@@ -40,49 +110,163 @@ class ConnectorEndpoint:
         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 | None:
-        try:
-            packet = self._packet_egress_queue.get(block=True, timeout=timeout)
-        except Empty:
-            return None
-        else:
-            return packet
+    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 | None:
-        try:
-            packet = self._packet_egress_queue.get_nowait()
-        except Empty:
-            return None
-        else:
-            return packet
+    def get_egress_packet_no_wait(self) -> ProtocolMessage:
+        return self._packet_egress_queue.get_nowait()
 
-    def _get_ingress_packet(self, timeout: float | None) -> ProtocolMessage | None:
-        try:
-            packet = self._packet_ingress_queue.get(block=True, timeout=timeout)
-        except Empty:
-            return None
-        else:
-            return packet
+    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 | None:
-        try:
-            packet = self._packet_ingress_queue.get_nowait()
-        except Empty:
-            return None
-        else:
-            return packet
+    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 _worker(self) -> None:
-        # TODO: build endpoint worker thread
-        pass
+    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
+
+                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,
+            },
+        )
+        while not self._stop_event.is_set():
+            try:
+                message = self._get_ingress_packet_no_wait()
+            except Empty:
+                self._stop_event.wait(0.01)
+                continue
+            else:
+                if isinstance(message, Heartbeat):
+                    self._heartbeat_ingress_queue.put_nowait(message)
+        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:
+        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 self._connector_hello.protocol_version == HELLO_PACKET.protocol_version:
+            self._put_egress_packet(HandshakeAccepted(accepted_signals=[]))
+            log.debug(f"Accepted handshake for endpoint {self.uuid()}")
+        else:
+            self._put_egress_packet(
+                HandshakeRejected(reason="Protocol versions mismatch")
+            )
+            log.debug(f"Rejected handshake for endpoint {self.uuid()}")
+
+        self._heartbeat_worker_thread.start()
+
+        while not self._stop_event.is_set():
+            messages: list[ProtocolMessage] = []
+            try:
+                messages.append(self._heartbeat_egress_queue.get_nowait())
+            except Empty:
+                self._stop_event.wait(0.01)
+                pass
+
+            for message in messages:
+                self._put_egress_packet(message)
+
+        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:
@@ -90,30 +274,99 @@ class ConnectorRegistry:
         self._endpoints: dict[UUID, ConnectorEndpoint] = {}
         self._lock = RLock()
 
-    def register(self, hello: ConnectorHello) -> None:
-        endpoint = ConnectorEndpoint(hello)
+    def register(
+        self, hello: ConnectorHello, timeout_event: threading.Event
+    ) -> ConnectorEndpoint:
+        endpoint = ConnectorEndpoint(hello, timeout_event)
 
         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(endpoint.uuid())
+            current = self._endpoints.get(connector_uuid)
 
             if current is endpoint:
-                del self._endpoints[endpoint.uuid()]
+                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:
-        for endpoint in self._endpoints.values():
+        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,
+            },
+        )