libervia-backend: sat/plugins/plugin_xep

comparison sat/plugins/plugin_xep_0060.py @ 3715:b9718216a1c0 0.9

merge bookmark 0.9

author	Goffi <goffi@goffi.org>
date	Wed, 01 Dec 2021 16:13:31 +0100
parents	5d108ce026d7
children	1cdb9d9fad6b

comparison

equal deleted inserted replaced

-:af09b5aaa5d7
+:b9718216a1c0
 # 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 typing import Optional
+from typing import Optional, List, Tuple
 from collections import namedtuple
 import urllib.request, urllib.parse, urllib.error
 from functools import reduce
 from zope.interface import implementer
 from twisted.words.xish import domish
 from wokkel import rsm
 from wokkel import mam
 from sat.core.i18n import _
 from sat.core.constants import Const as C
 from sat.core.log import getLogger
-from sat.core.xmpp import SatXMPPEntity
+from sat.core.core_types import SatXMPPEntity
 from sat.core import exceptions
+from sat.tools import utils
 from sat.tools import sat_defer
 from sat.tools import xml_tools
 from sat.tools.common import data_format
 OWPOL_ORIGINAL = "original_publisher"
 OWPOL_ANY_PUB = "any_publisher"
 ID_SINGLETON = "current"
 EXTRA_PUBLISH_OPTIONS = "publish_options"
 EXTRA_ON_PRECOND_NOT_MET = "on_precondition_not_met"
+# extra disco needed for RSM, cf. XEP-0060 § 6.5.4
+DISCO_RSM = "http://jabber.org/protocol/pubsub#rsm"
 def __init__(self, host):
 log.info(_("PubSub plugin initialization"))
 self.host = host
 self._rsm = host.plugins.get("XEP-0059")
 async_=True,
 )
 host.bridge.addMethod(
 "psItemsGet",
 ".plugin",
-in_sign="ssiassa{ss}s",
+in_sign="ssiassss",
 out_sign="s",
 method=self._getItems,
 async_=True,
 )
 host.bridge.addMethod(
 async_=True,
 )
 host.bridge.addMethod(
 "psGetFromMany",
 ".plugin",
-in_sign="a(ss)ia{ss}s",
+in_sign="a(ss)iss",
 out_sign="s",
 method=self._getFromMany,
 )
 host.bridge.addMethod(
 "psGetFromManyRTResult",
 all node *prefixed* with this one will be triggered
 @param **kwargs: method(s) to call when the node is found
 the method must be named after PubSub constants in lower case
 and suffixed with "_cb"
 e.g.: "items_cb" for C.PS_ITEMS, "delete_cb" for C.PS_DELETE
+note: only C.PS_ITEMS and C.PS_DELETE are implemented so far
 """
 assert node is not None
 assert kwargs
 callbacks = self._node_cb.setdefault(node, {})
 for event, cb in kwargs.items():
 profile_key=C.PROF_KEY_NONE):
 client = self.host.getClient(profile_key)
 service = None if not service else jid.JID(service)
 payload = xml_tools.parse(payload)
 extra = data_format.deserialise(extra_ser)
-d = self.sendItem(
+d = defer.ensureDeferred(self.sendItem(
 client, service, nodeIdentifier, payload, item_id or None, extra
-)
+))
 d.addCallback(lambda ret: ret or "")
 return d
 def _sendItems(self, service, nodeIdentifier, items, extra_ser=None,
 profile_key=C.PROF_KEY_NONE):
 items = [xml_tools.parse(item) for item in items]
 except Exception as e:
 raise exceptions.DataError(_("Can't parse items: {msg}").format(
 msg=e))
 extra = data_format.deserialise(extra_ser)
-d = self.sendItems(
+return defer.ensureDeferred(self.sendItems(
 client, service, nodeIdentifier, items, extra
-)
+))
-return d
+async def sendItem(
-def _getPublishedItemId(self, published_ids, original_id):
+self, client, service, nodeIdentifier, payload, item_id=None, extra=None
-"""Return item of published id if found in answer
+):
-if not found original_id is returned, which may be None
-"""
-try:
-return published_ids[0]
-except IndexError:
-return original_id
-def sendItem(self, client, service, nodeIdentifier, payload, item_id=None,
-extra=None):
 """High level method to send one item
 @param service(jid.JID, None): service to send the item to
 None to use PEP
 @param NodeIdentifier(unicode): PubSub node to use
 assert isinstance(payload, domish.Element)
 item_elt = domish.Element((pubsub.NS_PUBSUB, 'item'))
 if item_id is not None:
 item_elt['id'] = item_id
 item_elt.addChild(payload)
