# -*- coding: utf-8 -*- """ SyncConnection — Command-Only TCP-Verbindung zum Server. Leichtgewichtige Verbindung fuer Datensynchronisation. Nur Command-Socket (Port 2101), keine Audio-Ports. Handshake analog zu ClientConnection. """ from __future__ import annotations import asyncio import struct from dataclasses import dataclass from typing import TYPE_CHECKING, Callable, Awaitable from trixy_core.network.protocol import ( TrixyProtocol, ProtocolFlags, ProtocolMessage, MAGIC, MAGIC_LENGTH, COMMAND_HELLO, COMMAND_PING, COMMAND_PONG, ) from trixy_core.network.cmd import ( SatelliteConnect, SatelliteAccepted, SatelliteDenied, ) from trixy_core.utils.debug import pinfo, pdebug, perror, pwarn from trixy_core.utils.version import VERSION_STRING if TYPE_CHECKING: pass @dataclass class SyncConnectionConfig: """Konfiguration fuer die Sync-Verbindung.""" host: str = "" port: int = 2101 probe_interval_seconds: float = 30.0 sync_interval_seconds: float = 300.0 connect_timeout_seconds: float = 5.0 class SyncConnection: """ Command-Only TCP-Verbindung zum Server fuer Sync. Fuehrt den Standard-Handshake durch (TRXYHELO → SatelliteConnect → SatelliteAccepted), registriert den Standalone als Satellite. Kein Audio — nur Command-Socket. """ def __init__( self, config: SyncConnectionConfig, room: str, alias: str, mac_address: str, ) -> None: """ Initialisiert die SyncConnection. Args: config: Verbindungskonfiguration room: Raum-Kennung alias: Anzeigename mac_address: MAC-Adresse des Standalone """ self._config = config self._room = room self._alias = alias self._mac_address = mac_address # Protokoll self._protocol = TrixyProtocol() # Socket self._reader: asyncio.StreamReader | None = None self._writer: asyncio.StreamWriter | None = None # Zustand self._connected = False self._satellite_id: str = "" self._running = False # Tasks self._receive_task: asyncio.Task | None = None self._keepalive_task: asyncio.Task | None = None # Callbacks self._on_connected: Callable[[], Awaitable[None]] | None = None self._on_disconnected: Callable[[str], Awaitable[None]] | None = None self._on_message: Callable[[ProtocolMessage], Awaitable[None]] | None = None @property def is_connected(self) -> bool: """Ist die Verbindung aktiv?""" return self._connected @property def satellite_id(self) -> str: """Vom Server zugewiesene Satellite-ID.""" return self._satellite_id def on_connected(self, callback: Callable[[], Awaitable[None]]) -> None: """Registriert Callback fuer Verbindungsaufbau.""" self._on_connected = callback def on_disconnected(self, callback: Callable[[str], Awaitable[None]]) -> None: """Registriert Callback fuer Verbindungsabbruch.""" self._on_disconnected = callback def on_message(self, callback: Callable[[ProtocolMessage], Awaitable[None]]) -> None: """Registriert Callback fuer eingehende Nachrichten.""" self._on_message = callback async def connect(self) -> bool: """ Verbindet sich mit dem Server (Command-Socket + Handshake). Returns: True bei erfolgreicher Verbindung """ try: pdebug(f"SyncConnection: Verbinde zu {self._config.host}:{self._config.port}...") # Command-Socket verbinden try: self._reader, self._writer = await asyncio.wait_for( asyncio.open_connection(self._config.host, self._config.port), timeout=self._config.connect_timeout_seconds, ) except asyncio.TimeoutError: pdebug(f"SyncConnection: Timeout zu {self._config.host}:{self._config.port}") return False except OSError as e: pdebug(f"SyncConnection: Verbindungsfehler: {e}") return False # Handshake durchfuehren if not await self._perform_handshake(): await self._close_socket() return False # Empfangs-Tasks starten self._running = True self._receive_task = asyncio.create_task(self._receive_loop()) self._keepalive_task = asyncio.create_task(self._keepalive_loop()) self._connected = True pinfo(f"SyncConnection: Verbunden mit Server {self._config.host}:{self._config.port}") # Callback if self._on_connected: try: await self._on_connected() except Exception as e: perror(f"SyncConnection: on_connected Fehler: {e}") return True except Exception as e: perror(f"SyncConnection: Verbindungsfehler: {e}") await self._close_socket() return False async def disconnect(self, reason: str = "Sync-Disconnect") -> None: """ Trennt die Verbindung zum Server. Args: reason: Grund fuer die Trennung """ if not self._connected: return pdebug(f"SyncConnection: Trenne Verbindung: {reason}") self._running = False # Tasks beenden for task in [self._receive_task, self._keepalive_task]: if task and not task.done(): task.cancel() try: await task except asyncio.CancelledError: pass await self._close_socket() self._connected = False self._satellite_id = "" # Callback if self._on_disconnected: try: await self._on_disconnected(reason) except Exception as e: perror(f"SyncConnection: on_disconnected Fehler: {e}") async def send_command(self, cmd: object) -> bool: """ Sendet eine Nachricht zum Server. Args: cmd: Die zu sendende CommandMessage Returns: True bei Erfolg """ if not self._connected or not self._writer: return False return await self._send_message(cmd) # ========================================================================= # Handshake # ========================================================================= async def _perform_handshake(self) -> bool: """Fuehrt den Handshake mit dem Server durch.""" pdebug("SyncConnection: Starte Handshake...") # HELLO senden try: self._writer.write(COMMAND_HELLO) await self._writer.drain() except Exception as e: perror(f"SyncConnection: HELLO senden fehlgeschlagen: {e}") return False # SatelliteConnect senden connect_msg = SatelliteConnect( mac_address=self._mac_address, room=self._room, alias=self._alias, ip_address="", version=VERSION_STRING, ) if not await self._send_message(connect_msg): perror("SyncConnection: SatelliteConnect senden fehlgeschlagen") return False # Antwort empfangen try: response = await asyncio.wait_for( self._read_message(), timeout=self._config.connect_timeout_seconds, ) except asyncio.TimeoutError: perror("SyncConnection: Timeout beim Warten auf Server-Antwort") return False if response is None: perror("SyncConnection: Keine Antwort vom Server") return False # Antwort verarbeiten if response.class_name == "SatelliteDenied": data = response.data reason = "Unbekannt" if isinstance(data, dict): reason = data.get("reason", "Unbekannt") elif isinstance(data, SatelliteDenied): reason = data.reason pwarn(f"SyncConnection: Verbindung abgelehnt: {reason}") return False if response.class_name != "SatelliteAccepted": perror(f"SyncConnection: Unerwartete Antwort: {response.class_name}") return False # SatelliteAccepted verarbeiten data = response.data if isinstance(data, dict): self._satellite_id = data.get("satellite_id", "") elif isinstance(data, SatelliteAccepted): self._satellite_id = data.satellite_id else: perror(f"SyncConnection: Ungueltiges SatelliteAccepted: {type(data)}") return False pinfo(f"SyncConnection: Handshake erfolgreich, ID: {self._satellite_id}") return True # ========================================================================= # Empfangs-Loop # ========================================================================= async def _receive_loop(self) -> None: """Empfaengt Nachrichten vom Server.""" while self._running: try: message = await self._read_message() if message is None: if self._running: pwarn("SyncConnection: Verbindung zum Server verloren") asyncio.create_task(self._handle_disconnect("Verbindung verloren")) break await self._handle_message(message) except asyncio.CancelledError: break except Exception as e: if self._running: perror(f"SyncConnection: Empfangsfehler: {e}") asyncio.create_task(self._handle_disconnect(f"Fehler: {e}")) break async def _handle_message(self, message: ProtocolMessage) -> None: """Verarbeitet eine empfangene Nachricht.""" class_name = message.class_name # PING beantworten if class_name == "TRXIPING": await self._send_pong() return # PONG ignorieren if class_name == "TRXIPONG": return # An Callback weiterleiten if self._on_message: try: await self._on_message(message) except Exception as e: perror(f"SyncConnection: Message-Callback Fehler: {e}") async def _keepalive_loop(self) -> None: """Sendet regelmaessige Pings.""" while self._running: try: await asyncio.sleep(30.0) if not self._running: break await self._send_ping() except asyncio.CancelledError: break except Exception: pass async def _handle_disconnect(self, reason: str) -> None: """Behandelt einen Verbindungsabbruch.""" was_connected = self._connected self._connected = False self._running = False await self._close_socket() if was_connected and self._on_disconnected: try: await self._on_disconnected(reason) except Exception as e: perror(f"SyncConnection: on_disconnected Fehler: {e}") # ========================================================================= # Low-Level I/O # ========================================================================= async def _send_message(self, message: object, flags: ProtocolFlags = ProtocolFlags.PICKLE) -> bool: """Sendet eine Nachricht ueber den Command-Socket.""" if not self._writer: return False try: data = self._protocol.serialize(message, flags) self._writer.write(data) await self._writer.drain() return True except Exception as e: perror(f"SyncConnection: Sende-Fehler: {e}") return False async def _send_ping(self) -> None: """Sendet einen Ping.""" if self._writer: try: self._writer.write(COMMAND_PING) await self._writer.drain() except Exception: pass async def _send_pong(self) -> None: """Sendet einen Pong.""" if self._writer: try: self._writer.write(COMMAND_PONG) await self._writer.drain() except Exception: pass async def _read_message(self) -> ProtocolMessage | None: """Liest eine Nachricht vom Command-Socket.""" if not self._reader: return None try: header_size = MAGIC_LENGTH + 2 + 8 + 4 + 16 + 2 # 36 bytes # Prüfe auf Hard-coded Befehle (8 bytes) peek_data = await self._reader.readexactly(8) if self._protocol.is_hard_command(peek_data): return self._protocol.deserialize(peek_data) # Kein Hard-coded Befehl — lese Rest des Headers remaining_header = await self._reader.readexactly(header_size - 8) header = peek_data + remaining_header if header[:MAGIC_LENGTH] != MAGIC: perror(f"SyncConnection: Ungueltige Magic: {header[:MAGIC_LENGTH]}") return None # Parse Klassennamen-Laenge class_name_length = struct.unpack(">H", header[34:36])[0] # Lese Klassenname class_name = await self._reader.readexactly(class_name_length) # Lese Datenlaenge (4 bytes) data_length_bytes = await self._reader.readexactly(4) data_length = struct.unpack(">I", data_length_bytes)[0] # Lese Daten data = await self._reader.readexactly(data_length) if data_length > 0 else b"" # Baue vollstaendige Nachricht zusammen full_message = header + class_name + data_length_bytes + data return self._protocol.deserialize(full_message) except (ConnectionResetError, asyncio.IncompleteReadError): return None except Exception as e: perror(f"SyncConnection: Lese-Fehler: {e}") return None async def _close_socket(self) -> None: """Schliesst den Socket.""" if self._writer: try: self._writer.close() await self._writer.wait_closed() except Exception: pass self._reader = None self._writer = None