mirror of
https://github.com/blakeblackshear/frigate.git
synced 2026-10-13 10:12:47 +03:00
Refactor MQTT (#24010)
* refactor mqtt so that Frigate owns the transport lifecycle instead of delegating it to paho * release the shutdown barrier on worker crash and replay retained publishes the broker never acked * collapse in-flight retained values by topic and release the shutdown barrier from a finally * replay the outage buffer before the publish queue so newer values are not reverted
This commit is contained in:
Josh Hawkins
+668
-167
@@ -1,7 +1,12 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import queue
|
||||
import threading
|
||||
import time
|
||||
from collections.abc import Callable
|
||||
from typing import Any
|
||||
from dataclasses import dataclass
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
import paho.mqtt.client as mqtt
|
||||
from paho.mqtt.enums import CallbackAPIVersion
|
||||
@@ -9,8 +14,25 @@ from paho.mqtt.enums import CallbackAPIVersion
|
||||
from frigate.comms.base_communicator import Communicator
|
||||
from frigate.config import FrigateConfig, birdseye_modes_to_mqtt_payload
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from frigate.comms.dispatcher import Dispatcher
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
MQTT_LOOP_TIMEOUT = 1.0
|
||||
MQTT_RECONNECT_INTERVAL = 10.0
|
||||
MQTT_SHUTDOWN_FLUSH_TIMEOUT = 5.0
|
||||
MQTT_ON_CONNECT_RATE_LIMIT = 1.0
|
||||
MQTT_PUBLISH_WAIT_INTERVAL = 0.1
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class QueuedPublish:
|
||||
topic: str
|
||||
payload: Any
|
||||
retain: bool
|
||||
done: threading.Event | None = None
|
||||
|
||||
|
||||
class MqttClient(Communicator):
|
||||
"""Frigate wrapper for mqtt client."""
|
||||
@@ -19,28 +41,80 @@ class MqttClient(Communicator):
|
||||
self.config = config
|
||||
self.mqtt_config = config.mqtt
|
||||
self.connected = False
|
||||
self.client: mqtt.Client | None = None
|
||||
self._dispatcher: Callable[[str, Any], Any] | None = None
|
||||
self._command_router: Dispatcher | None = None
|
||||
self._worker: threading.Thread | None = None
|
||||
self._stop_event = threading.Event()
|
||||
self._publish_queue: queue.Queue[QueuedPublish] = queue.Queue()
|
||||
self._callback_queue: queue.Queue[tuple[Any, ...]] = queue.Queue()
|
||||
self._retained_lock = threading.Lock()
|
||||
self._pending_retained: dict[str, tuple[Any, bool]] = {}
|
||||
self._inflight_retained: dict[int, tuple[str, Any]] = {}
|
||||
self._subscription_mid: int | None = None
|
||||
self._subscription_ready = False
|
||||
self._next_connect_time = 0.0
|
||||
self._last_on_connect_dispatch = 0.0
|
||||
|
||||
def subscribe(self, receiver: Callable) -> None:
|
||||
"""Wrapper for allowing dispatcher to subscribe."""
|
||||
self._dispatcher = receiver
|
||||
self._start()
|
||||
|
||||
def attach_dispatcher(self, dispatcher: Dispatcher) -> None:
|
||||
"""Take Dispatcher's command surface and snapshot API."""
|
||||
self._command_router = dispatcher
|
||||
|
||||
def start(self) -> None:
|
||||
"""Start the MQTT worker after all receiver wiring is complete."""
|
||||
|
||||
if self._worker and self._worker.is_alive():
|
||||
return
|
||||
|
||||
self._stop_event.clear()
|
||||
self._start_worker()
|
||||
|
||||
def publish(self, topic: str, payload: Any, retain: bool = False) -> None:
|
||||
"""Wrapper for publishing when client is in valid state."""
|
||||
full_topic = f"{self.mqtt_config.topic_prefix}/{topic}"
|
||||
|
||||
if not self.connected:
|
||||
logger.debug(f"Unable to publish to {topic}: client is not connected")
|
||||
if retain:
|
||||
self._queue_retained(full_topic, payload, retain)
|
||||
else:
|
||||
logger.debug("Unable to publish to %s: client is not connected", topic)
|
||||
return
|
||||
|
||||
self.client.publish(
|
||||
f"{self.mqtt_config.topic_prefix}/{topic}",
|
||||
payload,
|
||||
qos=self.config.mqtt.qos,
|
||||
retain=retain,
|
||||
)
|
||||
self._publish_queue.put(QueuedPublish(full_topic, payload, retain))
|
||||
|
||||
def stop(self) -> None:
|
||||
self.publish("available", "stopped", retain=True)
|
||||
self.client.disconnect()
|
||||
if self._worker is None:
|
||||
return
|
||||
|
||||
if self.connected and self._subscription_ready:
|
||||
publish_done = threading.Event()
|
||||
self._publish_queue.put(
|
||||
QueuedPublish(
|
||||
f"{self.mqtt_config.topic_prefix}/available",
|
||||
"stopped",
|
||||
True,
|
||||
publish_done,
|
||||
)
|
||||
)
|
||||
publish_done.wait(MQTT_SHUTDOWN_FLUSH_TIMEOUT)
|
||||
|
||||
self._stop_event.set()
|
||||
|
||||
if self.client is not None:
|
||||
try:
|
||||
self.client.disconnect()
|
||||
except Exception:
|
||||
logger.debug("MQTT disconnect raised during shutdown", exc_info=True)
|
||||
|
||||
if self._worker.is_alive():
|
||||
self._worker.join(MQTT_SHUTDOWN_FLUSH_TIMEOUT + MQTT_LOOP_TIMEOUT)
|
||||
|
||||
self._cleanup_client()
|
||||
self._worker = None
|
||||
|
||||
def _notifications_enabled_in_config(self) -> bool:
|
||||
"""Whether notifications are configured globally or on any camera.
|
||||
@@ -54,17 +128,17 @@ class MqttClient(Communicator):
|
||||
for cam in self.config.cameras.values()
|
||||
)
|
||||
|
||||
def _set_initial_topics(self) -> None:
|
||||
"""Set initial state topics."""
|
||||
def _publish_retained_state(self) -> None:
|
||||
"""Publish retained MQTT state after a successful subscribe."""
|
||||
for camera_name, camera in self.config.cameras.items():
|
||||
self.publish(
|
||||
f"{camera_name}/enabled/state",
|
||||
"ON" if camera.enabled_in_config else "OFF",
|
||||
"ON" if camera.enabled else "OFF",
|
||||
retain=True,
|
||||
)
|
||||
self.publish(
|
||||
f"{camera_name}/recordings/state",
|
||||
"ON" if camera.record.enabled_in_config else "OFF",
|
||||
"ON" if camera.record.enabled else "OFF",
|
||||
retain=True,
|
||||
)
|
||||
self.publish(
|
||||
@@ -74,7 +148,7 @@ class MqttClient(Communicator):
|
||||
)
|
||||
self.publish(
|
||||
f"{camera_name}/audio/state",
|
||||
"ON" if camera.audio.enabled_in_config else "OFF",
|
||||
"ON" if camera.audio.enabled else "OFF",
|
||||
retain=True,
|
||||
)
|
||||
self.publish(
|
||||
@@ -89,7 +163,7 @@ class MqttClient(Communicator):
|
||||
)
|
||||
self.publish(
|
||||
f"{camera_name}/motion/state",
|
||||
"ON",
|
||||
"ON" if camera.motion.enabled else "OFF",
|
||||
retain=True,
|
||||
)
|
||||
self.publish(
|
||||
@@ -99,7 +173,7 @@ class MqttClient(Communicator):
|
||||
)
|
||||
self.publish(
|
||||
f"{camera_name}/ptz_autotracker/state",
|
||||
"ON" if camera.onvif.autotracking.enabled_in_config else "OFF",
|
||||
"ON" if camera.onvif.autotracking.enabled else "OFF",
|
||||
retain=True,
|
||||
)
|
||||
self.publish(
|
||||
@@ -133,22 +207,22 @@ class MqttClient(Communicator):
|
||||
)
|
||||
self.publish(
|
||||
f"{camera_name}/review_alerts/state",
|
||||
"ON" if camera.review.alerts.enabled_in_config else "OFF",
|
||||
"ON" if camera.review.alerts.enabled else "OFF",
|
||||
retain=True,
|
||||
)
|
||||
self.publish(
|
||||
f"{camera_name}/review_detections/state",
|
||||
"ON" if camera.review.detections.enabled_in_config else "OFF",
|
||||
"ON" if camera.review.detections.enabled else "OFF",
|
||||
retain=True,
|
||||
)
|
||||
self.publish(
|
||||
f"{camera_name}/object_descriptions/state",
|
||||
"ON" if camera.objects.genai.enabled_in_config else "OFF",
|
||||
"ON" if camera.objects.genai.enabled else "OFF",
|
||||
retain=True,
|
||||
)
|
||||
self.publish(
|
||||
f"{camera_name}/review_descriptions/state",
|
||||
"ON" if camera.review.genai.enabled_in_config else "OFF",
|
||||
"ON" if camera.review.genai.enabled else "OFF",
|
||||
retain=True,
|
||||
)
|
||||
|
||||
@@ -189,13 +263,521 @@ class MqttClient(Communicator):
|
||||
)
|
||||
self.publish("available", "online", retain=True)
|
||||
|
||||
def on_mqtt_command(
|
||||
self, client: mqtt.Client, userdata: Any, message: mqtt.MQTTMessage
|
||||
) -> None:
|
||||
self._dispatcher(
|
||||
message.topic.replace(f"{self.mqtt_config.topic_prefix}/", "", 1),
|
||||
message.payload.decode(),
|
||||
def _create_client(self) -> mqtt.Client:
|
||||
"""Build a fresh paho client for a single connect attempt."""
|
||||
client = mqtt.Client(
|
||||
callback_api_version=CallbackAPIVersion.VERSION2,
|
||||
client_id=self.mqtt_config.client_id,
|
||||
reconnect_on_failure=False,
|
||||
)
|
||||
client.on_connect = self._on_connect
|
||||
client.on_disconnect = self._on_disconnect
|
||||
client.on_message = self._on_message
|
||||
client.on_subscribe = self._on_subscribe
|
||||
client.on_publish = self._on_publish
|
||||
client.will_set(
|
||||
self.mqtt_config.topic_prefix + "/available",
|
||||
payload="offline",
|
||||
qos=1,
|
||||
retain=True,
|
||||
)
|
||||
|
||||
if self.mqtt_config.tls_ca_certs is not None:
|
||||
if (
|
||||
self.mqtt_config.tls_client_cert is not None
|
||||
and self.mqtt_config.tls_client_key is not None
|
||||
):
|
||||
client.tls_set(
|
||||
self.mqtt_config.tls_ca_certs,
|
||||
self.mqtt_config.tls_client_cert,
|
||||
self.mqtt_config.tls_client_key,
|
||||
)
|
||||
else:
|
||||
client.tls_set(self.mqtt_config.tls_ca_certs)
|
||||
|
||||
if self.mqtt_config.tls_insecure is not None:
|
||||
client.tls_insecure_set(self.mqtt_config.tls_insecure)
|
||||
|
||||
if self.mqtt_config.user is not None:
|
||||
client.username_pw_set(
|
||||
self.mqtt_config.user,
|
||||
password=self.mqtt_config.password,
|
||||
)
|
||||
|
||||
return client
|
||||
|
||||
def _start_worker(self) -> None:
|
||||
self._worker = threading.Thread(
|
||||
target=self._worker_main, name="mqtt", daemon=True
|
||||
)
|
||||
self._worker.start()
|
||||
logger.info("MQTT worker started")
|
||||
|
||||
def _worker_main(self) -> None:
|
||||
"""Run the worker loop.
|
||||
|
||||
An unexpected crash disables MQTT for this session rather than taking
|
||||
Frigate down with it, so it has to announce itself: without the offline
|
||||
publish, consumers keep the last retained values and see a healthy
|
||||
Frigate that has simply stopped updating.
|
||||
"""
|
||||
try:
|
||||
self._mqtt_loop_worker()
|
||||
except Exception:
|
||||
if not self._stop_event.is_set():
|
||||
logger.exception("MQTT worker crashed, disabling MQTT for this session")
|
||||
self._stop_event.set()
|
||||
self._subscription_ready = False
|
||||
self._publish_offline_availability()
|
||||
self.connected = False
|
||||
finally:
|
||||
# nothing drains the queue once the loop is gone, so release any
|
||||
# waiter here or stop() blocks for the full flush timeout
|
||||
self._requeue_disconnected_publishes()
|
||||
self._cleanup_client()
|
||||
|
||||
def _publish_offline_availability(self) -> None:
|
||||
"""Announce that MQTT is going away after a worker crash.
|
||||
|
||||
_cleanup_client() disconnects cleanly, which tells the broker to
|
||||
suppress the will, so the retained topic would otherwise stay "online".
|
||||
"""
|
||||
if self.client is None:
|
||||
return
|
||||
|
||||
try:
|
||||
message_info = self.client.publish(
|
||||
f"{self.mqtt_config.topic_prefix}/available",
|
||||
"offline",
|
||||
qos=self.config.mqtt.qos,
|
||||
retain=True,
|
||||
)
|
||||
|
||||
# pumped here rather than through _wait_for_publish() so the drain
|
||||
# that may have just crashed is not re-entered
|
||||
deadline = time.monotonic() + MQTT_SHUTDOWN_FLUSH_TIMEOUT
|
||||
while not message_info.is_published() and time.monotonic() < deadline:
|
||||
if (
|
||||
self.client.loop(timeout=MQTT_PUBLISH_WAIT_INTERVAL)
|
||||
!= mqtt.MQTT_ERR_SUCCESS
|
||||
):
|
||||
break
|
||||
except Exception:
|
||||
logger.warning(
|
||||
"MQTT is dormant and the broker could not be told Frigate is offline",
|
||||
exc_info=True,
|
||||
)
|
||||
|
||||
def _mqtt_loop_worker(self) -> None:
|
||||
# The worker owns all socket I/O so reconnect, subscribe, and publish
|
||||
# ordering stays serialized in one place.
|
||||
while not self._stop_event.is_set():
|
||||
if self.client is None:
|
||||
wait_time = self._next_connect_time - time.monotonic()
|
||||
if wait_time > 0:
|
||||
self._stop_event.wait(min(wait_time, MQTT_LOOP_TIMEOUT))
|
||||
continue
|
||||
|
||||
if not self._connect_client():
|
||||
self._next_connect_time = time.monotonic() + MQTT_RECONNECT_INTERVAL
|
||||
continue
|
||||
|
||||
assert self.client is not None
|
||||
try:
|
||||
result = self.client.loop(timeout=MQTT_LOOP_TIMEOUT)
|
||||
except (OSError, mqtt.WebsocketConnectionError) as err:
|
||||
logger.warning("MQTT loop error: %s", err)
|
||||
self._schedule_reconnect()
|
||||
continue
|
||||
|
||||
self._drain_callback_queue()
|
||||
self._drain_publish_queue()
|
||||
|
||||
if self._stop_event.is_set():
|
||||
break
|
||||
|
||||
if result != mqtt.MQTT_ERR_SUCCESS and self.client is not None:
|
||||
logger.error("MQTT loop returned error code: %s", result)
|
||||
self._schedule_reconnect()
|
||||
|
||||
def _connect_client(self) -> bool:
|
||||
"""Create and connect a new client instance owned by the worker thread."""
|
||||
try:
|
||||
self.client = self._create_client()
|
||||
self.client.connect(self.mqtt_config.host, self.mqtt_config.port, 60)
|
||||
except Exception as err:
|
||||
logger.error("Unable to connect to MQTT server: %s", err)
|
||||
self._cleanup_client()
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
def _cleanup_client(self) -> None:
|
||||
"""Drop session-specific state and release the current paho client."""
|
||||
self.connected = False
|
||||
self._subscription_ready = False
|
||||
self._subscription_mid = None
|
||||
self._requeue_inflight_retained()
|
||||
|
||||
client = self.client
|
||||
self.client = None
|
||||
|
||||
if client is None:
|
||||
return
|
||||
|
||||
try:
|
||||
client.disconnect()
|
||||
except Exception:
|
||||
logger.debug("MQTT client cleanup raised disconnect error", exc_info=True)
|
||||
|
||||
def _schedule_reconnect(self) -> None:
|
||||
"""Tear down the current session and arm the next reconnect attempt."""
|
||||
if self._stop_event.is_set():
|
||||
return
|
||||
|
||||
self.connected = False
|
||||
self._subscription_ready = False
|
||||
self._subscription_mid = None
|
||||
self._requeue_disconnected_publishes()
|
||||
self._next_connect_time = time.monotonic() + MQTT_RECONNECT_INTERVAL
|
||||
logger.info("MQTT reconnect scheduled in %.1fs", MQTT_RECONNECT_INTERVAL)
|
||||
self._cleanup_client()
|
||||
|
||||
def _requeue_inflight_retained(self) -> None:
|
||||
"""Rebuffer retained publishes paho took but the broker never acked.
|
||||
|
||||
Dropping the client drops paho's outbound queue with it, and the session
|
||||
is clean, so the broker will not resume delivery on the new one.
|
||||
"""
|
||||
with self._retained_lock:
|
||||
# mids are insertion ordered, so collapsing by topic keeps the
|
||||
# newest value when several updates to one topic were in flight
|
||||
latest = {
|
||||
topic: payload for topic, payload in self._inflight_retained.values()
|
||||
}
|
||||
self._inflight_retained.clear()
|
||||
|
||||
for topic, payload in latest.items():
|
||||
self._queue_retained(topic, payload, True, overwrite=False)
|
||||
|
||||
def _buffer_undelivered(
|
||||
self, queued_publish: QueuedPublish, overwrite: bool = True
|
||||
) -> None:
|
||||
"""Handle a publish that never reached the broker.
|
||||
|
||||
Releasing the waiter matters on every path: stop() blocks on it, so a
|
||||
broker error would otherwise stall shutdown for the full flush timeout.
|
||||
"""
|
||||
if queued_publish.retain:
|
||||
self._queue_retained(
|
||||
queued_publish.topic,
|
||||
queued_publish.payload,
|
||||
queued_publish.retain,
|
||||
overwrite=overwrite,
|
||||
)
|
||||
|
||||
if queued_publish.done is not None:
|
||||
queued_publish.done.set()
|
||||
|
||||
def _requeue_disconnected_publishes(self) -> None:
|
||||
while True:
|
||||
try:
|
||||
queued_publish = self._publish_queue.get_nowait()
|
||||
except queue.Empty:
|
||||
break
|
||||
|
||||
self._buffer_undelivered(queued_publish)
|
||||
|
||||
def _drain_callback_queue(self) -> None:
|
||||
# Paho callbacks only enqueue transport events; state transitions run
|
||||
# here on the worker thread.
|
||||
while True:
|
||||
try:
|
||||
event = self._callback_queue.get_nowait()
|
||||
except queue.Empty:
|
||||
break
|
||||
|
||||
event_type = event[0]
|
||||
|
||||
if event_type == "connect":
|
||||
self._handle_connect_event(event[1])
|
||||
elif event_type == "connect_failure":
|
||||
self._handle_connect_failure(event[1])
|
||||
elif event_type == "disconnect":
|
||||
self._handle_disconnect_event(event[1])
|
||||
elif event_type == "subscribed":
|
||||
self._handle_subscribe_event(event[1], event[2])
|
||||
elif event_type == "message":
|
||||
self._handle_inbound_message(event[1], event[2])
|
||||
elif event_type == "published":
|
||||
self._handle_publish_event(event[1])
|
||||
|
||||
def _drain_publish_queue(self) -> None:
|
||||
"""Publish queued work only after the session is fully subscribed.
|
||||
|
||||
Oldest first: the outage buffer replays before the queue, so a topic
|
||||
that changed since the reconnect ends up on its newest value rather
|
||||
than being reverted by the replay.
|
||||
"""
|
||||
if self.connected and not self._subscription_ready:
|
||||
return
|
||||
|
||||
self._flush_pending_retained()
|
||||
|
||||
while True:
|
||||
try:
|
||||
queued_publish = self._publish_queue.get_nowait()
|
||||
except queue.Empty:
|
||||
break
|
||||
|
||||
if not self.connected:
|
||||
self._buffer_undelivered(queued_publish)
|
||||
continue
|
||||
|
||||
self._publish_direct(queued_publish)
|
||||
|
||||
def _flush_pending_retained(self) -> None:
|
||||
"""Replay the latest retained state once the broker session is ready."""
|
||||
if not self.connected or not self._subscription_ready:
|
||||
return
|
||||
|
||||
with self._retained_lock:
|
||||
pending = list(self._pending_retained.items())
|
||||
self._pending_retained.clear()
|
||||
|
||||
for topic, (payload, retain) in pending:
|
||||
self._publish_direct(QueuedPublish(topic, payload, retain))
|
||||
|
||||
def _publish_direct(self, queued_publish: QueuedPublish) -> None:
|
||||
"""Publish a queued message from the worker thread's serialized context.
|
||||
|
||||
The waiter is released however this exits. The message is already off
|
||||
the queue by now, so nothing else can recover it for a stop() that is
|
||||
blocked waiting on it.
|
||||
"""
|
||||
try:
|
||||
if self.client is None:
|
||||
# never attempted, so anything already buffered for this topic
|
||||
# was written later and has to survive
|
||||
self._buffer_undelivered(queued_publish, overwrite=False)
|
||||
return
|
||||
|
||||
try:
|
||||
message_info = self.client.publish(
|
||||
queued_publish.topic,
|
||||
queued_publish.payload,
|
||||
qos=self.config.mqtt.qos,
|
||||
retain=queued_publish.retain,
|
||||
)
|
||||
except (OSError, mqtt.WebsocketConnectionError) as err:
|
||||
logger.warning(
|
||||
"MQTT publish failed for %s: %s", queued_publish.topic, err
|
||||
)
|
||||
# a newer buffered value for this topic wins over the failed one
|
||||
self._buffer_undelivered(queued_publish, overwrite=False)
|
||||
self._schedule_reconnect()
|
||||
return
|
||||
|
||||
if message_info.rc != mqtt.MQTT_ERR_SUCCESS:
|
||||
logger.error(
|
||||
"Unable to publish to %s: mqtt error %s",
|
||||
queued_publish.topic,
|
||||
message_info.rc,
|
||||
)
|
||||
self._buffer_undelivered(queued_publish, overwrite=False)
|
||||
self._schedule_reconnect()
|
||||
return
|
||||
|
||||
# a successful rc only means paho accepted the message; above qos 0
|
||||
# it is not durable until the broker acks, so keep a copy for replay
|
||||
if queued_publish.retain and not message_info.is_published():
|
||||
with self._retained_lock:
|
||||
self._inflight_retained[message_info.mid] = (
|
||||
queued_publish.topic,
|
||||
queued_publish.payload,
|
||||
)
|
||||
|
||||
if queued_publish.done is not None:
|
||||
self._wait_for_publish(message_info)
|
||||
finally:
|
||||
if queued_publish.done is not None:
|
||||
queued_publish.done.set()
|
||||
|
||||
def _handle_publish_event(self, mid: int) -> None:
|
||||
"""Drop the replay copy once the broker has acknowledged the message."""
|
||||
with self._retained_lock:
|
||||
self._inflight_retained.pop(mid, None)
|
||||
|
||||
def _wait_for_publish(self, message_info: mqtt.MQTTMessageInfo) -> None:
|
||||
"""Pump the loop until a shutdown-critical publish is acknowledged."""
|
||||
deadline = time.monotonic() + MQTT_SHUTDOWN_FLUSH_TIMEOUT
|
||||
|
||||
while not message_info.is_published() and time.monotonic() < deadline:
|
||||
if self.client is None:
|
||||
return
|
||||
|
||||
try:
|
||||
result = self.client.loop(timeout=MQTT_PUBLISH_WAIT_INTERVAL)
|
||||
except (OSError, mqtt.WebsocketConnectionError) as err:
|
||||
logger.warning("MQTT publish wait failed: %s", err)
|
||||
self._schedule_reconnect()
|
||||
return
|
||||
|
||||
self._drain_callback_queue()
|
||||
|
||||
if result != mqtt.MQTT_ERR_SUCCESS:
|
||||
logger.error(
|
||||
"MQTT loop returned error code while waiting for publish: %s",
|
||||
result,
|
||||
)
|
||||
self._schedule_reconnect()
|
||||
return
|
||||
|
||||
def _queue_retained(
|
||||
self,
|
||||
topic: str,
|
||||
payload: Any,
|
||||
retain: bool,
|
||||
overwrite: bool = True,
|
||||
) -> None:
|
||||
"""Store the last retained value per topic for replay after reconnect."""
|
||||
with self._retained_lock:
|
||||
if overwrite or topic not in self._pending_retained:
|
||||
self._pending_retained[topic] = (payload, retain)
|
||||
|
||||
def _handle_connect_event(self, reason_code: mqtt.ReasonCode) -> None: # type: ignore[name-defined]
|
||||
"""Begin a new session by subscribing before any replay is published."""
|
||||
if self.client is None:
|
||||
return
|
||||
|
||||
self.connected = True
|
||||
self._subscription_ready = False
|
||||
self._subscription_mid = None
|
||||
logger.debug("MQTT connected")
|
||||
|
||||
try:
|
||||
result, mid = self.client.subscribe(
|
||||
f"{self.mqtt_config.topic_prefix}/#",
|
||||
qos=self.config.mqtt.qos,
|
||||
)
|
||||
except (OSError, mqtt.WebsocketConnectionError) as err:
|
||||
logger.warning("MQTT subscribe failed: %s", err)
|
||||
self._schedule_reconnect()
|
||||
return
|
||||
|
||||
if result != mqtt.MQTT_ERR_SUCCESS:
|
||||
logger.error(
|
||||
"Unable to subscribe to MQTT command tree: mqtt error %s", result
|
||||
)
|
||||
self._schedule_reconnect()
|
||||
return
|
||||
|
||||
self._subscription_mid = mid
|
||||
|
||||
def _handle_connect_failure(self, reason_code: mqtt.ReasonCode) -> None: # type: ignore[name-defined]
|
||||
"""Record a failed connect attempt and transition into reconnect state."""
|
||||
self.connected = False
|
||||
logger.error(
|
||||
"Unable to connect to MQTT server: %s", self._reason_code_name(reason_code)
|
||||
)
|
||||
self._schedule_reconnect()
|
||||
|
||||
def _handle_disconnect_event(self, reason_code: mqtt.ReasonCode) -> None: # type: ignore[name-defined]
|
||||
"""Handle broker disconnects idempotently from the worker thread."""
|
||||
if not self.connected:
|
||||
return
|
||||
|
||||
self.connected = False
|
||||
self._subscription_ready = False
|
||||
self._subscription_mid = None
|
||||
|
||||
if self._stop_event.is_set():
|
||||
logger.debug("MQTT disconnected")
|
||||
self._cleanup_client()
|
||||
return
|
||||
|
||||
logger.error("MQTT disconnected: %s", self._reason_code_name(reason_code))
|
||||
self._schedule_reconnect()
|
||||
|
||||
def _handle_subscribe_event(
|
||||
self,
|
||||
mid: int,
|
||||
reason_codes: list[mqtt.ReasonCode], # type: ignore[name-defined]
|
||||
) -> None:
|
||||
"""Mark the session ready after SUBACK, then replay retained/runtime state."""
|
||||
if mid != self._subscription_mid:
|
||||
return
|
||||
|
||||
if any(
|
||||
getattr(reason_code, "is_failure", False) for reason_code in reason_codes
|
||||
):
|
||||
logger.error("MQTT subscription was rejected by the broker")
|
||||
self._schedule_reconnect()
|
||||
return
|
||||
|
||||
self._subscription_ready = True
|
||||
self._subscription_mid = None
|
||||
|
||||
# a bug in replay should cost a snapshot, not the MQTT session
|
||||
try:
|
||||
self._publish_retained_state()
|
||||
|
||||
if self._command_router is not None:
|
||||
self._command_router.publish_runtime_snapshot(self.publish)
|
||||
except Exception:
|
||||
logger.exception("Error replaying MQTT state after subscribe")
|
||||
|
||||
def _handle_inbound_message(self, topic: str, payload: str) -> None:
|
||||
"""Forward supported command topics into Dispatcher semantics."""
|
||||
if self._dispatcher is None:
|
||||
return
|
||||
|
||||
if not self._is_supported_command_topic(topic):
|
||||
return
|
||||
|
||||
if topic == "onConnect":
|
||||
now = time.monotonic()
|
||||
if now - self._last_on_connect_dispatch < MQTT_ON_CONNECT_RATE_LIMIT:
|
||||
logger.debug("Skipping MQTT onConnect replay request due to rate limit")
|
||||
return
|
||||
self._last_on_connect_dispatch = now
|
||||
|
||||
# a raise here used to end the network thread and take MQTT down
|
||||
try:
|
||||
self._dispatcher(topic, payload)
|
||||
except Exception:
|
||||
logger.exception("Error handling MQTT command topic %s", topic)
|
||||
|
||||
def _is_supported_command_topic(self, topic: str) -> bool:
|
||||
"""Filter the wildcard subscription down to Dispatcher's command surface.
|
||||
|
||||
Load-bearing rather than an optimization: the broker echoes Frigate's own
|
||||
publishes back through frigate/#, and Dispatcher republishes topics it
|
||||
does not recognize, so forwarding unfiltered would loop.
|
||||
"""
|
||||
if self._command_router is None:
|
||||
return False
|
||||
|
||||
# mirrors the gate on the state topic in _publish_retained_state()
|
||||
if topic == "notifications/set" and not self._notifications_enabled_in_config():
|
||||
return False
|
||||
|
||||
return self._command_router.is_command_topic(topic)
|
||||
|
||||
def _strip_topic_prefix(self, topic: str) -> str:
|
||||
return topic.replace(f"{self.mqtt_config.topic_prefix}/", "", 1)
|
||||
|
||||
def _is_success_reason_code(self, reason_code: mqtt.ReasonCode) -> bool: # type: ignore[name-defined]
|
||||
if hasattr(reason_code, "is_failure"):
|
||||
return not bool(reason_code.is_failure)
|
||||
|
||||
return bool(reason_code == 0)
|
||||
|
||||
def _reason_code_name(self, reason_code: mqtt.ReasonCode) -> str: # type: ignore[name-defined]
|
||||
if hasattr(reason_code, "getName"):
|
||||
return str(reason_code.getName())
|
||||
|
||||
return str(reason_code)
|
||||
|
||||
def _on_connect(
|
||||
self,
|
||||
@@ -205,29 +787,11 @@ class MqttClient(Communicator):
|
||||
reason_code: mqtt.ReasonCode, # type: ignore[name-defined]
|
||||
properties: Any,
|
||||
) -> None:
|
||||
"""Mqtt connection callback."""
|
||||
threading.current_thread().name = "mqtt"
|
||||
if reason_code != 0:
|
||||
if reason_code == "Server unavailable":
|
||||
logger.error(
|
||||
"Unable to connect to MQTT server: MQTT Server unavailable"
|
||||
)
|
||||
elif reason_code == "Bad user name or password":
|
||||
logger.error(
|
||||
"Unable to connect to MQTT server: MQTT Bad username or password"
|
||||
)
|
||||
elif reason_code == "Not authorized":
|
||||
logger.error("Unable to connect to MQTT server: MQTT Not authorized")
|
||||
else:
|
||||
logger.error(
|
||||
"Unable to connect to MQTT server: Connection refused. Error code: %s",
|
||||
reason_code.getName(),
|
||||
)
|
||||
|
||||
self.connected = True
|
||||
logger.debug("MQTT connected")
|
||||
client.subscribe(f"{self.mqtt_config.topic_prefix}/#", qos=self.config.mqtt.qos)
|
||||
self._set_initial_topics()
|
||||
"""Handle broker connect notifications from paho."""
|
||||
if self._is_success_reason_code(reason_code):
|
||||
self._callback_queue.put(("connect", reason_code))
|
||||
else:
|
||||
self._callback_queue.put(("connect_failure", reason_code))
|
||||
|
||||
def _on_disconnect(
|
||||
self,
|
||||
@@ -237,126 +801,63 @@ class MqttClient(Communicator):
|
||||
reason_code: mqtt.ReasonCode, # type: ignore[name-defined]
|
||||
properties: Any,
|
||||
) -> None:
|
||||
"""Mqtt disconnection callback."""
|
||||
self.connected = False
|
||||
logger.error("MQTT disconnected")
|
||||
"""Handle broker disconnect notifications from paho."""
|
||||
self._callback_queue.put(("disconnect", reason_code))
|
||||
|
||||
def _start(self) -> None:
|
||||
"""Start mqtt client."""
|
||||
self.client = mqtt.Client(
|
||||
callback_api_version=CallbackAPIVersion.VERSION2,
|
||||
client_id=self.mqtt_config.client_id,
|
||||
)
|
||||
self.client.on_connect = self._on_connect
|
||||
self.client.on_disconnect = self._on_disconnect
|
||||
self.client.will_set(
|
||||
self.mqtt_config.topic_prefix + "/available",
|
||||
payload="offline",
|
||||
qos=1,
|
||||
retain=True,
|
||||
)
|
||||
def _on_subscribe(
|
||||
self,
|
||||
client: mqtt.Client,
|
||||
userdata: Any,
|
||||
mid: int,
|
||||
reason_codes: list[mqtt.ReasonCode], # type: ignore[name-defined]
|
||||
properties: Any,
|
||||
) -> None:
|
||||
"""Handle subscribe acknowledgements from paho."""
|
||||
self._callback_queue.put(("subscribed", mid, reason_codes))
|
||||
|
||||
# register callbacks
|
||||
callback_types = [
|
||||
"enabled",
|
||||
"recordings",
|
||||
"snapshots",
|
||||
"detect",
|
||||
"audio",
|
||||
"audio_transcription",
|
||||
"motion",
|
||||
"improve_contrast",
|
||||
"ptz_autotracker",
|
||||
"motion_threshold",
|
||||
"motion_contour_area",
|
||||
"birdseye",
|
||||
"birdseye_modes",
|
||||
"review_alerts",
|
||||
"review_detections",
|
||||
"object_descriptions",
|
||||
"review_descriptions",
|
||||
"notifications",
|
||||
]
|
||||
def _on_publish(
|
||||
self,
|
||||
client: mqtt.Client,
|
||||
userdata: Any,
|
||||
mid: int,
|
||||
reason_code: mqtt.ReasonCode, # type: ignore[name-defined]
|
||||
properties: Any,
|
||||
) -> None:
|
||||
"""Handle publish acknowledgements from paho.
|
||||
|
||||
for name in self.config.cameras.keys():
|
||||
for callback in callback_types:
|
||||
self.client.message_callback_add(
|
||||
f"{self.mqtt_config.topic_prefix}/{name}/{callback}/set",
|
||||
self.on_mqtt_command,
|
||||
)
|
||||
Only tracked retained messages need an event. At the default qos 0
|
||||
nothing is tracked, so this stays off the hot publish path.
|
||||
"""
|
||||
with self._retained_lock:
|
||||
if mid not in self._inflight_retained:
|
||||
return
|
||||
|
||||
# notifications suspend doesn't follow the /set topic pattern
|
||||
self.client.message_callback_add(
|
||||
f"{self.mqtt_config.topic_prefix}/{name}/notifications/suspend",
|
||||
self.on_mqtt_command,
|
||||
)
|
||||
self._callback_queue.put(("published", mid))
|
||||
|
||||
if self.config.cameras[name].onvif.host:
|
||||
self.client.message_callback_add(
|
||||
f"{self.mqtt_config.topic_prefix}/{name}/ptz",
|
||||
self.on_mqtt_command,
|
||||
)
|
||||
def _on_message(
|
||||
self,
|
||||
client: mqtt.Client,
|
||||
userdata: Any,
|
||||
message: mqtt.MQTTMessage,
|
||||
) -> None:
|
||||
"""Queue inbound MQTT messages for processing in the worker loop."""
|
||||
topic = self._strip_topic_prefix(message.topic)
|
||||
|
||||
for mask_name in self.config.cameras[name].motion.mask.keys():
|
||||
self.client.message_callback_add(
|
||||
f"{self.mqtt_config.topic_prefix}/{name}/motion_mask/{mask_name}/set",
|
||||
self.on_mqtt_command,
|
||||
)
|
||||
|
||||
for mask_name in self.config.cameras[name].objects.mask.keys():
|
||||
self.client.message_callback_add(
|
||||
f"{self.mqtt_config.topic_prefix}/{name}/object_mask/{mask_name}/set",
|
||||
self.on_mqtt_command,
|
||||
)
|
||||
|
||||
for zone_name in self.config.cameras[name].zones.keys():
|
||||
self.client.message_callback_add(
|
||||
f"{self.mqtt_config.topic_prefix}/{name}/zone/{zone_name}/set",
|
||||
self.on_mqtt_command,
|
||||
)
|
||||
|
||||
if self._notifications_enabled_in_config():
|
||||
self.client.message_callback_add(
|
||||
f"{self.mqtt_config.topic_prefix}/notifications/set",
|
||||
self.on_mqtt_command,
|
||||
)
|
||||
|
||||
self.client.message_callback_add(
|
||||
f"{self.mqtt_config.topic_prefix}/profile/set",
|
||||
self.on_mqtt_command,
|
||||
)
|
||||
|
||||
self.client.message_callback_add(
|
||||
f"{self.mqtt_config.topic_prefix}/onConnect", self.on_mqtt_command
|
||||
)
|
||||
|
||||
self.client.message_callback_add(
|
||||
f"{self.mqtt_config.topic_prefix}/restart", self.on_mqtt_command
|
||||
)
|
||||
|
||||
if self.mqtt_config.tls_ca_certs is not None:
|
||||
if (
|
||||
self.mqtt_config.tls_client_cert is not None
|
||||
and self.mqtt_config.tls_client_key is not None
|
||||
):
|
||||
self.client.tls_set(
|
||||
self.mqtt_config.tls_ca_certs,
|
||||
self.mqtt_config.tls_client_cert,
|
||||
self.mqtt_config.tls_client_key,
|
||||
)
|
||||
else:
|
||||
self.client.tls_set(self.mqtt_config.tls_ca_certs)
|
||||
if self.mqtt_config.tls_insecure is not None:
|
||||
self.client.tls_insecure_set(self.mqtt_config.tls_insecure)
|
||||
if self.mqtt_config.user is not None:
|
||||
self.client.username_pw_set(
|
||||
self.mqtt_config.user, password=self.mqtt_config.password
|
||||
)
|
||||
try:
|
||||
# https://stackoverflow.com/a/55390477
|
||||
# with connect_async, retries are handled automatically
|
||||
self.client.connect_async(self.mqtt_config.host, self.mqtt_config.port, 60)
|
||||
self.client.loop_start()
|
||||
except Exception as e:
|
||||
logger.error(f"Unable to connect to MQTT server: {e}")
|
||||
# Ignore everything outside Frigate's command surface before decoding or
|
||||
# dispatching into the rest of the app.
|
||||
if not self._is_supported_command_topic(topic):
|
||||
return
|
||||
|
||||
try:
|
||||
payload = message.payload.decode()
|
||||
except UnicodeDecodeError:
|
||||
logger.debug("Ignoring non-UTF-8 MQTT payload for topic %s", topic)
|
||||
return
|
||||
|
||||
self._callback_queue.put(
|
||||
(
|
||||
"message",
|
||||
topic,
|
||||
payload,
|
||||
)
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user