Source code for edri.switch.status

"""Thin, read-only HTTP status endpoint for the Switch.

The Switch lives outside EDRI's router/manager/event topology — it only forwards
raw bytes between connected routers over a binary TCP protocol, so the API
``@api`` health-check machinery cannot be reused here. This module exposes the
same *kind* of operational snapshot (connections, backpressure, process vitals)
through a minimal stdlib HTTP server running in a daemon thread.

It is deliberately tiny: one route (``GET /status``), JSON only, zero new
dependencies, and read-only — it never mutates the structures it reads, so it
needs no locking.
"""
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from json import dumps
from logging import Logger, getLogger
from threading import Thread
from typing import TYPE_CHECKING, Any

if TYPE_CHECKING:
    from edri.switch.switch import Switch


class _StatusHandler(BaseHTTPRequestHandler):
    server: "_StatusServer"

    def do_GET(self) -> None:
        if self.path.rstrip("/") not in ("", "/status"):
            self.send_response(404)
            self.end_headers()
            return
        try:
            body = dumps(self.server.switch.status()).encode()
        except Exception:  # never let a status probe take the Switch down
            self.send_response(500)
            self.end_headers()
            return
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, *args: Any) -> None:
        # Silence the default per-request stderr logging; the Switch has its own.
        pass


class _StatusServer(ThreadingHTTPServer):
    daemon_threads = True

    def __init__(self, address: tuple[str, int], switch: "Switch") -> None:
        super().__init__(address, _StatusHandler)
        self.switch = switch




def serve_status(switch: "Switch", host: str, port: int, logger: Logger | None = None) -> _StatusServer:
    """Start the status HTTP server in a daemon thread and return it.

    :param switch: the running :class:`~edri.switch.switch.Switch` to introspect.
    :param host: bind host.
    :param port: bind port.
    :param logger: optional logger for the startup line.
    :return: the running server instance.
    """
    log = logger or getLogger(__name__)
    server = _StatusServer((host, port), switch)
    Thread(target=server.serve_forever, name="switch-status", daemon=True).start()
    log.info("Switch status endpoint on http://%s:%s/status", host, port)
    return server