diff --git a/src/dynalab_core/__init__.py b/src/dynalab_core/__init__.py index bbaff83..3f02431 100644 --- a/src/dynalab_core/__init__.py +++ b/src/dynalab_core/__init__.py @@ -9,7 +9,8 @@ from typing import Literal from dynalab_core.config import CoreConfig from dynalab_core.constants import CORE_VERSION from dynalab_core.errors import CoreStateMismatchError -from dynalab_core.protocols.json.common import VersionDescriptor +from dynalab_core.protocols.endpoint import ConnectorRegistry +from dynalab_core.protocols.common import VersionDescriptor from dynalab_core.protocols.json.server import JsonServer @@ -21,7 +22,8 @@ class Core: self._core_version: VersionDescriptor = CORE_VERSION self._stop_event: threading.Event = threading.Event() self._core_config: CoreConfig = config - self._json_server = JsonServer(self._core_config, self._stop_event) + self._json_server = JsonServer(self._core_config) + self._connector_registry = ConnectorRegistry() self._state = "initd" @@ -37,5 +39,7 @@ class Core: self._stop_event.wait(timeout) def stop(self) -> None: + self._json_server.stop() + self._connector_registry.stop() self._stop_event.set() self._state = "stopped" diff --git a/src/dynalab_core/constants.py b/src/dynalab_core/constants.py index 72e74b5..a43904f 100644 --- a/src/dynalab_core/constants.py +++ b/src/dynalab_core/constants.py @@ -2,7 +2,7 @@ # Copyright (C) 2026 Association Exergie # SPDX-License-Identifier: GPL-3.0-or-later -from dynalab_core.protocols.json.common import VersionDescriptor +from dynalab_core.protocols.common import VersionDescriptor CORE_VERSION = VersionDescriptor(type="alpha", major=0, minor=0, patch=1) diff --git a/src/dynalab_core/errors.py b/src/dynalab_core/errors.py index 1ba299c..7e399c7 100644 --- a/src/dynalab_core/errors.py +++ b/src/dynalab_core/errors.py @@ -9,20 +9,3 @@ class CoreError(Exception): class CoreStateMismatchError(CoreError): """DynaLab Core state error.""" - - -# JsonServer error declarations -class JsonServerError(Exception): - """Generic JsonServer Error""" - - -class JsonServerTimeoutError(JsonServerError): - """JsonServer timed out""" - - -class JsonServerStartupError(JsonServerError): - """JsonServer crashed on startup""" - - def __init__(self, error: Exception) -> None: - self.error = error - super().__init__(f"JSON server failed to start: {error}") diff --git a/src/dynalab_core/protocols/json/common.py b/src/dynalab_core/protocols/common.py similarity index 100% rename from src/dynalab_core/protocols/json/common.py rename to src/dynalab_core/protocols/common.py diff --git a/src/dynalab_core/protocols/endpoint.py b/src/dynalab_core/protocols/endpoint.py new file mode 100644 index 0000000..5fec425 --- /dev/null +++ b/src/dynalab_core/protocols/endpoint.py @@ -0,0 +1,119 @@ +# Copyright (C) 2026 Hector van der Aa +# Copyright (C) 2026 Association Exergie +# SPDX-License-Identifier: GPL-3.0-or-later + +from queue import Empty, Full, Queue +from threading import RLock, Thread +import threading +from uuid import UUID +from dynalab_core.protocols.errors import ( + ConnectorEndpointQueueFullError, + ConnectorRegistryAlreadyRegisteredError, +) +from dynalab_core.protocols.packets import ProtocolMessage +from dynalab_core.protocols.packets.handshake import ConnectorHello + + +class ConnectorEndpoint: + def __init__(self, hello: ConnectorHello) -> None: + self._packet_ingress_queue: Queue[ProtocolMessage] = Queue(524288) + self._packet_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}", + daemon=True, + ) + self._stop_event = threading.Event() + self._stopped_event = threading.Event() + + def stop(self) -> None: + self._stop_event.set() + if not self._worker_thread.is_alive(): + self._stopped_event.set() + self._stopped_event.wait(10) + + 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: + 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_no_wait(self) -> ProtocolMessage | None: + try: + packet = self._packet_egress_queue.get_nowait() + except Empty: + return None + else: + return packet + + 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_no_wait(self) -> ProtocolMessage | None: + try: + packet = self._packet_ingress_queue.get_nowait() + except Empty: + return None + else: + return packet + + def _put_egress_packet(self, packet: ProtocolMessage) -> None: + try: + self._packet_egress_queue.put_nowait(packet) + except Full: + raise ConnectorEndpointQueueFullError + + def _worker(self) -> None: + # TODO: build endpoint worker thread + pass + + +class ConnectorRegistry: + def __init__(self) -> None: + self._endpoints: dict[UUID, ConnectorEndpoint] = {} + self._lock = RLock() + + def register(self, hello: ConnectorHello) -> None: + endpoint = ConnectorEndpoint(hello) + + with self._lock: + current = self._endpoints.get(hello.connector_uuid) + + if current is not None: + raise ConnectorRegistryAlreadyRegisteredError + + self._endpoints[hello.connector_uuid] = endpoint + + return endpoint + + def unregister(self, endpoint: ConnectorEndpoint) -> None: + with self._lock: + current = self._endpoints.get(endpoint.uuid()) + + if current is endpoint: + del self._endpoints[endpoint.uuid()] + + 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(): + endpoint.stop() diff --git a/src/dynalab_core/protocols/errors.py b/src/dynalab_core/protocols/errors.py new file mode 100644 index 0000000..e443cf7 --- /dev/null +++ b/src/dynalab_core/protocols/errors.py @@ -0,0 +1,22 @@ +# Copyright (C) 2026 Hector van der Aa +# Copyright (C) 2026 Association Exergie +# SPDX-License-Identifier: GPL-3.0-or-later + +# ConnectorEndpoint error declarations +class ConnectorEndpointError(Exception): + """Generic ConnectorEndpoint Error""" + + +class ConnectorEndpointQueueFullError(ConnectorEndpointError): + """ConnectorEndpoing queue is full""" + + +# ConnectorRegistry error declarations + + +class ConnectorRegistryError(Exception): + """Generic ConnectorRegistry Error""" + + +class ConnectorRegistryAlreadyRegisteredError(ConnectorRegistryError): + """ConnectorRegistry endpoint already registered""" diff --git a/src/dynalab_core/protocols/json/constants.py b/src/dynalab_core/protocols/json/constants.py index 8718dc9..d0b4362 100644 --- a/src/dynalab_core/protocols/json/constants.py +++ b/src/dynalab_core/protocols/json/constants.py @@ -2,6 +2,6 @@ # Copyright (C) 2026 Association Exergie # SPDX-License-Identifier: GPL-3.0-or-later -from dynalab_core.protocols.json.common import VersionDescriptor +from dynalab_core.protocols.common import VersionDescriptor PROTOCOL_VERSION = VersionDescriptor(type="alpha", major=0, minor=0, patch=1) diff --git a/src/dynalab_core/protocols/json/errors.py b/src/dynalab_core/protocols/json/errors.py new file mode 100644 index 0000000..65a284d --- /dev/null +++ b/src/dynalab_core/protocols/json/errors.py @@ -0,0 +1,19 @@ +# Copyright (C) 2026 Hector van der Aa +# Copyright (C) 2026 Association Exergie +# SPDX-License-Identifier: GPL-3.0-or-later + +# JsonServer error declarations +class JsonServerError(Exception): + """Generic JsonServer Error""" + + +class JsonServerTimeoutError(JsonServerError): + """JsonServer timed out""" + + +class JsonServerStartupError(JsonServerError): + """JsonServer crashed on startup""" + + def __init__(self, error: Exception) -> None: + self.error = error + super().__init__(f"JSON server failed to start: {error}") diff --git a/src/dynalab_core/protocols/json/server.py b/src/dynalab_core/protocols/json/server.py index 444c83d..0a1c4f4 100644 --- a/src/dynalab_core/protocols/json/server.py +++ b/src/dynalab_core/protocols/json/server.py @@ -9,7 +9,7 @@ from threading import Thread from dynalab_core.config import CoreConfig -from dynalab_core.errors import ( +from dynalab_core.protocols.json.errors import ( JsonServerStartupError, JsonServerTimeoutError, ) @@ -19,7 +19,6 @@ class JsonServer: def __init__( self, config: CoreConfig, - stop_event: threading.Event, ) -> None: self._thread: Thread | None = None self._server: Server | None = None @@ -27,8 +26,9 @@ class JsonServer: self._startup_error: Exception | None = None self._debug_timeout_test: bool = False self._config = config - self._global_stop_event = stop_event self._timeout: float = 10 + self._stop_event = threading.Event() + self._stopped_event = threading.Event() def start(self) -> None: self._thread = Thread( @@ -44,6 +44,10 @@ class JsonServer: if self._startup_error is not None: raise JsonServerStartupError(self._startup_error) + def stop(self) -> None: + self._stop_event.set() + self._stopped_event.wait(10) + def _json_server_thread_main(self) -> None: asyncio.run(self._run_json_server()) @@ -64,14 +68,19 @@ class JsonServer: if not self._debug_timeout_test: self._started_event.set() + if self._startup_error is not None: + self._stopped_event.set() + return async with self._server: - await asyncio.to_thread(self._global_stop_event.wait) + await asyncio.to_thread(self._stop_event.wait) + self._stopped_event.set() async def _handle_json_connection( self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter ) -> None: print(f"Connected to: {writer.get_extra_info('peername')}") + # TODO: Build out handler to handshake, register peer and pipe packets into connector queues try: await reader.readline() finally: diff --git a/src/dynalab_core/protocols/packets/__init__.py b/src/dynalab_core/protocols/packets/__init__.py new file mode 100644 index 0000000..8da7159 --- /dev/null +++ b/src/dynalab_core/protocols/packets/__init__.py @@ -0,0 +1,16 @@ +from typing import Annotated + +from pydantic import Field + +from dynalab_core.protocols.packets.handshake import ( + ConnectorHello, + DynaLabHello, + HandshakeAccepted, + HandshakeRejected, +) + + +ProtocolMessage = Annotated[ + DynaLabHello | ConnectorHello | HandshakeAccepted | HandshakeRejected, + Field(discriminator="type"), +] diff --git a/src/dynalab_core/protocols/json/packets/handshake.py b/src/dynalab_core/protocols/packets/handshake.py similarity index 94% rename from src/dynalab_core/protocols/json/packets/handshake.py rename to src/dynalab_core/protocols/packets/handshake.py index 494de71..b4f0ccc 100644 --- a/src/dynalab_core/protocols/json/packets/handshake.py +++ b/src/dynalab_core/protocols/packets/handshake.py @@ -6,7 +6,7 @@ from typing import Literal from uuid import UUID from pydantic import BaseModel, Field -from dynalab_core.protocols.json.common import VersionDescriptor +from dynalab_core.protocols.common import VersionDescriptor from dynalab_core.protocols.json.constants import PROTOCOL_VERSION