🚧 WIP
This commit is contained in:
+49
-7
@@ -65,8 +65,15 @@ class DynsecClient:
|
||||
# -- connection lifecycle ------------------------------------------------
|
||||
def connect(self) -> None:
|
||||
with self._lock:
|
||||
if self._client is not None:
|
||||
if self._client is not None and self._connected.is_set():
|
||||
return
|
||||
if self._client is not None:
|
||||
# The previous client lost its connection (broker restart,
|
||||
# kicked credentials). paho's auto-reconnect retries with
|
||||
# the SAME credentials forever, so it can never recover
|
||||
# from a stale password — tear it down and start fresh,
|
||||
# which also re-reads the current settings.
|
||||
self._teardown_locked()
|
||||
settings = get_settings()
|
||||
client = mqtt.Client(
|
||||
callback_api_version=CallbackAPIVersion.VERSION2,
|
||||
@@ -76,24 +83,40 @@ class DynsecClient:
|
||||
settings.mosquitto_dynsec_username, settings.mosquitto_dynsec_password
|
||||
)
|
||||
client.on_connect = self._on_connect
|
||||
client.on_disconnect = self._on_disconnect
|
||||
client.on_message = self._on_message
|
||||
# Register BEFORE loop_start so the guard in the callbacks
|
||||
# (client is not self._client) recognises this client from
|
||||
# the very first callback onward.
|
||||
self._client = client
|
||||
client.connect_async(
|
||||
settings.mosquitto_host, settings.mosquitto_mgmt_port, keepalive=30
|
||||
)
|
||||
client.loop_start()
|
||||
if not self._connected.wait(timeout=_RESPONSE_TIMEOUT_S):
|
||||
client.loop_stop()
|
||||
self._teardown_locked()
|
||||
raise DynsecError("could not connect to broker management channel")
|
||||
self._client = client
|
||||
|
||||
def _teardown_locked(self) -> None:
|
||||
"""Stop and discard the current paho client. Call with _lock held."""
|
||||
client = self._client
|
||||
assert client is not None
|
||||
# Null the reference FIRST: _on_disconnect's guard then treats any
|
||||
# late callback from this client as stale and ignores it, so a
|
||||
# deliberate teardown cannot clear a successor's connected flag.
|
||||
self._client = None
|
||||
self._connected.clear()
|
||||
try:
|
||||
client.disconnect()
|
||||
except OSError:
|
||||
pass # socket may already be gone
|
||||
client.loop_stop()
|
||||
|
||||
def close(self) -> None:
|
||||
with self._lock:
|
||||
if self._client is None:
|
||||
return
|
||||
self._client.disconnect()
|
||||
self._client.loop_stop()
|
||||
self._client = None
|
||||
self._connected.clear()
|
||||
self._teardown_locked()
|
||||
|
||||
# -- paho callbacks -------------------------------------------------------
|
||||
def _on_connect(self, client, userdata, flags, reason_code, properties) -> None:
|
||||
@@ -105,6 +128,25 @@ class DynsecClient:
|
||||
else:
|
||||
logger.error("dynsec client connect failed: %s", reason_code)
|
||||
|
||||
def _on_disconnect(
|
||||
self, client, userdata, disconnect_flags, reason_code, properties
|
||||
) -> None:
|
||||
# paho VERSION2 signature: all params are required even when unused.
|
||||
if client is not self._client:
|
||||
return # stale callback from a client already torn down
|
||||
# Without this handler a connection loss went completely unnoticed:
|
||||
# the flag stayed set, every command published into the void and
|
||||
# burned its full response timeout (10 s per call). Clearing the
|
||||
# flag makes health/ready report honestly and lets connect()
|
||||
# rebuild the client on the next command. paho's auto-reconnect
|
||||
# still fires _on_connect on success for transient drops.
|
||||
self._connected.clear()
|
||||
logger.warning(
|
||||
"dynsec management channel disconnected (rc=%s); commands will "
|
||||
"fail until the connection is re-established",
|
||||
reason_code,
|
||||
)
|
||||
|
||||
def _on_message(self, client, userdata, msg) -> None:
|
||||
if msg.topic != RESPONSE_TOPIC:
|
||||
return
|
||||
|
||||
@@ -164,6 +164,12 @@ export function ClientsPage() {
|
||||
</div>
|
||||
)}
|
||||
{error && <div className="banner error">{error}</div>}
|
||||
{(clients.error || roles.error || groups.error) && (
|
||||
<div className="banner error">
|
||||
Failed to load clients/roles/groups from the broker management API:{" "}
|
||||
{((clients.error ?? roles.error ?? groups.error) as Error).message}
|
||||
</div>
|
||||
)}
|
||||
|
||||
<form className="stack" onSubmit={submit}>
|
||||
<label htmlFor="new-client">New client username</label>
|
||||
|
||||
@@ -102,6 +102,12 @@ export function GroupsPage() {
|
||||
group manages its own roles and members.
|
||||
</p>
|
||||
{error && <div className="banner error">{error}</div>}
|
||||
{(groups.error || roles.error || clients.error) && (
|
||||
<div className="banner error">
|
||||
Failed to load groups/roles/clients from the broker management API:{" "}
|
||||
{((groups.error ?? roles.error ?? clients.error) as Error).message}
|
||||
</div>
|
||||
)}
|
||||
|
||||
<form className="stack" onSubmit={submit}>
|
||||
<label htmlFor="group-name">New group name</label>
|
||||
|
||||
@@ -142,6 +142,12 @@ export function RolesPage() {
|
||||
username and client id.
|
||||
</p>
|
||||
{error && <div className="banner error">{error}</div>}
|
||||
{(roles.error || detail.error) && (
|
||||
<div className="banner error">
|
||||
Failed to load roles from the broker management API:{" "}
|
||||
{((roles.error ?? detail.error) as Error).message}
|
||||
</div>
|
||||
)}
|
||||
{banner && !error && <div className="banner ok">{banner}</div>}
|
||||
|
||||
<form className="stack" onSubmit={submit}>
|
||||
|
||||
Reference in New Issue
Block a user