diff --git a/CHANGELOG.md b/CHANGELOG.md index 9842de4..36a0bc1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,4 +1,47 @@ # CHANGELOG + +## Version 0.63.3 (July 2026) + +**Released**: July 14, 2026 + +This patch release improves WebSocket subscription reliability, isolates user callbacks from the receive loop, and adds stronger reconnect, rate-limit, and connection-health safeguards. (#26) + +--- + +### 1. NEW FEATURES + +#### **Subscription Acknowledgement Watchdog** +WebSocket clients now track `channel_subscribed` acknowledgements for every requested channel. The configurable `subscription_watchdog_timeout` closes an incomplete connection and reports a `TimeoutError` through `on_error`; when `auto_reconnect=True`, the client reconnects and retries the subscriptions. Explicit `subscribe_failed` events are also surfaced through `on_error` before reconnecting. + +#### **Ping and Pong Hooks** +Added `on_ping()` and `on_pong()` callbacks for applications that need visibility into WebSocket keepalive frames. + +--- + +### 2. IMPROVEMENTS + +#### **Non-Blocking Callback Delivery** +Incoming frames are now placed on a dedicated callback-worker queue so slow user callbacks no longer block the WebSocket receive loop. Pending callback messages are drained during shutdown, and `queue_maxsize` bounds both generator and callback delivery queues. + +#### **Reconnect and Rate-Limit Handling** +- HTTP 429 handshake failures use the configurable `rate_limit_backoff` as a minimum reconnect delay. +- Exponential reconnect state is reset only after all requested subscriptions are acknowledged, so persistent subscription failures respect `max_reconnect_attempts` and continue increasing the backoff. +- `SessionWithUrl` connections with no managed channels reset reconnect state when the transport opens. + +#### **Connection Health and Observability** +- Updated keepalive defaults to `ping_interval=60` and a derived `ping_timeout` of up to 45 seconds. +- Added periodic, thread-safe throughput and queue-depth logging through `throughput_log_interval`. +- Duplicate channels are removed while preserving order, high channel counts produce a warning, and connections above the 2,000-channel limit are rejected. + +--- + +### 3. BUG FIXES + +#### **Subscription Completion Accounting** +Unexpected `channel_subscribed` acknowledgements are now ignored before updating subscription progress, preventing them from cancelling the watchdog while requested channels are still missing. + +--- + ## Version 0.63.2 (July 2026) **Released**: July 14, 2026 diff --git a/README.md b/README.md index e488ea8..e3d80ae 100644 --- a/README.md +++ b/README.md @@ -583,19 +583,22 @@ All channel classes accept the following optional keyword arguments: | Parameter | Type | Default | Description | |-----------|------|---------|-------------| -| `ping_interval` | `int` | `30` | Seconds between automatic ping frames. Set to `0` to disable pings. | -| `ping_timeout` | `int` | `10` | Seconds to wait for a pong response before treating the connection as dead. | +| `ping_interval` | `int` | `60` | Seconds between automatic ping frames. Set to `0` to disable pings. | +| `ping_timeout` | `int \| None` | `None` | Seconds to wait for a pong response before treating the connection as dead. Defaults to `min(45, ping_interval - 1)` when pings are enabled, or `45` when `ping_interval=0` (unused since pings are disabled). When `ping_interval > 0`, this must be lower than `ping_interval`. | | `auto_reconnect` | `bool` | `False` | Automatically reconnect on transient failures using exponential backoff. | | `max_reconnect_attempts` | `int` | `5` | Maximum number of reconnect attempts before giving up. | -| `reconnect_backoff` | `float` | `2.0` | Base backoff delay in seconds. Doubles after each failed attempt (2s, 4s, 8s, ...). Resets on successful reconnection. | -| `queue_maxsize` | `int` | `0` | Maximum messages buffered in the internal queue for `receive()`. `0` means unbounded. When set, incoming messages are dropped with a warning when the queue is full, preventing memory growth on high-frequency streams. | +| `reconnect_backoff` | `float` | `2.0` | Base backoff delay in seconds. Doubles after each failed attempt (2s, 4s, 8s, ...). Resets once the connection is fully established and all requested subscriptions are acknowledged. | +| `queue_maxsize` | `int` | `0` | Maximum messages buffered in the internal queues used for both `receive()` and callback delivery. `0` means unbounded. When set, incoming messages are dropped with a warning when either queue is full, preventing memory growth on high-frequency streams. | +| `subscription_watchdog_timeout` | `float` | `10.0` | Maximum time to wait for all `channel_subscribed` acknowledgements after connect. On timeout, the error is reported to `on_error` and the connection is closed; with `auto_reconnect=True` this triggers a clean reconnect. | +| `rate_limit_backoff` | `float` | `30.0` | Minimum reconnect delay after a 429 rate-limit response. | +| `throughput_log_interval` | `int` | `100` | Logs queue depth and processed counts every N messages. Set to `0` to disable periodic throughput logs. | ```python ws = mistapi.websockets.sites.DeviceStatsEvents( apisession, site_ids=[""], ping_interval=60, # ping every 60 s - ping_timeout=20, # wait up to 20 s for pong + ping_timeout=45, # wait up to 45 s for pong auto_reconnect=True, # reconnect on transient failures ) ws.connect() @@ -609,6 +612,8 @@ ws.connect() | `ws.on_message(cb)` | `cb(data: dict)` | Register callback for incoming messages. Mutually exclusive with `receive()`. | | `ws.on_error(cb)` | `cb(error: Exception)` | Register callback for WebSocket errors | | `ws.on_close(cb)` | `cb(code: int \| None, msg: str \| None)` | Register callback for connection close. Safe to call `connect()` from within. | +| `ws.on_ping(cb)` | `cb(message: str \| bytes \| None)` | Register callback for received ping frames. | +| `ws.on_pong(cb)` | `cb(message: str \| bytes \| None)` | Register callback for received pong frames. | | `ws.connect(run_in_background)` | | Open the connection. `True` (default) runs in a daemon thread; `False` blocks. | | `ws.disconnect(wait, timeout)` | | Close the connection. `wait=True` blocks until the background thread finishes. | | `ws.receive()` | `-> Generator[dict]` | Blocking generator yielding messages. Mutually exclusive with `on_message`. | diff --git a/pyproject.toml b/pyproject.toml index f854d13..d36784b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "mistapi" -version = "0.63.2" +version = "0.63.3" authors = [{ name = "Thomas Munzer", email = "tmunzer@juniper.net" }] description = "Python package to simplify the Mist System APIs usage" keywords = ["Mist", "Juniper", "API"] diff --git a/src/mistapi/__version.py b/src/mistapi/__version.py index 71462fc..88d01e1 100644 --- a/src/mistapi/__version.py +++ b/src/mistapi/__version.py @@ -1,2 +1,2 @@ -__version__ = "0.63.2" +__version__ = "0.63.3" __author__ = "Thomas Munzer " diff --git a/src/mistapi/api/v1/sites/sle.py b/src/mistapi/api/v1/sites/sle.py index 7e478c8..2d044aa 100644 --- a/src/mistapi/api/v1/sites/sle.py +++ b/src/mistapi/api/v1/sites/sle.py @@ -18,7 +18,7 @@ @deprecation.deprecated( deprecated_in="0.59.2", removed_in="0.65.0", - current_version="0.63.2", + current_version="0.63.3", details="function replaced with getSiteSleClassifierSummaryTrend", ) def getSiteSleClassifierDetails( @@ -764,7 +764,7 @@ def listSiteSleImpactedWirelessClients( @deprecation.deprecated( deprecated_in="0.59.2", removed_in="0.65.0", - current_version="0.63.2", + current_version="0.63.3", details="function replaced with getSiteSleSummaryTrend", ) def getSiteSleSummary( diff --git a/src/mistapi/websockets/__ws_client.py b/src/mistapi/websockets/__ws_client.py index e351c5d..2422894 100644 --- a/src/mistapi/websockets/__ws_client.py +++ b/src/mistapi/websockets/__ws_client.py @@ -19,7 +19,7 @@ import ssl import threading from collections.abc import Callable, Generator -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any import websocket @@ -48,6 +48,17 @@ def filter(self, record: logging.LogRecord) -> bool: from mistapi import APISession +# Channel limits from the Mist WebSocket documentation: +# https://www.juniper.net/documentation/us/en/software/mist/api/http/guides/websockets/rate-limits +MAX_CHANNELS_PER_CONNECTION = 2000 +HIGH_CHANNEL_COUNT_WARNING = 1500 + +# Default pong timeout. When ping_timeout is not supplied it is derived as +# min(DEFAULT_PING_TIMEOUT, ping_interval - 1), or DEFAULT_PING_TIMEOUT when +# ping_interval is 0 (pings disabled, so the value is unused). +DEFAULT_PING_TIMEOUT = 45 + + class _MistWebsocket: """ Base class for Mist API WebSocket channels. @@ -64,14 +75,39 @@ def __init__( self, mist_session: "APISession", channels: list[str], - ping_interval: int = 30, - ping_timeout: int = 10, + ping_interval: int = 60, + ping_timeout: int | None = None, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, max_reconnect_backoff: float | None = None, queue_maxsize: int = 0, + subscription_watchdog_timeout: float = 10.0, + rate_limit_backoff: float = 30.0, + throughput_log_interval: int = 100, ) -> None: + if ping_interval < 0: + raise ValueError("ping_interval must be >= 0") + if ping_timeout is None: + # Derive a valid pong timeout so callers who only set + # ping_interval keep working (websocket-client requires + # ping_timeout < ping_interval). + ping_timeout = ( + min(DEFAULT_PING_TIMEOUT, ping_interval - 1) + if ping_interval > 0 + else DEFAULT_PING_TIMEOUT + ) + if ping_timeout < 1: + raise ValueError( + "ping_interval must be >= 2 to leave room for the pong " + "timeout, or set ping_interval=0 to disable pings" + ) + if ping_timeout <= 0: + raise ValueError("ping_timeout must be > 0") + if ping_interval and ping_interval <= ping_timeout: + raise ValueError( + "ping_interval must be greater than ping_timeout when enabled" + ) if max_reconnect_attempts < 0: raise ValueError("max_reconnect_attempts must be >= 0 (0 = unlimited)") if reconnect_backoff <= 0: @@ -80,32 +116,76 @@ def __init__( raise ValueError("max_reconnect_backoff must be > 0") if queue_maxsize < 0: raise ValueError("queue_maxsize must be >= 0") + if subscription_watchdog_timeout <= 0: + raise ValueError("subscription_watchdog_timeout must be > 0") + if rate_limit_backoff <= 0: + raise ValueError("rate_limit_backoff must be > 0") + if throughput_log_interval < 0: + raise ValueError("throughput_log_interval must be >= 0") + + deduped_channels = list(dict.fromkeys(channels)) + if len(deduped_channels) != len(channels): + logger.warning( + "Duplicate channels detected; using %d unique channels instead of %d", + len(deduped_channels), + len(channels), + ) + if len(deduped_channels) > MAX_CHANNELS_PER_CONNECTION: + raise ValueError( + f"Too many channels ({len(deduped_channels)}). " + f"Mist supports up to {MAX_CHANNELS_PER_CONNECTION} channels per connection" + ) + if len(deduped_channels) >= HIGH_CHANNEL_COUNT_WARNING: + logger.warning( + "High channel count (%d). Consider spreading subscriptions over multiple " + "WebSocket connections to reduce message backlog risk.", + len(deduped_channels), + ) self._mist_session = mist_session - self._channels = channels + self._channels = deduped_channels + self._expected_channels = set(deduped_channels) self._ping_interval = ping_interval self._ping_timeout = ping_timeout self._auto_reconnect = auto_reconnect self._max_reconnect_attempts = max_reconnect_attempts self._reconnect_backoff = reconnect_backoff self._max_reconnect_backoff = max_reconnect_backoff + self._subscription_watchdog_timeout = subscription_watchdog_timeout + self._rate_limit_backoff = rate_limit_backoff + self._throughput_log_interval = throughput_log_interval self._lock = threading.Lock() + self._subscription_lock = threading.Lock() + self._metrics_lock = threading.Lock() self._ws: websocket.WebSocketApp | None = None self._thread: threading.Thread | None = None + self._callback_thread: threading.Thread | None = None + self._subscription_watchdog: threading.Timer | None = None self._queue: queue.Queue[dict | None] = queue.Queue(maxsize=queue_maxsize) + self._callback_queue: queue.Queue[dict | None] = queue.Queue( + maxsize=queue_maxsize + ) self._connected = ( threading.Event() ) # tracks whether the WebSocket connection is currently open self._user_disconnect = threading.Event() + self._callback_stop = threading.Event() self._finished = threading.Event() self._finished.set() # not running initially self._reconnect_attempts = 0 self._last_close_code: int | None = None self._last_close_msg: str | None = None + self._last_http_status: int | None = None + self._subscribed_channels: set[str] = set() + self._messages_received = 0 + self._messages_dropped = 0 + self._messages_processed = 0 self._on_message_cb: Callable[[dict], None] | None = None self._on_error_cb: Callable[[Exception], None] | None = None self._on_open_cb: Callable[[], None] | None = None self._on_close_cb: Callable[[int | None, str | None], None] | None = None + self._on_ping_cb: Callable[[str | bytes | None], None] | None = None + self._on_pong_cb: Callable[[str | bytes | None], None] | None = None # ------------------------------------------------------------------ # Auth / URL helpers @@ -184,10 +264,229 @@ def on_close(self, callback: Callable[[int | None, str | None], None]) -> None: """Register a callback invoked when the connection closes.""" self._on_close_cb = callback + def on_ping(self, callback: Callable[[str | bytes | None], None]) -> None: + """Register a callback invoked when a ping frame is received.""" + self._on_ping_cb = callback + + def on_pong(self, callback: Callable[[str | bytes | None], None]) -> None: + """Register a callback invoked when a pong frame is received.""" + self._on_pong_cb = callback + + # ------------------------------------------------------------------ + # Internal helpers + + @staticmethod + def _extract_status_code(error: Exception) -> int | None: + status_code = getattr(error, "status_code", None) + if isinstance(status_code, int): + return status_code + response = getattr(error, "response", None) + if response is not None: + response_code = getattr(response, "status_code", None) + if isinstance(response_code, int): + return response_code + return None + + def _drain_queue(self, target_queue: queue.Queue[Any]) -> None: + while not target_queue.empty(): + try: + target_queue.get_nowait() + except queue.Empty: + break + + def _start_callback_worker(self) -> None: + if self._callback_stop.is_set(): + # Disconnecting/stopped: don't resurrect the worker for a late + # message. connect() clears the flag before starting a new run. + return + if self._callback_thread is not None and self._callback_thread.is_alive(): + return + self._callback_thread = threading.Thread( + target=self._run_callback_worker, daemon=True + ) + self._callback_thread.start() + + def _run_callback_worker(self) -> None: + # On stop, keep draining so every message received before the stop + # (i.e. before the None sentinel) is still delivered to the callback. + while True: + try: + item = self._callback_queue.get(timeout=1) + except queue.Empty: + if self._callback_stop.is_set(): + break # stop requested and queue fully drained + if self._finished.is_set() and self._callback_queue.empty(): + break + continue + if item is None: + break # end-of-stream sentinel; everything before it was delivered + callback = self._on_message_cb + if callback is None: + continue + try: + callback(item) + except Exception: + logger.exception("on_message callback raised") + with self._metrics_lock: + self._messages_processed += 1 + messages_processed = self._messages_processed + messages_dropped = self._messages_dropped + if ( + self._throughput_log_interval + and messages_processed % self._throughput_log_interval == 0 + ): + logger.info( + "WebSocket callback worker processed %d messages. " + "Callback queue size=%d dropped=%d", + messages_processed, + self._callback_queue.qsize(), + messages_dropped, + ) + + def _cancel_subscription_watchdog(self) -> None: + with self._lock: + timer = self._subscription_watchdog + self._subscription_watchdog = None + if timer is not None: + timer.cancel() + + def _arm_subscription_watchdog(self, ws: websocket.WebSocketApp) -> None: + if not self._expected_channels: + return + + def _watchdog_expired() -> None: + if self._user_disconnect.is_set(): + return + with self._lock: + current_ws = self._ws + if ws is not current_ws: + return + with self._subscription_lock: + missing = sorted(self._expected_channels - self._subscribed_channels) + if not missing: + return + preview = ", ".join(missing[:5]) + if len(missing) > 5: + preview = f"{preview}, ..." + self._last_close_code = 1008 + self._last_close_msg = ( + f"subscription watchdog timeout: missing {len(missing)} channels" + ) + # Surface through on_error so callers without auto_reconnect + # still get an actionable signal (the close alone reconnects + # only when auto_reconnect is enabled). + self._handle_error( + ws, + TimeoutError( + f"subscription watchdog timeout after " + f"{self._subscription_watchdog_timeout:.1f}s: received " + f"{len(self._expected_channels) - len(missing)}/" + f"{len(self._expected_channels)} subscriptions. " + f"Missing: {preview}" + ), + ) + ws.close() + + timer = threading.Timer(self._subscription_watchdog_timeout, _watchdog_expired) + timer.daemon = True + self._cancel_subscription_watchdog() + with self._lock: + self._subscription_watchdog = timer + timer.start() + + def _process_subscription_event( + self, ws: websocket.WebSocketApp, data: dict + ) -> None: + event = data.get("event") + channel = data.get("channel") + if not isinstance(channel, str): + channel = None + + if event == "channel_subscribed" and channel: + if channel not in self._expected_channels: + logger.warning( + "Received channel_subscribed for unexpected channel: %s", channel + ) + return + with self._subscription_lock: + self._subscribed_channels.add(channel) + subscribed_count = len(self._subscribed_channels) + expected_count = len(self._expected_channels) + logger.debug( + "Channel subscribed (%d/%d): %s", + subscribed_count, + expected_count, + channel, + ) + if expected_count and ( + subscribed_count == 1 + or subscribed_count % 100 == 0 + or subscribed_count >= expected_count + ): + logger.info( + "Subscription progress: received %d/%d channel acknowledgements", + subscribed_count, + expected_count, + ) + if subscribed_count == expected_count: + self._cancel_subscription_watchdog() + self._reconnect_attempts = 0 + self._last_close_code = None + self._last_close_msg = None + logger.info("All requested channels subscribed (%d)", expected_count) + return + + if event == "subscribe_failed": + detail = data.get("detail") + self._last_close_code = 1008 + self._last_close_msg = f"subscribe_failed channel={channel} detail={detail}" + self._cancel_subscription_watchdog() + # Surface through on_error so callers without auto_reconnect + # still get an actionable signal before the connection closes. + self._handle_error( + ws, + ConnectionError(f"subscription failed for channel {channel}: {detail}"), + ) + ws.close() + + def _enqueue_message(self, message: dict, to_callback_queue: bool) -> None: + target_queue = self._callback_queue if to_callback_queue else self._queue + queue_name = "callback" if to_callback_queue else "receive" + with self._metrics_lock: + self._messages_received += 1 + messages_received = self._messages_received + try: + target_queue.put_nowait(message) + except queue.Full: + with self._metrics_lock: + self._messages_dropped += 1 + logger.warning("%s queue full; dropping message", queue_name.capitalize()) + return + if ( + self._throughput_log_interval + and messages_received % self._throughput_log_interval == 0 + ): + with self._metrics_lock: + messages_dropped = self._messages_dropped + logger.info( + "WebSocket received %d messages. %s queue size=%d dropped=%d", + messages_received, + queue_name.capitalize(), + target_queue.qsize(), + messages_dropped, + ) + # ------------------------------------------------------------------ # Internal WebSocketApp handlers def _handle_open(self, ws: websocket.WebSocketApp) -> None: + logger.info( + "WebSocket opened. Requesting %d channel subscription(s)", + len(self._channels), + ) + self._last_http_status = None + with self._subscription_lock: + self._subscribed_channels.clear() try: for channel in self._channels: ws.send(json.dumps({"subscribe": channel})) @@ -195,9 +494,15 @@ def _handle_open(self, ws: websocket.WebSocketApp) -> None: logger.error("Subscription send failed: %s", exc) ws.close() return - self._reconnect_attempts = 0 - self._last_close_code = None - self._last_close_msg = None + if self._expected_channels: + self._arm_subscription_watchdog(ws) + else: + # Clients without managed subscriptions (for example + # SessionWithUrl) are fully established as soon as the transport + # opens because no channel acknowledgements will arrive. + self._reconnect_attempts = 0 + self._last_close_code = None + self._last_close_msg = None self._connected.set() if self._on_open_cb: try: @@ -212,33 +517,72 @@ def _handle_message(self, ws: websocket.WebSocketApp, message: str | bytes) -> N data = json.loads(message) except (json.JSONDecodeError, TypeError): data = {"raw": message} + if not isinstance(data, dict): + data = {"data": data} + + self._process_subscription_event(ws, data) + if self._on_message_cb: - try: - self._on_message_cb(data) - except Exception: - logger.exception("on_message callback raised") - else: - try: - self._queue.put_nowait(data) - except queue.Full: - logger.warning("Receive queue full; dropping message") + self._start_callback_worker() + self._enqueue_message(data, to_callback_queue=True) + return + + self._enqueue_message(data, to_callback_queue=False) - def _handle_error(self, ws: websocket.WebSocketApp, error: Exception) -> None: + def _handle_error(self, _ws: websocket.WebSocketApp, error: Exception) -> None: + status_code = self._extract_status_code(error) + self._last_http_status = status_code + if status_code == 429: + logger.warning( + "WebSocket received HTTP 429 (rate limit). " + "Reconnect backoff will be raised to at least %.1fs", + self._rate_limit_backoff, + ) + else: + logger.error("WebSocket error: %s", error) if self._on_error_cb: try: self._on_error_cb(error) except Exception: logger.exception("on_error callback raised") + def _handle_ping( + self, _ws: websocket.WebSocketApp, message: str | bytes | None + ) -> None: + logger.debug("WebSocket ping received") + if self._on_ping_cb: + try: + self._on_ping_cb(message) + except Exception: + logger.exception("on_ping callback raised") + + def _handle_pong( + self, _ws: websocket.WebSocketApp, message: str | bytes | None + ) -> None: + logger.debug("WebSocket pong received") + if self._on_pong_cb: + try: + self._on_pong_cb(message) + except Exception: + logger.exception("on_pong callback raised") + def _handle_close( self, - ws: websocket.WebSocketApp, + _ws: websocket.WebSocketApp, close_status_code: int | None, close_msg: str | None, ) -> None: self._connected.clear() - self._last_close_code = close_status_code - self._last_close_msg = close_msg + self._cancel_subscription_watchdog() + if close_status_code is not None: + self._last_close_code = close_status_code + if close_msg not in (None, ""): + self._last_close_msg = close_msg + logger.info( + "WebSocket closed. code=%s message=%s", + self._last_close_code, + self._last_close_msg, + ) # ------------------------------------------------------------------ # Lifecycle @@ -253,6 +597,8 @@ def _create_ws_app(self) -> websocket.WebSocketApp: on_message=self._handle_message, on_error=self._handle_error, on_close=self._handle_close, + on_ping=self._handle_ping, + on_pong=self._handle_pong, ) def connect(self, run_in_background: bool = True) -> None: @@ -270,15 +616,18 @@ def connect(self, run_in_background: bool = True) -> None: raise RuntimeError("Already connected; call disconnect() first") self._finished.clear() self._user_disconnect.clear() + self._callback_stop.clear() self._reconnect_attempts = 0 + self._messages_received = 0 + self._messages_dropped = 0 + self._messages_processed = 0 # Drain stale sentinel from previous connection - while not self._queue.empty(): - try: - self._queue.get_nowait() - except queue.Empty: - break + self._drain_queue(self._queue) + self._drain_queue(self._callback_queue) self._ws = self._create_ws_app() + if self._on_message_cb: + self._start_callback_worker() if run_in_background: self._thread = threading.Thread( target=self._run_forever_safe, daemon=True @@ -294,6 +643,8 @@ def _run_forever_safe(self) -> None: while True: with self._lock: ws = self._ws + if ws is None: + break try: sslopt = self._build_sslopt() ws.run_forever( @@ -322,6 +673,8 @@ def _run_forever_safe(self) -> None: delay = self._reconnect_backoff * (2 ** (self._reconnect_attempts - 1)) if self._max_reconnect_backoff is not None: delay = min(delay, self._max_reconnect_backoff) + if self._last_http_status == 429: + delay = max(delay, self._rate_limit_backoff) if self._max_reconnect_attempts > 0: logger.info( "Reconnecting in %.1fs (attempt %d/%d)", @@ -353,10 +706,16 @@ def _run_forever_safe(self) -> None: pass finally: + self._cancel_subscription_watchdog() + self._callback_stop.set() try: self._queue.put_nowait(None) # sentinel — unblocks receive() except queue.Full: pass # _finished.set() below will unblock receive() independently + try: + self._callback_queue.put_nowait(None) # sentinel — unblocks worker + except queue.Full: + pass self._finished.set() # mark as not running — unblocks connect() if self._on_close_cb: try: @@ -376,13 +735,22 @@ def disconnect(self, wait: bool = False, timeout: float | None = None) -> None: when *wait* is True). ``None`` means wait indefinitely. """ self._user_disconnect.set() + self._callback_stop.set() + self._cancel_subscription_watchdog() with self._lock: ws = self._ws if ws: ws.close() + try: + self._callback_queue.put_nowait(None) + except queue.Full: + pass if wait and self._thread is not None: if self._thread is not threading.current_thread(): self._thread.join(timeout=timeout) + if wait and self._callback_thread is not None: + if self._callback_thread is not threading.current_thread(): + self._callback_thread.join(timeout=timeout) def receive(self) -> Generator[dict, None, None]: """ @@ -442,4 +810,4 @@ def __exit__(self, *args) -> None: def ready(self) -> bool: """Returns True if the WebSocket connection is open and ready.""" - return self._ws is not None and self._ws.ready() + return bool(self._ws is not None and self._ws.ready()) diff --git a/src/mistapi/websockets/location.py b/src/mistapi/websockets/location.py index ad2773c..f75fe30 100644 --- a/src/mistapi/websockets/location.py +++ b/src/mistapi/websockets/location.py @@ -29,10 +29,14 @@ class BleAssetsEvents(_MistWebsocket): UUID of the site to stream events from. map_ids : list[str] UUIDs of the maps to stream events from. - ping_interval : int, default 30 + ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 10 - Time in seconds to wait for a ping response before considering the connection dead. + ping_timeout : int | None, default None + Time in seconds to wait for a ping response before considering the + connection dead. Defaults to ``min(45, ping_interval - 1)`` when + pings are enabled, or ``45`` when ``ping_interval`` is 0 (unused + since pings are disabled). Must be lower than ping_interval when + pings are enabled. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -42,10 +46,22 @@ class BleAssetsEvents(_MistWebsocket): max_reconnect_backoff : float | None, default None Maximum backoff delay in seconds. If None, backoff grows indefinitely. queue_maxsize : int, default 0 - Maximum number of messages buffered in the internal queue for the - ``receive()`` generator. ``0`` means unbounded. When set, - incoming messages are dropped with a warning when the queue is - full, preventing memory growth on high-frequency streams. + Maximum number of messages buffered in each of the internal queues + used for the ``receive()`` generator and callback delivery. ``0`` + means unbounded. When set, incoming messages are dropped with a + warning when the target queue is full, preventing memory growth on + high-frequency streams. + subscription_watchdog_timeout : float, default 10.0 + Maximum time in seconds to wait for all channel subscription + acknowledgements after connect. On timeout, the error is reported to + ``on_error`` and the connection is closed (auto_reconnect, when + enabled, then reconnects). + rate_limit_backoff : float, default 30.0 + Minimum reconnect delay in seconds after an HTTP 429 rate-limit + response. + throughput_log_interval : int, default 100 + Log queue depth and processed counts every N messages. ``0`` disables + periodic throughput logs. EXAMPLE ----------- @@ -77,13 +93,16 @@ def __init__( mist_session: APISession, site_id: str, map_ids: list[str], - ping_interval: int = 30, - ping_timeout: int = 10, + ping_interval: int = 60, + ping_timeout: int | None = None, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, max_reconnect_backoff: float | None = None, queue_maxsize: int = 0, + subscription_watchdog_timeout: float = 10.0, + rate_limit_backoff: float = 30.0, + throughput_log_interval: int = 100, ) -> None: channels = [f"/sites/{site_id}/stats/maps/{mid}/assets" for mid in map_ids] super().__init__( @@ -96,6 +115,9 @@ def __init__( reconnect_backoff=reconnect_backoff, max_reconnect_backoff=max_reconnect_backoff, queue_maxsize=queue_maxsize, + subscription_watchdog_timeout=subscription_watchdog_timeout, + rate_limit_backoff=rate_limit_backoff, + throughput_log_interval=throughput_log_interval, ) @@ -113,10 +135,14 @@ class ConnectedClientsEvents(_MistWebsocket): UUID of the site to stream events from. map_ids : list[str] UUIDs of the maps to stream events from. - ping_interval : int, default 30 + ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 10 - Time in seconds to wait for a ping response before considering the connection dead. + ping_timeout : int | None, default None + Time in seconds to wait for a ping response before considering the + connection dead. Defaults to ``min(45, ping_interval - 1)`` when + pings are enabled, or ``45`` when ``ping_interval`` is 0 (unused + since pings are disabled). Must be lower than ping_interval when + pings are enabled. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -126,10 +152,22 @@ class ConnectedClientsEvents(_MistWebsocket): max_reconnect_backoff : float | None, default None Maximum backoff delay in seconds. If None, backoff grows indefinitely. queue_maxsize : int, default 0 - Maximum number of messages buffered in the internal queue for the - ``receive()`` generator. ``0`` means unbounded. When set, - incoming messages are dropped with a warning when the queue is - full, preventing memory growth on high-frequency streams. + Maximum number of messages buffered in each of the internal queues + used for the ``receive()`` generator and callback delivery. ``0`` + means unbounded. When set, incoming messages are dropped with a + warning when the target queue is full, preventing memory growth on + high-frequency streams. + subscription_watchdog_timeout : float, default 10.0 + Maximum time in seconds to wait for all channel subscription + acknowledgements after connect. On timeout, the error is reported to + ``on_error`` and the connection is closed (auto_reconnect, when + enabled, then reconnects). + rate_limit_backoff : float, default 30.0 + Minimum reconnect delay in seconds after an HTTP 429 rate-limit + response. + throughput_log_interval : int, default 100 + Log queue depth and processed counts every N messages. ``0`` disables + periodic throughput logs. EXAMPLE ----------- @@ -161,13 +199,16 @@ def __init__( mist_session: APISession, site_id: str, map_ids: list[str], - ping_interval: int = 30, - ping_timeout: int = 10, + ping_interval: int = 60, + ping_timeout: int | None = None, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, max_reconnect_backoff: float | None = None, queue_maxsize: int = 0, + subscription_watchdog_timeout: float = 10.0, + rate_limit_backoff: float = 30.0, + throughput_log_interval: int = 100, ) -> None: channels = [f"/sites/{site_id}/stats/maps/{mid}/clients" for mid in map_ids] super().__init__( @@ -180,6 +221,9 @@ def __init__( reconnect_backoff=reconnect_backoff, max_reconnect_backoff=max_reconnect_backoff, queue_maxsize=queue_maxsize, + subscription_watchdog_timeout=subscription_watchdog_timeout, + rate_limit_backoff=rate_limit_backoff, + throughput_log_interval=throughput_log_interval, ) @@ -197,10 +241,14 @@ class SdkClientsEvents(_MistWebsocket): UUID of the site to stream events from. map_ids : list[str] UUIDs of the maps to stream events from. - ping_interval : int, default 30 + ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 10 - Time in seconds to wait for a ping response before considering the connection dead. + ping_timeout : int | None, default None + Time in seconds to wait for a ping response before considering the + connection dead. Defaults to ``min(45, ping_interval - 1)`` when + pings are enabled, or ``45`` when ``ping_interval`` is 0 (unused + since pings are disabled). Must be lower than ping_interval when + pings are enabled. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -210,10 +258,22 @@ class SdkClientsEvents(_MistWebsocket): max_reconnect_backoff : float | None, default None Maximum backoff delay in seconds. If None, backoff grows indefinitely. queue_maxsize : int, default 0 - Maximum number of messages buffered in the internal queue for the - ``receive()`` generator. ``0`` means unbounded. When set, - incoming messages are dropped with a warning when the queue is - full, preventing memory growth on high-frequency streams. + Maximum number of messages buffered in each of the internal queues + used for the ``receive()`` generator and callback delivery. ``0`` + means unbounded. When set, incoming messages are dropped with a + warning when the target queue is full, preventing memory growth on + high-frequency streams. + subscription_watchdog_timeout : float, default 10.0 + Maximum time in seconds to wait for all channel subscription + acknowledgements after connect. On timeout, the error is reported to + ``on_error`` and the connection is closed (auto_reconnect, when + enabled, then reconnects). + rate_limit_backoff : float, default 30.0 + Minimum reconnect delay in seconds after an HTTP 429 rate-limit + response. + throughput_log_interval : int, default 100 + Log queue depth and processed counts every N messages. ``0`` disables + periodic throughput logs. EXAMPLE ----------- @@ -245,13 +305,16 @@ def __init__( mist_session: APISession, site_id: str, map_ids: list[str], - ping_interval: int = 30, - ping_timeout: int = 10, + ping_interval: int = 60, + ping_timeout: int | None = None, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, max_reconnect_backoff: float | None = None, queue_maxsize: int = 0, + subscription_watchdog_timeout: float = 10.0, + rate_limit_backoff: float = 30.0, + throughput_log_interval: int = 100, ) -> None: channels = [f"/sites/{site_id}/stats/maps/{mid}/sdkclients" for mid in map_ids] super().__init__( @@ -264,6 +327,9 @@ def __init__( reconnect_backoff=reconnect_backoff, max_reconnect_backoff=max_reconnect_backoff, queue_maxsize=queue_maxsize, + subscription_watchdog_timeout=subscription_watchdog_timeout, + rate_limit_backoff=rate_limit_backoff, + throughput_log_interval=throughput_log_interval, ) @@ -281,10 +347,14 @@ class UnconnectedClientsEvents(_MistWebsocket): UUID of the site to stream events from. map_ids : list[str] UUIDs of the maps to stream events from. - ping_interval : int, default 30 + ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 10 - Time in seconds to wait for a ping response before considering the connection dead. + ping_timeout : int | None, default None + Time in seconds to wait for a ping response before considering the + connection dead. Defaults to ``min(45, ping_interval - 1)`` when + pings are enabled, or ``45`` when ``ping_interval`` is 0 (unused + since pings are disabled). Must be lower than ping_interval when + pings are enabled. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -294,10 +364,22 @@ class UnconnectedClientsEvents(_MistWebsocket): max_reconnect_backoff : float | None, default None Maximum backoff delay in seconds. If None, backoff grows indefinitely. queue_maxsize : int, default 0 - Maximum number of messages buffered in the internal queue for the - ``receive()`` generator. ``0`` means unbounded. When set, - incoming messages are dropped with a warning when the queue is - full, preventing memory growth on high-frequency streams. + Maximum number of messages buffered in each of the internal queues + used for the ``receive()`` generator and callback delivery. ``0`` + means unbounded. When set, incoming messages are dropped with a + warning when the target queue is full, preventing memory growth on + high-frequency streams. + subscription_watchdog_timeout : float, default 10.0 + Maximum time in seconds to wait for all channel subscription + acknowledgements after connect. On timeout, the error is reported to + ``on_error`` and the connection is closed (auto_reconnect, when + enabled, then reconnects). + rate_limit_backoff : float, default 30.0 + Minimum reconnect delay in seconds after an HTTP 429 rate-limit + response. + throughput_log_interval : int, default 100 + Log queue depth and processed counts every N messages. ``0`` disables + periodic throughput logs. EXAMPLE ----------- @@ -329,13 +411,16 @@ def __init__( mist_session: APISession, site_id: str, map_ids: list[str], - ping_interval: int = 30, - ping_timeout: int = 10, + ping_interval: int = 60, + ping_timeout: int | None = None, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, max_reconnect_backoff: float | None = None, queue_maxsize: int = 0, + subscription_watchdog_timeout: float = 10.0, + rate_limit_backoff: float = 30.0, + throughput_log_interval: int = 100, ) -> None: channels = [ f"/sites/{site_id}/stats/maps/{mid}/unconnected_clients" for mid in map_ids @@ -350,6 +435,9 @@ def __init__( reconnect_backoff=reconnect_backoff, max_reconnect_backoff=max_reconnect_backoff, queue_maxsize=queue_maxsize, + subscription_watchdog_timeout=subscription_watchdog_timeout, + rate_limit_backoff=rate_limit_backoff, + throughput_log_interval=throughput_log_interval, ) @@ -367,10 +455,14 @@ class DiscoveredBleAssetsEvents(_MistWebsocket): UUID of the site to stream events from. map_ids : list[str] UUIDs of the maps to stream events from. - ping_interval : int, default 30 + ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 10 - Time in seconds to wait for a ping response before considering the connection dead. + ping_timeout : int | None, default None + Time in seconds to wait for a ping response before considering the + connection dead. Defaults to ``min(45, ping_interval - 1)`` when + pings are enabled, or ``45`` when ``ping_interval`` is 0 (unused + since pings are disabled). Must be lower than ping_interval when + pings are enabled. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -380,10 +472,22 @@ class DiscoveredBleAssetsEvents(_MistWebsocket): max_reconnect_backoff : float | None, default None Maximum backoff delay in seconds. If None, backoff grows indefinitely. queue_maxsize : int, default 0 - Maximum number of messages buffered in the internal queue for the - ``receive()`` generator. ``0`` means unbounded. When set, - incoming messages are dropped with a warning when the queue is - full, preventing memory growth on high-frequency streams. + Maximum number of messages buffered in each of the internal queues + used for the ``receive()`` generator and callback delivery. ``0`` + means unbounded. When set, incoming messages are dropped with a + warning when the target queue is full, preventing memory growth on + high-frequency streams. + subscription_watchdog_timeout : float, default 10.0 + Maximum time in seconds to wait for all channel subscription + acknowledgements after connect. On timeout, the error is reported to + ``on_error`` and the connection is closed (auto_reconnect, when + enabled, then reconnects). + rate_limit_backoff : float, default 30.0 + Minimum reconnect delay in seconds after an HTTP 429 rate-limit + response. + throughput_log_interval : int, default 100 + Log queue depth and processed counts every N messages. ``0`` disables + periodic throughput logs. EXAMPLE ----------- @@ -415,13 +519,16 @@ def __init__( mist_session: APISession, site_id: str, map_ids: list[str], - ping_interval: int = 30, - ping_timeout: int = 10, + ping_interval: int = 60, + ping_timeout: int | None = None, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, max_reconnect_backoff: float | None = None, queue_maxsize: int = 0, + subscription_watchdog_timeout: float = 10.0, + rate_limit_backoff: float = 30.0, + throughput_log_interval: int = 100, ) -> None: channels = [ f"/sites/{site_id}/stats/maps/{mid}/discovered_assets" for mid in map_ids @@ -436,4 +543,7 @@ def __init__( reconnect_backoff=reconnect_backoff, max_reconnect_backoff=max_reconnect_backoff, queue_maxsize=queue_maxsize, + subscription_watchdog_timeout=subscription_watchdog_timeout, + rate_limit_backoff=rate_limit_backoff, + throughput_log_interval=throughput_log_interval, ) diff --git a/src/mistapi/websockets/orgs.py b/src/mistapi/websockets/orgs.py index 9a04e7f..65463f5 100644 --- a/src/mistapi/websockets/orgs.py +++ b/src/mistapi/websockets/orgs.py @@ -27,10 +27,14 @@ class InsightsEvents(_MistWebsocket): Authenticated API session. org_id : str UUID of the organization to stream events from. - ping_interval : int, default 30 + ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 10 - Time in seconds to wait for a ping response before considering the connection dead. + ping_timeout : int | None, default None + Time in seconds to wait for a ping response before considering the + connection dead. Defaults to ``min(45, ping_interval - 1)`` when + pings are enabled, or ``45`` when ``ping_interval`` is 0 (unused + since pings are disabled). Must be lower than ping_interval when + pings are enabled. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -40,10 +44,22 @@ class InsightsEvents(_MistWebsocket): max_reconnect_backoff : float | None, default None Maximum backoff delay in seconds. If None, backoff grows indefinitely. queue_maxsize : int, default 0 - Maximum number of messages buffered in the internal queue for the - ``receive()`` generator. ``0`` means unbounded. When set, - incoming messages are dropped with a warning when the queue is - full, preventing memory growth on high-frequency streams. + Maximum number of messages buffered in each of the internal queues + used for the ``receive()`` generator and callback delivery. ``0`` + means unbounded. When set, incoming messages are dropped with a + warning when the target queue is full, preventing memory growth on + high-frequency streams. + subscription_watchdog_timeout : float, default 10.0 + Maximum time in seconds to wait for all channel subscription + acknowledgements after connect. On timeout, the error is reported to + ``on_error`` and the connection is closed (auto_reconnect, when + enabled, then reconnects). + rate_limit_backoff : float, default 30.0 + Minimum reconnect delay in seconds after an HTTP 429 rate-limit + response. + throughput_log_interval : int, default 100 + Log queue depth and processed counts every N messages. ``0`` disables + periodic throughput logs. EXAMPLE ----------- @@ -74,13 +90,16 @@ def __init__( self, mist_session: APISession, org_id: str, - ping_interval: int = 30, - ping_timeout: int = 10, + ping_interval: int = 60, + ping_timeout: int | None = None, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, max_reconnect_backoff: float | None = None, queue_maxsize: int = 0, + subscription_watchdog_timeout: float = 10.0, + rate_limit_backoff: float = 30.0, + throughput_log_interval: int = 100, ) -> None: super().__init__( mist_session, @@ -92,6 +111,9 @@ def __init__( reconnect_backoff=reconnect_backoff, max_reconnect_backoff=max_reconnect_backoff, queue_maxsize=queue_maxsize, + subscription_watchdog_timeout=subscription_watchdog_timeout, + rate_limit_backoff=rate_limit_backoff, + throughput_log_interval=throughput_log_interval, ) @@ -107,10 +129,14 @@ class MxEdgesStatsEvents(_MistWebsocket): Authenticated API session. org_id : str UUID of the organization to stream events from. - ping_interval : int, default 30 + ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 10 - Time in seconds to wait for a ping response before considering the connection dead. + ping_timeout : int | None, default None + Time in seconds to wait for a ping response before considering the + connection dead. Defaults to ``min(45, ping_interval - 1)`` when + pings are enabled, or ``45`` when ``ping_interval`` is 0 (unused + since pings are disabled). Must be lower than ping_interval when + pings are enabled. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -120,10 +146,22 @@ class MxEdgesStatsEvents(_MistWebsocket): max_reconnect_backoff : float | None, default None Maximum backoff delay in seconds. If None, backoff grows indefinitely. queue_maxsize : int, default 0 - Maximum number of messages buffered in the internal queue for the - ``receive()`` generator. ``0`` means unbounded. When set, - incoming messages are dropped with a warning when the queue is - full, preventing memory growth on high-frequency streams. + Maximum number of messages buffered in each of the internal queues + used for the ``receive()`` generator and callback delivery. ``0`` + means unbounded. When set, incoming messages are dropped with a + warning when the target queue is full, preventing memory growth on + high-frequency streams. + subscription_watchdog_timeout : float, default 10.0 + Maximum time in seconds to wait for all channel subscription + acknowledgements after connect. On timeout, the error is reported to + ``on_error`` and the connection is closed (auto_reconnect, when + enabled, then reconnects). + rate_limit_backoff : float, default 30.0 + Minimum reconnect delay in seconds after an HTTP 429 rate-limit + response. + throughput_log_interval : int, default 100 + Log queue depth and processed counts every N messages. ``0`` disables + periodic throughput logs. EXAMPLE ----------- @@ -154,13 +192,16 @@ def __init__( self, mist_session: APISession, org_id: str, - ping_interval: int = 30, - ping_timeout: int = 10, + ping_interval: int = 60, + ping_timeout: int | None = None, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, max_reconnect_backoff: float | None = None, queue_maxsize: int = 0, + subscription_watchdog_timeout: float = 10.0, + rate_limit_backoff: float = 30.0, + throughput_log_interval: int = 100, ) -> None: super().__init__( mist_session, @@ -172,6 +213,9 @@ def __init__( reconnect_backoff=reconnect_backoff, max_reconnect_backoff=max_reconnect_backoff, queue_maxsize=queue_maxsize, + subscription_watchdog_timeout=subscription_watchdog_timeout, + rate_limit_backoff=rate_limit_backoff, + throughput_log_interval=throughput_log_interval, ) @@ -187,10 +231,14 @@ class MxEdgesEvents(_MistWebsocket): Authenticated API session. org_id : str UUID of the org to stream events from. - ping_interval : int, default 30 + ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 10 - Time in seconds to wait for a ping response before considering the connection dead. + ping_timeout : int | None, default None + Time in seconds to wait for a ping response before considering the + connection dead. Defaults to ``min(45, ping_interval - 1)`` when + pings are enabled, or ``45`` when ``ping_interval`` is 0 (unused + since pings are disabled). Must be lower than ping_interval when + pings are enabled. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -200,10 +248,22 @@ class MxEdgesEvents(_MistWebsocket): max_reconnect_backoff : float | None, default None Maximum backoff delay in seconds. If None, backoff grows indefinitely. queue_maxsize : int, default 0 - Maximum number of messages buffered in the internal queue for the - ``receive()`` generator. ``0`` means unbounded. When set, - incoming messages are dropped with a warning when the queue is - full, preventing memory growth on high-frequency streams. + Maximum number of messages buffered in each of the internal queues + used for the ``receive()`` generator and callback delivery. ``0`` + means unbounded. When set, incoming messages are dropped with a + warning when the target queue is full, preventing memory growth on + high-frequency streams. + subscription_watchdog_timeout : float, default 10.0 + Maximum time in seconds to wait for all channel subscription + acknowledgements after connect. On timeout, the error is reported to + ``on_error`` and the connection is closed (auto_reconnect, when + enabled, then reconnects). + rate_limit_backoff : float, default 30.0 + Minimum reconnect delay in seconds after an HTTP 429 rate-limit + response. + throughput_log_interval : int, default 100 + Log queue depth and processed counts every N messages. ``0`` disables + periodic throughput logs. EXAMPLE ----------- @@ -234,13 +294,16 @@ def __init__( self, mist_session: APISession, org_id: str, - ping_interval: int = 30, - ping_timeout: int = 10, + ping_interval: int = 60, + ping_timeout: int | None = None, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, max_reconnect_backoff: float | None = None, queue_maxsize: int = 0, + subscription_watchdog_timeout: float = 10.0, + rate_limit_backoff: float = 30.0, + throughput_log_interval: int = 100, ) -> None: super().__init__( mist_session, @@ -252,4 +315,7 @@ def __init__( reconnect_backoff=reconnect_backoff, max_reconnect_backoff=max_reconnect_backoff, queue_maxsize=queue_maxsize, + subscription_watchdog_timeout=subscription_watchdog_timeout, + rate_limit_backoff=rate_limit_backoff, + throughput_log_interval=throughput_log_interval, ) diff --git a/src/mistapi/websockets/session.py b/src/mistapi/websockets/session.py index e7365dc..2a5f56c 100644 --- a/src/mistapi/websockets/session.py +++ b/src/mistapi/websockets/session.py @@ -36,10 +36,14 @@ class SessionWithUrl(_MistWebsocket): The session's authentication credentials (API token or cookies) are sent to whatever host is specified in this URL. Only use trusted URLs — never pass user-supplied or untrusted input. - ping_interval : int, default 30 + ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 10 - Time in seconds to wait for a ping response before considering the connection dead. + ping_timeout : int | None, default None + Time in seconds to wait for a ping response before considering the + connection dead. Defaults to ``min(45, ping_interval - 1)`` when + pings are enabled, or ``45`` when ``ping_interval`` is 0 (unused + since pings are disabled). Must be lower than ping_interval when + pings are enabled. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -49,10 +53,22 @@ class SessionWithUrl(_MistWebsocket): max_reconnect_backoff : float | None, default None Maximum backoff delay in seconds. If None, backoff grows indefinitely. queue_maxsize : int, default 0 - Maximum number of messages buffered in the internal queue for the - ``receive()`` generator. ``0`` means unbounded. When set, - incoming messages are dropped with a warning when the queue is - full, preventing memory growth on high-frequency streams. + Maximum number of messages buffered in each of the internal queues + used for the ``receive()`` generator and callback delivery. ``0`` + means unbounded. When set, incoming messages are dropped with a + warning when the target queue is full, preventing memory growth on + high-frequency streams. + subscription_watchdog_timeout : float, default 10.0 + Maximum time in seconds to wait for all channel subscription + acknowledgements after connect. On timeout, the error is reported to + ``on_error`` and the connection is closed (auto_reconnect, when + enabled, then reconnects). + rate_limit_backoff : float, default 30.0 + Minimum reconnect delay in seconds after an HTTP 429 rate-limit + response. + throughput_log_interval : int, default 100 + Log queue depth and processed counts every N messages. ``0`` disables + periodic throughput logs. EXAMPLE ----------- @@ -83,13 +99,16 @@ def __init__( self, mist_session: APISession, url: str, - ping_interval: int = 30, - ping_timeout: int = 10, + ping_interval: int = 60, + ping_timeout: int | None = None, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, max_reconnect_backoff: float | None = None, queue_maxsize: int = 0, + subscription_watchdog_timeout: float = 10.0, + rate_limit_backoff: float = 30.0, + throughput_log_interval: int = 100, ) -> None: parsed = urlparse(url) if parsed.scheme.lower() != "wss" or not parsed.netloc: @@ -105,6 +124,9 @@ def __init__( reconnect_backoff=reconnect_backoff, max_reconnect_backoff=max_reconnect_backoff, queue_maxsize=queue_maxsize, + subscription_watchdog_timeout=subscription_watchdog_timeout, + rate_limit_backoff=rate_limit_backoff, + throughput_log_interval=throughput_log_interval, ) def _build_ws_url(self) -> str: diff --git a/src/mistapi/websockets/sites.py b/src/mistapi/websockets/sites.py index 64b2b88..df194c2 100644 --- a/src/mistapi/websockets/sites.py +++ b/src/mistapi/websockets/sites.py @@ -27,10 +27,14 @@ class ClientsStatsEvents(_MistWebsocket): Authenticated API session. site_ids : list[str] UUIDs of the sites to stream events from. - ping_interval : int, default 30 + ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 10 - Time in seconds to wait for a ping response before considering the connection dead. + ping_timeout : int | None, default None + Time in seconds to wait for a ping response before considering the + connection dead. Defaults to ``min(45, ping_interval - 1)`` when + pings are enabled, or ``45`` when ``ping_interval`` is 0 (unused + since pings are disabled). Must be lower than ping_interval when + pings are enabled. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -40,10 +44,22 @@ class ClientsStatsEvents(_MistWebsocket): max_reconnect_backoff : float | None, default None Maximum backoff delay in seconds. If None, backoff grows indefinitely. queue_maxsize : int, default 0 - Maximum number of messages buffered in the internal queue for the - ``receive()`` generator. ``0`` means unbounded. When set, - incoming messages are dropped with a warning when the queue is - full, preventing memory growth on high-frequency streams. + Maximum number of messages buffered in each of the internal queues + used for the ``receive()`` generator and callback delivery. ``0`` + means unbounded. When set, incoming messages are dropped with a + warning when the target queue is full, preventing memory growth on + high-frequency streams. + subscription_watchdog_timeout : float, default 10.0 + Maximum time in seconds to wait for all channel subscription + acknowledgements after connect. On timeout, the error is reported to + ``on_error`` and the connection is closed (auto_reconnect, when + enabled, then reconnects). + rate_limit_backoff : float, default 30.0 + Minimum reconnect delay in seconds after an HTTP 429 rate-limit + response. + throughput_log_interval : int, default 100 + Log queue depth and processed counts every N messages. ``0`` disables + periodic throughput logs. EXAMPLE ----------- @@ -74,13 +90,16 @@ def __init__( self, mist_session: APISession, site_ids: list[str], - ping_interval: int = 30, - ping_timeout: int = 10, + ping_interval: int = 60, + ping_timeout: int | None = None, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, max_reconnect_backoff: float | None = None, queue_maxsize: int = 0, + subscription_watchdog_timeout: float = 10.0, + rate_limit_backoff: float = 30.0, + throughput_log_interval: int = 100, ) -> None: channels = [f"/sites/{site_id}/stats/clients" for site_id in site_ids] super().__init__( @@ -93,6 +112,9 @@ def __init__( reconnect_backoff=reconnect_backoff, max_reconnect_backoff=max_reconnect_backoff, queue_maxsize=queue_maxsize, + subscription_watchdog_timeout=subscription_watchdog_timeout, + rate_limit_backoff=rate_limit_backoff, + throughput_log_interval=throughput_log_interval, ) @@ -116,10 +138,14 @@ class DeviceCmdEvents(_MistWebsocket): UUID of the site to stream events from. device_ids : list[str] UUIDs of the devices to stream events from. - ping_interval : int, default 30 + ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 10 - Time in seconds to wait for a ping response before considering the connection dead. + ping_timeout : int | None, default None + Time in seconds to wait for a ping response before considering the + connection dead. Defaults to ``min(45, ping_interval - 1)`` when + pings are enabled, or ``45`` when ``ping_interval`` is 0 (unused + since pings are disabled). Must be lower than ping_interval when + pings are enabled. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -129,10 +155,22 @@ class DeviceCmdEvents(_MistWebsocket): max_reconnect_backoff : float | None, default None Maximum backoff delay in seconds. If None, backoff grows indefinitely. queue_maxsize : int, default 0 - Maximum number of messages buffered in the internal queue for the - ``receive()`` generator. ``0`` means unbounded. When set, - incoming messages are dropped with a warning when the queue is - full, preventing memory growth on high-frequency streams. + Maximum number of messages buffered in each of the internal queues + used for the ``receive()`` generator and callback delivery. ``0`` + means unbounded. When set, incoming messages are dropped with a + warning when the target queue is full, preventing memory growth on + high-frequency streams. + subscription_watchdog_timeout : float, default 10.0 + Maximum time in seconds to wait for all channel subscription + acknowledgements after connect. On timeout, the error is reported to + ``on_error`` and the connection is closed (auto_reconnect, when + enabled, then reconnects). + rate_limit_backoff : float, default 30.0 + Minimum reconnect delay in seconds after an HTTP 429 rate-limit + response. + throughput_log_interval : int, default 100 + Log queue depth and processed counts every N messages. ``0`` disables + periodic throughput logs. EXAMPLE ----------- @@ -164,13 +202,16 @@ def __init__( mist_session: APISession, site_id: str, device_ids: list[str], - ping_interval: int = 30, - ping_timeout: int = 10, + ping_interval: int = 60, + ping_timeout: int | None = None, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, max_reconnect_backoff: float | None = None, queue_maxsize: int = 0, + subscription_watchdog_timeout: float = 10.0, + rate_limit_backoff: float = 30.0, + throughput_log_interval: int = 100, ) -> None: channels = [ f"/sites/{site_id}/devices/{device_id}/cmd" for device_id in device_ids @@ -185,6 +226,9 @@ def __init__( reconnect_backoff=reconnect_backoff, max_reconnect_backoff=max_reconnect_backoff, queue_maxsize=queue_maxsize, + subscription_watchdog_timeout=subscription_watchdog_timeout, + rate_limit_backoff=rate_limit_backoff, + throughput_log_interval=throughput_log_interval, ) @@ -200,10 +244,14 @@ class DeviceStatsEvents(_MistWebsocket): Authenticated API session. site_ids : list[str] UUIDs of the sites to stream events from. - ping_interval : int, default 30 + ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 10 - Time in seconds to wait for a ping response before considering the connection dead. + ping_timeout : int | None, default None + Time in seconds to wait for a ping response before considering the + connection dead. Defaults to ``min(45, ping_interval - 1)`` when + pings are enabled, or ``45`` when ``ping_interval`` is 0 (unused + since pings are disabled). Must be lower than ping_interval when + pings are enabled. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -213,10 +261,22 @@ class DeviceStatsEvents(_MistWebsocket): max_reconnect_backoff : float | None, default None Maximum backoff delay in seconds. If None, backoff grows indefinitely. queue_maxsize : int, default 0 - Maximum number of messages buffered in the internal queue for the - ``receive()`` generator. ``0`` means unbounded. When set, - incoming messages are dropped with a warning when the queue is - full, preventing memory growth on high-frequency streams. + Maximum number of messages buffered in each of the internal queues + used for the ``receive()`` generator and callback delivery. ``0`` + means unbounded. When set, incoming messages are dropped with a + warning when the target queue is full, preventing memory growth on + high-frequency streams. + subscription_watchdog_timeout : float, default 10.0 + Maximum time in seconds to wait for all channel subscription + acknowledgements after connect. On timeout, the error is reported to + ``on_error`` and the connection is closed (auto_reconnect, when + enabled, then reconnects). + rate_limit_backoff : float, default 30.0 + Minimum reconnect delay in seconds after an HTTP 429 rate-limit + response. + throughput_log_interval : int, default 100 + Log queue depth and processed counts every N messages. ``0`` disables + periodic throughput logs. EXAMPLE ----------- @@ -247,13 +307,16 @@ def __init__( self, mist_session: APISession, site_ids: list[str], - ping_interval: int = 30, - ping_timeout: int = 10, + ping_interval: int = 60, + ping_timeout: int | None = None, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, max_reconnect_backoff: float | None = None, queue_maxsize: int = 0, + subscription_watchdog_timeout: float = 10.0, + rate_limit_backoff: float = 30.0, + throughput_log_interval: int = 100, ) -> None: channels = [f"/sites/{site_id}/stats/devices" for site_id in site_ids] super().__init__( @@ -266,6 +329,9 @@ def __init__( reconnect_backoff=reconnect_backoff, max_reconnect_backoff=max_reconnect_backoff, queue_maxsize=queue_maxsize, + subscription_watchdog_timeout=subscription_watchdog_timeout, + rate_limit_backoff=rate_limit_backoff, + throughput_log_interval=throughput_log_interval, ) @@ -281,10 +347,14 @@ class DeviceEvents(_MistWebsocket): Authenticated API session. site_ids : list[str] UUIDs of the sites to stream events from. - ping_interval : int, default 30 + ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 10 - Time in seconds to wait for a ping response before considering the connection dead. + ping_timeout : int | None, default None + Time in seconds to wait for a ping response before considering the + connection dead. Defaults to ``min(45, ping_interval - 1)`` when + pings are enabled, or ``45`` when ``ping_interval`` is 0 (unused + since pings are disabled). Must be lower than ping_interval when + pings are enabled. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -294,10 +364,22 @@ class DeviceEvents(_MistWebsocket): max_reconnect_backoff : float | None, default None Maximum backoff delay in seconds. If None, backoff grows indefinitely. queue_maxsize : int, default 0 - Maximum number of messages buffered in the internal queue for the - ``receive()`` generator. ``0`` means unbounded. When set, - incoming messages are dropped with a warning when the queue is - full, preventing memory growth on high-frequency streams. + Maximum number of messages buffered in each of the internal queues + used for the ``receive()`` generator and callback delivery. ``0`` + means unbounded. When set, incoming messages are dropped with a + warning when the target queue is full, preventing memory growth on + high-frequency streams. + subscription_watchdog_timeout : float, default 10.0 + Maximum time in seconds to wait for all channel subscription + acknowledgements after connect. On timeout, the error is reported to + ``on_error`` and the connection is closed (auto_reconnect, when + enabled, then reconnects). + rate_limit_backoff : float, default 30.0 + Minimum reconnect delay in seconds after an HTTP 429 rate-limit + response. + throughput_log_interval : int, default 100 + Log queue depth and processed counts every N messages. ``0`` disables + periodic throughput logs. EXAMPLE ----------- @@ -328,13 +410,16 @@ def __init__( self, mist_session: APISession, site_ids: list[str], - ping_interval: int = 30, - ping_timeout: int = 10, + ping_interval: int = 60, + ping_timeout: int | None = None, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, max_reconnect_backoff: float | None = None, queue_maxsize: int = 0, + subscription_watchdog_timeout: float = 10.0, + rate_limit_backoff: float = 30.0, + throughput_log_interval: int = 100, ) -> None: channels = [f"/sites/{site_id}/devices" for site_id in site_ids] super().__init__( @@ -347,6 +432,9 @@ def __init__( reconnect_backoff=reconnect_backoff, max_reconnect_backoff=max_reconnect_backoff, queue_maxsize=queue_maxsize, + subscription_watchdog_timeout=subscription_watchdog_timeout, + rate_limit_backoff=rate_limit_backoff, + throughput_log_interval=throughput_log_interval, ) @@ -362,10 +450,14 @@ class MxEdgesStatsEvents(_MistWebsocket): Authenticated API session. site_ids : list[str] UUIDs of the sites to stream events from. - ping_interval : int, default 30 + ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 10 - Time in seconds to wait for a ping response before considering the connection dead. + ping_timeout : int | None, default None + Time in seconds to wait for a ping response before considering the + connection dead. Defaults to ``min(45, ping_interval - 1)`` when + pings are enabled, or ``45`` when ``ping_interval`` is 0 (unused + since pings are disabled). Must be lower than ping_interval when + pings are enabled. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -375,10 +467,22 @@ class MxEdgesStatsEvents(_MistWebsocket): max_reconnect_backoff : float | None, default None Maximum backoff delay in seconds. If None, backoff grows indefinitely. queue_maxsize : int, default 0 - Maximum number of messages buffered in the internal queue for the - ``receive()`` generator. ``0`` means unbounded. When set, - incoming messages are dropped with a warning when the queue is - full, preventing memory growth on high-frequency streams. + Maximum number of messages buffered in each of the internal queues + used for the ``receive()`` generator and callback delivery. ``0`` + means unbounded. When set, incoming messages are dropped with a + warning when the target queue is full, preventing memory growth on + high-frequency streams. + subscription_watchdog_timeout : float, default 10.0 + Maximum time in seconds to wait for all channel subscription + acknowledgements after connect. On timeout, the error is reported to + ``on_error`` and the connection is closed (auto_reconnect, when + enabled, then reconnects). + rate_limit_backoff : float, default 30.0 + Minimum reconnect delay in seconds after an HTTP 429 rate-limit + response. + throughput_log_interval : int, default 100 + Log queue depth and processed counts every N messages. ``0`` disables + periodic throughput logs. EXAMPLE ----------- @@ -409,13 +513,16 @@ def __init__( self, mist_session: APISession, site_ids: list[str], - ping_interval: int = 30, - ping_timeout: int = 10, + ping_interval: int = 60, + ping_timeout: int | None = None, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, max_reconnect_backoff: float | None = None, queue_maxsize: int = 0, + subscription_watchdog_timeout: float = 10.0, + rate_limit_backoff: float = 30.0, + throughput_log_interval: int = 100, ) -> None: channels = [f"/sites/{site_id}/stats/mxedges" for site_id in site_ids] super().__init__( @@ -428,6 +535,9 @@ def __init__( reconnect_backoff=reconnect_backoff, max_reconnect_backoff=max_reconnect_backoff, queue_maxsize=queue_maxsize, + subscription_watchdog_timeout=subscription_watchdog_timeout, + rate_limit_backoff=rate_limit_backoff, + throughput_log_interval=throughput_log_interval, ) @@ -443,10 +553,14 @@ class MxEdgesEvents(_MistWebsocket): Authenticated API session. site_ids : list[str] UUIDs of the sites to stream events from. - ping_interval : int, default 30 + ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 10 - Time in seconds to wait for a ping response before considering the connection dead. + ping_timeout : int | None, default None + Time in seconds to wait for a ping response before considering the + connection dead. Defaults to ``min(45, ping_interval - 1)`` when + pings are enabled, or ``45`` when ``ping_interval`` is 0 (unused + since pings are disabled). Must be lower than ping_interval when + pings are enabled. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -456,10 +570,22 @@ class MxEdgesEvents(_MistWebsocket): max_reconnect_backoff : float | None, default None Maximum backoff delay in seconds. If None, backoff grows indefinitely. queue_maxsize : int, default 0 - Maximum number of messages buffered in the internal queue for the - ``receive()`` generator. ``0`` means unbounded. When set, - incoming messages are dropped with a warning when the queue is - full, preventing memory growth on high-frequency streams. + Maximum number of messages buffered in each of the internal queues + used for the ``receive()`` generator and callback delivery. ``0`` + means unbounded. When set, incoming messages are dropped with a + warning when the target queue is full, preventing memory growth on + high-frequency streams. + subscription_watchdog_timeout : float, default 10.0 + Maximum time in seconds to wait for all channel subscription + acknowledgements after connect. On timeout, the error is reported to + ``on_error`` and the connection is closed (auto_reconnect, when + enabled, then reconnects). + rate_limit_backoff : float, default 30.0 + Minimum reconnect delay in seconds after an HTTP 429 rate-limit + response. + throughput_log_interval : int, default 100 + Log queue depth and processed counts every N messages. ``0`` disables + periodic throughput logs. EXAMPLE ----------- @@ -490,13 +616,16 @@ def __init__( self, mist_session: APISession, site_ids: list[str], - ping_interval: int = 30, - ping_timeout: int = 10, + ping_interval: int = 60, + ping_timeout: int | None = None, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, max_reconnect_backoff: float | None = None, queue_maxsize: int = 0, + subscription_watchdog_timeout: float = 10.0, + rate_limit_backoff: float = 30.0, + throughput_log_interval: int = 100, ) -> None: channels = [f"/sites/{site_id}/mxedges" for site_id in site_ids] super().__init__( @@ -509,6 +638,9 @@ def __init__( reconnect_backoff=reconnect_backoff, max_reconnect_backoff=max_reconnect_backoff, queue_maxsize=queue_maxsize, + subscription_watchdog_timeout=subscription_watchdog_timeout, + rate_limit_backoff=rate_limit_backoff, + throughput_log_interval=throughput_log_interval, ) @@ -524,10 +656,14 @@ class PcapEvents(_MistWebsocket): Authenticated API session. site_id : str UUID of the site to stream events from. - ping_interval : int, default 30 + ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 10 - Time in seconds to wait for a ping response before considering the connection dead. + ping_timeout : int | None, default None + Time in seconds to wait for a ping response before considering the + connection dead. Defaults to ``min(45, ping_interval - 1)`` when + pings are enabled, or ``45`` when ``ping_interval`` is 0 (unused + since pings are disabled). Must be lower than ping_interval when + pings are enabled. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -537,10 +673,22 @@ class PcapEvents(_MistWebsocket): max_reconnect_backoff : float | None, default None Maximum backoff delay in seconds. If None, backoff grows indefinitely. queue_maxsize : int, default 0 - Maximum number of messages buffered in the internal queue for the - ``receive()`` generator. ``0`` means unbounded. When set, - incoming messages are dropped with a warning when the queue is - full, preventing memory growth on high-frequency streams. + Maximum number of messages buffered in each of the internal queues + used for the ``receive()`` generator and callback delivery. ``0`` + means unbounded. When set, incoming messages are dropped with a + warning when the target queue is full, preventing memory growth on + high-frequency streams. + subscription_watchdog_timeout : float, default 10.0 + Maximum time in seconds to wait for all channel subscription + acknowledgements after connect. On timeout, the error is reported to + ``on_error`` and the connection is closed (auto_reconnect, when + enabled, then reconnects). + rate_limit_backoff : float, default 30.0 + Minimum reconnect delay in seconds after an HTTP 429 rate-limit + response. + throughput_log_interval : int, default 100 + Log queue depth and processed counts every N messages. ``0`` disables + periodic throughput logs. EXAMPLE ----------- @@ -571,13 +719,16 @@ def __init__( self, mist_session: APISession, site_id: str, - ping_interval: int = 30, - ping_timeout: int = 10, + ping_interval: int = 60, + ping_timeout: int | None = None, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, max_reconnect_backoff: float | None = None, queue_maxsize: int = 0, + subscription_watchdog_timeout: float = 10.0, + rate_limit_backoff: float = 30.0, + throughput_log_interval: int = 100, ) -> None: channels = [f"/sites/{site_id}/pcaps"] super().__init__( @@ -590,4 +741,7 @@ def __init__( reconnect_backoff=reconnect_backoff, max_reconnect_backoff=max_reconnect_backoff, queue_maxsize=queue_maxsize, + subscription_watchdog_timeout=subscription_watchdog_timeout, + rate_limit_backoff=rate_limit_backoff, + throughput_log_interval=throughput_log_interval, ) diff --git a/tests/unit/test_websocket_client.py b/tests/unit/test_websocket_client.py index 7177476..dccc7fe 100644 --- a/tests/unit/test_websocket_client.py +++ b/tests/unit/test_websocket_client.py @@ -363,17 +363,41 @@ def test_wraps_invalid_json_in_raw_key(self, ws_client) -> None: def test_calls_on_message_callback_with_parsed_data(self, ws_client) -> None: cb = Mock() - ws_client.on_message(cb) + called = threading.Event() + + def cb_wrapper(data): + cb(data) + called.set() + + ws_client.on_message(cb_wrapper) + ws_client._finished.clear() # keep worker alive for this assertion + ws_client._start_callback_worker() payload = {"type": "event"} ws_client._handle_message(Mock(), json.dumps(payload)) + + assert called.wait(timeout=1), "callback was not invoked by worker" cb.assert_called_once_with(payload) + ws_client.disconnect(wait=True, timeout=1) + def test_calls_on_message_callback_with_raw_fallback(self, ws_client) -> None: cb = Mock() - ws_client.on_message(cb) + called = threading.Event() + + def cb_wrapper(data): + cb(data) + called.set() + + ws_client.on_message(cb_wrapper) + ws_client._finished.clear() # keep worker alive for this assertion + ws_client._start_callback_worker() ws_client._handle_message(Mock(), "plain text") + + assert called.wait(timeout=1), "callback was not invoked by worker" cb.assert_called_once_with({"raw": "plain text"}) + ws_client.disconnect(wait=True, timeout=1) + def test_no_error_without_on_message_callback(self, ws_client) -> None: ws_client._handle_message(Mock(), '{"ok": true}') # Should not raise @@ -461,6 +485,8 @@ def test_connect_creates_websocket_app(self, mock_ws_cls, ws_client) -> None: on_message=ws_client._handle_message, on_error=ws_client._handle_error, on_close=ws_client._handle_close, + on_ping=ws_client._handle_ping, + on_pong=ws_client._handle_pong, ) mock_ws_instance.run_forever.assert_called_once() @@ -538,8 +564,8 @@ def test_passes_sslopt_when_verify_false(self, mock_session) -> None: client._ws = mock_ws client._run_forever_safe() mock_ws.run_forever.assert_called_once_with( - ping_interval=30, - ping_timeout=10, + ping_interval=60, + ping_timeout=45, sslopt={"cert_reqs": ssl.CERT_NONE, "check_hostname": False}, ) @@ -687,8 +713,8 @@ class TestInit: """Tests for __init__ defaults.""" def test_default_ping_interval_and_timeout(self, ws_client) -> None: - assert ws_client._ping_interval == 30 - assert ws_client._ping_timeout == 10 + assert ws_client._ping_interval == 60 + assert ws_client._ping_timeout == 45 def test_custom_ping_interval_and_timeout(self, single_channel_client) -> None: assert single_channel_client._ping_interval == 15 @@ -729,6 +755,37 @@ def test_negative_queue_maxsize_raises(self, mock_session) -> None: with pytest.raises(ValueError, match="queue_maxsize must be >= 0"): _MistWebsocket(mock_session, channels=["/ch"], queue_maxsize=-1) + def test_ping_interval_must_be_greater_than_ping_timeout( + self, mock_session + ) -> None: + with pytest.raises(ValueError, match="ping_interval must be greater"): + _MistWebsocket( + mock_session, + channels=["/ch"], + ping_interval=10, + ping_timeout=10, + ) + + def test_ping_timeout_derived_from_ping_interval(self, mock_session) -> None: + client = _MistWebsocket(mock_session, channels=["/ch"], ping_interval=30) + assert client._ping_timeout == 29 + + client = _MistWebsocket(mock_session, channels=["/ch"], ping_interval=120) + assert client._ping_timeout == 45 + + def test_ping_timeout_derived_with_pings_disabled(self, mock_session) -> None: + client = _MistWebsocket(mock_session, channels=["/ch"], ping_interval=0) + assert client._ping_timeout == 45 + + def test_tiny_ping_interval_without_timeout_raises(self, mock_session) -> None: + with pytest.raises(ValueError, match="ping_interval must be >= 2"): + _MistWebsocket(mock_session, channels=["/ch"], ping_interval=1) + + def test_channel_limit_enforced(self, mock_session) -> None: + channels = [f"/sites/{i}/stats/devices" for i in range(2001)] + with pytest.raises(ValueError, match="Too many channels"): + _MistWebsocket(mock_session, channels=channels) + def test_negative_max_reconnect_backoff_raises(self, mock_session) -> None: with pytest.raises(ValueError, match="max_reconnect_backoff must be > 0"): _MistWebsocket(mock_session, channels=["/ch"], max_reconnect_backoff=-1.0) @@ -787,6 +844,18 @@ def test_inherits_from_mist_websocket(self, mock_session) -> None: ws = DeviceCmdEvents(mock_session, site_id="s1", device_ids=["d1"]) assert isinstance(ws, _MistWebsocket) + def test_supports_reliability_kwargs(self, mock_session) -> None: + ws = DeviceStatsEvents( + mock_session, + site_ids=["s1"], + subscription_watchdog_timeout=3.0, + rate_limit_backoff=12.0, + throughput_log_interval=250, + ) + assert ws._subscription_watchdog_timeout == 3.0 + assert ws._rate_limit_backoff == 12.0 + assert ws._throughput_log_interval == 250 + class TestOrgChannels: """Tests for public org-level WebSocket channel classes.""" @@ -807,6 +876,18 @@ def test_inherits_from_mist_websocket(self, mock_session) -> None: ws = InsightsEvents(mock_session, org_id="o1") assert isinstance(ws, _MistWebsocket) + def test_supports_reliability_kwargs(self, mock_session) -> None: + ws = InsightsEvents( + mock_session, + org_id="o1", + subscription_watchdog_timeout=5.0, + rate_limit_backoff=20.0, + throughput_log_interval=400, + ) + assert ws._subscription_watchdog_timeout == 5.0 + assert ws._rate_limit_backoff == 20.0 + assert ws._throughput_log_interval == 400 + class TestLocationChannels: """Tests for public location-level WebSocket channel classes.""" @@ -838,6 +919,19 @@ def test_inherits_from_mist_websocket(self, mock_session) -> None: ws = BleAssetsEvents(mock_session, site_id="s1", map_ids=["m1"]) assert isinstance(ws, _MistWebsocket) + def test_supports_reliability_kwargs(self, mock_session) -> None: + ws = ConnectedClientsEvents( + mock_session, + site_id="s1", + map_ids=["m1"], + subscription_watchdog_timeout=4.0, + rate_limit_backoff=18.0, + throughput_log_interval=150, + ) + assert ws._subscription_watchdog_timeout == 4.0 + assert ws._rate_limit_backoff == 18.0 + assert ws._throughput_log_interval == 150 + class TestSessionChannel: """Tests for the SessionWithUrl WebSocket channel class.""" @@ -851,6 +945,18 @@ def test_inherits_from_mist_websocket(self, mock_session) -> None: ws = SessionWithUrl(mock_session, url="wss://example.com/custom") assert isinstance(ws, _MistWebsocket) + def test_supports_reliability_kwargs(self, mock_session) -> None: + ws = SessionWithUrl( + mock_session, + url="wss://example.com/custom", + subscription_watchdog_timeout=6.0, + rate_limit_backoff=25.0, + throughput_log_interval=300, + ) + assert ws._subscription_watchdog_timeout == 6.0 + assert ws._rate_limit_backoff == 25.0 + assert ws._throughput_log_interval == 300 + # --------------------------------------------------------------------------- # Auto-reconnect @@ -951,11 +1057,83 @@ def disconnect_when_ready(): assert entered_backoff.is_set(), "Backoff was never entered" assert client._queue.get_nowait() is None - def test_handle_open_resets_reconnect_attempts(self, mock_session) -> None: + def test_reconnect_counter_resets_after_all_channels_subscribed( + self, mock_session + ) -> None: client = self._make_client(mock_session, max_reconnect_attempts=3) client._reconnect_attempts = 5 + client._last_close_code = 1006 + client._last_close_msg = "previous failure" + mock_ws = Mock() + + client._handle_open(mock_ws) + + assert client._reconnect_attempts == 5 + assert client._last_close_code == 1006 + assert client._last_close_msg == "previous failure" + + client._process_subscription_event( + mock_ws, {"event": "channel_subscribed", "channel": "/ch"} + ) + + assert client._reconnect_attempts == 0 + assert client._last_close_code is None + assert client._last_close_msg is None + + def test_handle_open_resets_reconnect_attempts_without_channels( + self, mock_session + ) -> None: + client = self._make_client(mock_session, channels=[], max_reconnect_attempts=3) + client._reconnect_attempts = 5 + client._last_close_code = 1006 + client._last_close_msg = "previous failure" + client._handle_open(Mock()) + assert client._reconnect_attempts == 0 + assert client._last_close_code is None + assert client._last_close_msg is None + + def test_subscription_failures_honor_retry_limit_and_backoff( + self, mock_session + ) -> None: + client = self._make_client( + mock_session, max_reconnect_attempts=2, reconnect_backoff=0.01 + ) + call_count = 0 + observed_delays: list[float] = [] + + def fake_run_forever(**kwargs): + nonlocal call_count + call_count += 1 + client._handle_open(client._ws) + client._process_subscription_event( + client._ws, + { + "event": "subscribe_failed", + "channel": "/ch", + "detail": "denied", + }, + ) + client._handle_close(client._ws, None, None) + + def capture_delay(timeout=None): + if timeout is not None: + observed_delays.append(timeout) + return False + + mock_ws = Mock() + mock_ws.run_forever.side_effect = fake_run_forever + with ( + patch.object(client, "_create_ws_app", return_value=mock_ws), + patch.object(client, "_arm_subscription_watchdog"), + patch.object(client._user_disconnect, "wait", side_effect=capture_delay), + ): + client._ws = mock_ws + client._run_forever_safe() + + assert call_count == 3 # initial connection plus two retries + assert observed_delays == pytest.approx([0.01, 0.02]) def test_successful_reconnect_resets_counter(self, mock_session) -> None: client = self._make_client(mock_session, max_reconnect_attempts=2) @@ -970,10 +1148,18 @@ def fake_run_forever(**kwargs): elif call_count == 2: # Reconnect succeeds, then simulate open + later drop client._handle_open(client._ws) + client._process_subscription_event( + client._ws, + {"event": "channel_subscribed", "channel": "/ch"}, + ) client._handle_close(client._ws, 1006, "drop again") elif call_count == 3: # Another reconnect succeeds, then clean exit client._handle_open(client._ws) + client._process_subscription_event( + client._ws, + {"event": "channel_subscribed", "channel": "/ch"}, + ) client._handle_close(client._ws, 1006, "drop again") elif call_count == 4: # Final reconnect succeeds then user disconnects @@ -1209,12 +1395,23 @@ class TestQueueCallbackBehavior: def test_message_callback_skips_queue(self, ws_client) -> None: cb = Mock() - ws_client.on_message(cb) + called = threading.Event() + + def cb_wrapper(data): + cb(data) + called.set() + + ws_client.on_message(cb_wrapper) + ws_client._finished.clear() # keep worker alive for this assertion + ws_client._start_callback_worker() ws_client._handle_message(Mock(), '{"event": "data"}') + assert called.wait(timeout=1), "callback was not invoked by worker" cb.assert_called_once_with({"event": "data"}) assert ws_client._queue.empty() + ws_client.disconnect(wait=True, timeout=1) + def test_no_callback_uses_queue(self, ws_client) -> None: ws_client._handle_message(Mock(), '{"event": "data"}') assert not ws_client._queue.empty() @@ -1457,3 +1654,120 @@ def blocking_run_forever(**kwargs): barrier.set() # release blocking thread t.join(timeout=5) assert not t.is_alive() + + +# --------------------------------------------------------------------------- +# Reliability behaviors (callback worker, watchdog, subscribe_failed) +# --------------------------------------------------------------------------- + + +class TestReliabilityBehavior: + """Tests for the callback worker lifecycle and subscription safeguards.""" + + def test_callback_worker_drains_queue_on_stop(self, ws_client) -> None: + received = [] + ws_client.on_message(received.append) + for i in range(3): + ws_client._callback_queue.put_nowait({"n": i}) + ws_client._callback_queue.put_nowait(None) # end-of-stream sentinel + ws_client._callback_stop.set() + + ws_client._run_callback_worker() # run synchronously + + assert received == [{"n": 0}, {"n": 1}, {"n": 2}] + + def test_start_callback_worker_refuses_after_stop(self, ws_client) -> None: + ws_client._callback_stop.set() + ws_client._start_callback_worker() + assert ws_client._callback_thread is None + + def test_watchdog_timeout_reports_on_error_and_closes(self, mock_session) -> None: + client = _MistWebsocket( + mock_session, + channels=["/ch1", "/ch2"], + subscription_watchdog_timeout=0.05, + ) + errors = [] + error_seen = threading.Event() + + def on_error(exc): + errors.append(exc) + error_seen.set() + + client.on_error(on_error) + mock_ws = Mock() + client._ws = mock_ws + + client._arm_subscription_watchdog(mock_ws) + + assert error_seen.wait(timeout=2), "watchdog did not fire" + assert isinstance(errors[0], TimeoutError) + assert "subscription watchdog timeout" in str(errors[0]) + mock_ws.close.assert_called_once() + assert client._last_close_code == 1008 + + def test_watchdog_cancelled_when_all_channels_subscribed( + self, mock_session + ) -> None: + client = _MistWebsocket( + mock_session, + channels=["/ch1"], + subscription_watchdog_timeout=0.05, + ) + errors = [] + client.on_error(errors.append) + mock_ws = Mock() + client._ws = mock_ws + + client._arm_subscription_watchdog(mock_ws) + client._process_subscription_event( + mock_ws, {"event": "channel_subscribed", "channel": "/ch1"} + ) + + threading.Event().wait(timeout=0.2) # let a stray timer fire if any + assert errors == [] + mock_ws.close.assert_not_called() + + def test_unexpected_ack_does_not_count_toward_subscription_completion( + self, mock_session + ) -> None: + client = _MistWebsocket( + mock_session, + channels=["/ch1", "/ch2"], + ) + mock_ws = Mock() + watchdog = Mock() + client._subscription_watchdog = watchdog + + client._process_subscription_event( + mock_ws, + {"event": "channel_subscribed", "channel": "/unexpected"}, + ) + client._process_subscription_event( + mock_ws, + {"event": "channel_subscribed", "channel": "/ch1"}, + ) + + assert client._subscribed_channels == {"/ch1"} + assert client._subscription_watchdog is watchdog + watchdog.cancel.assert_not_called() + + def test_subscribe_failed_reports_on_error_and_closes(self, ws_client) -> None: + errors = [] + ws_client.on_error(errors.append) + mock_ws = Mock() + + ws_client._process_subscription_event( + mock_ws, + { + "event": "subscribe_failed", + "channel": "/test/channel1", + "detail": "denied", + }, + ) + + assert len(errors) == 1 + assert isinstance(errors[0], ConnectionError) + assert "/test/channel1" in str(errors[0]) + mock_ws.close.assert_called_once() + assert ws_client._last_close_code == 1008 diff --git a/uv.lock b/uv.lock index 2233da5..8866d8e 100644 --- a/uv.lock +++ b/uv.lock @@ -552,7 +552,7 @@ wheels = [ [[package]] name = "mistapi" -version = "0.63.2" +version = "0.63.3" source = { editable = "." } dependencies = [ { name = "deprecation" },