Mercurial > libervia-backend

comparison src/plugins/plugin_xep_0060.py @ 1777:8b18e5f55a90

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
plugin XEP-0060: MAM integration: - removed useless XEP-0059 recommendation - added XEP-0313 as a recommended plugin - parseExtra now manage mam_* keys (mam_filter_[name] to use a filter) - if a mam request is in 'mam' key of extra parameted in getItems, a MAM request is done, but the result is returned in the same way as for normal getItems, making like more easy for other plugins and for frontends
author Goffi <goffi@goffi.org>
date Tue, 05 Jan 2016 23:20:22 +0100
parents 6e867caf4621
children 442303b62a16
comparison
equal deleted inserted replaced
1776:4fc1bf1af48f 1777:8b18e5f55a90
26 from sat.tools import sat_defer 26 from sat.tools import sat_defer
27 27
28 from twisted.words.protocols.jabber import jid, error 28 from twisted.words.protocols.jabber import jid, error
29 from twisted.internet import defer 29 from twisted.internet import defer
30 from wokkel import disco 30 from wokkel import disco
31 # XXX: tmp.pubsub is actually use instead of wokkel version 31 from wokkel import data_form
32 # same thing for rsm 32 from zope.interface import implements
33 from collections import namedtuple
34 import datetime
35 from dateutil import tz
36 # XXX: tmp.wokkel.pubsub is actually use instead of wokkel version
37 # mam and rsm come from tmp.wokkel too
33 from wokkel import pubsub 38 from wokkel import pubsub
34 from wokkel import rsm 39 from wokkel import rsm
35 from zope.interface import implements 40 from wokkel import mam
36 from collections import namedtuple
37
38
39 UNSPECIFIED = "unspecified error"
40 41
41 42
42 PLUGIN_INFO = { 43 PLUGIN_INFO = {
43 "name": "Publish-Subscribe", 44 "name": "Publish-Subscribe",
44 "import_name": "XEP-0060", 45 "import_name": "XEP-0060",
45 "type": "XEP", 46 "type": "XEP",
46 "protocols": ["XEP-0060"], 47 "protocols": ["XEP-0060"],
47 "dependencies": [], 48 "dependencies": [],
48 "recommendations": ["XEP-0059"], 49 "recommendations": ["XEP-0313"],
49 "main": "XEP_0060", 50 "main": "XEP_0060",
50 "handler": "yes", 51 "handler": "yes",
51 "description": _("""Implementation of PubSub Protocol""") 52 "description": _("""Implementation of PubSub Protocol""")
52 } 53 }
54
55 UNSPECIFIED = "unspecified error"
56 MAM_FILTER = "mam_filter_"
53 57
54 58
55 Extra = namedtuple('Extra', ('rsm_request', 'extra')) 59 Extra = namedtuple('Extra', ('rsm_request', 'extra'))
56 # rsm_request is the rsm.RSMRequest build with rsm_ prefixed keys, or None 60 # rsm_request is the rsm.RSMRequest build with rsm_ prefixed keys, or None
57 # extra is a potentially empty dict 61 # extra is a potentially empty dict
75 ACCESS_WHITELIST = 'whitelist' 79 ACCESS_WHITELIST = 'whitelist'
76 80
77 def __init__(self, host): 81 def __init__(self, host):
78 log.info(_(u"PubSub plugin initialization")) 82 log.info(_(u"PubSub plugin initialization"))
79 self.host = host 83 self.host = host
84 self._mam = host.plugins.get('XEP-0313')
80 self._node_cb = {} # dictionnary of callbacks for node (key: node, value: list of callbacks) 85 self._node_cb = {} # dictionnary of callbacks for node (key: node, value: list of callbacks)
81 self.rt_sessions = sat_defer.RTDeferredSessions() 86 self.rt_sessions = sat_defer.RTDeferredSessions()
82 host.bridge.addMethod("psDeleteNode", ".plugin", in_sign='sss', out_sign='', method=self._deleteNode, async=True) 87 host.bridge.addMethod("psDeleteNode", ".plugin", in_sign='sss', out_sign='', method=self._deleteNode, async=True)
83 host.bridge.addMethod("psRetractItem", ".plugin", in_sign='sssbs', out_sign='', method=self._retractItem, async=True) 88 host.bridge.addMethod("psRetractItem", ".plugin", in_sign='sssbs', out_sign='', method=self._retractItem, async=True)
84 host.bridge.addMethod("psRetractItems", ".plugin", in_sign='ssasbs', out_sign='', method=self._retractItems, async=True) 89 host.bridge.addMethod("psRetractItems", ".plugin", in_sign='ssasbs', out_sign='', method=self._retractItems, async=True)
123 used bridge's extra dictionnaries 128 used bridge's extra dictionnaries
124 @param extra(dict): extra data used to configure request 129 @param extra(dict): extra data used to configure request
125 @return(Extra): filled Extra instance 130 @return(Extra): filled Extra instance
126 """ 131 """
127 if extra is not None: 132 if extra is not None:
128 rsm_dict = { key[4:]: value for key, value in extra.iteritems() if key.startswith('rsm_') } 133 # rsm
129 if rsm_dict: 134 rsm_args = {}
135 for arg in ('max', 'after', 'before', 'index'):
130 try: 136 try:
131 rsm_dict['max_'] = rsm_dict.pop('max') 137 argname = "max_" if arg == 'max' else arg
138 rsm_args[argname] = extra.pop('rsm_{}'.format(arg))
132 except KeyError: 139 except KeyError:
133 pass 140 continue
134 rsm_request = rsm.RSMRequest(**rsm_dict) 141
142 if rsm_args:
143 rsm_request = rsm.RSMRequest(**rsm_args)
135 else: 144 else:
136 rsm_request = None 145 rsm_request = None
146
147 # mam
148 mam_args = {}
149 for arg in ('start', 'end'):
150 try:
151 mam_args[arg] = datetime.datetime.fromtimestamp(int(extra.pop('{}{}'.format(MAM_FILTER, arg))), tz.tzutc())
152 except (TypeError, ValueError):
153 log.warning(u"Bad value for {} filter".format(arg))
154 except KeyError:
155 continue
156
157 try:
158 mam_args['with_jid'] = jid.JID(extra.pop('{}jid'.format(MAM_FILTER)))
159 except (jid.InvalidFormat):
160 log.warning(u"Bad value for jid filter")
161 except KeyError:
162 pass
163
164 for name, value in extra:
165 if name.startswith(MAM_FILTER):
166 var = name[len[MAM_FILTER]:]
167 extra_fields = mam_args.setdefault('extra_fields', [])
168 extra_fields.append(data_form.Field(var=var, value=value))
169
170 if mam_args:
171 assert 'mam' not in extra
172 extra['mam'] = mam.MAMRequest(mam.buildForm(**mam_args))
137 else: 173 else:
138 rsm_request = None 174 rsm_request = None
139 extra = {} 175 extra = {}
140 return Extra(rsm_request, extra) 176 return Extra(rsm_request, extra)
141 177
216 252
217 def publish(self, service, nodeIdentifier, items=None, profile_key=C.PROF_KEY_NONE): 253 def publish(self, service, nodeIdentifier, items=None, profile_key=C.PROF_KEY_NONE):
218 client = self.host.getClient(profile_key) 254 client = self.host.getClient(profile_key)
219 return client.pubsub_client.publish(service, nodeIdentifier, items, client.pubsub_client.parent.jid) 255 return client.pubsub_client.publish(service, nodeIdentifier, items, client.pubsub_client.parent.jid)
220 256
257 def _unwrapMAMMessage(self, message_elt):
258 try:
259 item_elt = (message_elt.elements(mam.NS_MAM, 'result').next()
260 .elements(C.NS_FORWARD, 'forwarded').next()
261 .elements(C.NS_CLIENT, 'message').next()
262 .elements('http://jabber.org/protocol/pubsub#event', 'event').next()
263 .elements('http://jabber.org/protocol/pubsub#event', 'items').next()
264 .elements('http://jabber.org/protocol/pubsub#event', 'item').next())
265 except StopIteration:
266 raise exceptions.DataError(u"Can't find Item in MAM message element")
267 return item_elt
268
221 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): 269 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):
222 """Retrieve pubsub items from a node. 270 """Retrieve pubsub items from a node.
223 271
224 @param service (JID): pubsub service. 272 @param service (JID): pubsub service.
225 @param node (str): node id. 273 @param node (str): node id.
236 if rsm_request and item_ids: 284 if rsm_request and item_ids:
237 raise ValueError(u"items_id can't be used with rsm") 285 raise ValueError(u"items_id can't be used with rsm")
238 if extra is None: 286 if extra is None:
239 extra = {} 287 extra = {}
240 client = self.host.getClient(profile_key) 288 client = self.host.getClient(profile_key)
241 d = client.pubsub_client.items(service, node, max_items, item_ids, sub_id, client.pubsub_client.parent.jid, rsm_request) 289 try:
290 mam_query = extra['mam']
291 except KeyError:
292 d = client.pubsub_client.items(service, node, max_items, item_ids, sub_id, None, rsm_request)
293 else:
294 # if mam is requested, we have to do a totally different query
295 if self._mam is None:
296 raise exceptions.NotFound(u"MAM (XEP-0313) plugin is not available")
297 if max_items is not None:
298 raise exceptions.DataError(u"max_items parameter can't be used with MAM")
299 if item_ids:
300 raise exceptions.DataError(u"items_ids parameter can't be used with MAM")
301 if mam_query.node is None:
302 mam_query.node = node
303 elif mam_query.node != node:
304 raise exceptions.DataError(u"MAM query node is incoherent with getItems's node")
305 if mam_query.rsm is None:
306 mam_query.rsm = rsm_request
307 else:
308 if mam_query.rsm != rsm_request:
309 raise exceptions.DataError(u"Conflict between RSM request and MAM's RSM request")
310 d = self._mam.getArchives(client, mam_query, service, self._unwrapMAMMessage)
242 311
243 try: 312 try:
244 subscribe = C.bool(extra['subscribe']) 313 subscribe = C.bool(extra['subscribe'])
245 except KeyError: 314 except KeyError:
246 subscribe = False 315 subscribe = False