-d = defer.ensureDeferred(self.sendItems(
+published_ids = await self.sendItems(
 client,
 service,
 nodeIdentifier,
 [item_elt],
 extra
-))
+)
-d.addCallback(self._getPublishedItemId, item_id)
+try:
-return d
+return published_ids[0]
+except IndexError:
+return item_id
 async def sendItems(self, client, service, nodeIdentifier, items, extra=None):
 """High level method to send several items at once
 @param service(jid.JID, None): service to send the item to
 return [item['id']
 for item in iq_result.pubsub.publish.elements(pubsub.NS_PUBSUB, 'item')]
 except AttributeError:
 return []
-def publish(self, client, service, nodeIdentifier, items=None, options=None):
+async def publish(
-return client.pubsub_client.publish(
+self,
+client: SatXMPPEntity,
+service: jid.JID,
+nodeIdentifier: str,
+items: Optional[List[domish.Element]] = None,
+options: Optional[dict] = None
+) -> List[str]:
+published_ids = await client.pubsub_client.publish(
 service, nodeIdentifier, items, client.pubsub_client.parent.jid,
 options=options
 )
+await self.host.trigger.asyncPoint(
+"XEP-0060_publish", client, service, nodeIdentifier, items, options,
+published_ids
+)
+return published_ids
 def _unwrapMAMMessage(self, message_elt):
 try:
 item_elt = reduce(
 lambda elt, ns_name: next(elt.elements(*ns_name)),
 items, metadata = items_data
 metadata['items'] = items
 return data_format.serialise(metadata)
 def _getItems(self, service="", node="", max_items=10, item_ids=None, sub_id=None,
-extra_dict=None, profile_key=C.PROF_KEY_NONE):
+extra="", profile_key=C.PROF_KEY_NONE):
 """Get items from pubsub node
 @param max_items(int): maximum number of item to get, C.NO_LIMIT for no limit
 """
 client = self.host.getClient(profile_key)
 service = jid.JID(service) if service else None
 max_items = None if max_items == C.NO_LIMIT else max_items
-extra = self.parseExtra(extra_dict)
+extra = self.parseExtra(data_format.deserialise(extra))
-d = self.getItems(
+d = defer.ensureDeferred(self.getItems(
 client,
 service,
 node or None,
-max_items or None,
+max_items,
 item_ids,
 sub_id or None,
 extra.rsm_request,
 extra.extra,
-)
+))
 d.addCallback(self.transItemsData)
 d.addCallback(self.serialiseItems)
 return d
-def getItems(self, client, service, node, max_items=None, item_ids=None, sub_id=None,
+async def getItems(
-rsm_request=None, extra=None):
+self,
+client: SatXMPPEntity,
+service: Optional[jid.JID],
+node: str,
+max_items: Optional[int] = None,
+item_ids: Optional[List[str]] = None,
+sub_id: Optional[str] = None,
+rsm_request: Optional[rsm.RSMRequest] = None,
+extra: Optional[dict] = None
+) -> Tuple[List[dict], dict]:
 """Retrieve pubsub items from a node.
 @param service (JID, None): pubsub service.
 @param node (str): node id.
 @param max_items (int): optional limit on the number of retrieved items.
 max_items = None
 if rsm_request and item_ids:
 raise ValueError("items_id can't be used with rsm")
 if extra is None:
 extra = {}
+cont, ret = await self.host.trigger.asyncReturnPoint(
+"XEP-0060_getItems", client, service, node, max_items, item_ids, sub_id,
+rsm_request, extra
+)
+if not cont:
+return ret
 try:
 mam_query = extra["mam"]
 except KeyError:
 d = client.pubsub_client.items(
 service = service,
 itemIdentifiers = item_ids,
 orderBy = extra.get(C.KEY_ORDER_BY),
 rsm_request = rsm_request
 )
 # we have no MAM data here, so we add None
-d.addCallback(lambda data: data + (None,))
 d.addErrback(sat_defer.stanza2NotFound)
 d.addTimeout(TIMEOUT, reactor)
+items, rsm_response = await d
+mam_response = None
 else:
 # if mam is requested, we have to do a totally different query
 if self._mam is None:
 raise exceptions.NotFound("MAM (XEP-0313) plugin is not available")
 if max_items is not None:
 else:
 if mam_query.rsm != rsm_request:
 raise exceptions.DataError(
 "Conflict between RSM request and MAM's RSM request"
 )
-d = self._mam.getArchives(client, mam_query, service, self._unwrapMAMMessage)
+items, rsm_response, mam_response = await self._mam.getArchives(
+client, mam_query, service, self._unwrapMAMMessage
+)
 try:
 subscribe = C.bool(extra["subscribe"])
 except KeyError:
 subscribe = False
