libervia-backend/src/plugins/plugin_xep_0060.py

#!/usr/bin/python
# -*- coding: utf-8 -*-

# SAT plugin for Publish-Subscribe (xep-0060)
# Copyright (C) 2009, 2010, 2011, 2012, 2013, 2014, 2015 Jérôme Poisson (goffi@goffi.org)

# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.

# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.

# 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 sat.core.i18n import _
from sat.core.constants import Const as C
from sat.core.log import getLogger
log = getLogger(__name__)

from sat.tools import sat_defer

from twisted.words.protocols.jabber import jid
from twisted.internet import defer
from wokkel import disco
from wokkel import pubsub
from wokkel import rsm
from zope.interface import implements
# from twisted.internet import defer
import uuid

UNSPECIFIED = "unspecified error"


PLUGIN_INFO = {
    "name": "Publish-Subscribe",
    "import_name": "XEP-0060",
    "type": "XEP",
    "protocols": ["XEP-0060"],
    "dependencies": [],
    "recommendations": ["XEP-0059"],
    "main": "XEP_0060",
    "handler": "yes",
    "description": _("""Implementation of PubSub Protocol""")
}


class XEP_0060(object):
    OPT_ACCESS_MODEL = 'pubsub#access_model'
    OPT_PERSIST_ITEMS = 'pubsub#persist_items'
    OPT_MAX_ITEMS = 'pubsub#max_items'
    OPT_DELIVER_PAYLOADS = 'pubsub#deliver_payloads'
    OPT_SEND_ITEM_SUBSCRIBE = 'pubsub#send_item_subscribe'
    OPT_NODE_TYPE = 'pubsub#node_type'
    OPT_SUBSCRIPTION_TYPE = 'pubsub#subscription_type'
    OPT_SUBSCRIPTION_DEPTH = 'pubsub#subscription_depth'
    OPT_ROSTER_GROUPS_ALLOWED = 'pubsub#roster_groups_allowed'
    OPT_PUBLISH_MODEL = 'pubsub#publish_model'

    def __init__(self, host):
        log.info(_(u"PubSub plugin initialization"))
        self.host = host
        self.managedNodes = []
        self.rt_sessions = sat_defer.RTDeferredSessions()
        host.bridge.addMethod("subscribeToMany", ".plugin", in_sign='a(ss)sa{ss}s', out_sign='s', method=self._subscribeToMany)
        host.bridge.addMethod("getSubscribeRTResult", ".plugin", in_sign='ss', out_sign='(ua(sss))', method=self._manySubscribeRTResult, async=True)

    def getHandler(self, profile):
        client = self.host.getClient(profile)
        client.pubsub_client = SatPubSubClient(self.host, self)
        return client.pubsub_client

    def addManagedNode(self, node_name, callback):
        """Add a handler for a namespace

        @param namespace: NS of the handler (will appear in disco info)
        @param callback: method to call when the handler is found
        @param profile: profile which manage this handler"""
        self.managedNodes.append((node_name, callback))

    # 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 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 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]
    #     """
    #     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)
        return client.pubsub_client.publish(service, nodeIdentifier, items, client.pubsub_client.parent.jid)

    def getItems(self, service, node, max_items=None, item_ids=None, sub_id=None, rsm=None, profile_key=C.PROF_KEY_NONE):
        """Retrieve pubsub items from a node.

        @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 (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
            - 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)
        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.

    #     @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)
        return client.pubsub_client.getOptions(service, nodeIdentifier, subscriber, subscriptionIdentifier)

    def setOptions(self, service, nodeIdentifier, subscriber, options, subscriptionIdentifier=None, profile_key=C.PROF_KEY_NONE):
        client = self.host.getClient(profile_key)
        return client.pubsub_client.setOptions(service, nodeIdentifier, subscriber, options, subscriptionIdentifier)

    def createNode(self, service, nodeIdentifier, options, profile_key=C.PROF_KEY_NONE):
        client = self.host.getClient(profile_key)
        return client.pubsub_client.createNode(service, nodeIdentifier, options)

    def deleteNode(self, service, nodeIdentifier, profile_key=C.PROF_KEY_NONE):
        client = self.host.getClient(profile_key)
        return client.pubsub_client.deleteNode(service, nodeIdentifier)

    def retractItems(self, service, nodeIdentifier, itemIdentifiers, profile_key=C.PROF_KEY_NONE):
        client = self.host.getClient(profile_key)
        return client.pubsub_client.retractItems(service, nodeIdentifier, itemIdentifiers)

    def subscribe(self, service, nodeIdentifier, sub_jid=None, options=None, profile_key=C.PROF_KEY_NONE):
        client = self.host.getClient(profile_key)
        return client.pubsub_client.subscribe(service, nodeIdentifier, sub_jid or client.pubsub_client.parent.jid.userhostJID(), options=options)

    def subscriptions(self, service, nodeIdentifier='', profile_key=C.PROF_KEY_NONE):
        client = self.host.getClient(profile_key)
        return client.pubsub_client.subscriptions(service, nodeIdentifier)

    ## methods to manage several stanzas/jids at once ##

    def serItemsData(self, items_data, item_cb=lambda item: item.toXml()):
        """Helper method to serialise result from [getItems]

        the items_data must be a tuple(list[domish.Element], dict[unicode, unicode])
        as returned by [getItems]. metadata values are then casted to unicode and
        each item is passed to items_cb
        @param items_data(tuple): tuple returned by [getItems]
        @param item_cb(callable): method to transform each item
        @return (tuple): a serialised form ready to go throught bridge
        """
        items, metadata = items_data
        return [item_cb(item) for item in items], {key: unicode(value) for key, value in metadata.iteritems()}

    def serItemsDataD(self, items_data, item_cb):
        """Helper method to serialise result from [getItems], deferred version

        the items_data must be a tuple(list[domish.Element], dict[unicode, unicode])
        as returned by [getItems]. metadata values are then casted to unicode and
        each item is passed to items_cb
        An errback is added to item_cb, and when it is fired the value is filtered from final items
        @param items_data(tuple): tuple returned by [getItems]
        @param item_cb(callable): method to transform each item (must return a deferred)
        @return (tuple): a deferred which fire a serialised form ready to go throught bridge
        """
        items, metadata = items_data
        def eb(failure):
            log.warning("Error while serialising/parsing item: {}".format(unicode(failure.value)))
        d = defer.gatherResults([item_cb(item).addErrback(eb) for item in items])
        def finishSerialisation(serialised_items):
            return [item for item in serialised_items if item is not None], {key: unicode(value) for key, value in metadata.iteritems()}
        d.addCallback(finishSerialisation)
        return d

    def serDList(self, results, failure_result=None):
        """Serialise a DeferredList result

        @param results: DeferredList results
        @param failure_result: value to use as value for failed Deferred
            (default: empty tuple)
        @return (list): list with:
            - failure: empty in case of success, else error message
            - result
        """
        if failure_result is None:
            failure_result = ()
        return [('', result) if success else (unicode(result.result) or UNSPECIFIED, failure_result) for success, result in results]

    # subscribe #

    def _manySubscribeRTResult(self, session_id, profile_key=C.PROF_KEY_DEFAULT):
        """Get real-time results for subcribeToManu session

        @param session_id: id of the real-time deferred session
        @param return (tuple): (remaining, results) where:
            - remaining is the number of still expected results
            - results is a list of tuple(unicode, unicode, bool, unicode) with:
                - service: pubsub service
                - and node: pubsub node
                - failure(unicode): empty string in case of success, error message else
        @param profile_key: %(doc_profile_key)s
        """
        profile = self.host.getClient(profile_key).profile
        d = self.rt_sessions.getResults(session_id, on_success=lambda result:'', on_error=lambda failure:unicode(failure.value), profile=profile)
        # we need to convert jid.JID to unicode with full() to serialise it for the bridge
        d.addCallback(lambda ret: (ret[0], [(service.full(), node, '' if success else failure or UNSPECIFIED)
                                            for (service, node), (success, failure) in ret[1].iteritems()]))
        return d

    def _subscribeToMany(self, node_data, subscriber=None, options=None, profile_key=C.PROF_KEY_NONE):
        return self.subscribeToMany([(jid.JID(service), unicode(node)) for service, node in node_data], jid.JID(subscriber), options, profile_key)

    def subscribeToMany(self, node_data, subscriber, options=None, profile_key=C.PROF_KEY_NONE):
        """Subscribe to several nodes at once.

        @param node_data (iterable[tuple]): iterable of tuple (service, node) where:
            - service (jid.JID) is the pubsub service
            - node (unicode) is the node to subscribe to
        @param subscriber (jid.JID): optional subscription identifier.
        @param options (dict): subscription options
        @param profile_key (str): %(doc_profile_key)s
        @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(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 = []
        # 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)


class SatPubSubClient(rsm.PubSubClient):
    implements(disco.IDisco)

    def __init__(self, host, parent_plugin):
        self.host = host
        self.parent_plugin = parent_plugin
        rsm.PubSubClient.__init__(self)

    def connectionInitialized(self):
        rsm.PubSubClient.connectionInitialized(self)

    def itemsReceived(self, event):
        if not self.host.trigger.point("PubSubItemsReceived", event, self.parent.profile):
            return
        for node in self.parent_plugin.managedNodes:
            if event.nodeIdentifier == node[0]:
                node[1](event, self.parent.profile)

    def deleteReceived(self, event):
        #TODO: manage delete event
        log.debug(_(u"Publish node deleted"))

    # def purgeReceived(self, event):

    def subscriptions(self, service, nodeIdentifier, sender=None):
        """Return the list of subscriptions to the given service and node.

        @param service: The publish subscribe service to retrieve the subscriptions from.
        @type service: L{JID<twisted.words.protocols.jabber.jid.JID>}
        @param nodeIdentifier: The identifier of the node (leave empty to retrieve all subscriptions).
        @type nodeIdentifier: C{unicode}
        """
        request = pubsub.PubSubRequest('subscriptions')
        request.recipient = service
        request.nodeIdentifier = nodeIdentifier
        request.sender = sender
        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')]

        return d.addCallback(cb)

    def getDiscoInfo(self, requestor, service, nodeIdentifier=''):
        disco_info = []
        self.host.trigger.point("PubSub Disco Info", disco_info, self.parent.profile)
        return disco_info

    def getDiscoItems(self, requestor, service, nodeIdentifier=''):
        return []