HMS MediaEngine Agent
260 lines
10 KiB
Python
260 lines
10 KiB
Python
"""IPC-Verbindung zwischen Control Core und Renderer (PLAN.md §6.2).
|
||
|
||
Server (Renderer-seitig) und Client (Control-Core-seitig) auf 127.0.0.1.
|
||
Verbindungsablauf:
|
||
|
||
1. Client verbindet, sendet hello mit Protokollversion + Capabilities
|
||
2. Server prüft Version, antwortet welcome mit eigenen Capabilities
|
||
3. Server sendet vollständigen state snapshot
|
||
4. danach inkrementelle Deltas mit monotoner Revision
|
||
5. Heartbeat mindestens alle 500 ms in beide Richtungen
|
||
6. Ack für jedes zustandsändernde Command (idempotent über message_id)
|
||
7. Nach Reconnect: Re-Sync (neuer Snapshot), Deltas erst danach akzeptiert
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import time
|
||
from dataclasses import dataclass
|
||
|
||
from hms_protocol.envelope import PROTOCOL_VERSION, Envelope, MessageType
|
||
from hms_protocol.framing import encode_frame, read_frame_async
|
||
|
||
HEARTBEAT_INTERVAL_S = 0.5 # §6.2: mindestens alle 500 ms
|
||
|
||
|
||
@dataclass
|
||
class ProtocolError(Exception):
|
||
"""Protokollverstoß; Verbindung wird getrennt."""
|
||
|
||
code: str
|
||
message: str
|
||
|
||
def __str__(self) -> str:
|
||
return f"{self.code}: {self.message}"
|
||
|
||
|
||
@dataclass
|
||
class HandshakeInfo:
|
||
"""Ergebnis des Handshakes mit Capabilities der Gegenseite."""
|
||
|
||
peer_name: str
|
||
peer_capabilities: dict
|
||
protocol_version: int = PROTOCOL_VERSION
|
||
|
||
|
||
class IpcServer:
|
||
"""Renderer-seitiger IPC-Server. Lauscht ausschließlich auf 127.0.0.1.
|
||
|
||
Liefert dem Renderer:
|
||
- accept(handler) → wartet auf Client, führt Handshake durch
|
||
- receive() → nächste Nachricht (command/event/heartbeat)
|
||
- send(envelope) → Nachricht an Control Core
|
||
"""
|
||
|
||
def __init__(self, port: int = 0, host: str = "127.0.0.1") -> None:
|
||
if host not in ("127.0.0.1", "localhost", "::1"):
|
||
raise ValueError("IPC-Server darf nur auf Loopback lauschen (§6.2)")
|
||
self._host = host
|
||
self._port = port
|
||
self._server: asyncio.Server | None = None
|
||
self._reader: asyncio.StreamReader | None = None
|
||
self._writer: asyncio.StreamWriter | None = None
|
||
self._handler_task: asyncio.Task | None = None
|
||
self._heartbeat_task: asyncio.Task | None = None
|
||
self._last_peer_heartbeat_ns: int = 0
|
||
self.capabilities: dict = {}
|
||
self.peer_name: str = ""
|
||
self.peer_capabilities: dict = {}
|
||
self.name: str = "hms-renderer"
|
||
|
||
@property
|
||
def port(self) -> int:
|
||
if self._server is None:
|
||
return self._port
|
||
return self._server.sockets[0].getsockname()[1] if self._server.sockets else self._port
|
||
|
||
async def start(self) -> int:
|
||
"""Startet den Listener; gibt den tatsächlichen Port zurück."""
|
||
self._server = await asyncio.start_server(self._on_client, self._host, self._port)
|
||
return self.port
|
||
|
||
async def stop(self) -> None:
|
||
if self._heartbeat_task:
|
||
self._heartbeat_task.cancel()
|
||
if self._handler_task:
|
||
self._handler_task.cancel()
|
||
if self._writer:
|
||
self._writer.close()
|
||
if self._server:
|
||
self._server.close()
|
||
await self._server.wait_closed()
|
||
|
||
async def _on_client(
|
||
self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter
|
||
) -> None:
|
||
"""Nimmt genau einen Client an (V1: eine Verbindung)."""
|
||
self._reader = reader
|
||
self._writer = writer
|
||
# auf hello warten
|
||
hello_raw = await read_frame_async(reader)
|
||
action = hello_raw.get("payload", {}).get("action")
|
||
if hello_raw.get("type") != "command" or action != "hello":
|
||
got = hello_raw.get("type")
|
||
raise ProtocolError("EXPECTED_HELLO", f"got {got}")
|
||
peer_caps = hello_raw.get("payload", {}).get("capabilities", {})
|
||
peer_name = hello_raw.get("payload", {}).get("name", "unknown")
|
||
self.peer_name = peer_name
|
||
self.peer_capabilities = peer_caps
|
||
if hello_raw.get("protocol_version") != PROTOCOL_VERSION:
|
||
err = Envelope(
|
||
type=MessageType.ERROR,
|
||
payload={"code": "VERSION_MISMATCH", "expected": PROTOCOL_VERSION},
|
||
)
|
||
writer.write(_encode_envelope(err))
|
||
await writer.drain()
|
||
writer.close()
|
||
raise ProtocolError("VERSION_MISMATCH", "client protocol mismatch")
|
||
# welcome senden
|
||
welcome = Envelope(
|
||
type=MessageType.EVENT,
|
||
payload={"action": "welcome", "name": self.name, "capabilities": self.capabilities},
|
||
)
|
||
writer.write(_encode_envelope(welcome))
|
||
await writer.drain()
|
||
self._last_peer_heartbeat_ns = time.monotonic_ns()
|
||
self._heartbeat_task = asyncio.get_running_loop().create_task(self._send_heartbeats())
|
||
|
||
async def _send_heartbeats(self) -> None:
|
||
while True:
|
||
await asyncio.sleep(HEARTBEAT_INTERVAL_S)
|
||
if self._writer is None:
|
||
return
|
||
hb = Envelope(type=MessageType.HEARTBEAT, payload={"source": self.name})
|
||
self._writer.write(_encode_envelope(hb))
|
||
await self._writer.drain()
|
||
|
||
async def receive(self) -> Envelope | None:
|
||
"""Liest die nächste Nachricht; None bei Verbindungsabbruch."""
|
||
if self._reader is None:
|
||
return None
|
||
try:
|
||
raw = await read_frame_async(self._reader)
|
||
except (asyncio.IncompleteReadError, ConnectionError):
|
||
return None
|
||
if raw.get("type") == "heartbeat":
|
||
self._last_peer_heartbeat_ns = time.monotonic_ns()
|
||
return Envelope.model_validate(raw)
|
||
|
||
async def send(self, envelope: Envelope) -> None:
|
||
if self._writer is None:
|
||
raise ConnectionError("IPC client not connected")
|
||
self._writer.write(_encode_envelope(envelope))
|
||
await self._writer.drain()
|
||
|
||
@property
|
||
def peer_alive(self) -> bool:
|
||
"""True, wenn letzter Peer-Heartbeat < 2× Intervall zurückliegt."""
|
||
if self._last_peer_heartbeat_ns == 0:
|
||
return False
|
||
return (time.monotonic_ns() - self._last_peer_heartbeat_ns) < 2 * HEARTBEAT_INTERVAL_S * 1e9
|
||
|
||
|
||
class IpcClient:
|
||
"""Control-Core-seitiger IPC-Client. Verbindet sich mit 127.0.0.1:port.
|
||
|
||
- connect() → Handshake, gibt HandshakeInfo zurück
|
||
- receive() → nächste Nachricht (snapshot/event/ack/heartbeat)
|
||
- send(envelope) → Nachricht an Renderer
|
||
- Nach Reconnect: connect() erneut → neuer Snapshot (Re-Sync)
|
||
"""
|
||
|
||
def __init__(self, port: int, host: str = "127.0.0.1", name: str = "control-core") -> None:
|
||
if host not in ("127.0.0.1", "localhost", "::1"):
|
||
raise ValueError("IPC-Client darf nur Loopback verbinden (§6.2)")
|
||
self._host = host
|
||
self._port = port
|
||
self._name = name
|
||
self._reader: asyncio.StreamReader | None = None
|
||
self._writer: asyncio.StreamWriter | None = None
|
||
self._heartbeat_task: asyncio.Task | None = None
|
||
self._last_peer_heartbeat_ns: int = 0
|
||
self.capabilities: dict = {}
|
||
|
||
async def connect(self) -> HandshakeInfo:
|
||
"""Baut Verbindung auf, führt Handshake durch."""
|
||
self._reader, self._writer = await asyncio.open_connection(self._host, self._port)
|
||
hello = Envelope(
|
||
type=MessageType.COMMAND,
|
||
payload={"action": "hello", "name": self._name, "capabilities": self.capabilities},
|
||
)
|
||
self._writer.write(_encode_envelope(hello))
|
||
await self._writer.drain()
|
||
raw = await read_frame_async(self._reader)
|
||
if raw.get("type") == "error":
|
||
raise ProtocolError(
|
||
raw.get("payload", {}).get("code", "REMOTE_ERROR"),
|
||
str(raw.get("payload", {})),
|
||
)
|
||
if raw.get("type") != "event" or raw.get("payload", {}).get("action") != "welcome":
|
||
raise ProtocolError("EXPECTED_WELCOME", f"got {raw.get('type')}")
|
||
if raw.get("protocol_version") != PROTOCOL_VERSION:
|
||
sent = raw.get("protocol_version")
|
||
raise ProtocolError("VERSION_MISMATCH", f"server sent version {sent}")
|
||
self._last_peer_heartbeat_ns = time.monotonic_ns()
|
||
self._heartbeat_task = asyncio.get_running_loop().create_task(self._send_heartbeats())
|
||
return HandshakeInfo(
|
||
peer_name=raw.get("payload", {}).get("name", "unknown"),
|
||
peer_capabilities=raw.get("payload", {}).get("capabilities", {}),
|
||
)
|
||
|
||
async def disconnect(self) -> None:
|
||
if self._heartbeat_task:
|
||
self._heartbeat_task.cancel()
|
||
self._heartbeat_task = None
|
||
if self._writer:
|
||
self._writer.close()
|
||
try:
|
||
await self._writer.wait_closed()
|
||
except (ConnectionError, asyncio.CancelledError):
|
||
pass
|
||
self._writer = None
|
||
self._reader = None
|
||
|
||
async def receive(self) -> Envelope | None:
|
||
if self._reader is None:
|
||
return None
|
||
try:
|
||
raw = await read_frame_async(self._reader)
|
||
except (asyncio.IncompleteReadError, ConnectionError):
|
||
return None
|
||
if raw.get("type") == "heartbeat":
|
||
self._last_peer_heartbeat_ns = time.monotonic_ns()
|
||
return Envelope.model_validate(raw)
|
||
|
||
async def send(self, envelope: Envelope) -> None:
|
||
if self._writer is None:
|
||
raise ConnectionError("IPC not connected")
|
||
self._writer.write(_encode_envelope(envelope))
|
||
await self._writer.drain()
|
||
|
||
async def _send_heartbeats(self) -> None:
|
||
while True:
|
||
await asyncio.sleep(HEARTBEAT_INTERVAL_S)
|
||
if self._writer is None:
|
||
return
|
||
hb = Envelope(type=MessageType.HEARTBEAT, payload={"source": self._name})
|
||
self._writer.write(_encode_envelope(hb))
|
||
await self._writer.drain()
|
||
|
||
@property
|
||
def peer_alive(self) -> bool:
|
||
if self._last_peer_heartbeat_ns == 0:
|
||
return False
|
||
return (time.monotonic_ns() - self._last_peer_heartbeat_ns) < 2 * HEARTBEAT_INTERVAL_S * 1e9
|
||
|
||
|
||
def _encode_envelope(envelope: Envelope) -> bytes:
|
||
return encode_frame(envelope.model_dump(mode="json"))
|