-def subscribeEb(failure, service, node):
+if subscribe:
-failure.trap(error.StanzaError)
+try:
-log.warning(
+await self.subscribe(client, service, node)
-"Could not subscribe to node {} on service {}: {}".format(
+except error.StanzaError as e:
-node, str(service), str(failure.value)
+log.warning(
+f"Could not subscribe to node {node} on service {service}: {e}"
 )
-)
+# TODO: handle mam_response
-def doSubscribe(data):
+service_jid = service if service else client.jid.userhostJID()
-self.subscribe(client, service, node).addErrback(
+metadata = {
-subscribeEb, service, node
+"service": service_jid,
-)
+"node": node,
-return data
+"uri": self.getNodeURI(service_jid, node),
+}
-if subscribe:
+if mam_response is not None:
-d.addCallback(doSubscribe)
+# mam_response is a dict with "complete" and "stable" keys
+# we can put them directly in metadata
-def addMetadata(result):
+metadata.update(mam_response)
-# TODO: handle the third argument (mam_response)
+if rsm_request is not None and rsm_response is not None:
-items, rsm_response, mam_response = result
+metadata['rsm'] = rsm_response.toDict()
-service_jid = service if service else client.jid.userhostJID()
+if mam_response is None:
-metadata = {
+index = rsm_response.index
-"service": service_jid,
+count = rsm_response.count
-"node": node,
+if index is None or count is None:
-"uri": self.getNodeURI(service_jid, node),
+# we don't have enough information to know if the data is complete
-}
+# or not
-if mam_response is not None:
+metadata["complete"] = None
-# mam_response is a dict with "complete" and "stable" keys
+else:
-# we can put them directly in metadata
+# normally we have a strict equality here but XEP-0059 states
-metadata.update(mam_response)
+# that index MAY be approximative, so just in case…
-if rsm_request is not None and rsm_response is not None:
+metadata["complete"] = index + len(items) >= count
-metadata['rsm'] = rsm_response.toDict()
-if mam_response is None:
+return (items, metadata)
-index = rsm_response.index
-count = rsm_response.count
-if index is None or count is None:
-# we don't have enough information to know if the data is complete
-# or not
-metadata["complete"] = None
-else:
-# normally we have a strict equality here but XEP-0059 states
-# that index MAY be approximative, so just in case…
-metadata["complete"] = index + len(items) >= count
-return (items, metadata)
-d.addCallback(addMetadata)
-return d
 # @defer.inlineCallbacks
 # def getItemsFromMany(self, service, data, max_items=None, sub_id=None, rsm=None, profile_key=C.PROF_KEY_NONE):
 #     """Massively retrieve pubsub items from many nodes.
 #     @param service (JID): target service.
 nodeIdentifier,
 itemIdentifiers,
 notify=True,
 ):
 return client.pubsub_client.retractItems(
-service, nodeIdentifier, itemIdentifiers, notify=True
+service, nodeIdentifier, itemIdentifiers, notify=notify
 )
 def _renameItem(
 self,
 service,
 await self.retractItems(client, service, node, [item_id])
 def _subscribe(self, service, nodeIdentifier, options, profile_key=C.PROF_KEY_NONE):
 client = self.host.getClient(profile_key)
 service = None if not service else jid.JID(service)
-d = self.subscribe(client, service, nodeIdentifier, options=options or None)
+d = defer.ensureDeferred(
+self.subscribe(client, service, nodeIdentifier, options=options or None)
+)
 d.addCallback(lambda subscription: subscription.subscriptionIdentifier or "")
 return d
-def subscribe(self, client, service, nodeIdentifier, sub_jid=None, options=None):
+async def subscribe(
+self,
+client: SatXMPPEntity,
+service: jid.JID,
+nodeIdentifier: str,
+sub_jid: Optional[jid.JID] = None,
+options: Optional[dict] = None
+) -> pubsub.Subscription:
 # TODO: reimplement a subscribtion cache, checking that we have not subscription before trying to subscribe
-return client.pubsub_client.subscribe(
+subscription = await client.pubsub_client.subscribe(
 service, nodeIdentifier, sub_jid or client.jid.userhostJID(), options=options
 )
+await self.host.trigger.asyncPoint(
+"XEP-0060_subscribe", client, service, nodeIdentifier, sub_jid, options,
+subscription
+)
+return subscription
 def _unsubscribe(self, service, nodeIdentifier, profile_key=C.PROF_KEY_NONE):
 client = self.host.getClient(profile_key)
 service = None if not service else jid.JID(service)
-return self.unsubscribe(client, service, nodeIdentifier)
+return defer.ensureDeferred(self.unsubscribe(client, service, nodeIdentifier))
-def unsubscribe(
+async def unsubscribe(
 self,
-client,
+client: SatXMPPEntity,
-service,
+service: jid.JID,
-nodeIdentifier,
+nodeIdentifier: str,
 sub_jid=None,
 subscriptionIdentifier=None,
 sender=None,
 ):
-return client.pubsub_client.unsubscribe(
+await client.pubsub_client.unsubscribe(
 service,
 nodeIdentifier,
 sub_jid or client.jid.userhostJID(),
 subscriptionIdentifier,
 sender,
+)
+await self.host.trigger.asyncPoint(
+"XEP-0060_unsubscribe", client, service, nodeIdentifier, sub_jid,
+subscriptionIdentifier, sender
 )
 def _subscriptions(self, service, nodeIdentifier="", profile_key=C.PROF_KEY_NONE):
 client = self.host.getClient(profile_key)
 service = None if not service else jid.JID(service)
 @return (str): RT Deferred session id
 """
 client = self.host.getClient(profile_key)
 deferreds = {}
 for service, node in node_data:
-deferreds[(service, node)] = client.pubsub_client.subscribe(
+deferreds[(service, node)] = defer.ensureDeferred(
-service, node, subscriber, options=options
+client.pubsub_client.subscribe(
+service, node, subscriber, options=options
+)
 )
 return self.rt_sessions.newSession(deferreds, client.profile)
 # found_nodes = yield self.listNodes(service, profile=client.profile)
 # subscribed_nodes = yield self.listSubscribedNodes(service, profile=client.profile)
 # d_list = []
 )
 )
 return d
 def _getFromMany(
-self, node_data, max_item=10, extra_dict=None, profile_key=C.PROF_KEY_NONE
+self, node_data, max_item=10, extra="", profile_key=C.PROF_KEY_NONE
 ):
 """
 @param max_item(int): maximum number of item to get, C.NO_LIMIT for no limit
 """
 max_item = None if max_item == C.NO_LIMIT else max_item
-extra = self.parseExtra(extra_dict)
+extra = self.parseExtra(data_format.deserialise(extra))
 return self.getFromMany(
 [(jid.JID(service), str(node)) for service, node in node_data],
 max_item,
 extra.rsm_request,
 extra.extra,
 @return (str): RT Deferred session id
 """
 client = self.host.getClient(profile_key)
 deferreds = {}
 for service, node in node_data:
-deferreds[(service, node)] = self.getItems(
+deferreds[(service, node)] = defer.ensureDeferred(self.getItems(
 client, service, node, max_item, rsm_request=rsm_request, extra=extra
-)
+))
 return self.rt_sessions.newSession(deferreds, client.profile)
 @implementer(disco.IDisco)
 class SatPubSubClient(rsm.PubSubClient):
 def itemsReceived(self, event):
 log.debug("Pubsub items received")
 for callback in self._getNodeCallbacks(event.nodeIdentifier, C.PS_ITEMS):
-callback(self.parent, event)
+d = utils.asDeferred(callback, self.parent, event)
+d.addErrback(lambda f: log.error(
+f"Error while running items event callback {callback}: {f}"
+))
 client = self.parent
 if (event.sender, event.nodeIdentifier) in client.pubsub_watching:
 raw_items = [i.toXml() for i in event.items]
 self.host.bridge.psEventRaw(
 event.sender.full(),
 )
 def deleteReceived(self, event):
 log.debug(("Publish node deleted"))
 for callback in self._getNodeCallbacks(event.nodeIdentifier, C.PS_DELETE):
-callback(self.parent, event)
+d = utils.asDeferred(callback, self.parent, event)
+d.addErrback(lambda f: log.error(
+f"Error while running delete event callback {callback}: {f}"
+))
 client = self.parent
 if (event.sender, event.nodeIdentifier) in client.pubsub_watching:
 self.host.bridge.psEventRaw(
 event.sender.full(), event.nodeIdentifier, C.PS_DELETE, [], client.profile
+)
+def purgeReceived(self, event):
+log.debug(("Publish node purged"))
+for callback in self._getNodeCallbacks(event.nodeIdentifier, C.PS_PURGE):
+d = utils.asDeferred(callback, self.parent, event)
+d.addErrback(lambda f: log.error(
+f"Error while running purge event callback {callback}: {f}"
+))
+client = self.parent
+if (event.sender, event.nodeIdentifier) in client.pubsub_watching:
+self.host.bridge.psEventRaw(
+event.sender.full(), event.nodeIdentifier, C.PS_PURGE, [], client.profile
 )
 def subscriptions(self, service, nodeIdentifier, sender=None):
 """Return the list of subscriptions to the given service and node.

Mercurial > libervia-backend

comparison sat/plugins/plugin_xep_0060.py @ 3715:b9718216a1c0 0.9