# -*- coding: utf-8 -*- """ Connection-Wrapper für Pool-Verwaltung. Kapselt asyncio-Verbindungen mit Metadaten und Lebenszyklus-Tracking. """ import asyncio import time from dataclasses import dataclass, field from datetime import datetime from enum import IntEnum, auto from typing import Any, TYPE_CHECKING if TYPE_CHECKING: from asyncio import StreamReader, StreamWriter class ConnectionState(IntEnum): """Zustand einer Verbindung.""" IDLE = auto() # Verfügbar im Pool IN_USE = auto() # Wird verwendet CONNECTING = auto() # Verbindungsaufbau CLOSING = auto() # Wird geschlossen CLOSED = auto() # Geschlossen ERROR = auto() # Fehlerhaft @dataclass class ConnectionInfo: """ Metadaten einer Verbindung. Attributes: host: Zielhost port: Zielport created_at: Erstellungszeitpunkt last_used_at: Letzter Verwendungszeitpunkt use_count: Anzahl Verwendungen bytes_sent: Gesendete Bytes bytes_received: Empfangene Bytes errors: Anzahl Fehler """ host: str port: int created_at: float = field(default_factory=time.monotonic) last_used_at: float = field(default_factory=time.monotonic) use_count: int = 0 bytes_sent: int = 0 bytes_received: int = 0 errors: int = 0 @property def age(self) -> float: """Alter der Verbindung in Sekunden.""" return time.monotonic() - self.created_at @property def idle_time(self) -> float: """Zeit seit letzter Verwendung in Sekunden.""" return time.monotonic() - self.last_used_at class Connection: """ Wrapper für eine gepoolte Verbindung. Verwaltet den Lebenszyklus und Zustand einer Netzwerkverbindung. Example: async with pool.acquire() as conn: await conn.write(b"Hello") data = await conn.read(1024) """ def __init__( self, host: str, port: int, reader: "StreamReader | None" = None, writer: "StreamWriter | None" = None ) -> None: """ Initialisiert eine Connection. Args: host: Zielhost port: Zielport reader: Optional vorhandener StreamReader writer: Optional vorhandener StreamWriter """ self._reader = reader self._writer = writer self._state = ConnectionState.IDLE if reader else ConnectionState.CLOSED self._info = ConnectionInfo(host=host, port=port) self._lock = asyncio.Lock() self._id: str = f"{host}:{port}:{id(self)}" @property def id(self) -> str: """Eindeutige Verbindungs-ID.""" return self._id @property def state(self) -> ConnectionState: """Aktueller Zustand.""" return self._state @property def info(self) -> ConnectionInfo: """Verbindungsmetadaten.""" return self._info @property def is_usable(self) -> bool: """Prüft, ob die Verbindung verwendbar ist.""" return self._state == ConnectionState.IDLE and self._writer is not None @property def host(self) -> str: """Zielhost.""" return self._info.host @property def port(self) -> int: """Zielport.""" return self._info.port async def connect(self, ssl_context: Any = None, timeout: float = 10.0) -> bool: """ Stellt die Verbindung her. Args: ssl_context: Optionaler SSL-Kontext timeout: Verbindungs-Timeout Returns: True bei Erfolg """ if self._state not in (ConnectionState.CLOSED, ConnectionState.ERROR): return self._state in (ConnectionState.IDLE, ConnectionState.IN_USE) self._state = ConnectionState.CONNECTING try: self._reader, self._writer = await asyncio.wait_for( asyncio.open_connection( self._info.host, self._info.port, ssl=ssl_context ), timeout=timeout ) self._state = ConnectionState.IDLE self._info.created_at = time.monotonic() return True except Exception as e: self._state = ConnectionState.ERROR self._info.errors += 1 return False async def close(self) -> None: """Schließt die Verbindung.""" if self._state == ConnectionState.CLOSED: return self._state = ConnectionState.CLOSING if self._writer is not None: try: self._writer.close() await self._writer.wait_closed() except Exception: pass self._reader = None self._writer = None self._state = ConnectionState.CLOSED def acquire(self) -> None: """Markiert die Verbindung als in Verwendung.""" if self._state == ConnectionState.IDLE: self._state = ConnectionState.IN_USE self._info.use_count += 1 self._info.last_used_at = time.monotonic() def release(self) -> None: """Gibt die Verbindung zurück in den Pool.""" if self._state == ConnectionState.IN_USE: self._state = ConnectionState.IDLE self._info.last_used_at = time.monotonic() async def write(self, data: bytes) -> None: """ Sendet Daten über die Verbindung. Args: data: Zu sendende Daten Raises: ConnectionError: Wenn nicht verbunden """ if self._writer is None: raise ConnectionError("Verbindung nicht hergestellt") self._writer.write(data) await self._writer.drain() self._info.bytes_sent += len(data) async def read(self, n: int = -1) -> bytes: """ Liest Daten von der Verbindung. Args: n: Anzahl zu lesender Bytes (-1 = bis EOF) Returns: Gelesene Daten Raises: ConnectionError: Wenn nicht verbunden """ if self._reader is None: raise ConnectionError("Verbindung nicht hergestellt") if n < 0: data = await self._reader.read(n) else: data = await self._reader.read(n) self._info.bytes_received += len(data) return data async def readline(self) -> bytes: """ Liest eine Zeile von der Verbindung. Returns: Gelesene Zeile """ if self._reader is None: raise ConnectionError("Verbindung nicht hergestellt") data = await self._reader.readline() self._info.bytes_received += len(data) return data async def readexactly(self, n: int) -> bytes: """ Liest exakt n Bytes. Args: n: Anzahl Bytes Returns: Gelesene Daten """ if self._reader is None: raise ConnectionError("Verbindung nicht hergestellt") data = await self._reader.readexactly(n) self._info.bytes_received += len(data) return data def at_eof(self) -> bool: """Prüft, ob Ende des Streams erreicht ist.""" if self._reader is None: return True return self._reader.at_eof() async def __aenter__(self) -> "Connection": """Context-Manager Entry.""" self.acquire() return self async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None: """Context-Manager Exit.""" if exc_type is not None: self._info.errors += 1 self._state = ConnectionState.ERROR else: self.release() def to_dict(self) -> dict[str, Any]: """Konvertiert zu Dictionary.""" return { "id": self._id, "host": self._info.host, "port": self._info.port, "state": self._state.name, "age": self._info.age, "idle_time": self._info.idle_time, "use_count": self._info.use_count, "bytes_sent": self._info.bytes_sent, "bytes_received": self._info.bytes_received, "errors": self._info.errors, }