Trixy7/trixy_core/network/pool/connection.py

# -*- 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,
        }