erebion/libervia-backend
Watch
1
0
Fork
You've already forked libervia-backend
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

plugins xep-0060, xep-0277: code simplification/cleaning/fix:

Browse source 
- 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
This commit is contained in:
Goffi 2015-08-15 22:13:27 +02:00
commit fd3e9ce044
1 changed files with 223 additions and 216 deletions
Download patch file Download diff file Expand all files Collapse all files

190
src/plugins/plugin_xep_0060.py
View file

@ -21,11 +21,10 @@ from sat.core.i18n import _
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 @@ class XEP_0060(object):
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 @@ class XEP_0060(object):
@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'])
# @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
cache = {'init': init, 'waiting': [], 'nodes': None}
self.node_cache.newSession(cache, session_id=session_id, profile=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 cb(nodes):
cache['nodes'] = nodes
for d in cache['waiting']:
d.callback(nodes)
return nodes
return init.addCallback(cb)
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.
@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 @@ class XEP_0060(object):
@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 @@ class XEP_0060(object):
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 @@ class SatPubSubClient(rsm.PubSubClient):
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')]