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-0163, xep-0277, groupblog: bloging improvments (huge patch, sorry):

Browse source 
/!\ not everything is working yet, and specially groupblogs are broken /!\
    - renamed bridge api to use prefixed methods (e.g. psSubscribeToMany instead of subscribeToMany in PubSub)
    - (xep-0060): try to find a default PubSub service, and put it in client.pubsub_service
    - (xep-0060): extra dictionary can be used in bridge method for RSM and other options
    - (xep-0060): XEP_0060.addManagedNode and XEP_0060.removeManagedNode allow to easily catch notifications for a specific node
    - (xep-0060): retractItem manage "notify" attribute
    - (xep-0060): new signal psEvent will be used to transmit notifications to frontends
    - (xep-0060, constants): added a bunch of useful constants
    - (xep-0163): removed personalEvent in favor of psEvent
    - (xep-0163): addPEPEvent now filter non PEP events for in_callback
    - (xep-0277): use of new XEP-0060 plugin's addManagedNode
    - (xep-0277): fixed author handling for incoming blogs: author is the human readable name, author_jid it jid, and author_jid_verified is set to True is the jid is checked
    - (xep-0277): reworked data2entry with Twisted instead of feed, item_id can now be specified, <content/> is changed to  <title/> if there is only content
    - (xep-0277): comments are now managed here (core removed from groupblog)
    - (xep-0277): (comments) node is created if needed, default pubsub service is used if available, else PEP
    - (xep-0277): retract is managed
This commit is contained in:
Goffi 2015-08-16 00:39:44 +02:00
commit b7409e0488
1 changed files with 544 additions and 263 deletions
Download patch file Download diff file Expand all files Collapse all files

179
src/plugins/plugin_xep_0060.py
View file

