libervia-backend: sat/plugins/plugin_xep

comparison sat/plugins/plugin_xep_0060.py @ 3584:edc79cefe968

plugin XEP-0060: `getItem(s)`, `publish` and `(un)subscribe` are now coroutines

author	Goffi <goffi@goffi.org>
date	Wed, 30 Jun 2021 16:19:14 +0200
parents	02eec2a5b5f9
children	5f65f4e9f8cb

comparison

equal deleted inserted replaced

-:16ade4ad63f3
+:edc79cefe968
 # 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 sat_defer
 from sat.tools import xml_tools
 from sat.tools.common import data_format
 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(self, client, service, nodeIdentifier, items=None, options=None):
-return client.pubsub_client.publish(
+return await client.pubsub_client.publish(
 service, nodeIdentifier, items, client.pubsub_client.parent.jid,
 options=options
 )
 def _unwrapMAMMessage(self, message_elt):
 """
 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)
-d = self.getItems(
+d = defer.ensureDeferred(self.getItems(
 client,
 service,
 node or None,
 max_items or None,
 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.
 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.
 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, service, nodeIdentifier, sub_jid=None, options=None):
 # TODO: reimplement a subscribtion cache, checking that we have not subscription before trying to subscribe
-return client.pubsub_client.subscribe(
+return await client.pubsub_client.subscribe(
 service, nodeIdentifier, sub_jid or client.jid.userhostJID(), options=options
 )
 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,
 service,
 nodeIdentifier,
 sub_jid=None,
 subscriptionIdentifier=None,
 sender=None,
 ):
-return client.pubsub_client.unsubscribe(
+return await client.pubsub_client.unsubscribe(
 service,
 nodeIdentifier,
 sub_jid or client.jid.userhostJID(),
 subscriptionIdentifier,
 sender,
 @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 (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):

Mercurial > libervia-backend

comparison sat/plugins/plugin_xep_0060.py @ 3584:edc79cefe968