Mercurial > libervia-backend
diff src/plugins/plugin_xep_0060.py @ 1446:e8c8e467964b
plugins xep-0060, xep-0277: code simplification/cleaning/fix:
- plugin xep-0060: moved rsm data to a more general metadata dict, which will contain all data relative to the node/items set. RSM metadata are prefixed with "rsm_"
- plugin xep-0060: minor docstring fixes
- plugin xep-0060: removed cache to simplify code base
- fixed broken getLastMicroblogs
- added _getLastMicroblogs as wrapper to getLastMicroblogs, for bridge
- removed lxml dependecy for this plugin, use native twisted instead
- several improvments/fixes in item2mbdata
author | Goffi <goffi@goffi.org> |
---|---|
date | Sat, 15 Aug 2015 22:13:27 +0200 |
parents | 7c0acb966fd6 |
children | 389357fd79ce |
line wrap: on
line diff
--- a/src/plugins/plugin_xep_0060.py Wed Jul 22 11:42:37 2015 +0200 +++ b/src/plugins/plugin_xep_0060.py Sat Aug 15 22:13:27 2015 +0200 @@ -21,11 +21,10 @@ from sat.core.constants import Const as C from sat.core.log import getLogger log = getLogger(__name__) -from sat.memory.memory import Sessions from wokkel import disco, pubsub, rsm from zope.interface import implements -from twisted.internet import defer +# from twisted.internet import defer import uuid @@ -58,7 +57,6 @@ log.info(_(u"PubSub plugin initialization")) self.host = host self.managedNodes = [] - self.node_cache = Sessions(timeout=60, resettable_timeout=False) def getHandler(self, profile): client = self.host.getClient(profile) @@ -73,64 +71,34 @@ @param profile: profile which manage this handler""" self.managedNodes.append((node_name, callback)) - def _getDeferredNodeCache(self, session_id, init, profile): - """Manage a node cache with deferred initialisation and concurrent access. + # def listNodes(self, service, nodeIdentifier='', profile=C.PROF_KEY_NONE): + # """Retrieve the name of the nodes that are accessible on the target service. - @param session_id (string): node cache session ID - @param init (Deferred): deferred list of strings to initialise the cache. - @param profile (str): %(doc_profile)s - @return: Deferred list[str] - """ - if session_id in self.node_cache: - cache = self.node_cache.profileGet(session_id, profile) - if cache['nodes'] is None: - # init is still running - d = defer.Deferred() - cache['waiting'].append(d) - return d - return defer.succeed(cache['nodes']) - - cache = {'init': init, 'waiting': [], 'nodes': None} - self.node_cache.newSession(cache, session_id=session_id, profile=profile) - - def cb(nodes): - cache['nodes'] = nodes - for d in cache['waiting']: - d.callback(nodes) - return nodes - - return init.addCallback(cb) + # @param service (JID): target service + # @param nodeIdentifier (str): the parent node name (leave empty to retrieve first-level nodes) + # @param profile (str): %(doc_profile)s + # @return: deferred which fire a list of nodes + # """ + # d = self.host.getDiscoItems(service, nodeIdentifier, profile_key=profile) + # d.addCallback(lambda result: [item.getAttribute('node') for item in result.toElement().children if item.hasAttribute('node')]) + # return d - def listNodes(self, service, nodeIdentifier='', profile=C.PROF_KEY_NONE): - """Retrieve the name of the nodes that are accessible on the target service. - - @param service (JID): target service - @param nodeIdentifier (str): the parent node name (leave empty to retrieve first-level nodes) - @param profile (str): %(doc_profile)s - @return: Deferred list[str] - """ - session_id = profile + '@found@' + service.userhost() - d = self.host.getDiscoItems(service, nodeIdentifier, profile_key=profile) - d.addCallback(lambda result: [item.getAttribute('node') for item in result.toElement().children if item.hasAttribute('node')]) - return self._getDeferredNodeCache(session_id, d, profile) + # def listSubscribedNodes(self, service, nodeIdentifier='', filter_='subscribed', profile=C.PROF_KEY_NONE): + # """Retrieve the name of the nodes to which the profile is subscribed on the target service. - def listSubscribedNodes(self, service, nodeIdentifier='', filter_='subscribed', profile=C.PROF_KEY_NONE): - """Retrieve the name of the nodes to which the profile is subscribed on the target service. - - @param service (JID): target service - @param nodeIdentifier (str): the parent node name (leave empty to retrieve all subscriptions) - @param filter_ (str): filter the result according to the given subscription type: - - None: do not filter - - 'pending': subscription has not been approved yet by the node owner - - 'unconfigured': subscription options have not been configured yet - - 'subscribed': subscription is complete - @param profile (str): %(doc_profile)s - @return: Deferred list[str] - """ - session_id = profile + '@subscriptions@' + service.userhost() - d = self.subscriptions(service, nodeIdentifier, profile_key=profile) - d.addCallback(lambda subs: [sub.getAttribute('node') for sub in subs if sub.getAttribute('subscription') == filter_]) - return self._getDeferredNodeCache(session_id, d, profile) + # @param service (JID): target service + # @param nodeIdentifier (str): the parent node name (leave empty to retrieve all subscriptions) + # @param filter_ (str): filter the result according to the given subscription type: + # - None: do not filter + # - 'pending': subscription has not been approved yet by the node owner + # - 'unconfigured': subscription options have not been configured yet + # - 'subscribed': subscription is complete + # @param profile (str): %(doc_profile)s + # @return: Deferred list[str] + # """ + # d = self.subscriptions(service, nodeIdentifier, profile_key=profile) + # d.addCallback(lambda subs: [sub.getAttribute('node') for sub in subs if sub.getAttribute('subscription') == filter_]) + # return d def publish(self, service, nodeIdentifier, items=None, profile_key=C.PROF_KEY_NONE): client = self.host.getClient(profile_key) @@ -142,45 +110,56 @@ @param service (JID): target service. @param node (str): node id. @param max_items (int): optional limit on the number of retrieved items. - @param item_ids (list[str]): identifiers of the items to be retrieved (should not be used). + @param item_ids (list[str]): identifiers of the items to be retrieved (can't be used with rsm). @param sub_id (str): optional subscription identifier. @param rsm (dict): RSM request data @param profile_key (str): %(doc_profile_key)s @return: a deferred couple (list[dict], dict) containing: - list of items - - RSM response data + - metadata with the following keys: + - rsm_first, rsm_last, rsm_count, rsm_index: first, last, count and index value of RSMResponse """ + if rsm and item_ids: + raise ValueError("items_id can't be used with rsm") client = self.host.getClient(profile_key) ext_data = {'id': unicode(uuid.uuid4()), 'rsm': rsm} if rsm else None d = client.pubsub_client.items(service, node, max_items, item_ids, sub_id, client.pubsub_client.parent.jid, ext_data) - d.addCallback(lambda items: (items, client.pubsub_client.getRSMResponse(ext_data['id']) if rsm else {})) + def addMetadata(items): + if not rsm: + metadata = {} + else: + rsm_data = client.pubsub_client.getRSMResponse(ext_data['id']) + metadata = {'rsm_{}'.format(key): value for key, value in rsm_data} + 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. + # @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. - @param data (dict): dictionnary binding some arbitrary keys to the node identifiers. - @param max_items (int): optional limit on the number of retrieved items *per node*. - @param sub_id (str): optional subscription identifier. - @param rsm (dict): RSM request data - @param profile_key (str): %(doc_profile_key)s - @return: a deferred dict with: - - key: a value in (a subset of) data.keys() - - couple (list[dict], dict) containing: - - list of items - - RSM response data - """ - client = self.host.getClient(profile_key) - found_nodes = yield self.listNodes(service, profile=client.profile) - d_dict = {} - for publisher, node in data.items(): - if node not in found_nodes: - log.debug(u"Skip the items retrieval for [{node}]: node doesn't exist".format(node=node)) - continue # avoid pubsub "item-not-found" error - d_dict[publisher] = self.getItems(service, node, max_items, None, sub_id, rsm, client.profile) - defer.returnValue(d_dict) + # @param service (JID): target service. + # @param data (dict): dictionnary binding some arbitrary keys to the node identifiers. + # @param max_items (int): optional limit on the number of retrieved items *per node*. + # @param sub_id (str): optional subscription identifier. + # @param rsm (dict): RSM request data + # @param profile_key (str): %(doc_profile_key)s + # @return: a deferred dict with: + # - key: a value in (a subset of) data.keys() + # - couple (list[dict], dict) containing: + # - list of items + # - RSM response data + # """ + # client = self.host.getClient(profile_key) + # found_nodes = yield self.listNodes(service, profile=client.profile) + # d_dict = {} + # for publisher, node in data.items(): + # if node not in found_nodes: + # log.debug(u"Skip the items retrieval for [{node}]: node doesn't exist".format(node=node)) + # continue # avoid pubsub "item-not-found" error + # d_dict[publisher] = self.getItems(service, node, max_items, None, sub_id, rsm, client.profile) + # defer.returnValue(d_dict) def getOptions(self, service, nodeIdentifier, subscriber, subscriptionIdentifier=None, profile_key=C.PROF_KEY_NONE): client = self.host.getClient(profile_key) @@ -206,27 +185,27 @@ client = self.host.getClient(profile_key) return client.pubsub_client.subscribe(service, nodeIdentifier, sub_jid or client.pubsub_client.parent.jid.userhostJID(), options=options) - @defer.inlineCallbacks - def subscribeToMany(self, service, nodeIdentifiers, sub_jid=None, options=None, profile_key=C.PROF_KEY_NONE): - """Massively subscribe to many nodes. + # @defer.inlineCallbacks + # def subscribeToMany(self, service, nodeIdentifiers, sub_jid=None, options=None, profile_key=C.PROF_KEY_NONE): + # """Massively subscribe to many nodes. - @param service (JID): target service. - @param nodeIdentifiers (list): the list of node identifiers to subscribe to. - @param sub_id (str): optional subscription identifier. - @param options (list): optional list of subscription options - @param profile_key (str): %(doc_profile_key)s - @return: list of Deferred instances. - """ - client = self.host.getClient(profile_key) - found_nodes = yield self.listNodes(service, profile=client.profile) - subscribed_nodes = yield self.listSubscribedNodes(service, profile=client.profile) - d_list = [] - for nodeIdentifier in (set(nodeIdentifiers) - set(subscribed_nodes)): - if nodeIdentifier not in found_nodes: - log.debug(u"Skip the subscription to [{node}]: node doesn't exist".format(node=nodeIdentifier)) - continue # avoid sat-pubsub "SubscriptionExists" error - d_list.append(client.pubsub_client.subscribe(service, nodeIdentifier, sub_jid or client.pubsub_client.parent.jid.userhostJID(), options=options)) - defer.returnValue(d_list) + # @param service (JID): target service. + # @param nodeIdentifiers (list): the list of node identifiers to subscribe to. + # @param sub_id (str): optional subscription identifier. + # @param options (list): optional list of subscription options + # @param profile_key (str): %(doc_profile_key)s + # @return: list of Deferred instances. + # """ + # client = self.host.getClient(profile_key) + # found_nodes = yield self.listNodes(service, profile=client.profile) + # subscribed_nodes = yield self.listSubscribedNodes(service, profile=client.profile) + # d_list = [] + # for nodeIdentifier in (set(nodeIdentifiers) - set(subscribed_nodes)): + # if nodeIdentifier not in found_nodes: + # log.debug(u"Skip the subscription to [{node}]: node doesn't exist".format(node=nodeIdentifier)) + # continue # avoid sat-pubsub "SubscriptionExists" error + # d_list.append(client.pubsub_client.subscribe(service, nodeIdentifier, sub_jid or client.pubsub_client.parent.jid.userhostJID(), options=options)) + # defer.returnValue(d_list) def subscriptions(self, service, nodeIdentifier='', profile_key=C.PROF_KEY_NONE): client = self.host.getClient(profile_key) @@ -272,6 +251,7 @@ d = request.send(self.xmlstream) def cb(iq): + # FIXME: to be checked return [sub for sub in iq.pubsub.subscriptions.elements() if (sub.uri == pubsub.NS_PUBSUB and sub.name == 'subscription')]