from __future__ import annotations
import asyncio
import functools
import json
import logging
import random
import re
import ssl
import time
import xml.sax.saxutils
from collections.abc import Callable, Coroutine
from typing import TYPE_CHECKING, Any
import slixmpp
import slixmpp.exceptions
from slixmpp import JID, ElementBase
from slixmpp.xmlstream import ET
from slixmpp.xmlstream.handler import Callback
from slixmpp.xmlstream.matcher import MatchXMLMask
from pyobs.comm import Comm
from pyobs.events import Event, LogEvent, ModuleClosedEvent, ModuleOpenedEvent
from pyobs.events.event import EventFactory
from pyobs.interfaces import Interface
from pyobs.interfaces.interface import get_registered_interface
from pyobs.utils import exceptions as exc
from pyobs.utils.enums import ModuleState
from .rpc import RPC
from .serializer import _dataclass_to_xml, _event_schema_to_xml, _interface_schema_to_xml, _xml_to_dataclass
from .xmppclient import XmppClient
if TYPE_CHECKING:
from pyobs.modules import Module
log = logging.getLogger(__name__)
def _event_role(ev_cls: type[Event], events_sent: set[type[Event]], events_subscribed: set[type[Event]]) -> str:
"""Space-separated role(s) ("send", "subscribe", or both) for an event class, for the
disco#info `role` attribute -- lets consumers with no access to the pyobs.events catalog
(e.g. pyobs-web-client) tell producers from consumers instead of guessing from the union."""
roles = []
if ev_cls in events_sent:
roles.append("send")
if ev_cls in events_subscribed:
roles.append("subscribe")
return " ".join(roles)
def _retry_delay(attempt: int, cap: float = 30.0, base: float = 1.0) -> float:
"""Capped exponential backoff with full jitter.
Used for retry loops that can end up running on every module in the fleet at once (e.g.
every module reconnecting to ejabberd simultaneously after a mass restart). Fixed-interval
retries stay in lockstep across all of them, repeatedly hammering the server at the same
instants; jitter decorrelates that so retries spread out over time instead.
The `min(attempt, 60)` doesn't affect real backoff behavior -- with the defaults, `cap`
already clamps the delay from attempt 5 onward. It only exists so `2 ** attempt` can't grow
past what a float can hold for retry loops with no attempt limit (#824: attempt 1024 raised
OverflowError here and killed the retrying task).
"""
return random.uniform(0, min(cap, base * (2 ** min(attempt, 60))))
def _log_task_exception(task: asyncio.Task[Any]) -> None:
"""Retrieve and log a background task's exception, if it failed.
Results of asyncio.create_task() that are never awaited or callback-ed have their
exceptions reported by the event loop's default handler as "Task exception was never
retrieved" -- a noisy ERROR traceback that carries no context. Retrieving the exception
here turns that into a normal log line with the task's failure attached.
"""
if task.cancelled():
return
exc = task.exception()
if exc is not None:
log.error("Unhandled exception in XMPP event handler task", exc_info=exc)
# anchored at both ends -- re.match alone doesn't anchor the end, so e.g. "user@domain/res/extra"
# would otherwise still "match" as a valid prefix
_JID_RE = re.compile(r"([\w_\-\.]+)@([\w_\-\.]+)/([\w_\-\.]+)$")
def is_valid_jid(jid: str, resource: str = "pyobs") -> bool:
"""Whether jid is a valid user@domain or user@domain/resource JID -- exactly what
XmppComm.__init__ requires, so callers taking raw user input (e.g. a login window) can
validate it up front instead of finding out via a raised, less specific exception.
Args:
jid: JID to check, with or without a resource part.
resource: Resource to assume if jid doesn't already include one -- must match whatever
XmppComm itself would be constructed with, since a resource-less jid is only valid
if *some* resource ends up attached to it.
"""
candidate = jid if "/" in jid else f"{jid}/{resource}"
return bool(_JID_RE.match(candidate))
class EventStanza(ElementBase):
name = "event"
namespace = "pyobs:event"
class StateStanza(ElementBase):
name = "state"
namespace = "pyobs:state"
class XmppComm(Comm):
"""A Comm class using XMPP.
This Comm class uses an XMPP server (e.g. `ejabberd <https://www.ejabberd.im>`_) for communication between modules.
Essentially required for a connection to the server is a JID, a JabberID. It can be specified in the configuration
like this::
comm:
class: pyobs.comm.xmpp.XmppComm
jid: someuser@example.com/pyobs
Using this, *pyobs* tries to connect to example.com as user ``someuser`` with resource ``pyobs``. Since ``pyobs``
is the default resource, it can be omitted::
jid: someuser@example.com
Alternatively, one can split the user, domain, and resource (if required) into three different parameters::
user: someuser
domain: example.com
This comes in handy, if one wants to put the basic Comm configuration into a separate file. Imagine a ``_comm.yaml``
in the same directory as the module config::
comm_cfg: &comm
class: pyobs.comm.sleekxmpp.XmppComm
domain: example.com
Now in the module configuration, one can simply do this::
{include _comm.yaml}
comm:
<<: *comm
user: someuser
password: supersecret
This allows for a super easy change of the domain for all configurations, which especially makes developing on
different machines a lot easier.
The ``server`` parameter can be used, when the server's hostname is different from the XMPP domain. This might,
e.g., be the case, when connecting to a server via SSH port forwarding::
jid: someuser@example.com/pyobs
server: localhost:52222
Finally, always make sure that ``use_tls`` is set according to the server's settings, i.e. if it uses TLS, this
parameter must be True, and False otherwise. Cryptic error messages will follow, if one does not set this properly.
``ping_interval``/``ping_timeout`` control the XEP-0199 keepalive ping that detects a dead connection.
``ping_timeout`` in particular may need raising above its 30s default on a server whose shaper can delay an IQ
reply that long under load -- otherwise a merely-slow reply is indistinguishable from a dead connection and
triggers a reconnect that didn't need to happen::
comm:
class: pyobs.comm.xmpp.XmppComm
jid: someuser@example.com/pyobs
ping_timeout: 60
"""
__module__ = "pyobs.comm.xmpp"
def __init__(
self,
jid: str | None = None,
user: str | None = None,
domain: str | None = None,
resource: str = "pyobs",
password: str = "",
server: str | None = None,
use_tls: bool = False,
ignore_cert_errors: bool = False,
ping_interval: float = 300.0,
ping_timeout: float = 30.0,
*args: Any,
**kwargs: Any,
):
"""Create a new XMPP Comm module.
Either a fill JID needs to be provided, or a set of user/domian/resource, from which a JID is built.
Args:
jid: JID to connect as.
user: Username part of the JID.
domain: Domain part of the JID.
resource: Resource part of the JID.
password: Password for given JID.
server: Server to connect to. If not given, domain from JID is used.
use_tls: Whether to use TLS.
ping_interval: Seconds between XEP-0199 keepalive pings.
ping_timeout: Seconds to wait for a ping reply before reconnecting.
"""
Comm.__init__(self, *args, **kwargs)
# variables
self._connected = False
self._online_clients: list[str] = []
self._interface_cache: dict[str, asyncio.Future[list[type[Interface]]]] = {}
self._interface_features: dict[str, list[str]] = {}
self._peer_sent_events: dict[str, set[tuple[str, int]]] = {}
self._warned_version_mismatches: set[tuple[str, str]] = set()
self._user = user
self._password = password
self._domain = domain
self._resource = resource
self._server = server
self._use_tls = use_tls
self._ignore_cert_errors = ignore_cert_errors
self._ping_interval = ping_interval
self._ping_timeout = ping_timeout
self._loop = asyncio.get_event_loop()
self._safe_send_attempts = 5
self._safe_send_wait = 1
self._safe_send_timeout = 15.0
# build jid
if jid:
# resource given in jid?
if "/" not in jid:
jid += "/" + resource
# get user/domain/resource and write it back to config
m = _JID_RE.match(jid)
if not m:
log.error("Invalid JID format: %r (expected user@domain or user@domain/resource).", jid)
raise ValueError(f"Invalid JID format: {jid!r} (expected user@domain or user@domain/resource).")
self._user = m.group(1)
self._domain = m.group(2)
self._resource = m.group(3)
# set jid itself
self._jid = jid
else:
self._jid = f"{self._user}@{self._domain}/{self._resource}"
# client and RPC handler
self._xmpp: XmppClient | None = None
self._rpc: RPC | None = None
# module readiness (see mark_ready()) -- lives here rather than solely on XmppClient
# because _connect() replaces self._xmpp with a brand-new instance on every reconnect,
# and a reconnect after the module is already READY must announce presence immediately
# rather than re-gating it
self._module_ready = False
# pubsub for states
self._pubsub_service = f"pubsub.{self._domain}"
self._state_node_handlers: dict[str, tuple[type[Interface], list[Callable[[Any], None]]]] = {}
self._client_states: dict[str, tuple[ModuleState, str]] = {} # jid -> (state, error_string)
# pubsub for events -- desired (peer module name, event class) subscriptions. A key's
# presence drives _subscribe_event_with_retry's retry loop; removing it (unregister, or
# never re-added on a fresh connect) is what stops a still-retrying subscribe.
self._event_subscriptions: set[tuple[str, type[Event]]] = set()
self._capabilities: dict[type, Any] = {} # interface → Capabilities instance
self._own_states: dict[type, Any] = {} # interface → last published state for this module
self._presence_callbacks: dict[str, list[Callable[[ModuleState, str], None]]] = {}
def _set_module(self, module: Module) -> None:
"""Called, when the module connected to this Comm changes.
Args:
module: The module.
"""
self._module = module
[docs]
async def open(self) -> None:
"""Open the connection to the XMPP server.
Returns:
Whether opening was successful.
"""
# connect
await self._connect()
# subscribe to events
await self.register_event(LogEvent)
# open Comm
await Comm.open(self)
async def _connect(self) -> None:
# abort any previous client instead of just dropping the reference —
# otherwise its socket/tasks keep running in the background and it
# can still try to reconnect itself, fighting the new client below
# for the same JID resource
if self._xmpp is not None:
self._xmpp.abort()
# create client
self._xmpp = XmppClient(
self._jid, self._password, ping_interval=self._ping_interval, ping_timeout=self._ping_timeout
)
# presence gating (see mark_ready()) only applies to a comm with an actual Module
# attached that hasn't finished starting yet -- a module-less XmppComm (a GUI, an
# admin tool, a bare observer in tests) has no such lifecycle and must announce
# itself immediately as before, or peers relying on presence-based discovery would
# never see it. Likewise, if the module already reached READY on a previous
# connection, tell the new client right away -- it isn't connected yet, so this
# doesn't send anything itself, but it means session_start() will announce presence
# immediately once (re)connected instead of holding it back as if this were the
# initial startup.
if self._module_ready or not self.has_module:
self._xmpp.mark_ready()
# self._xmpp = slixmpp.ClientXMPP(self._jid, password)
# Register directly with an XML mask rather than via pubsub_publish.
# slixmpp's StanzaPath matcher requires plugins to be lazily loaded
# before matching — for live notifications (not triggered by an IQ)
# this never happens, so pubsub_publish is never fired. MatchXMLMask
# works on raw XML and fires reliably for every pubsub notification.
self._xmpp.register_handler(
Callback(
"pyobs pubsub event",
MatchXMLMask(
'<message xmlns="jabber:client">'
'<event xmlns="http://jabber.org/protocol/pubsub#event">'
"<items /></event></message>"
),
self._handle_event_sync,
)
)
self._xmpp.add_event_handler("got_online", self._got_online)
self._xmpp.add_event_handler("changed_status", self._got_presence_update)
self._xmpp.add_event_handler("got_offline", self._got_offline)
self._xmpp.add_event_handler("disconnected", functools.partial(self._disconnected, client=self._xmpp))
# server given?
server: str = "localhost"
port: int = 5222
if self._server is not None:
if ":" in self._server:
server, sport = self._server.split(":")
port = int(sport)
else:
server, port = self._server, 5222
elif self._domain is not None:
server, port = self._domain, 5222
# add features
if self._module is not None:
for i in self._module.interfaces:
self._xmpp.plugin["xep_0030"].add_feature(f"urn:pyobs:interface:{i.__name__}:{i.version}")
if i.has_own_state():
self._xmpp.plugin["xep_0030"].add_feature(f"urn:pyobs:state:{i.__name__}:{i.version}")
# register custom disco#info handler to inject <capability> elements
if self._module is not None:
self._xmpp.plugin["xep_0030"].set_node_handler("get_info", None, None, self._get_disco_info)
# RPC
self._rpc = RPC(self, self._xmpp, None)
self._rpc.set_handler(self._module)
# connect
self._xmpp.enable_starttls = self._use_tls
self._xmpp.enable_direct_tls = self._use_tls
self._xmpp.enable_plaintext = not self._use_tls
self._xmpp.plugin["feature_mechanisms"].unencrypted_scram = not self._use_tls # type: ignore[typeddict-item]
# Without this, slixmpp still refuses PLAIN over a non-TLS connection even
# with enable_plaintext set above -- it only affects the stream feature
# advertisement, not SASL mechanism selection. Matters in practice: this
# server's SCRAM implementation fails unencrypted SCRAM auth with "Invalid
# channel binding" (confirmed against ejabberd 26.4.0), so without this
# flag every non-TLS connection falls through all mechanisms and fails
# with "No appropriate login method" rather than falling back to PLAIN.
self._xmpp.plugin["feature_mechanisms"].unencrypted_plain = not self._use_tls # type: ignore[typeddict-item]
if self._ignore_cert_errors:
self._xmpp.ssl_context.check_hostname = False
self._xmpp.ssl_context.verify_mode = ssl.CERT_NONE
# connect
await self._xmpp.connect(host=server, port=port)
self._xmpp.init_plugins()
# wait for connected
if not await self._xmpp.wait_connect():
if self._module is not None:
self._module.quit()
return
# wait a little and finished
await asyncio.sleep(1)
self._connected = True
[docs]
async def close(self) -> None:
"""Close connection."""
# close parent class
await Comm.close(self)
# disconnect from sleekxmpp server
if self._xmpp is not None:
await self._xmpp.disconnect()
async def _reconnect(self) -> None:
"""Sleep a little and reconnect"""
await asyncio.sleep(2)
await self._connect()
def _disconnected(self, event: Any, client: XmppClient) -> None:
"""Reset connection after disconnect."""
if self._closing.is_set():
return
if client is not self._xmpp:
# stale event from a client that's already been replaced/aborted
return
self._capabilities = {} # clear capabilities on reconnect
# disconnect all clients
for jid in self._online_clients:
self._jid_got_offline(jid)
if client.kicked_by_conflict:
# another session took over our JID/resource -- reconnecting would just
# race that session for the resource again, so shut down instead
reason = client.conflict_reason or "conflict, no reason given"
log.error("Kicked from server (%s), shutting down module.", reason)
if self._module is not None:
self._module.quit()
return
log.info("Disconnected from server, waiting for reconnect...")
# reconnect
asyncio.create_task(self._reconnect())
@property
def name(self) -> str | None:
"""Name of this client."""
return self._user
def _failed_auth(self, event: Any) -> None:
"""Authentication failed.
Args:
event: XMPP event.
"""
print("Authorization at server failed.")
def _get_full_client_name(self, name: str) -> str:
"""Builds full JID from a given username.
Args:
name: Username to build JID for.
Returns:
Full JID for given user.
"""
return name if "@" in name else f"{name}@{self._domain}/{self._resource}"
[docs]
async def get_interfaces(self, client: str) -> list[type[Interface]]:
"""Returns list of interfaces for given client.
Args:
client: Name of client.
Returns:
List of supported interfaces.
Raises:
IndexError: If client cannot be found.
"""
# full JID given?
if "@" not in client:
client = f"{client}@{self._domain}/{self._resource}"
# the client's cache entry is only created once its presence has been processed
# (_got_online) -- that can lag slightly behind this module's own startup (e.g. a
# background task's very first tick can run before presence has even arrived), so wait
# briefly for it instead of failing instantly on a transient "not discovered yet"
for _ in range(20):
if client in self._interface_cache:
break
await asyncio.sleep(0.25)
else:
raise IndexError(f"Client {client} not found.")
# return them from cache
return await self._interface_cache[client]
async def _get_interfaces(self, jid: str, attempts: int = 3) -> list[str]:
"""Return list of interfaces for the given JID.
Args:
jid: JID to get interfaces for.
Returns:
List of interface names or empty list, if an error occurred.
"""
# request features
try:
info = await self._safe_send(self.client.plugin["xep_0030"].get_info, jid=jid, cached=False)
except (slixmpp.exceptions.IqError, slixmpp.exceptions.IqTimeout):
return []
# extract pyobs interface names
if info is None:
return []
try:
if isinstance(info, slixmpp.stanza.iq.Iq):
info = info["disco_info"]
prefix = "urn:pyobs:interface:"
features = [i for i in info["features"] if i.startswith(prefix)]
except TypeError:
raise IndexError()
# cache raw features for this JID, so a later version-mismatch can be diagnosed
self._interface_features[jid] = features
# cache which event types this peer actually publishes (role="send" in the rich <event>
# schema elements from _get_disco_info) -- distinct from the plain urn:pyobs:event:
# feature list above, which covers subscribe-only registrations too and can't tell a
# publisher apart from a mere consumer. Used to gate event-node subscriptions so we don't
# retry forever against peers that will never publish to that node (see _got_online,
# _register_events).
self._peer_sent_events[jid] = self._parse_peer_sent_events(info)
# keep only names whose remote-published version matches what this client expects --
# a mismatch is treated the same as the interface not being there at all, rather than
# silently using the local (possibly incompatible) class
interface_names = []
for feature in features:
name, _, version = feature[len(prefix) :].rpartition(":")
local_cls = get_registered_interface(name)
if local_cls is not None and str(local_cls.version) == version:
interface_names.append(name)
# IModule not in list?
if "IModule" not in interface_names:
# try again or quit?
if attempts == 0:
return []
else:
await asyncio.sleep(5)
interface_names = await self._get_interfaces(jid, attempts - 1)
# finished
return interface_names
@staticmethod
def _parse_peer_sent_events(info: Any) -> set[tuple[str, int]]:
"""Extract (event class name, version) pairs a peer's disco#info marks role="send" for.
Reads the rich <{urn:pyobs:event:name:version}event role="..."> elements _get_disco_info
appends (see _event_role), not the plain urn:pyobs:event: feature list -- that list only
says the peer registered the event at all, sent or subscribe-only, and can't tell a
publisher apart from a mere consumer.
"""
sent: set[tuple[str, int]] = set()
xml = getattr(info, "xml", None)
if xml is None:
return sent
ns_prefix = "urn:pyobs:event:"
for elem in xml:
tag = elem.tag
if not tag.startswith("{") or "}" not in tag:
continue
ns, _, local = tag[1:].partition("}")
if local != "event" or not ns.startswith(ns_prefix):
continue
if "send" not in elem.get("role", "").split():
continue
name = elem.get("name")
_, _, version_str = ns[len(ns_prefix) :].rpartition(":")
if not name or not version_str.isdigit():
continue
sent.add((name, int(version_str)))
return sent
def _diagnose_missing_interface(self, client: str, obj_type: type[Any]) -> str | None:
"""Checks the disco#info features already cached for client for a version of obj_type
other than the one this client expects, to tell a version mismatch apart from
obj_type genuinely not being implemented at all.
"""
jid = self._get_full_client_name(client)
features = self._interface_features.get(jid, [])
prefix = f"urn:pyobs:interface:{obj_type.__name__}:"
other = [f for f in features if f.startswith(prefix)]
if not other:
return None
remote_version = other[0][len(prefix) :]
try:
remote_version_int = int(remote_version)
except ValueError:
return None
direction = "upgrade the remote module" if remote_version_int < obj_type.version else "upgrade this client"
pair = (jid, obj_type.__name__)
if pair not in self._warned_version_mismatches:
self._warned_version_mismatches.add(pair)
log.warning(
'"%s" implements %s at v%s, this client expects v%s (%s).',
client,
obj_type.__name__,
remote_version,
obj_type.version,
direction,
)
return f"Remote implements it at v{remote_version}, this client expects v{obj_type.version} ({direction})."
async def _supports_interface(self, client: str, interface: type[Interface]) -> bool:
"""Checks, whether the given client supports the given interface.
Args:
client: Client to check.
interface: Interface to check.
Returns:
Whether or not interface is supported.
"""
# full JID given?
if "@" not in client:
client = f"{client}@{self._domain}/{self._resource}"
# update interface cache and get interface names
interfaces = await self.get_interfaces(client)
# supported?
return interface in interfaces
[docs]
async def execute(self, client: str, method: str, annotation: dict[str, Any], *args: Any) -> Any:
"""Execute a given method on a remote client.
Args:
client (str): ID of client.
method (str): Method to call.
annotation: Method annotation.
*args: List of parameters for given method.
Returns:
Passes through return from method call.
"""
# prepare
if self._rpc is None:
raise ValueError("No RPC.")
jid = self._get_full_client_name(client)
# call
try:
return await self._rpc.call(jid, method, annotation, *args)
except slixmpp.exceptions.IqError as e:
# slixmpp's own Iq.send() future resolves on the reply's stanza id before the
# RPC layer's jabber_rpc_error event handling ever gets a chance to act on it,
# so the IQ-level "forbidden" condition (see Module ACLs) has to be read here.
# This branch is a back-compat fallback for a peer running an older pyobs-core that
# still sends a raw XEP-0009 forbidden IQ error instead of a normal fault -- a current
# server routes ACL denials through the same fault path as every other domain
# exception (see Module.execute()/comm/xmpp/rpc.py), so a denied call to a peer
# running this fix never reaches this branch at all.
if e.iq["error"]["condition"] == "forbidden":
call_id: str | slixmpp.JID = e.iq["id"]
if isinstance(call_id, slixmpp.JID):
call_id = call_id.node
forbidden = exc.ForbiddenError(f"Forbidden to invoke {method} on {client}.", module=client)
setattr(forbidden, "call_id", call_id or None)
raise forbidden
raise exc.RemoteError(f"Could not call {method} on {client}.", module=client)
except slixmpp.exceptions.IqTimeout:
raise exc.RemoteTimeoutError(f"Call to {method} on {client} timed out.", module=client)
async def _got_online(self, msg: Any) -> None:
"""If a new client connects, add it to list.
Args:
msg: XMPP message.
"""
# get jid, ignore event if it's myself
jid = msg["from"].full
if jid == self._jid:
return
# clear interface cache, just in case there is something there
if jid in self._interface_cache:
del self._interface_cache[jid]
# create future for interfaces
self._interface_cache[jid] = asyncio.get_running_loop().create_future()
# request interfaces
interface_names = await self._get_interfaces(jid)
# if no interfaces are implemented (not even IModule), quit here
if len(interface_names) == 0:
module = jid[: jid.index("@")]
log.debug("Module %s does not seem to implement IModule, ignoring.", module)
# resolve the future we created above, otherwise it's left pending forever and
# any later get_interfaces()/proxy() call for this JID hangs indefinitely
future = self._interface_cache.get(jid)
if future is not None and not future.done():
future.set_result([])
return
# store interfaces — guard against a second _got_online for the same JID
# (ejabberd may send multiple presence stanzas) racing against the first
future = self._interface_cache.get(jid)
if future is not None and not future.done():
future.set_result(self._interface_names_to_classes(interface_names))
# store incoming presence state
show = msg.get("show", "")
status = msg.get("status", "")
if show == "dnd":
client_state = ModuleState.ERROR
elif show == "away":
client_state = ModuleState.LOCAL
else:
client_state = ModuleState.READY
self._client_states[jid] = (client_state, status)
# fire presence callbacks
module_name = jid[: jid.index("@")]
self._fire_presence_callbacks(module_name, client_state, status)
# append to list
if jid not in self._online_clients:
self._online_clients.append(jid)
# subscribe to this peer's event nodes for every event type we currently handle -- a
# handler registered before this peer came online never got a chance to subscribe to it
# from _register_events, since it wasn't in self._online_clients yet.
# Skip local events (e.g. ModuleOpenedEvent/ModuleClosedEvent, registered by every
# module via module.py/comm.py) -- they're synthesized locally here and in
# _jid_got_offline, never published to a node, so subscribing would retry forever
# against something that will never exist. Also skip event classes whose last handler
# was already removed -- unregister_event() discards from _events_subscribed but leaves
# an empty list in _event_handlers. Also skip event types this peer doesn't actually
# publish (per its disco#info role="send" list, cached above by _get_interfaces) --
# otherwise every module subscribes to every peer for every event type it handles, e.g. a
# camera's BadWeatherEvent handler retry-subscribing to admin:BadWeatherEvent forever,
# since that node will never be created.
peer_sent_events = self._peer_sent_events.get(jid, set())
for ev, handlers in self._event_handlers.items():
if ev.local or not handlers:
continue
if (ev.__name__, ev.version) not in peer_sent_events:
continue
task = asyncio.create_task(self._subscribe_event_with_retry(module_name, ev))
task.add_done_callback(_log_task_exception)
# send event
self._send_event_to_module(ModuleOpenedEvent(), msg["from"].username)
def _fire_presence_callbacks(self, module_name: str, state: ModuleState, status: str) -> None:
"""Call every presence callback registered for a module, isolating failures.
A callback belongs to whoever subscribed (e.g. a GUI widget) and may be stale --
subscribers are expected to unsubscribe on disconnect, but a callback raising here
must never abort the caller: this runs inline inside got_online/got_offline presence
handling, and an uncaught exception here previously meant the module's reconnect was
silently dropped (online_clients never updated, ModuleOpenedEvent never sent).
"""
for cb in list(self._presence_callbacks.get(module_name, [])):
try:
cb(state, status)
except Exception:
log.exception("Presence callback for module %s raised, ignoring.", module_name)
def _get_client_state(self, module: str) -> tuple[ModuleState, str] | None:
"""Return cached presence state for a connected module."""
for jid, state in self._client_states.items():
if jid.startswith(f"{module}@"):
return state
return None
def _got_presence_update(self, msg: Any) -> None:
"""Handle presence changes from already-connected modules."""
jid = msg["from"].full
if jid == self._jid or jid not in self._online_clients:
return
show = msg.get("show", "")
status = msg.get("status", "")
if show == "dnd":
client_state = ModuleState.ERROR
elif show == "away":
client_state = ModuleState.LOCAL
else:
client_state = ModuleState.READY
self._client_states[jid] = (client_state, status)
# fire presence callbacks
module_name = jid[: jid.index("@")]
self._fire_presence_callbacks(module_name, client_state, status)
def _got_offline(self, msg: Any) -> None:
"""If a new client disconnects, remove it from list.
Args:
msg: XMPP message.
"""
self._jid_got_offline(msg["from"].full)
def _jid_got_offline(self, jid: str) -> None:
"""If a new client disconnects, remove it from list.
Args:
jid: JID that got offline.
"""
# remove from list
if jid in self._online_clients:
self._online_clients.remove(jid)
self._client_states.pop(jid, None)
# notify presence subscribers that the module is gone
module_name = jid[: jid.find("@")]
self._fire_presence_callbacks(module_name, ModuleState.CLOSED, "")
# clear interface cache
if jid in self._interface_cache:
del self._interface_cache[jid]
self._interface_features.pop(jid, None)
# send event
self._send_event_to_module(ModuleClosedEvent(), module_name)
@property
def clients(self) -> list[str]:
"""Returns list of currently connected clients.
Returns:
(list) List of currently connected clients.
"""
return [c[: c.find("@")] for c in self._online_clients]
@property
def client(self) -> XmppClient:
"""Returns the XMPP client.
Returns:
The XMPP client.
"""
if self._xmpp is None:
raise ValueError("No XMPP client.")
return self._xmpp
[docs]
async def send_event(self, event: Event) -> None:
"""Send an event to other clients.
Args:
event (Event): Event to send
"""
# create stanza
stanza = EventStanza()
# dump event to JSON and escape it
body = xml.sax.saxutils.escape(json.dumps(event.to_json()))
# set xml and send event
stanza.xml = ET.fromstring(f'<event xmlns="pyobs:event">{body}</event>')
# publish to the shared pubsub service, same mechanism as state (see _set_state) --
# the node id encodes the publishing module, since notifications from this service
# arrive "from" the service itself, not from us (see _event_node/_handle_event).
# A module-less XmppComm (GUI, admin tool, observer) has no self._module -- fall back to
# the JID's own username, same identity peers would derive from our JID anyway.
publisher = self._module.name if self._module is not None else self.client.boundjid.user
node = self._event_node(publisher, event.__class__)
await self._safe_send(
self.client.plugin["xep_0060"].publish,
self._pubsub_service,
node,
payload=stanza,
callback=functools.partial(self._send_event_callback, event=event),
)
# send it to local module
if self._module is not None:
self._send_event_to_module(event, self._module.name)
@staticmethod
def _send_event_callback(iq: Any, event: Event | None = None) -> None:
"""Called when an event has been successfully sent.
Args:
iq: Response package.
event: Sent event.
"""
log.debug("%s successfully sent.", event.__class__.__name__)
async def _register_events(
self, events: list[type[Event]], handler: Callable[[Event, str], Coroutine[Any, Any, bool]] | None = None
) -> None:
# loop events
for ev in events:
# register event at XMPP (disco advertising only -- unrelated to delivery, used by
# e.g. pyobs-web-client to distinguish producers from consumers, see _event_role)
self.client.plugin["xep_0030"].add_feature(f"urn:pyobs:event:{ev.__name__}:{ev.version}")
# if we have a handler, subscribe to this event's node on every peer already online
# that actually publishes it (see _got_online for why this is gated). A peer coming
# online later is covered by _got_online subscribing to every event currently in
# self._event_handlers.
if handler:
for peer_jid in list(self._online_clients):
if (ev.__name__, ev.version) not in self._peer_sent_events.get(peer_jid, set()):
continue
peer_module = peer_jid[: peer_jid.index("@")]
task = asyncio.create_task(self._subscribe_event_with_retry(peer_module, ev))
task.add_done_callback(_log_task_exception)
elif self._module is not None:
# send-only declaration -- pre-create the node now, before this module announces
# presence, so peers reacting to it in _got_online land their subscribe on the
# first attempt instead of falling back to the retry loop (see #824). Sequential
# and awaited inside open(): each _create_node can burn up to ~90s against an
# unresponsive pubsub service (_safe_send's 5 x 15s timeout budget plus jittered
# waits between attempts), times the number of send-only events this module
# declares. Accepted since connect fails anyway in that scenario.
await self._create_node(self._event_node(self._module.name, ev))
# update caps and send presence
await self._safe_send(self.client.plugin["xep_0115"].update_caps)
self.client.send_presence()
async def _unregister_events(self, events: list[type[Event]]) -> None:
# unsubscribe goes to self._pubsub_service, not to the peer -- it doesn't depend on the
# peer being online, so this has to cover every (peer, ev) we're actually subscribed to,
# not just currently-online ones (a peer offline at unregister time would otherwise keep
# its server-side subscription, and we'd go on receiving that event's wire message)
for ev in events:
for peer_module, key_ev in list(self._event_subscriptions):
if key_ev is not ev:
continue
self._event_subscriptions.discard((peer_module, ev))
node = self._event_node(peer_module, ev)
try:
await self._safe_send(self.client.plugin["xep_0060"].unsubscribe, self._pubsub_service, node)
except (slixmpp.exceptions.IqError, slixmpp.exceptions.IqTimeout):
pass # already gone server-side
@staticmethod
def _event_node(module: str, event_class: type[Event]) -> str:
return f"pyobs:event:{module}:{event_class.__name__}:{event_class.version}"
@staticmethod
def _event_node_module(node: str) -> str | None:
"""Recover the publishing module's name from an event node id, since notifications from
the shared pubsub service arrive "from" the service itself, not from the publisher."""
parts = node.split(":")
if len(parts) == 5 and parts[0] == "pyobs" and parts[1] == "event":
return parts[2]
return None
async def _create_node(self, node: str) -> None:
"""Pre-create a pubsub node so a subscriber doesn't have to wait for the first publish.
The realistic error here is <conflict/> on restart (nodes persist server-side), and a
permission denial should degrade gracefully to today's lazy auto-create, never block
startup -- so IqError is swallowed. IqTimeout is already retried inside _safe_send; if it
still runs out, a dead server shouldn't hang open() any longer than that existing budget,
so it's swallowed here too rather than propagating.
"""
try:
await self._safe_send(self.client.plugin["xep_0060"].create_node, self._pubsub_service, node)
except (slixmpp.exceptions.IqError, slixmpp.exceptions.IqTimeout) as e:
log.debug("Could not pre-create pubsub node %s (%s), will lazy-create on first publish.", node, e)
async def _subscribe_event_with_retry(self, peer_module: str, event_class: type[Event]) -> None:
"""Subscribe to a peer's event node, retrying until the node exists.
Mirrors _subscribe_with_retry (state). Runs as a background task; retries indefinitely
with capped backoff since the peer may not have published (and thus auto-created the
node) yet. Stops early if _unregister_events drops the (peer_module, event_class) key
in the meantime.
"""
key = (peer_module, event_class)
if key in self._event_subscriptions:
return
self._event_subscriptions.add(key)
node = self._event_node(peer_module, event_class)
try:
attempt = 0
while key in self._event_subscriptions:
try:
await self._safe_send(self.client.plugin["xep_0060"].subscribe, self._pubsub_service, node)
return
except (slixmpp.exceptions.IqError, slixmpp.exceptions.IqTimeout):
attempt += 1
if attempt == 30:
log.warning(
"Still failing to subscribe to event node %s after %d attempts, will keep retrying",
node,
attempt,
)
await asyncio.sleep(_retry_delay(attempt))
except Exception:
# an unexpected (non-IqError/IqTimeout) failure must not leave key permanently
# marking (peer_module, event_class) as subscribed while nothing is subscribed and
# nothing is retrying -- discard it so a later register_event()/_got_online()
# re-subscribes from scratch instead of short-circuiting on the stale key (#824).
self._event_subscriptions.discard(key)
raise
def _handle_event_sync(self, msg: Any) -> None:
"""Synchronous entry point for the MatchXMLMask Callback.
State-node dispatch is done synchronously here (no awaits needed).
Non-state messages (pyobs events) are dispatched via asyncio.create_task
since they need JSON parsing and event routing.
"""
pubsub_ns = "http://jabber.org/protocol/pubsub#event"
event_xml = msg.xml.find(f"{{{pubsub_ns}}}event")
if event_xml is None:
return
items_xml = event_xml.find(f"{{{pubsub_ns}}}items")
if items_xml is None:
return
node = items_xml.get("node", "")
if node.startswith("pyobs:state:"):
if node in self._state_node_handlers:
if len(msg.xml.findall("{urn:xmpp:delay}delay")) == 0:
interface, callbacks = self._state_node_handlers[node]
item_xml = items_xml.find(f"{{{pubsub_ns}}}item")
payload = list(item_xml)[0] if item_xml is not None and len(item_xml) > 0 else None
if payload is not None and interface.state is not None:
state_obj = _xml_to_dataclass(payload, interface.state)
for callback in callbacks:
callback(state_obj)
else:
# Non-state notifications are handled asynchronously. Attach a done-callback so the
# task's exception is retrieved: without it, a failure inside _handle_event is never
# retrieved and asyncio reports it as "Task exception was never retrieved" (see
# _handle_event for the payload-less notification case that used to trigger this).
task = asyncio.create_task(self._handle_event(msg, node))
task.add_done_callback(_log_task_exception)
async def _handle_event(self, msg: Any, node: str) -> None:
"""Handles an event.
Args:
msg: Received XMPP message.
node: pubsub node id the event arrived on. Notifications from the shared pubsub
service come "from" that service, not from the publisher, so the publishing
module's name has to come from the node id (see _event_node) instead of msg["from"].
"""
# get body, unescape it, parse it
# State-node messages are handled synchronously in _handle_event_sync
# before this async task runs. By the time we get here it's a pyobs event.
# Not every notification carries an item with a payload: retract stanzas,
# node purges, and nodes with deliver_payloads off all arrive payload-less.
# The state-node path copes via _fetch_and_dispatch_state; for events there is
# nothing to refetch, so just drop the notification.
items = msg["pubsub_event"]["items"]
item = items["item"] if items is not None else None
payload = item["payload"] if item is not None else None
if payload is None or payload.text is None:
return
body = json.loads(xml.sax.saxutils.unescape(payload.text))
# do we have a <delay> element?
delay = msg.xml.findall("{urn:sleekxmpp:delay}delay")
if len(delay) > 0:
# ignore this message
return
from_module = self._event_node_module(node)
if from_module is None:
return
# did we send this? (we never subscribe to our own node, so this shouldn't happen, but
# stay defensive rather than assume)
if self._module is not None and from_module == self._module.name:
return
# create event and check timestamp
event = EventFactory.from_dict(body)
if event is None:
return
if time.time() - event.timestamp > 30:
# event is more than 30 seconds old, ignore it
# we do this do avoid resent events after a reconnect
return
# send it to module
self._send_event_to_module(event, from_module)
async def _safe_send(self, method: Callable[..., Coroutine[Any, Any, Any]], *args: Any, **kwargs: Any) -> Any:
"""Safely send an XMPP message.
Args:
method: Method to call.
*args: Parameters for method.
**kwargs: Parameters for method.
Returns:
Return value from method.
"""
# try multiple times -- unlike _get_capabilities/_subscribe_with_retry this still gives up
# after a fixed budget instead of retrying indefinitely, since some callers run from open()
# and need to know a send failed rather than hang forever (see #664/#666). But the wait
# between attempts is jittered for the same reason as those: many modules' _safe_send calls
# can end up retrying around the same moment (e.g. a fleet-wide restart), and a fixed wait
# keeps them all retrying in lockstep instead of spreading the load out.
iq = None
for i in range(self._safe_send_attempts):
try:
# execute method and return result, but never wait longer than our own
# timeout -- some XMPP servers/slixmpp's own IQ timeout can fail to fire,
# which would otherwise hang the caller (and, if called from open(), the
# whole module) indefinitely
return await asyncio.wait_for(method(*args, **kwargs), timeout=self._safe_send_timeout)
except slixmpp.exceptions.IqTimeout as timeout:
# timeout occurred, try again after some wait
iq = timeout.iq
await asyncio.sleep(_retry_delay(i + 1, cap=self._safe_send_wait * 4, base=self._safe_send_wait))
except TimeoutError:
# our own timeout fired instead of slixmpp's, try again after some wait
await asyncio.sleep(_retry_delay(i + 1, cap=self._safe_send_wait * 4, base=self._safe_send_wait))
# never should reach this
raise slixmpp.exceptions.IqTimeout(iq)
@staticmethod
def _state_namespace(interface: type[Interface]) -> str:
return f"urn:pyobs:state:{interface.__name__}:{interface.version}"
@staticmethod
def _state_node(module: str, interface: type[Interface]) -> str:
return f"pyobs:state:{module}:{interface.__name__}:{interface.version}"
async def _fetch_and_dispatch_state(
self, node: str, interface: type[Interface], callback: Callable[[Any], None]
) -> None:
"""Fetch the current item for *node* and dispatch it to *callback*.
Called when a live notification arrives without a payload — either a
retract stanza or a node whose deliver_payloads flag is off.
"""
try:
result = await self._safe_send(
self.client.plugin["xep_0060"].get_items, self._pubsub_service, node, max_items=1
)
# Use raw XML to avoid lazy-loading issues with plugin_multi_attrib
pubsub_ns = "http://jabber.org/protocol/pubsub"
pubsub_xml = result.xml.find(f"{{{pubsub_ns}}}pubsub")
items_xml = pubsub_xml.find(f"{{{pubsub_ns}}}items") if pubsub_xml is not None else None
item_xml = items_xml.find(f"{{{pubsub_ns}}}item") if items_xml is not None else None
payload = list(item_xml)[0] if item_xml is not None and len(item_xml) > 0 else None
if payload is not None and interface.state is not None:
callback(_xml_to_dataclass(payload, interface.state))
except (slixmpp.exceptions.IqError, slixmpp.exceptions.IqTimeout):
pass
def _get_own_state(self, interface: type[Interface]) -> Any:
return self._own_states.get(interface)
async def _set_state(self, interface: type[Interface], state: Any) -> None:
self._own_states[interface] = state
node = self._state_node(self._module.name, interface) # type: ignore[union-attr]
stanza = StateStanza()
stanza.xml = _dataclass_to_xml(state, self._state_namespace(interface))
await self._safe_send(self.client.plugin["xep_0060"].publish, self._pubsub_service, node, payload=stanza)
async def _get_disco_info(self, jid, node, ifrom, data):
"""Custom disco#info handler that adds <capability> elements for static module values."""
# Get the default info from the static handler
info = self._xmpp.plugin["xep_0030"].static.get_info(jid, node, ifrom, data)
if info is None:
from slixmpp.plugins.xep_0030.stanza import DiscoInfo
info = DiscoInfo()
# Remove any previously appended capability and interface schema elements
# (info.xml is cached by slixmpp and reused across calls — without this,
# each query appends another copy of every element)
for old_elem in list(info.xml):
local = old_elem.tag.split("}")[-1]
ns = old_elem.tag[1 : old_elem.tag.index("}")] if "}" in old_elem.tag else ""
if local == "capabilities":
info.xml.remove(old_elem)
elif local == "interface" and ns.startswith("urn:pyobs:interface:"):
info.xml.remove(old_elem)
elif local == "event" and ns.startswith("urn:pyobs:event:"):
info.xml.remove(old_elem)
# Append current capabilities
for interface, caps in self._capabilities.items():
ns = f"urn:pyobs:capabilities:{interface.__name__}:{interface.version}"
cap_xml = _dataclass_to_xml(caps, ns, tag="capabilities")
info.xml.append(cap_xml)
# Append interface schemas (<command>, <state>, <types> blocks)
if self._module is not None:
for interface in self._module.interfaces:
info.xml.append(_interface_schema_to_xml(interface))
# Append event schemas, tagged with a role attribute so consumers that have no access
# to the pyobs.events catalog themselves (e.g. pyobs-web-client) can tell which events
# a module actually sends vs. which it only subscribes to receive.
registered_events = self._events_sent | self._events_subscribed
for ev_cls in sorted(registered_events, key=lambda e: e.__name__):
if ev_cls.local:
continue
event_xml = _event_schema_to_xml(ev_cls)
event_xml.attrib["role"] = _event_role(ev_cls, self._events_sent, self._events_subscribed)
info.xml.append(event_xml)
return info
async def _set_capabilities(self, interface: type[Interface], capabilities: Any) -> None:
"""Store published capabilities for inclusion in disco#info responses.
Capabilities are set once in open() and never mutated afterwards, so advertising a
disco feature per interface (mirrors _register_events) is enough to change the caps
hash the first -- and only -- time an interface's capability appears. Without a feature
that a peer hasn't seen advertised before, update_caps() would recompute the same hash
(capability payloads aren't part of what feeds it, see XEP_0115.generate_verstring), and
ejabberd's mod_caps would keep serving whatever disco#info response it cached under that
unchanged hash -- potentially one from before this capability was ever set (#888).
"""
self._capabilities[interface] = capabilities
self.client.plugin["xep_0030"].add_feature(f"urn:pyobs:capabilities:{interface.__name__}:{interface.version}")
await self._safe_send(self.client.plugin["xep_0115"].update_caps)
self.client.send_presence()
log.info("Published capabilities for %s", interface.__name__)
def _get_own_capabilities(self, interface: type[Interface]) -> Any:
"""Return this client's own published capabilities."""
return self._capabilities.get(interface)
async def _get_capabilities(self, module: str, interface: type[Interface]) -> Any | None:
"""Fetch and deserialize capabilities for a remote module's interface.
Retries indefinitely (capped exponential backoff with jitter) instead of giving up after
a fixed budget -- a peer that's merely slow to respond (e.g. every module in the fleet
reconnecting to ejabberd at once) must still eventually get its capabilities fetched
without requiring a full disconnect/reconnect of that peer to retrigger discovery. Only
stops if the peer itself goes offline in the meantime, since a fresh fetch is triggered
from scratch the next time it comes back online (see _got_online).
"""
if interface.capabilities is None:
return None
ns = f"urn:pyobs:capabilities:{interface.__name__}:{interface.version}"
result = None
last_error: BaseException | None = None
attempt = 0
while True:
# Use full JID (with resource) if we know it — bare JID may not route correctly
full_jid = next((jid for jid in self._online_clients if jid.startswith(f"{module}@")), None)
if full_jid is None and attempt > 0:
log.debug(
"Giving up fetching capabilities for %s from %s: peer went offline",
interface.__name__,
module,
)
return None
try:
result = await asyncio.wait_for(
self.client.plugin["xep_0030"].get_info(jid=JID(full_jid or f"{module}@{self._domain}")),
timeout=10.0,
)
break
except Exception as e:
last_error = e
attempt += 1
if attempt == 3:
log.warning(
"Still failing to get capabilities for %s from %s after %d attempts (%r), "
"will keep retrying",
interface.__name__,
module,
attempt,
last_error,
)
await asyncio.sleep(_retry_delay(attempt))
log.debug("get_capabilities disco result XML: %s", ET.tostring(result.xml).decode()[:500])
# result.xml is the <iq> — the <query> is its child, capabilities are grandchildren
for child in result.xml:
for elem in child:
tag = elem.tag.split("}")[-1]
if tag == "capabilities" and f"{{{ns}}}" in elem.tag:
return _xml_to_dataclass(elem, interface.capabilities)
return None
async def _set_presence(self, state: ModuleState, error_string: str = "") -> None:
"""Send XMPP presence stanza reflecting the module lifecycle state.
ModuleState maps onto XMPP <show>:
READY → no <show> (available, the XMPP default)
ERROR → dnd
LOCAL → away
CLOSED → handled by normal disconnect (unavailable)
error_string rides as <status> text when state is ERROR.
"""
# publishing any lifecycle state at all is itself a readiness signal -- covers direct
# set_presence() callers that don't go through Module.set_state() (see mark_ready()).
# Idempotent: unlocks send_presence() once, then this call's own send below goes through.
# Goes through self._mark_ready() (not just self.client.mark_ready()) so self._module_ready
# is set too -- otherwise a later reconnect (_connect() creates a brand-new XmppClient)
# would re-gate presence on the new client despite this module already being announced.
await self._mark_ready()
_show_map: dict[ModuleState, str | None] = {
ModuleState.READY: None,
ModuleState.ERROR: "dnd",
ModuleState.LOCAL: "away",
}
show = _show_map.get(state)
status = error_string if state == ModuleState.ERROR and error_string else None
self.client.send_presence(pshow=show, pstatus=status)
async def _mark_ready(self) -> None:
"""See Comm.mark_ready(). Remembers readiness on self (survives client recreation on
reconnect, see _connect()) and lets the live client announce presence now."""
self._module_ready = True
self.client.mark_ready()
async def _subscribe_presence(self, module: str, callback: Callable[[ModuleState, str], None]) -> None:
self._presence_callbacks.setdefault(module, []).append(callback)
result = self._get_client_state(module)
if result is not None:
callback(*result)
async def _unsubscribe_presence(self, module: str, callback: Callable[[ModuleState, str], None]) -> None:
callbacks = self._presence_callbacks.get(module)
if callbacks is not None and callback in callbacks:
callbacks.remove(callback)
async def _subscribe_with_retry(self, node: str, interface: type[Interface]) -> None:
"""Subscribe to a pubsub node, retrying until the node exists.
Runs as a background task so _subscribe_state returns immediately. Retries indefinitely
(capped exponential backoff with jitter) instead of giving up after a fixed budget -- a
node that's merely slow to appear (e.g. every module in the fleet reconnecting to
ejabberd at once, or the publisher hasn't started up yet) must still eventually get
subscribed without manual intervention. Stops only if _unsubscribe_state removes the
last callback for this node in the meantime -- the while condition then goes False,
which exits the loop into the else clause (not the break path) and returns.
Once subscribed, fetches the current value and dispatches it.
"""
try:
attempt = 0
while node in self._state_node_handlers:
try:
await self._safe_send(self.client.plugin["xep_0060"].subscribe, self._pubsub_service, node)
break
except (slixmpp.exceptions.IqError, slixmpp.exceptions.IqTimeout):
attempt += 1
if attempt == 30:
log.warning(
"Still failing to subscribe to state node %s after %d attempts, will keep retrying",
node,
attempt,
)
await asyncio.sleep(_retry_delay(attempt))
else:
return
except Exception:
# an unexpected failure in the subscribe loop itself must not leave a
# _state_node_handlers entry behind forever -- that would permanently short-circuit
# _subscribe_state (see its "already subscribed" branch) while nothing is actually
# subscribed and nothing is retrying (#824). A later _subscribe_state call starts a
# fresh subscribe from scratch instead; note it only carries the callback that
# triggered that fresh call, not any earlier ones -- accepted as strictly better than
# permanent silent loss for this catastrophic, unexpected case. Logged by the task's
# _log_task_exception done-callback, not here, to avoid printing the traceback twice.
self._state_node_handlers.pop(node, None)
raise
# Fetch current value immediately after subscribing. Deliberately outside the block
# above: a bad payload (_xml_to_dataclass) or a raising callback here must not be treated
# as a failed subscription -- the server-side subscription is live either way, so
# discarding the handler would just recreate #824's permanent-silent-loss failure via a
# different trigger while making it look like a subscribe failure.
try:
result = await self._safe_send(
self.client.plugin["xep_0060"].get_items, self._pubsub_service, node, max_items=1
)
pubsub_ns = "http://jabber.org/protocol/pubsub"
pubsub_xml = result.xml.find(f"{{{pubsub_ns}}}pubsub")
items_xml = pubsub_xml.find(f"{{{pubsub_ns}}}items") if pubsub_xml is not None else None
item_xml = items_xml.find(f"{{{pubsub_ns}}}item") if items_xml is not None else None
payload = list(item_xml)[0] if item_xml is not None and len(item_xml) > 0 else None
if payload is not None and node in self._state_node_handlers and interface.state is not None:
_, callbacks = self._state_node_handlers[node]
state_obj = _xml_to_dataclass(payload, interface.state)
for cb in callbacks:
cb(state_obj)
except (slixmpp.exceptions.IqError, slixmpp.exceptions.IqTimeout):
pass
async def _subscribe_state(self, module: str, interface: type[Interface], callback: Callable[[Any], None]) -> None:
node = self._state_node(module, interface)
if node in self._state_node_handlers:
# Node already subscribed — just add the callback
self._state_node_handlers[node][1].append(callback)
# Deliver the current cached value immediately to the new callback
asyncio.create_task(self._fetch_and_dispatch_state(node, interface, callback))
else:
self._state_node_handlers[node] = (interface, [callback])
# Subscribe in a background task so _get_client() returns immediately.
# The retry loop handles the case where the publisher hasn't created
# the node yet (e.g. GUI connects before camera's first publish).
task = asyncio.create_task(self._subscribe_with_retry(node, interface))
task.add_done_callback(_log_task_exception)
# Initial value is fetched in _subscribe_with_retry after the subscribe IQ succeeds
async def _unsubscribe_state(
self, module: str, interface: type[Interface], callback: Callable[[Any], None]
) -> None:
node = self._state_node(module, interface)
if node in self._state_node_handlers:
_, callbacks = self._state_node_handlers[node]
try:
callbacks.remove(callback)
except ValueError:
pass
if not callbacks:
# Last subscriber — unsubscribe from ejabberd and remove handler
del self._state_node_handlers[node]
try:
await self._safe_send(self.client.plugin["xep_0060"].unsubscribe, self._pubsub_service, node)
except (slixmpp.exceptions.IqError, slixmpp.exceptions.IqTimeout):
pass # already gone server-side
__all__ = ["XmppComm", "is_valid_jid"]