libervia-backend: src/plugins/plugin_xep

comparison src/plugins/plugin_xep_0060.py @ 1452:5116d70ddd1c

plugin xep-0060,xep-0277: added getFromMany using new RTDeferredSession mechanism

author	Goffi <goffi@goffi.org>
date	Sat, 15 Aug 2015 22:22:34 +0200
parents	9b88b19b1ca8
children	4c4f88d7b156

comparison

equal deleted inserted replaced

-:9b88b19b1ca8
+:5116d70ddd1c
 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)
+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)
 def getHandler(self, profile):
 client = self.host.getClient(profile)
 client.pubsub_client = SatPubSubClient(self.host, self)
 return client.pubsub_client
 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):
+def getItems(self, service, node, max_items=None, item_ids=None, sub_id=None, rsm_request=None, profile_key=C.PROF_KEY_NONE):
 """Retrieve pubsub items from a node.
-@param service (JID): target service.
+@param service (JID): pubsub 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 item_ids (list[str]): identifiers of the items to be retrieved (can't be used with rsm_request).
 @param sub_id (str): optional subscription identifier.
-@param rsm (dict): RSM request data
+@param rsm_request (rsm.RSMRequest): RSM request data
-@param profile_key (str): %(doc_profile_key)s
+@param profile_key (unicode): %(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:
+if rsm_request 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
+ext_data = {'id': unicode(uuid.uuid4()), 'rsm': rsm_request} if rsm_request 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 = {}
-metadata = {}
+if rsm_request:
-else:
 rsm_data = client.pubsub_client.getRSMResponse(ext_data['id'])
-metadata = {'rsm_{}'.format(key): value for key, value in rsm_data}
+metadata.update({'rsm_{}'.format(key): value for key, value in rsm_data})
 return (items, metadata)
 d.addCallback(addMetadata)
 return d
 client = self.host.getClient(profile_key)
 return client.pubsub_client.subscriptions(service, nodeIdentifier)
 ## methods to manage several stanzas/jids at once ##
+# generic #
+def getRTResults(self, session_id, on_success=None, on_error=None, profile=C.PROF_KEY_NONE):
+return self.rt_sessions.getResults(session_id, on_success, on_error, profile)
 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
 #         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)
+# get #
+def _getFromManyRTResult(self, session_id, profile_key=C.PROF_KEY_DEFAULT):
+"""Get real-time results for getFromMany session
+@param session_id: id of the real-time deferred session
+@param profile_key: %(doc_profile_key)s
+@param return (tuple): (remaining, results) where:
+- remaining is the number of still expected results
+- results is a list of tuple with
+- service (unicode): pubsub service
+- node (unicode): pubsub node
+- failure (unicode): empty string in case of success, error message else
+- items (list[s]): raw XML of items
+- metadata(dict): serialised metadata
+"""
+profile = self.host.getClient(profile_key).profile
+d = self.rt_sessions.getResults(session_id,
+on_success=lambda result: ('', self.serItemsData(result)),
+on_error=lambda failure: (unicode(failure.value) or UNSPECIFIED, ([],{})),
+profile=profile)
+d.addCallback(lambda ret: (ret[0],
+[(service.full(), node, failure, items, metadata)
+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):
+"""
+@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)
+def getFromMany(self, node_data, max_item=None, rsm_request=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
+- node (unicode) is the node to get items from
+@param max_items (int): optional limit on the number of retrieved items.
+@param rsm_request (RSMRequest): RSM request data
+@param profile_key (unicode): %(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)] = self.getItems(service, node, max_item, rsm_request=rsm_request, profile_key=profile_key)
+return self.rt_sessions.newSession(deferreds, client.profile)
 class SatPubSubClient(rsm.PubSubClient):
 implements(disco.IDisco)
 def __init__(self, host, parent_plugin):

Mercurial > libervia-backend

comparison src/plugins/plugin_xep_0060.py @ 1452:5116d70ddd1c