Mercurial > libervia-backend
view libervia/backend/plugins/plugin_xep_0199.py @ 4334:111dce64dcb5
plugins XEP-0300, XEP-0446, XEP-0447, XEP0448 and others: Refactoring to use Pydantic:
Pydantic models are used more and more in Libervia, for the bridge API, and also to
convert `domish.Element` to internal representation.
Type hints have also been added in many places.
rel 453
author | Goffi <goffi@goffi.org> |
---|---|
date | Tue, 03 Dec 2024 00:12:38 +0100 |
parents | 0d7bb4df2343 |
children |
line wrap: on
line source
#!/usr/bin/env python3 # SAT plugin for Delayed Delivery (XEP-0199) # Copyright (C) 2009-2021 Jérôme Poisson (goffi@goffi.org) # Copyright (C) 2013-2016 Adrien Cossa (souliane@mailoo.org) # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. from libervia.backend.core.i18n import _, D_ from libervia.backend.core.log import getLogger log = getLogger(__name__) from libervia.backend.core.constants import Const as C from wokkel import disco, iwokkel from twisted.words.protocols.jabber import xmlstream, jid from zope.interface import implementer import time PLUGIN_INFO = { C.PI_NAME: "XMPP PING", C.PI_IMPORT_NAME: "XEP-0199", C.PI_TYPE: "XEP", C.PI_PROTOCOLS: ["XEP-199"], C.PI_MAIN: "XEP_0199", C.PI_HANDLER: "yes", C.PI_DESCRIPTION: D_("""Implementation of XMPP Ping"""), } NS_PING = "urn:xmpp:ping" PING_REQUEST = C.IQ_GET + '/ping[@xmlns="' + NS_PING + '"]' class XEP_0199(object): def __init__(self, host): log.info(_("XMPP Ping plugin initialization")) self.host = host host.bridge.add_method( "ping", ".plugin", in_sign="ss", out_sign="d", method=self._ping, async_=True ) try: self.text_cmds = self.host.plugins[C.TEXT_CMDS] except KeyError: log.info(_("Text commands not available")) else: self.text_cmds.register_text_commands(self) def get_handler(self, client): return XEP_0199_handler(self) def _ping_raise_if_failure(self, pong): """If ping didn't succeed, raise the failure, else return pong delay""" if pong[0] != "PONG": raise pong[0] return pong[1] def _ping(self, jid_s, profile): client = self.host.get_client(profile) entity_jid = jid.JID(jid_s) d = self.ping(client, entity_jid) d.addCallback(self._ping_raise_if_failure) return d def _ping_cb(self, iq_result, send_time): receive_time = time.time() return ("PONG", receive_time - send_time) def _ping_eb(self, failure_, send_time): receive_time = time.time() return (failure_.value, receive_time - send_time) def ping(self, client, entity_jid): """Ping an XMPP entity @param entity_jid(jid.JID): entity to ping @return (tuple[(unicode,failure), float]): pong data: - either u"PONG" if it was successful, or failure - delay between sending time and reception time """ iq_elt = client.IQ("get") iq_elt["to"] = entity_jid.full() iq_elt.addElement((NS_PING, "ping")) d = iq_elt.send() send_time = time.time() d.addCallback(self._ping_cb, send_time) d.addErrback(self._ping_eb, send_time) return d def _cmd_ping_fb(self, pong, client, mess_data): """Send feedback to client when pong data is received""" txt_cmd = self.host.plugins[C.TEXT_CMDS] if pong[0] == "PONG": txt_cmd.feed_back(client, "PONG ({time} s)".format(time=pong[1]), mess_data) else: txt_cmd.feed_back( client, _("ping error ({err_msg}). Response time: {time} s").format( err_msg=pong[0], time=pong[1] ), mess_data, ) def cmd_ping(self, client, mess_data): """ping an entity @command (all): [JID] - JID: jid of the entity to ping """ if mess_data["unparsed"].strip(): try: entity_jid = jid.JID(mess_data["unparsed"].strip()) except RuntimeError: txt_cmd = self.host.plugins[C.TEXT_CMDS] txt_cmd.feed_back( client, _('Invalid jid: "{entity_jid}"').format( entity_jid=mess_data["unparsed"].strip() ), mess_data, ) return False else: entity_jid = mess_data["to"] d = self.ping(client, entity_jid) d.addCallback(self._cmd_ping_fb, client, mess_data) return False def on_ping_request(self, iq_elt, client): log.info( _("XMPP PING received from {from_jid} [{profile}]").format( from_jid=iq_elt["from"], profile=client.profile ) ) iq_elt.handled = True iq_result_elt = xmlstream.toResponse(iq_elt, "result") client.send(iq_result_elt) @implementer(iwokkel.IDisco) class XEP_0199_handler(xmlstream.XMPPHandler): def __init__(self, plugin_parent): self.plugin_parent = plugin_parent def connectionInitialized(self): self.xmlstream.addObserver( PING_REQUEST, self.plugin_parent.on_ping_request, client=self.parent ) def getDiscoInfo(self, requestor, target, nodeIdentifier=""): return [disco.DiscoFeature(NS_PING)] def getDiscoItems(self, requestor, target, nodeIdentifier=""): return []