Mercurial > libervia-backend
comparison sat/plugins/plugin_xep_0060.py @ 2562:26edcf3a30eb
core, setup: huge cleaning:
- moved directories from src and frontends/src to sat and sat_frontends, which is the recommanded naming convention
- move twisted directory to root
- removed all hacks from setup.py, and added missing dependencies, it is now clean
- use https URL for website in setup.py
- removed "Environment :: X11 Applications :: GTK", as wix is deprecated and removed
- renamed sat.sh to sat and fixed its installation
- added python_requires to specify Python version needed
- replaced glib2reactor which use deprecated code by gtk3reactor
sat can now be installed directly from virtualenv without using --system-site-packages anymore \o/
author | Goffi <goffi@goffi.org> |
---|---|
date | Mon, 02 Apr 2018 19:44:50 +0200 |
parents | src/plugins/plugin_xep_0060.py@785b6a1cef0a |
children | 56f94936df1e |
comparison
equal
deleted
inserted
replaced
2561:bd30dc3ffe5a | 2562:26edcf3a30eb |
---|---|
1 #!/usr/bin/env python2 | |
2 # -*- coding: utf-8 -*- | |
3 | |
4 # SAT plugin for Publish-Subscribe (xep-0060) | |
5 # Copyright (C) 2009-2018 Jérôme Poisson (goffi@goffi.org) | |
6 | |
7 # This program is free software: you can redistribute it and/or modify | |
8 # it under the terms of the GNU Affero General Public License as published by | |
9 # the Free Software Foundation, either version 3 of the License, or | |
10 # (at your option) any later version. | |
11 | |
12 # This program is distributed in the hope that it will be useful, | |
13 # but WITHOUT ANY WARRANTY; without even the implied warranty of | |
14 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the | |
15 # GNU Affero General Public License for more details. | |
16 | |
17 # You should have received a copy of the GNU Affero General Public License | |
18 # along with this program. If not, see <http://www.gnu.org/licenses/>. | |
19 | |
20 from sat.core.i18n import _ | |
21 from sat.core.constants import Const as C | |
22 from sat.core.log import getLogger | |
23 log = getLogger(__name__) | |
24 from sat.core import exceptions | |
25 from sat.tools import sat_defer | |
26 | |
27 from twisted.words.protocols.jabber import jid, error | |
28 from twisted.internet import defer | |
29 from wokkel import disco | |
30 from wokkel import data_form | |
31 from zope.interface import implements | |
32 from collections import namedtuple | |
33 import urllib | |
34 import datetime | |
35 from dateutil import tz | |
36 # XXX: sat_tmp.wokkel.pubsub is actually use instead of wokkel version | |
37 # mam and rsm come from sat_tmp.wokkel too | |
38 from wokkel import pubsub | |
39 from wokkel import rsm | |
40 from wokkel import mam | |
41 | |
42 | |
43 PLUGIN_INFO = { | |
44 C.PI_NAME: "Publish-Subscribe", | |
45 C.PI_IMPORT_NAME: "XEP-0060", | |
46 C.PI_TYPE: "XEP", | |
47 C.PI_PROTOCOLS: ["XEP-0060"], | |
48 C.PI_DEPENDENCIES: [], | |
49 C.PI_RECOMMENDATIONS: ["XEP-0313"], | |
50 C.PI_MAIN: "XEP_0060", | |
51 C.PI_HANDLER: "yes", | |
52 C.PI_DESCRIPTION: _("""Implementation of PubSub Protocol""") | |
53 } | |
54 | |
55 UNSPECIFIED = "unspecified error" | |
56 MAM_FILTER = "mam_filter_" | |
57 | |
58 | |
59 Extra = namedtuple('Extra', ('rsm_request', 'extra')) | |
60 # rsm_request is the rsm.RSMRequest build with rsm_ prefixed keys, or None | |
61 # extra is a potentially empty dict | |
62 | |
63 | |
64 class XEP_0060(object): | |
65 OPT_ACCESS_MODEL = 'pubsub#access_model' | |
66 OPT_PERSIST_ITEMS = 'pubsub#persist_items' | |
67 OPT_MAX_ITEMS = 'pubsub#max_items' | |
68 OPT_DELIVER_PAYLOADS = 'pubsub#deliver_payloads' | |
69 OPT_SEND_ITEM_SUBSCRIBE = 'pubsub#send_item_subscribe' | |
70 OPT_NODE_TYPE = 'pubsub#node_type' | |
71 OPT_SUBSCRIPTION_TYPE = 'pubsub#subscription_type' | |
72 OPT_SUBSCRIPTION_DEPTH = 'pubsub#subscription_depth' | |
73 OPT_ROSTER_GROUPS_ALLOWED = 'pubsub#roster_groups_allowed' | |
74 OPT_PUBLISH_MODEL = 'pubsub#publish_model' | |
75 ACCESS_OPEN = 'open' | |
76 ACCESS_PRESENCE = 'presence' | |
77 ACCESS_ROSTER = 'roster' | |
78 ACCESS_PUBLISHER_ROSTER = 'publisher-roster' | |
79 ACCESS_AUTHORIZE = 'authorize' | |
80 ACCESS_WHITELIST = 'whitelist' | |
81 | |
82 def __init__(self, host): | |
83 log.info(_(u"PubSub plugin initialization")) | |
84 self.host = host | |
85 self._mam = host.plugins.get('XEP-0313') | |
86 self._node_cb = {} # dictionnary of callbacks for node (key: node, value: list of callbacks) | |
87 self.rt_sessions = sat_defer.RTDeferredSessions() | |
88 host.bridge.addMethod("psNodeCreate", ".plugin", in_sign='ssa{ss}s', out_sign='s', method=self._createNode, async=True) | |
89 host.bridge.addMethod("psNodeConfigurationGet", ".plugin", in_sign='sss', out_sign='a{ss}', method=self._getNodeConfiguration, async=True) | |
90 host.bridge.addMethod("psNodeConfigurationSet", ".plugin", in_sign='ssa{ss}s', out_sign='', method=self._setNodeConfiguration, async=True) | |
91 host.bridge.addMethod("psNodeAffiliationsGet", ".plugin", in_sign='sss', out_sign='a{ss}', method=self._getNodeAffiliations, async=True) | |
92 host.bridge.addMethod("psNodeAffiliationsSet", ".plugin", in_sign='ssa{ss}s', out_sign='', method=self._setNodeAffiliations, async=True) | |
93 host.bridge.addMethod("psNodeSubscriptionsGet", ".plugin", in_sign='sss', out_sign='a{ss}', method=self._getNodeSubscriptions, async=True) | |
94 host.bridge.addMethod("psNodeSubscriptionsSet", ".plugin", in_sign='ssa{ss}s', out_sign='', method=self._setNodeSubscriptions, async=True) | |
95 host.bridge.addMethod("psNodeDelete", ".plugin", in_sign='sss', out_sign='', method=self._deleteNode, async=True) | |
96 host.bridge.addMethod("psNodeWatchAdd", ".plugin", in_sign='sss', out_sign='', method=self._addWatch, async=False) | |
97 host.bridge.addMethod("psNodeWatchRemove", ".plugin", in_sign='sss', out_sign='', method=self._removeWatch, async=False) | |
98 host.bridge.addMethod("psAffiliationsGet", ".plugin", in_sign='sss', out_sign='a{ss}', method=self._getAffiliations, async=True) | |
99 host.bridge.addMethod("psItemsGet", ".plugin", in_sign='ssiassa{ss}s', out_sign='(asa{ss})', method=self._getItems, async=True) | |
100 host.bridge.addMethod("psItemSend", ".plugin", in_sign='ssssa{ss}s', out_sign='s', method=self._sendItem, async=True) | |
101 host.bridge.addMethod("psRetractItem", ".plugin", in_sign='sssbs', out_sign='', method=self._retractItem, async=True) | |
102 host.bridge.addMethod("psRetractItems", ".plugin", in_sign='ssasbs', out_sign='', method=self._retractItems, async=True) | |
103 host.bridge.addMethod("psSubscribe", ".plugin", in_sign='ssa{ss}s', out_sign='s', method=self._subscribe, async=True) | |
104 host.bridge.addMethod("psUnsubscribe", ".plugin", in_sign='sss', out_sign='', method=self._unsubscribe, async=True) | |
105 host.bridge.addMethod("psSubscriptionsGet", ".plugin", in_sign='sss', out_sign='aa{ss}', method=self._subscriptions, async=True) | |
106 host.bridge.addMethod("psSubscribeToMany", ".plugin", in_sign='a(ss)sa{ss}s', out_sign='s', method=self._subscribeToMany) | |
107 host.bridge.addMethod("psGetSubscribeRTResult", ".plugin", in_sign='ss', out_sign='(ua(sss))', method=self._manySubscribeRTResult, async=True) | |
108 host.bridge.addMethod("psGetFromMany", ".plugin", in_sign='a(ss)ia{ss}s', out_sign='s', method=self._getFromMany) | |
109 host.bridge.addMethod("psGetFromManyRTResult", ".plugin", in_sign='ss', out_sign='(ua(sssasa{ss}))', method=self._getFromManyRTResult, async=True) | |
110 | |
111 # high level observer method | |
112 host.bridge.addSignal("psEvent", ".plugin", signature='ssssa{ss}s') # args: category, service(jid), node, type (C.PS_ITEMS, C.PS_DELETE), data, profile | |
113 | |
114 # low level observer method, used if service/node is in watching list (see psNodeWatch* methods) | |
115 host.bridge.addSignal("psEventRaw", ".plugin", signature='sssass') # args: service(jid), node, type (C.PS_ITEMS, C.PS_DELETE), list of item_xml, profile | |
116 | |
117 def getHandler(self, client): | |
118 client.pubsub_client = SatPubSubClient(self.host, self) | |
119 return client.pubsub_client | |
120 | |
121 @defer.inlineCallbacks | |
122 def profileConnected(self, client): | |
123 client.pubsub_watching = set() | |
124 try: | |
125 client.pubsub_service = jid.JID(self.host.memory.getConfig('', 'pubsub_service')) | |
126 except RuntimeError: | |
127 log.info(_(u"Can't retrieve pubsub_service from conf, we'll use first one that we find")) | |
128 client.pubsub_service = yield self.host.findServiceEntity(client, "pubsub", "service") | |
129 | |
130 def getFeatures(self, profile): | |
131 try: | |
132 client = self.host.getClient(profile) | |
133 except exceptions.ProfileNotSetError: | |
134 return {} | |
135 try: | |
136 return {'service': client.pubsub_service.full() if client.pubsub_service is not None else ''} | |
137 except AttributeError: | |
138 if self.host.isConnected(profile): | |
139 log.debug("Profile is not connected, service is not checked yet") | |
140 else: | |
141 log.error("Service should be available !") | |
142 return {} | |
143 | |
144 def parseExtra(self, extra): | |
145 """Parse extra dictionnary | |
146 | |
147 used bridge's extra dictionnaries | |
148 @param extra(dict): extra data used to configure request | |
149 @return(Extra): filled Extra instance | |
150 """ | |
151 if extra is None: | |
152 rsm_request = None | |
153 extra = {} | |
154 else: | |
155 # rsm | |
156 rsm_args = {} | |
157 for arg in ('max', 'after', 'before', 'index'): | |
158 try: | |
159 argname = "max_" if arg == 'max' else arg | |
160 rsm_args[argname] = extra.pop('rsm_{}'.format(arg)) | |
161 except KeyError: | |
162 continue | |
163 | |
164 if rsm_args: | |
165 rsm_request = rsm.RSMRequest(**rsm_args) | |
166 else: | |
167 rsm_request = None | |
168 | |
169 # mam | |
170 mam_args = {} | |
171 for arg in ('start', 'end'): | |
172 try: | |
173 mam_args[arg] = datetime.datetime.fromtimestamp(int(extra.pop('{}{}'.format(MAM_FILTER, arg))), tz.tzutc()) | |
174 except (TypeError, ValueError): | |
175 log.warning(u"Bad value for {} filter".format(arg)) | |
176 except KeyError: | |
177 continue | |
178 | |
179 try: | |
180 mam_args['with_jid'] = jid.JID(extra.pop('{}jid'.format(MAM_FILTER))) | |
181 except (jid.InvalidFormat): | |
182 log.warning(u"Bad value for jid filter") | |
183 except KeyError: | |
184 pass | |
185 | |
186 for name, value in extra.iteritems(): | |
187 if name.startswith(MAM_FILTER): | |
188 var = name[len(MAM_FILTER):] | |
189 extra_fields = mam_args.setdefault('extra_fields', []) | |
190 extra_fields.append(data_form.Field(var=var, value=value)) | |
191 | |
192 if mam_args: | |
193 assert 'mam' not in extra | |
194 extra['mam'] = mam.MAMRequest(mam.buildForm(**mam_args)) | |
195 return Extra(rsm_request, extra) | |
196 | |
197 def addManagedNode(self, node, **kwargs): | |
198 """Add a handler for a node | |
199 | |
200 @param node(unicode): node to monitor | |
201 all node *prefixed* with this one will be triggered | |
202 @param **kwargs: method(s) to call when the node is found | |
203 the method must be named after PubSub constants in lower case | |
204 and suffixed with "_cb" | |
205 e.g.: "items_cb" for C.PS_ITEMS, "delete_cb" for C.PS_DELETE | |
206 """ | |
207 assert node is not None | |
208 assert kwargs | |
209 callbacks = self._node_cb.setdefault(node, {}) | |
210 for event, cb in kwargs.iteritems(): | |
211 event_name = event[:-3] | |
212 assert event_name in C.PS_EVENTS | |
213 callbacks.setdefault(event_name,[]).append(cb) | |
214 | |
215 def removeManagedNode(self, node, *args): | |
216 """Add a handler for a node | |
217 | |
218 @param node(unicode): node to monitor | |
219 @param *args: callback(s) to remove | |
220 """ | |
221 assert args | |
222 try: | |
223 registred_cb = self._node_cb[node] | |
224 except KeyError: | |
225 pass | |
226 else: | |
227 for callback in args: | |
228 for event, cb_list in registred_cb.iteritems(): | |
229 try: | |
230 cb_list.remove(callback) | |
231 except ValueError: | |
232 pass | |
233 else: | |
234 log.debug(u"removed callback {cb} for event {event} on node {node}".format( | |
235 cb=callback, event=event, node=node)) | |
236 if not cb_list: | |
237 del registred_cb[event] | |
238 if not registred_cb: | |
239 del self._node_cb[node] | |
240 return | |
241 log.error(u"Trying to remove inexistant callback {cb} for node {node}".format(cb=callback, node=node)) | |
242 | |
243 # def listNodes(self, service, nodeIdentifier='', profile=C.PROF_KEY_NONE): | |
244 # """Retrieve the name of the nodes that are accessible on the target service. | |
245 | |
246 # @param service (JID): target service | |
247 # @param nodeIdentifier (str): the parent node name (leave empty to retrieve first-level nodes) | |
248 # @param profile (str): %(doc_profile)s | |
249 # @return: deferred which fire a list of nodes | |
250 # """ | |
251 # client = self.host.getClient(profile) | |
252 # d = self.host.getDiscoItems(client, service, nodeIdentifier) | |
253 # d.addCallback(lambda result: [item.getAttribute('node') for item in result.toElement().children if item.hasAttribute('node')]) | |
254 # return d | |
255 | |
256 # def listSubscribedNodes(self, service, nodeIdentifier='', filter_='subscribed', profile=C.PROF_KEY_NONE): | |
257 # """Retrieve the name of the nodes to which the profile is subscribed on the target service. | |
258 | |
259 # @param service (JID): target service | |
260 # @param nodeIdentifier (str): the parent node name (leave empty to retrieve all subscriptions) | |
261 # @param filter_ (str): filter the result according to the given subscription type: | |
262 # - None: do not filter | |
263 # - 'pending': subscription has not been approved yet by the node owner | |
264 # - 'unconfigured': subscription options have not been configured yet | |
265 # - 'subscribed': subscription is complete | |
266 # @param profile (str): %(doc_profile)s | |
267 # @return: Deferred list[str] | |
268 # """ | |
269 # d = self.subscriptions(service, nodeIdentifier, profile_key=profile) | |
270 # d.addCallback(lambda subs: [sub.getAttribute('node') for sub in subs if sub.getAttribute('subscription') == filter_]) | |
271 # return d | |
272 | |
273 def _sendItem(self, service, nodeIdentifier, payload, item_id=None, extra=None, profile_key=C.PROF_KEY_NONE): | |
274 client = self.host.getClient(profile_key) | |
275 service = None if not service else jid.JID(service) | |
276 d = self.sendItem(client, service, nodeIdentifier, payload, item_id or None, extra) | |
277 d.addCallback(lambda ret: ret or u'') | |
278 return d | |
279 | |
280 def _getPublishedItemId(self, iq_elt, original_id): | |
281 """return item of published id if found in answer | |
282 | |
283 if not found original_id is returned, or empty string if it is None or empty string | |
284 """ | |
285 try: | |
286 item_id = iq_elt.pubsub.publish.item['id'] | |
287 except (AttributeError, KeyError): | |
288 item_id = None | |
289 return item_id or original_id | |
290 | |
291 def sendItem(self, client, service, nodeIdentifier, payload, item_id=None, extra=None): | |
292 """high level method to send one item | |
293 | |
294 @param service(jid.JID, None): service to send the item to | |
295 None to use PEP | |
296 @param NodeIdentifier(unicode): PubSub node to use | |
297 @param item_id(unicode, None): id to use or None to create one | |
298 @param payload(domish.Element, unicode): payload of the item to send | |
299 @param extra(dict, None): extra option, not used yet | |
300 @return (unicode, None): id of the created item | |
301 """ | |
302 item_elt = pubsub.Item(id=item_id, payload=payload) | |
303 d = self.publish(client, service, nodeIdentifier, [item_elt]) | |
304 d.addCallback(self._getPublishedItemId, item_id) | |
305 return d | |
306 | |
307 def publish(self, client, service, nodeIdentifier, items=None): | |
308 return client.pubsub_client.publish(service, nodeIdentifier, items, client.pubsub_client.parent.jid) | |
309 | |
310 def _unwrapMAMMessage(self, message_elt): | |
311 try: | |
312 item_elt = (message_elt.elements(mam.NS_MAM, 'result').next() | |
313 .elements(C.NS_FORWARD, 'forwarded').next() | |
314 .elements(C.NS_CLIENT, 'message').next() | |
315 .elements('http://jabber.org/protocol/pubsub#event', 'event').next() | |
316 .elements('http://jabber.org/protocol/pubsub#event', 'items').next() | |
317 .elements('http://jabber.org/protocol/pubsub#event', 'item').next()) | |
318 except StopIteration: | |
319 raise exceptions.DataError(u"Can't find Item in MAM message element") | |
320 return item_elt | |
321 | |
322 def _getItems(self, service='', node='', max_items=10, item_ids=None, sub_id=None, extra_dict=None, profile_key=C.PROF_KEY_NONE): | |
323 """Get items from pubsub node | |
324 | |
325 @param max_items(int): maximum number of item to get, C.NO_LIMIT for no limit | |
326 """ | |
327 client = self.host.getClient(profile_key) | |
328 service = jid.JID(service) if service else None | |
329 max_items = None if max_items == C.NO_LIMIT else max_items | |
330 extra = self.parseExtra(extra_dict) | |
331 d = self.getItems(client, service, node or None, max_items or None, item_ids, sub_id or None, extra.rsm_request, extra.extra) | |
332 d.addCallback(self.serItemsData) | |
333 return d | |
334 | |
335 def getItems(self, client, service, node, max_items=None, item_ids=None, sub_id=None, rsm_request=None, extra=None): | |
336 """Retrieve pubsub items from a node. | |
337 | |
338 @param service (JID, None): pubsub service. | |
339 @param node (str): node id. | |
340 @param max_items (int): optional limit on the number of retrieved items. | |
341 @param item_ids (list[str]): identifiers of the items to be retrieved (can't be used with rsm_request). | |
342 @param sub_id (str): optional subscription identifier. | |
343 @param rsm_request (rsm.RSMRequest): RSM request data | |
344 @return: a deferred couple (list[dict], dict) containing: | |
345 - list of items | |
346 - metadata with the following keys: | |
347 - rsm_first, rsm_last, rsm_count, rsm_index: first, last, count and index value of RSMResponse | |
348 - service, node: service and node used | |
349 """ | |
350 if item_ids and max_items is not None: | |
351 max_items = None | |
352 if rsm_request and item_ids: | |
353 raise ValueError(u"items_id can't be used with rsm") | |
354 if extra is None: | |
355 extra = {} | |
356 try: | |
357 mam_query = extra['mam'] | |
358 except KeyError: | |
359 d = client.pubsub_client.items(service, node, max_items, item_ids, sub_id, None, rsm_request) | |
360 else: | |
361 # if mam is requested, we have to do a totally different query | |
362 if self._mam is None: | |
363 raise exceptions.NotFound(u"MAM (XEP-0313) plugin is not available") | |
364 if max_items is not None: | |
365 raise exceptions.DataError(u"max_items parameter can't be used with MAM") | |
366 if item_ids: | |
367 raise exceptions.DataError(u"items_ids parameter can't be used with MAM") | |
368 if mam_query.node is None: | |
369 mam_query.node = node | |
370 elif mam_query.node != node: | |
371 raise exceptions.DataError(u"MAM query node is incoherent with getItems's node") | |
372 if mam_query.rsm is None: | |
373 mam_query.rsm = rsm_request | |
374 else: | |
375 if mam_query.rsm != rsm_request: | |
376 raise exceptions.DataError(u"Conflict between RSM request and MAM's RSM request") | |
377 d = self._mam.getArchives(client, mam_query, service, self._unwrapMAMMessage) | |
378 | |
379 try: | |
380 subscribe = C.bool(extra['subscribe']) | |
381 except KeyError: | |
382 subscribe = False | |
383 | |
384 def subscribeEb(failure, service, node): | |
385 failure.trap(error.StanzaError) | |
386 log.warning(u"Could not subscribe to node {} on service {}: {}".format(node, unicode(service), unicode(failure.value))) | |
387 | |
388 def doSubscribe(items): | |
389 self.subscribe(service, node, profile_key=client.profile).addErrback(subscribeEb, service, node) | |
390 return items | |
391 | |
392 if subscribe: | |
393 d.addCallback(doSubscribe) | |
394 | |
395 def addMetadata(result): | |
396 items, rsm_response = result | |
397 service_jid = service if service else client.jid.userhostJID() | |
398 metadata = {'service': service_jid, | |
399 'node': node, | |
400 'uri': self.getNodeURI(service_jid, node), | |
401 } | |
402 if rsm_request is not None and rsm_response is not None: | |
403 metadata.update({'rsm_{}'.format(key): value for key, value in rsm_response.toDict().iteritems()}) | |
404 return (items, metadata) | |
405 | |
406 d.addCallback(addMetadata) | |
407 return d | |
408 | |
409 # @defer.inlineCallbacks | |
410 # def getItemsFromMany(self, service, data, max_items=None, sub_id=None, rsm=None, profile_key=C.PROF_KEY_NONE): | |
411 # """Massively retrieve pubsub items from many nodes. | |
412 # @param service (JID): target service. | |
413 # @param data (dict): dictionnary binding some arbitrary keys to the node identifiers. | |
414 # @param max_items (int): optional limit on the number of retrieved items *per node*. | |
415 # @param sub_id (str): optional subscription identifier. | |
416 # @param rsm (dict): RSM request data | |
417 # @param profile_key (str): %(doc_profile_key)s | |
418 # @return: a deferred dict with: | |
419 # - key: a value in (a subset of) data.keys() | |
420 # - couple (list[dict], dict) containing: | |
421 # - list of items | |
422 # - RSM response data | |
423 # """ | |
424 # client = self.host.getClient(profile_key) | |
425 # found_nodes = yield self.listNodes(service, profile=client.profile) | |
426 # d_dict = {} | |
427 # for publisher, node in data.items(): | |
428 # if node not in found_nodes: | |
429 # log.debug(u"Skip the items retrieval for [{node}]: node doesn't exist".format(node=node)) | |
430 # continue # avoid pubsub "item-not-found" error | |
431 # d_dict[publisher] = self.getItems(service, node, max_items, None, sub_id, rsm, client.profile) | |
432 # defer.returnValue(d_dict) | |
433 | |
434 def getOptions(self, service, nodeIdentifier, subscriber, subscriptionIdentifier=None, profile_key=C.PROF_KEY_NONE): | |
435 client = self.host.getClient(profile_key) | |
436 return client.pubsub_client.getOptions(service, nodeIdentifier, subscriber, subscriptionIdentifier) | |
437 | |
438 def setOptions(self, service, nodeIdentifier, subscriber, options, subscriptionIdentifier=None, profile_key=C.PROF_KEY_NONE): | |
439 client = self.host.getClient(profile_key) | |
440 return client.pubsub_client.setOptions(service, nodeIdentifier, subscriber, options, subscriptionIdentifier) | |
441 | |
442 def _createNode(self, service_s, nodeIdentifier, options, profile_key): | |
443 client = self.host.getClient(profile_key) | |
444 return self.createNode(client, jid.JID(service_s) if service_s else None, nodeIdentifier, options) | |
445 | |
446 def createNode(self, client, service, nodeIdentifier=None, options=None): | |
447 """Create a new node | |
448 | |
449 @param service(jid.JID): PubSub service, | |
450 @param NodeIdentifier(unicode, None): node name | |
451 use None to create instant node (identifier will be returned by this method) | |
452 @param option(dict[unicode, unicode], None): node configuration options | |
453 @return (unicode): identifier of the created node (may be different from requested name) | |
454 """ | |
455 # TODO: if pubsub service doesn't hande publish-options, configure it in a second time | |
456 return client.pubsub_client.createNode(service, nodeIdentifier, options) | |
457 | |
458 @defer.inlineCallbacks | |
459 def createIfNewNode(self, client, service, nodeIdentifier, options=None): | |
460 """Helper method similar to createNode, but will not fail in case of conflict""" | |
461 try: | |
462 yield self.createNode(client, service, nodeIdentifier, options) | |
463 except error.StanzaError as e: | |
464 if e.condition == 'conflict': | |
465 pass | |
466 else: | |
467 raise e | |
468 | |
469 def _getNodeConfiguration(self, service_s, nodeIdentifier, profile_key): | |
470 client = self.host.getClient(profile_key) | |
471 d = self.getConfiguration(client, jid.JID(service_s) if service_s else None, nodeIdentifier) | |
472 def serialize(form): | |
473 # FIXME: better more generic dataform serialisation should be available in SàT | |
474 return {f.var: unicode(f.value) for f in form.fields.values()} | |
475 d.addCallback(serialize) | |
476 return d | |
477 | |
478 def getConfiguration(self, client, service, nodeIdentifier): | |
479 request = pubsub.PubSubRequest('configureGet') | |
480 request.recipient = service | |
481 request.nodeIdentifier = nodeIdentifier | |
482 | |
483 def cb(iq): | |
484 form = data_form.findForm(iq.pubsub.configure, | |
485 pubsub.NS_PUBSUB_NODE_CONFIG) | |
486 form.typeCheck() | |
487 return form | |
488 | |
489 d = request.send(client.xmlstream) | |
490 d.addCallback(cb) | |
491 return d | |
492 | |
493 def _setNodeConfiguration(self, service_s, nodeIdentifier, options, profile_key): | |
494 client = self.host.getClient(profile_key) | |
495 d = self.setConfiguration(client, jid.JID(service_s) if service_s else None, nodeIdentifier, options) | |
496 return d | |
497 | |
498 def setConfiguration(self, client, service, nodeIdentifier, options): | |
499 request = pubsub.PubSubRequest('configureSet') | |
500 request.recipient = service | |
501 request.nodeIdentifier = nodeIdentifier | |
502 | |
503 form = data_form.Form(formType='submit', | |
504 formNamespace=pubsub.NS_PUBSUB_NODE_CONFIG) | |
505 form.makeFields(options) | |
506 request.options = form | |
507 | |
508 d = request.send(client.xmlstream) | |
509 return d | |
510 | |
511 def _getAffiliations(self, service_s, nodeIdentifier, profile_key): | |
512 client = self.host.getClient(profile_key) | |
513 d = self.getAffiliations(client, jid.JID(service_s) if service_s else None, nodeIdentifier or None) | |
514 return d | |
515 | |
516 def getAffiliations(self, client, service, nodeIdentifier=None): | |
517 """Retrieve affiliations of an entity | |
518 | |
519 @param nodeIdentifier(unicode, None): node to get affiliation from | |
520 None to get all nodes affiliations for this service | |
521 """ | |
522 request = pubsub.PubSubRequest('affiliations') | |
523 request.recipient = service | |
524 request.nodeIdentifier = nodeIdentifier | |
525 | |
526 def cb(iq_elt): | |
527 try: | |
528 affiliations_elt = next(iq_elt.pubsub.elements((pubsub.NS_PUBSUB, 'affiliations'))) | |
529 except StopIteration: | |
530 raise ValueError(_(u"Invalid result: missing <affiliations> element: {}").format(iq_elt.toXml)) | |
531 try: | |
532 return {e['node']: e['affiliation'] for e in affiliations_elt.elements((pubsub.NS_PUBSUB, 'affiliation'))} | |
533 except KeyError: | |
534 raise ValueError(_(u"Invalid result: bad <affiliation> element: {}").format(iq_elt.toXml)) | |
535 | |
536 d = request.send(client.xmlstream) | |
537 d.addCallback(cb) | |
538 return d | |
539 | |
540 def _getNodeAffiliations(self, service_s, nodeIdentifier, profile_key): | |
541 client = self.host.getClient(profile_key) | |
542 d = self.getNodeAffiliations(client, jid.JID(service_s) if service_s else None, nodeIdentifier) | |
543 d.addCallback(lambda affiliations: {j.full(): a for j, a in affiliations.iteritems()}) | |
544 return d | |
545 | |
546 def getNodeAffiliations(self, client, service, nodeIdentifier): | |
547 """Retrieve affiliations of a node owned by profile""" | |
548 request = pubsub.PubSubRequest('affiliationsGet') | |
549 request.recipient = service | |
550 request.nodeIdentifier = nodeIdentifier | |
551 | |
552 def cb(iq_elt): | |
553 try: | |
554 affiliations_elt = next(iq_elt.pubsub.elements((pubsub.NS_PUBSUB_OWNER, 'affiliations'))) | |
555 except StopIteration: | |
556 raise ValueError(_(u"Invalid result: missing <affiliations> element: {}").format(iq_elt.toXml)) | |
557 try: | |
558 return {jid.JID(e['jid']): e['affiliation'] for e in affiliations_elt.elements((pubsub.NS_PUBSUB_OWNER, 'affiliation'))} | |
559 except KeyError: | |
560 raise ValueError(_(u"Invalid result: bad <affiliation> element: {}").format(iq_elt.toXml)) | |
561 | |
562 d = request.send(client.xmlstream) | |
563 d.addCallback(cb) | |
564 return d | |
565 | |
566 def _setNodeAffiliations(self, service_s, nodeIdentifier, affiliations, profile_key=C.PROF_KEY_NONE): | |
567 client = self.host.getClient(profile_key) | |
568 affiliations = {jid.JID(jid_): affiliation for jid_, affiliation in affiliations.iteritems()} | |
569 d = self.setNodeAffiliations(client, jid.JID(service_s) if service_s else None, nodeIdentifier, affiliations) | |
570 return d | |
571 | |
572 def setNodeAffiliations(self, client, service, nodeIdentifier, affiliations): | |
573 """Update affiliations of a node owned by profile | |
574 | |
575 @param affiliations(dict[jid.JID, unicode]): affiliations to set | |
576 check https://xmpp.org/extensions/xep-0060.html#affiliations for a list of possible affiliations | |
577 """ | |
578 request = pubsub.PubSubRequest('affiliationsSet') | |
579 request.recipient = service | |
580 request.nodeIdentifier = nodeIdentifier | |
581 request.affiliations = affiliations | |
582 d = request.send(client.xmlstream) | |
583 return d | |
584 | |
585 def _deleteNode(self, service_s, nodeIdentifier, profile_key): | |
586 client = self.host.getClient(profile_key) | |
587 return self.deleteNode(client, jid.JID(service_s) if service_s else None, nodeIdentifier) | |
588 | |
589 def deleteNode(self, client, service, nodeIdentifier): | |
590 return client.pubsub_client.deleteNode(service, nodeIdentifier) | |
591 | |
592 def _addWatch(self, service_s, node, profile_key): | |
593 """watch modifications on a node | |
594 | |
595 This method should only be called from bridge | |
596 """ | |
597 client = self.host.getClient(profile_key) | |
598 service = jid.JID(service_s) if service_s else client.jid.userhostJID() | |
599 client.pubsub_watching.add((service, node)) | |
600 | |
601 def _removeWatch(self, service_s, node, profile_key): | |
602 """remove a node watch | |
603 | |
604 This method should only be called from bridge | |
605 """ | |
606 client = self.host.getClient(profile_key) | |
607 service = jid.JID(service_s) if service_s else client.jid.userhostJID() | |
608 client.pubsub_watching.remove((service, node)) | |
609 | |
610 def _retractItem(self, service_s, nodeIdentifier, itemIdentifier, notify, profile_key): | |
611 return self._retractItems(service_s, nodeIdentifier, (itemIdentifier,), notify, profile_key) | |
612 | |
613 def _retractItems(self, service_s, nodeIdentifier, itemIdentifiers, notify, profile_key): | |
614 return self.retractItems(jid.JID(service_s) if service_s else None, nodeIdentifier, itemIdentifiers, notify, profile_key) | |
615 | |
616 def retractItems(self, service, nodeIdentifier, itemIdentifiers, notify=True, profile_key=C.PROF_KEY_NONE): | |
617 client = self.host.getClient(profile_key) | |
618 return client.pubsub_client.retractItems(service, nodeIdentifier, itemIdentifiers, notify=True) | |
619 | |
620 def _subscribe(self, service, nodeIdentifier, options, profile_key=C.PROF_KEY_NONE): | |
621 client = self.host.getClient(profile_key) | |
622 service = None if not service else jid.JID(service) | |
623 d = self.subscribe(client, service, nodeIdentifier, options=options or None) | |
624 d.addCallback(lambda subscription: subscription.subscriptionIdentifier or u'') | |
625 return d | |
626 | |
627 def subscribe(self, client, service, nodeIdentifier, sub_jid=None, options=None): | |
628 # TODO: reimplement a subscribtion cache, checking that we have not subscription before trying to subscribe | |
629 return client.pubsub_client.subscribe(service, nodeIdentifier, sub_jid or client.jid.userhostJID(), options=options) | |
630 | |
631 def _unsubscribe(self, service, nodeIdentifier, profile_key=C.PROF_KEY_NONE): | |
632 client = self.host.getClient(profile_key) | |
633 service = None if not service else jid.JID(service) | |
634 return self.unsubscribe(client, service, nodeIdentifier) | |
635 | |
636 def unsubscribe(self, client, service, nodeIdentifier, sub_jid=None, subscriptionIdentifier=None, sender=None): | |
637 return client.pubsub_client.unsubscribe(service, nodeIdentifier, sub_jid or client.jid.userhostJID(), subscriptionIdentifier, sender) | |
638 | |
639 def _subscriptions(self, service, nodeIdentifier='', profile_key=C.PROF_KEY_NONE): | |
640 client = self.host.getClient(profile_key) | |
641 service = None if not service else jid.JID(service) | |
642 | |
643 def gotSubscriptions(subscriptions): | |
644 # we replace pubsub.Subscription instance by dict that we can serialize | |
645 for idx, sub in enumerate(subscriptions): | |
646 sub_dict = {'node': sub.nodeIdentifier, | |
647 'subscriber': sub.subscriber.full(), | |
648 'state': sub.state | |
649 } | |
650 if sub.subscriptionIdentifier is not None: | |
651 sub_dict['id'] = sub.subscriptionIdentifier | |
652 subscriptions[idx] = sub_dict | |
653 | |
654 return subscriptions | |
655 | |
656 d = self.subscriptions(client, service, nodeIdentifier or None) | |
657 d.addCallback(gotSubscriptions) | |
658 return d | |
659 | |
660 def subscriptions(self, client, service, nodeIdentifier=None): | |
661 """retrieve subscriptions from a service | |
662 | |
663 @param service(jid.JID): PubSub service | |
664 @param nodeIdentifier(unicode, None): node to check | |
665 None to get all subscriptions | |
666 """ | |
667 return client.pubsub_client.subscriptions(service, nodeIdentifier) | |
668 | |
669 ## misc tools ## | |
670 | |
671 def getNodeURI(self, service, node, item=None): | |
672 """Return XMPP URI of a PubSub node | |
673 | |
674 @param service(jid.JID): PubSub service | |
675 @param node(unicode): node | |
676 @return (unicode): URI of the node | |
677 """ | |
678 assert service is not None | |
679 # XXX: urllib.urlencode use "&" to separate value, while XMPP URL (cf. RFC 5122) | |
680 # use ";" as a separator. So if more than one value is used in query_data, | |
681 # urlencode MUST NOT BE USED. | |
682 query_data = [('node', node.encode('utf-8'))] | |
683 if item is not None: | |
684 query_data.append(('item', item.encode('utf-8'))) | |
685 return "xmpp:{service}?;{query}".format( | |
686 service=service.userhost(), | |
687 query=urllib.urlencode(query_data) | |
688 ).decode('utf-8') | |
689 | |
690 ## methods to manage several stanzas/jids at once ## | |
691 | |
692 # generic # | |
693 | |
694 def getRTResults(self, session_id, on_success=None, on_error=None, profile=C.PROF_KEY_NONE): | |
695 return self.rt_sessions.getResults(session_id, on_success, on_error, profile) | |
696 | |
697 def serItemsData(self, items_data, item_cb=lambda item: item.toXml()): | |
698 """Helper method to serialise result from [getItems] | |
699 | |
700 the items_data must be a tuple(list[domish.Element], dict[unicode, unicode]) | |
701 as returned by [getItems]. metadata values are then casted to unicode and | |
702 each item is passed to items_cb | |
703 @param items_data(tuple): tuple returned by [getItems] | |
704 @param item_cb(callable): method to transform each item | |
705 @return (tuple): a serialised form ready to go throught bridge | |
706 """ | |
707 items, metadata = items_data | |
708 return [item_cb(item) for item in items], {key: unicode(value) for key, value in metadata.iteritems()} | |
709 | |
710 def serItemsDataD(self, items_data, item_cb): | |
711 """Helper method to serialise result from [getItems], deferred version | |
712 | |
713 the items_data must be a tuple(list[domish.Element], dict[unicode, unicode]) | |
714 as returned by [getItems]. metadata values are then casted to unicode and | |
715 each item is passed to items_cb | |
716 An errback is added to item_cb, and when it is fired the value is filtered from final items | |
717 @param items_data(tuple): tuple returned by [getItems] | |
718 @param item_cb(callable): method to transform each item (must return a deferred) | |
719 @return (tuple): a deferred which fire a serialised form ready to go throught bridge | |
720 """ | |
721 items, metadata = items_data | |
722 def eb(failure): | |
723 log.warning("Error while serialising/parsing item: {}".format(unicode(failure.value))) | |
724 d = defer.gatherResults([item_cb(item).addErrback(eb) for item in items]) | |
725 def finishSerialisation(serialised_items): | |
726 return [item for item in serialised_items if item is not None], {key: unicode(value) for key, value in metadata.iteritems()} | |
727 d.addCallback(finishSerialisation) | |
728 return d | |
729 | |
730 def serDList(self, results, failure_result=None): | |
731 """Serialise a DeferredList result | |
732 | |
733 @param results: DeferredList results | |
734 @param failure_result: value to use as value for failed Deferred | |
735 (default: empty tuple) | |
736 @return (list): list with: | |
737 - failure: empty in case of success, else error message | |
738 - result | |
739 """ | |
740 if failure_result is None: | |
741 failure_result = () | |
742 return [('', result) if success else (unicode(result.result) or UNSPECIFIED, failure_result) for success, result in results] | |
743 | |
744 # subscribe # | |
745 | |
746 def _getNodeSubscriptions(self, service_s, nodeIdentifier, profile_key): | |
747 client = self.host.getClient(profile_key) | |
748 d = self.getNodeSubscriptions(client, jid.JID(service_s) if service_s else None, nodeIdentifier) | |
749 d.addCallback(lambda subscriptions: {j.full(): a for j, a in subscriptions.iteritems()}) | |
750 return d | |
751 | |
752 def getNodeSubscriptions(self, client, service, nodeIdentifier): | |
753 """Retrieve subscriptions to a node | |
754 | |
755 @param nodeIdentifier(unicode): node to get subscriptions from | |
756 """ | |
757 if not nodeIdentifier: | |
758 raise exceptions.DataError("node identifier can't be empty") | |
759 request = pubsub.PubSubRequest('subscriptionsGet') | |
760 request.recipient = service | |
761 request.nodeIdentifier = nodeIdentifier | |
762 | |
763 def cb(iq_elt): | |
764 try: | |
765 subscriptions_elt = next(iq_elt.pubsub.elements((pubsub.NS_PUBSUB, 'subscriptions'))) | |
766 except StopIteration: | |
767 raise ValueError(_(u"Invalid result: missing <subscriptions> element: {}").format(iq_elt.toXml)) | |
768 except AttributeError as e: | |
769 raise ValueError(_(u"Invalid result: {}").format(e)) | |
770 try: | |
771 return {jid.JID(s['jid']): s['subscription'] for s in subscriptions_elt.elements((pubsub.NS_PUBSUB, 'subscription'))} | |
772 except KeyError: | |
773 raise ValueError(_(u"Invalid result: bad <subscription> element: {}").format(iq_elt.toXml)) | |
774 | |
775 d = request.send(client.xmlstream) | |
776 d.addCallback(cb) | |
777 return d | |
778 | |
779 def _setNodeSubscriptions(self, service_s, nodeIdentifier, subscriptions, profile_key=C.PROF_KEY_NONE): | |
780 client = self.host.getClient(profile_key) | |
781 subscriptions = {jid.JID(jid_): subscription for jid_, subscription in subscriptions.iteritems()} | |
782 d = self.setNodeSubscriptions(client, jid.JID(service_s) if service_s else None, nodeIdentifier, subscriptions) | |
783 return d | |
784 | |
785 def setNodeSubscriptions(self, client, service, nodeIdentifier, subscriptions): | |
786 """Set or update subscriptions of a node owned by profile | |
787 | |
788 @param subscriptions(dict[jid.JID, unicode]): subscriptions to set | |
789 check https://xmpp.org/extensions/xep-0060.html#substates for a list of possible subscriptions | |
790 """ | |
791 request = pubsub.PubSubRequest('subscriptionsSet') | |
792 request.recipient = service | |
793 request.nodeIdentifier = nodeIdentifier | |
794 request.subscriptions = {pubsub.Subscription(nodeIdentifier, jid_, state) for jid_, state in subscriptions.iteritems()} | |
795 d = request.send(client.xmlstream) | |
796 return d | |
797 | |
798 def _manySubscribeRTResult(self, session_id, profile_key=C.PROF_KEY_DEFAULT): | |
799 """Get real-time results for subcribeToManu session | |
800 | |
801 @param session_id: id of the real-time deferred session | |
802 @param return (tuple): (remaining, results) where: | |
803 - remaining is the number of still expected results | |
804 - results is a list of tuple(unicode, unicode, bool, unicode) with: | |
805 - service: pubsub service | |
806 - and node: pubsub node | |
807 - failure(unicode): empty string in case of success, error message else | |
808 @param profile_key: %(doc_profile_key)s | |
809 """ | |
810 profile = self.host.getClient(profile_key).profile | |
811 d = self.rt_sessions.getResults(session_id, on_success=lambda result:'', on_error=lambda failure:unicode(failure.value), profile=profile) | |
812 # we need to convert jid.JID to unicode with full() to serialise it for the bridge | |
813 d.addCallback(lambda ret: (ret[0], [(service.full(), node, '' if success else failure or UNSPECIFIED) | |
814 for (service, node), (success, failure) in ret[1].iteritems()])) | |
815 return d | |
816 | |
817 def _subscribeToMany(self, node_data, subscriber=None, options=None, profile_key=C.PROF_KEY_NONE): | |
818 return self.subscribeToMany([(jid.JID(service), unicode(node)) for service, node in node_data], jid.JID(subscriber), options, profile_key) | |
819 | |
820 def subscribeToMany(self, node_data, subscriber, options=None, profile_key=C.PROF_KEY_NONE): | |
821 """Subscribe to several nodes at once. | |
822 | |
823 @param node_data (iterable[tuple]): iterable of tuple (service, node) where: | |
824 - service (jid.JID) is the pubsub service | |
825 - node (unicode) is the node to subscribe to | |
826 @param subscriber (jid.JID): optional subscription identifier. | |
827 @param options (dict): subscription options | |
828 @param profile_key (str): %(doc_profile_key)s | |
829 @return (str): RT Deferred session id | |
830 """ | |
831 client = self.host.getClient(profile_key) | |
832 deferreds = {} | |
833 for service, node in node_data: | |
834 deferreds[(service, node)] = client.pubsub_client.subscribe(service, node, subscriber, options=options) | |
835 return self.rt_sessions.newSession(deferreds, client.profile) | |
836 # found_nodes = yield self.listNodes(service, profile=client.profile) | |
837 # subscribed_nodes = yield self.listSubscribedNodes(service, profile=client.profile) | |
838 # d_list = [] | |
839 # for nodeIdentifier in (set(nodeIdentifiers) - set(subscribed_nodes)): | |
840 # if nodeIdentifier not in found_nodes: | |
841 # log.debug(u"Skip the subscription to [{node}]: node doesn't exist".format(node=nodeIdentifier)) | |
842 # continue # avoid sat-pubsub "SubscriptionExists" error | |
843 # d_list.append(client.pubsub_client.subscribe(service, nodeIdentifier, sub_jid or client.pubsub_client.parent.jid.userhostJID(), options=options)) | |
844 # defer.returnValue(d_list) | |
845 | |
846 # get # | |
847 | |
848 def _getFromManyRTResult(self, session_id, profile_key=C.PROF_KEY_DEFAULT): | |
849 """Get real-time results for getFromMany session | |
850 | |
851 @param session_id: id of the real-time deferred session | |
852 @param profile_key: %(doc_profile_key)s | |
853 @param return (tuple): (remaining, results) where: | |
854 - remaining is the number of still expected results | |
855 - results is a list of tuple with | |
856 - service (unicode): pubsub service | |
857 - node (unicode): pubsub node | |
858 - failure (unicode): empty string in case of success, error message else | |
859 - items (list[s]): raw XML of items | |
860 - metadata(dict): serialised metadata | |
861 """ | |
862 profile = self.host.getClient(profile_key).profile | |
863 d = self.rt_sessions.getResults(session_id, | |
864 on_success=lambda result: ('', self.serItemsData(result)), | |
865 on_error=lambda failure: (unicode(failure.value) or UNSPECIFIED, ([],{})), | |
866 profile=profile) | |
867 d.addCallback(lambda ret: (ret[0], | |
868 [(service.full(), node, failure, items, metadata) | |
869 for (service, node), (success, (failure, (items, metadata))) in ret[1].iteritems()])) | |
870 return d | |
871 | |
872 def _getFromMany(self, node_data, max_item=10, extra_dict=None, profile_key=C.PROF_KEY_NONE): | |
873 """ | |
874 @param max_item(int): maximum number of item to get, C.NO_LIMIT for no limit | |
875 """ | |
876 max_item = None if max_item == C.NO_LIMIT else max_item | |
877 extra = self.parseExtra(extra_dict) | |
878 return self.getFromMany([(jid.JID(service), unicode(node)) for service, node in node_data], max_item, extra.rsm_request, extra.extra, profile_key) | |
879 | |
880 def getFromMany(self, node_data, max_item=None, rsm_request=None, extra=None, profile_key=C.PROF_KEY_NONE): | |
881 """Get items from many nodes at once | |
882 | |
883 @param node_data (iterable[tuple]): iterable of tuple (service, node) where: | |
884 - service (jid.JID) is the pubsub service | |
885 - node (unicode) is the node to get items from | |
886 @param max_items (int): optional limit on the number of retrieved items. | |
887 @param rsm_request (RSMRequest): RSM request data | |
888 @param profile_key (unicode): %(doc_profile_key)s | |
889 @return (str): RT Deferred session id | |
890 """ | |
891 client = self.host.getClient(profile_key) | |
892 deferreds = {} | |
893 for service, node in node_data: | |
894 deferreds[(service, node)] = self.getItems(client, service, node, max_item, rsm_request=rsm_request, extra=extra) | |
895 return self.rt_sessions.newSession(deferreds, client.profile) | |
896 | |
897 | |
898 class SatPubSubClient(rsm.PubSubClient): | |
899 implements(disco.IDisco) | |
900 | |
901 def __init__(self, host, parent_plugin): | |
902 self.host = host | |
903 self.parent_plugin = parent_plugin | |
904 rsm.PubSubClient.__init__(self) | |
905 | |
906 def connectionInitialized(self): | |
907 rsm.PubSubClient.connectionInitialized(self) | |
908 | |
909 def _getNodeCallbacks(self, node, event): | |
910 """Generate callbacks from given node and event | |
911 | |
912 @param node(unicode): node used for the item | |
913 any registered node which prefix the node will match | |
914 @param event(unicode): one of C.PS_ITEMS, C.PS_RETRACT, C.PS_DELETE | |
915 @return (iterator[callable]): callbacks for this node/event | |
916 """ | |
917 for registered_node, callbacks_dict in self.parent_plugin._node_cb.iteritems(): | |
918 if not node.startswith(registered_node): | |
919 continue | |
920 try: | |
921 for callback in callbacks_dict[event]: | |
922 yield callback | |
923 except KeyError: | |
924 continue | |
925 | |
926 def itemsReceived(self, event): | |
927 log.debug(u"Pubsub items received") | |
928 for callback in self._getNodeCallbacks(event.nodeIdentifier, C.PS_ITEMS): | |
929 callback(self.parent, event) | |
930 client = self.parent | |
931 if (event.sender, event.nodeIdentifier) in client.pubsub_watching: | |
932 raw_items = [i.toXml() for i in event.items] | |
933 self.host.bridge.psEventRaw(event.sender.full(), event.nodeIdentifier, C.PS_ITEMS, raw_items, client.profile) | |
934 | |
935 def deleteReceived(self, event): | |
936 log.debug((u"Publish node deleted")) | |
937 for callback in self._getNodeCallbacks(event.nodeIdentifier, C.PS_DELETE): | |
938 callback(self.parent, event) | |
939 client = self.parent | |
940 if (event.sender, event.nodeIdentifier) in client.pubsub_watching: | |
941 self.host.bridge.psEventRaw(event.sender.full(), event.nodeIdentifier, C.PS_DELETE, [], client.profile) | |
942 | |
943 def subscriptions(self, service, nodeIdentifier, sender=None): | |
944 """Return the list of subscriptions to the given service and node. | |
945 | |
946 @param service: The publish subscribe service to retrieve the subscriptions from. | |
947 @type service: L{JID<twisted.words.protocols.jabber.jid.JID>} | |
948 @param nodeIdentifier: The identifier of the node (leave empty to retrieve all subscriptions). | |
949 @type nodeIdentifier: C{unicode} | |
950 @return (list[pubsub.Subscription]): list of subscriptions | |
951 """ | |
952 request = pubsub.PubSubRequest('subscriptions') | |
953 request.recipient = service | |
954 request.nodeIdentifier = nodeIdentifier | |
955 request.sender = sender | |
956 d = request.send(self.xmlstream) | |
957 | |
958 def cb(iq): | |
959 subs = [] | |
960 for subscription_elt in iq.pubsub.subscriptions.elements(pubsub.NS_PUBSUB, 'subscription'): | |
961 subscription = pubsub.Subscription(subscription_elt['node'], | |
962 jid.JID(subscription_elt['jid']), | |
963 subscription_elt['subscription'], | |
964 subscriptionIdentifier=subscription_elt.getAttribute('subid')) | |
965 subs.append(subscription) | |
966 return subs | |
967 | |
968 return d.addCallback(cb) | |
969 | |
970 def getDiscoInfo(self, requestor, service, nodeIdentifier=''): | |
971 disco_info = [] | |
972 self.host.trigger.point("PubSub Disco Info", disco_info, self.parent.profile) | |
973 return disco_info | |
974 | |
975 def getDiscoItems(self, requestor, service, nodeIdentifier=''): | |
976 return [] |