73 lines
2.6 KiB
Python
73 lines
2.6 KiB
Python
"""Synchronous Server-Sent Events consumer for live routing decisions.
|
|
|
|
Runs in its own background thread (a network reader must never block the
|
|
Textual event loop) and reconnects after transient failures. It is decoupled
|
|
from Textual on purpose: the callback it receives is expected to marshal
|
|
itself onto the UI thread (the app passes ``App.call_from_thread``), so this
|
|
module never needs to import ``textual``.
|
|
|
|
The SSE body is line-based: comments (``:heartbeat``) are ignored and only
|
|
``data: <json>`` lines are handed to the callback.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import threading
|
|
from typing import Any, Callable
|
|
|
|
import requests
|
|
|
|
import logs
|
|
|
|
SSE_PATH = "/events/decisions"
|
|
STREAM_TIMEOUT = 15
|
|
RECONNECT_SECONDS = 5
|
|
|
|
|
|
class DecisionStream(threading.Thread):
|
|
"""Background thread that yields routing decisions from the SSE endpoint."""
|
|
|
|
def __init__(
|
|
self,
|
|
base_url: str,
|
|
callback: Callable[[dict[str, Any]], None],
|
|
*,
|
|
timeout: float = STREAM_TIMEOUT,
|
|
reconnect_seconds: float = RECONNECT_SECONDS,
|
|
) -> None:
|
|
super().__init__(daemon=True)
|
|
self.url = base_url.rstrip("/") + SSE_PATH
|
|
self.callback = callback
|
|
self.timeout = timeout
|
|
self.reconnect_seconds = reconnect_seconds
|
|
self._stopped = threading.Event()
|
|
|
|
def run(self) -> None:
|
|
while not self._stopped.is_set():
|
|
try:
|
|
with requests.get(
|
|
self.url, stream=True, timeout=self.timeout * 2
|
|
) as resp:
|
|
resp.raise_for_status()
|
|
for line in resp.iter_lines(decode_unicode=True):
|
|
if self._stopped.is_set():
|
|
break
|
|
if line.startswith("data: "):
|
|
self.callback(json.loads(line[len("data: ") :]))
|
|
except RuntimeError as exc:
|
|
# callback failure (call_from_thread after the Textual app
|
|
# event loop has closed). Log it; a stopped thread exits on
|
|
# the guard below without scheduling a pointless reconnect.
|
|
logs.warning("SSE callback failed", message=str(exc))
|
|
except (requests.RequestException, ValueError) as exc:
|
|
# Transient network / parse error: back off and retry. A
|
|
# dashboard must survive a router restart or a dropped SSE.
|
|
logs.warning("SSE connection error", message=str(exc))
|
|
if self._stopped.is_set():
|
|
break
|
|
self._stopped.wait(self.reconnect_seconds)
|
|
|
|
def stop(self) -> None:
|
|
self._stopped.set()
|