@ -27,10 +27,12 @@ from sat.tools import sat_defer
from twisted.words.protocols.jabber import jid
from twisted.internet import defer
from wokkel import disco
# XXX: tmp.pubsub is actually use instead of wokkel version
# same thing for rsm
from wokkel import pubsub
from wokkel import rsm
from zope.interface import implements
# from twisted.internet import defer
from collections import namedtuple
import uuid
UNSPECIFIED = "unspecified error"
@ -49,6 +51,11 @@ PLUGIN_INFO = {
}
Extra = namedtuple('Extra', ('rsm_request', 'extra'))
# rsm_request is the rsm.RSMRequest build with rsm_ prefixed keys, or None
# extra is a potentially empty dict
class XEP_0060(object):
OPT_ACCESS_MODEL = 'pubsub#access_model'
OPT_PERSIST_ITEMS = 'pubsub#persist_items'
@ -60,29 +67,106 @@ class XEP_0060(object):
OPT_SUBSCRIPTION_DEPTH = 'pubsub#subscription_depth'
OPT_ROSTER_GROUPS_ALLOWED = 'pubsub#roster_groups_allowed'
OPT_PUBLISH_MODEL = 'pubsub#publish_model'
ACCESS_OPEN = 'open'
ACCESS_PRESENCE = 'presence'
ACCESS_ROSTER = 'roster'
ACCESS_AUTHORIZE = 'authorize'
ACCESS_WHITELIST = 'whitelist'
def __init__(self, host):
log.info(_(u"PubSub plugin initialization"))
self.host = host
self.managedNodes = []
self._node_cb = {} # dictionnary of callbacks for node (key: node, value: list of callbacks)
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)
host.bridge.addMethod("getFromMany", ".plugin", in_sign='a(ss)ia{ss}s', out_sign='s', method=self._getFromMany)
host.bridge.addMethod("getFromManyRTResult", ".plugin", in_sign='ss', out_sign='(ua(sssasa{ss}))', method=self._getFromManyRTResult, async=True)
host.bridge.addMethod("psDeleteNode", ".plugin", in_sign='sss', out_sign='', method=self._deleteNode, async=True)
host.bridge.addMethod("psRetractItem", ".plugin", in_sign='sssbs', out_sign='', method=self._retractItem, async=True)
host.bridge.addMethod("psRetractItems", ".plugin", in_sign='ssasbs', out_sign='', method=self._retractItems, async=True)
host.bridge.addMethod("psSubscribeToMany", ".plugin", in_sign='a(ss)sa{ss}s', out_sign='s', method=self._subscribeToMany)
host.bridge.addMethod("psGetSubscribeRTResult", ".plugin", in_sign='ss', out_sign='(ua(sss))', method=self._manySubscribeRTResult, async=True)
host.bridge.addMethod("psGetFromMany", ".plugin", in_sign='a(ss)ia{ss}s', out_sign='s', method=self._getFromMany)
host.bridge.addMethod("psGetFromManyRTResult", ".plugin", in_sign='ss', out_sign='(ua(sssasa{ss}))', method=self._getFromManyRTResult, async=True)
host.bridge.addSignal("psEvent", ".plugin", signature='ssssa{ss}s') # args: category, service(jid), node, type (C.PS_ITEMS, C.PS_DELETE), data, profile
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
@defer.inlineCallbacks
def profileConnected(self, profile):
client = self.host.getClient(profile)
pubsub_services = yield self.host.findServiceEntities("pubsub", "service", profile_key = profile)
if pubsub_services:
# we use one of the found services as our default pubsub service
client.pubsub_service = pubsub_services.pop()
else:
client.pubsub_service = None
@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 parseExtra(self, extra):
"""Parse extra dictionnary
used bridge's extra dictionnaries
@param extra(dict): extra data used to configure request
@return(Extra): filled Extra instance
"""
if extra is not None:
rsm_dict = { key[4:]: value for key, value in extra.iteritems() if key.startswith('rsm_') }
if rsm_dict:
try:
rsm_dict['max_'] = rsm_dict.pop('max')
except KeyError:
pass
rsm_request = rsm.RSMRequest(**rsm_dict)
else:
rsm_request = None
else:
rsm_request = None
extra = {}
return Extra(rsm_request, extra)
def addManagedNode(self, node, **kwargs):
"""Add a handler for a node
@param node(unicode): node to monitor, or None to monitor all
@param **kwargs: method(s) to call when the node is found
the methode must be named after PubSub constants in lower case
and suffixed with "_cb"
e.g.: "publish_cb" for C.PS_PUBLISH, "delete_cb" for C.PS_DELETE
"""
assert kwargs
callbacks = self._node_cb.setdefault(node, {})
for event, cb in kwargs.iteritems():
event_name = event[:-3]
assert event_name in C.PS_EVENTS
callbacks.setdefault(event_name,[]).append(cb)
def removeManagedNode(self, node, *args):
"""Add a handler for a node
@param node(unicode): node to monitor
@param *args: callback(s) to remove
"""
assert args
try:
registred_cb = self._node_cb[node]
except KeyError:
pass
else:
for callback in args:
for event, cb_list in registred_cb.iteritems():
try:
cb_list.remove(callback)
except ValueError:
pass
else:
log.debug(u"removed callback {cb} for event {event} on node {node}".format(
cb=callback, event=event, node=node))
if not cb_list:
del registred_cb[event]
if not registred_cb:
del self._node_cb[node]
return
log.error(u"Trying to remove inexistant callback {cb} for node {node}".format(cb=callback, node=node))
# def listNodes(self, service, nodeIdentifier='', profile=C.PROF_KEY_NONE):
# """Retrieve the name of the nodes that are accessible on the target service.
@ -117,7 +201,7 @@ class XEP_0060(object):
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_request=None, profile_key=C.PROF_KEY_NONE):
def getItems(self, service, node, max_items=None, item_ids=None, sub_id=None, rsm_request=None, extra=None, profile_key=C.PROF_KEY_NONE):
"""Retrieve pubsub items from a node.
@param service (JID): pubsub service.
@ -134,12 +218,27 @@ class XEP_0060(object):
"""
if rsm_request and item_ids:
raise ValueError("items_id can't be used with rsm")
if extra is None:
extra = {}
client = self.host.getClient(profile_key)
ext_data = {'id': unicode(uuid.uuid4()), 'rsm': rsm_request} if rsm_request else None
ext_data = {'id': unicode(uuid.uuid4()), 'rsm': rsm_request} if rsm_request is not None else None
d = client.pubsub_client.items(service, node, max_items, item_ids, sub_id, client.pubsub_client.parent.jid, ext_data)
try:
subscribe = C.bool(extra['subscribe'])
except KeyError:
subscribe = False
def doSubscribe(items):
self.subscribe(service, node, profile_key=profile_key)
return items
if subscribe:
d.addCallback(doSubscribe)
def addMetadata(items):
metadata = {}
if rsm_request:
if rsm_request is not None:
rsm_data = client.pubsub_client.getRSMResponse(ext_data['id'])
metadata.update({'rsm_{}'.format(key): value for key, value in rsm_data})
return (items, metadata)
@ -185,15 +284,25 @@ class XEP_0060(object):
client = self.host.getClient(profile_key)
return client.pubsub_client.createNode(service, nodeIdentifier, options)
def _deleteNode(self, service_s, nodeIdentifier, profile_key):
return self.deleteNode(jid.JID(service_s) if service_s else None, nodeIdentifier, profile_key)
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):
def _retractItem(self, service_s, nodeIdentifier, itemIdentifier, notify, profile_key):
return self._retractItems(service_s, nodeIdentifier, (itemIdentifier,), notify, profile_key)
def _retractItems(self, service_s, nodeIdentifier, itemIdentifiers, notify, profile_key):
return self.retractItems(jid.JID(service_s) if service_s else None, nodeIdentifier, itemIdentifiers, notify, profile_key)
def retractItems(self, service, nodeIdentifier, itemIdentifiers, notify=True, profile_key=C.PROF_KEY_NONE):
client = self.host.getClient(profile_key)
return client.pubsub_client.retractItems(service, nodeIdentifier, itemIdentifiers)
return client.pubsub_client.retractItems(service, nodeIdentifier, itemIdentifiers, notify=True)
def subscribe(self, service, nodeIdentifier, sub_jid=None, options=None, profile_key=C.PROF_KEY_NONE):
# TODO: reimplement a subscribtion cache, checking that we have not subscription before trying to subscribe
client = self.host.getClient(profile_key)
return client.pubsub_client.subscribe(service, nodeIdentifier, sub_jid or client.pubsub_client.parent.jid.userhostJID(), options=options)
@ -331,14 +440,15 @@ class XEP_0060(object):
for (service, node), (success, (failure, (items, metadata))) in ret[1].iteritems()]))
return d
def _getFromMany(self, node_data, max_item=10, rsm_dict=None, profile_key=C.PROF_KEY_NONE):
def _getFromMany(self, node_data, max_item=10, extra_dict=None, profile_key=C.PROF_KEY_NONE):
"""
@param max_item(int): maximum number of item to get, C.NO_LIMIT for no limit
"""
max_item = None if max_item == C.NO_LIMIT else max_item
return self.getFromMany([(jid.JID(service), unicode(node)) for service, node in node_data], max_item, rsm.RSMRequest(**rsm_dict) if rsm_dict else None, profile_key)
extra = self.parseExtra(extra_dict)
return self.getFromMany([(jid.JID(service), unicode(node)) for service, node in node_data], max_item, extra.rsm_request, extra.extra, profile_key)
def getFromMany(self, node_data, max_item=None, rsm_request=None, profile_key=C.PROF_KEY_NONE):
def getFromMany(self, node_data, max_item=None, rsm_request=None, extra=None, profile_key=C.PROF_KEY_NONE):
"""Get items from many nodes at once
@param node_data (iterable[tuple]): iterable of tuple (service, node) where:
- service (jid.JID) is the pubsub service
@ -351,7 +461,7 @@ class XEP_0060(object):
client = self.host.getClient(profile_key)
deferreds = {}
for service, node in node_data:
deferreds[(service, node)] = self.getItems(service, node, max_item, rsm_request=rsm_request, profile_key=profile_key)
deferreds[(service, node)] = self.getItems(service, node, max_item, rsm_request=rsm_request, extra=extra, profile_key=profile_key)
return self.rt_sessions.newSession(deferreds, client.profile)
@ -367,17 +477,26 @@ class SatPubSubClient(rsm.PubSubClient):
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)
log.debug(u"Pubsub items received")
for node in (event.nodeIdentifier, None):
try:
callbacks = self.parent_plugin._node_cb[node][C.PS_ITEMS]
except KeyError:
pass
else:
for callback in callbacks:
callback(event, self.parent.profile)
def deleteReceived(self, event):
#TODO: manage delete event
log.debug(_(u"Publish node deleted"))
# def purgeReceived(self, event):
log.debug((u"Publish node deleted"))
for node in (event.nodeIdentifier, None):
try:
callbacks = self.parent_plugin._node_cb[node][C.PS_DELETE]
except KeyError:
pass
else:
for callback in callbacks:
callback(event, self.parent.profile)
def subscriptions(self, service, nodeIdentifier, sender=None):
"""Return the list of subscriptions to the given service and node.