orxtra v0.13.0 /transport.src.orxtra.transport._liveness
Updated
On this page

LivenessMonitor: tracks time since the last streaming SSE event and reports healthy/warning/stuck against configurable thresholds, so callers can detect a stalled API stream.

#transport.src.orxtra.transport._liveness

#transport.src.orxtra.transport._liveness

#LivenessMonitor

Monitors streaming response liveness by tracking time between events.

Call record_event() each time an SSE event arrives. Call check() to get the current status: "healthy", "warning", "stuck", or None if no event has been recorded yet.

#record_event

python
def record_event(self) -> None

Call on each SSE event received.

#check

python
def check(self) -> str | None

Returns 'healthy', 'warning', 'stuck', or None (no events yet).

#elapsed

python
def elapsed(self) -> float | None

Seconds since the last event, or None if no event recorded.

#reset

python
def reset(self) -> None

Clear recorded state.

Search