diff --git a/src/plugins/plugin_xep_0060.py b/src/plugins/plugin_xep_0060.py index 6e03d169..ac861c7b 100755 --- a/src/plugins/plugin_xep_0060.py +++ b/src/plugins/plugin_xep_0060.py @@ -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')] diff --git a/src/plugins/plugin_xep_0277.py b/src/plugins/plugin_xep_0277.py index 29872f76..abd67e62 100755 --- a/src/plugins/plugin_xep_0277.py +++ b/src/plugins/plugin_xep_0277.py @@ -23,20 +23,21 @@ from sat.core.log import getLogger log = getLogger(__name__) from twisted.words.protocols.jabber import jid from twisted.internet import defer +from twisted.python import failure from sat.core import exceptions from sat.tools.xml_tools import ElementParser from wokkel import pubsub from feed import atom, date -from lxml import etree import uuid from time import time import urlparse from cgi import escape NS_MICROBLOG = 'urn:xmpp:microblog:0' +NS_ATOM = 'http://www.w3.org/2005/Atom' NS_XHTML = 'http://www.w3.org/1999/xhtml' -NS_PUBSUB = 'http://jabber.org/protocol/pubsub' +NS_PUBSUB_EVENT = "{}{}".format(pubsub.NS_PUBSUB, "#event") PLUGIN_INFO = { "name": "Microblogging over XMPP Plugin", @@ -60,10 +61,10 @@ class XEP_0277(object): def __init__(self, host): log.info(_("Microblogging plugin initialization")) self.host = host - self.host.plugins["XEP-0163"].addPEPEvent("MICROBLOG", NS_MICROBLOG, self.microblogCB, self.sendMicroblog) + self.host.plugins["XEP-0163"].addPEPEvent("MICROBLOG", NS_MICROBLOG, self.microblogCB, self.sendMicroblog, notify=False) host.bridge.addMethod("getLastMicroblogs", ".plugin", - in_sign='sis', out_sign='aa{ss}', - method=self.getLastMicroblogs, + in_sign='sis', out_sign='(aa{ss}a{ss})', + method=self._getLastMicroblogs, async=True, doc={'summary': 'retrieve items', 'param_0': 'jid: publisher of wanted microblog', @@ -87,16 +88,6 @@ class XEP_0277(object): ## data/item transformation ## - def _getDomishInnerContent(self, elt): - """Return the inner content of a domish.Element.""" - result = '' - for child in elt.children: - try: - result += child.toXml() # child id a domish.Element - except AttributeError: - result += child # child is unicode - return result - def _removeXHTMLMarkups(self, xhtml): """Remove XHTML markups from the given string. @@ -109,107 +100,131 @@ class XEP_0277(object): False) @defer.inlineCallbacks - def item2mbdata(self, item): + def item2mbdata(self, item_elt): """Convert an XML Item to microblog data used in bridge API - @param item: domish.Element of microblog item + @param item_elt: domish.Element of microblog item @return: microblog data (dictionary) """ - - def xpath(elt, path): - """Return the XPATH result of an entry element or its descendance.""" - # XXX: use a wildcard to work with all and even undefined namespaces - return elt.xpath('/'.join(["*[local-name() = '%s']" % tag for tag in path.split('/')])) - - def date2float(elt, path): - """Convert a date string to float without dealing with the date format.""" - return unicode(date.rfc3339.tf_from_timestamp(xpath(elt, path)[0].text)) - - item_elt = etree.fromstring(item.toXml().encode('utf-8')) - item_id = item_elt.get('id', '') - - # XXX: when you raise an exception from inline callbacks, do defer.returnValue(Exception()) - # to make it catchable by an eventual errback. If you do raise Exception, raise Exception() - # or defer.returnValue(Exception), it will explode and then the normal callback is ran. - - if item.uri not in (NS_PUBSUB, NS_PUBSUB + "#event"): - log.error(_(u"Unsupported namespace {ns} in pubsub item {id}").format(ns=item.uri, id=item_id)) - defer.returnValue(exceptions.DataError()) - - try: - entry_elt = xpath(item_elt, 'entry')[0] - except IndexError: - log.error(_(u'No atom entry found in the pubsub item %s') % item_id) - defer.returnValue(exceptions.DataError()) - microblog_data = {} - for key in ['title', 'content']: # process the textual elements - for attr_elt in xpath(entry_elt, key): - # Return the inner content of a lxml.etree.Element. It is not - # trivial because the lxml tostring method would return the full - # content including elt's tag and attributes, and elt.getchildren() - # would skip a text value which is not within an element... - attr_content = self._getDomishInnerContent(ElementParser()(etree.tostring(attr_elt))) - if not attr_content.strip(): - continue # element with empty value - content_type = attr_elt.get('type', 'text').lower() - if content_type == 'xhtml': - # Check for XHTML namespace and decapsulate the content so the user - # who wants to modify an entry will see the text that he entered. Also - # this avoids successive encapsulation with a new