Phase 1: IPC-Verbindungsschicht + Build-first-Strategie (ADR-0008)
- ADR-0008: Build-first, gesammelte Windows-Validierung (Auftraggeber-Freigabe) - ADR-0004 vorläufig: Rust auf GStreamer-D3D11-Elementpfad - ADR-0005 vorläufig: Nuitka Onefolder - hms_protocol: IpcServer/IpcClient mit Handshake (hello/welcome), Capabilities-Austausch, Heartbeat 500 ms bidirektional, Snapshot/Delta, Loopback-only-Enforcement, Re-Sync nach Reconnect, Version-Mismatch-Fehler - 9 Integrationstests: echter TCP-Loopback, kein Mock
This commit is contained in:
HMS MediaEngine Agent
@@ -0,0 +1,252 @@
|
||||
"""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.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)
|
||||
if hello_raw.get("type") != "command" or hello_raw.get("payload", {}).get("action") != "hello":
|
||||
raise ProtocolError("EXPECTED_HELLO", f"got {hello_raw.get('type')}")
|
||||
peer_caps = hello_raw.get("payload", {}).get("capabilities", {})
|
||||
peer_name = hello_raw.get("payload", {}).get("name", "unknown")
|
||||
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:
|
||||
raise ProtocolError("VERSION_MISMATCH", f"server sent version {raw.get('protocol_version')}")
|
||||
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"))
|
||||
Reference in New Issue
Block a user