# HG changeset patch # User Goffi # Date 1439670154 -7200 # Node ID 5116d70ddd1cc844146be271cb2cf86d550f0de2 # Parent 9b88b19b1ca817b339389176ab0da516c471b52e plugin xep-0060,xep-0277: added getFromMany using new RTDeferredSession mechanism diff -r 9b88b19b1ca8 -r 5116d70ddd1c src/plugins/plugin_misc_groupblog.py --- a/src/plugins/plugin_misc_groupblog.py Sat Aug 15 22:20:56 2015 +0200 +++ b/src/plugins/plugin_misc_groupblog.py Sat Aug 15 22:22:34 2015 +0200 @@ -137,7 +137,6 @@ method=self.subscribeGroupBlog, async=True) - host.trigger.add("PubSubItemsReceived", self.pubSubItemsReceivedTrigger) ## plugin management methods ## diff -r 9b88b19b1ca8 -r 5116d70ddd1c src/plugins/plugin_xep_0060.py --- a/src/plugins/plugin_xep_0060.py Sat Aug 15 22:20:56 2015 +0200 +++ b/src/plugins/plugin_xep_0060.py Sat Aug 15 22:22:34 2015 +0200 @@ -68,6 +68,8 @@ 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) @@ -115,32 +117,31 @@ 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 profile_key (str): %(doc_profile_key)s + @param rsm_request (rsm.RSMRequest): RSM request data + @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 = {} - else: + metadata = {} + if rsm_request: 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) @@ -202,6 +203,11 @@ ## 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] @@ -299,6 +305,55 @@ # 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) diff -r 9b88b19b1ca8 -r 5116d70ddd1c src/plugins/plugin_xep_0277.py --- a/src/plugins/plugin_xep_0277.py Sat Aug 15 22:20:56 2015 +0200 +++ b/src/plugins/plugin_xep_0277.py Sat Aug 15 22:22:34 2015 +0200 @@ -28,6 +28,7 @@ from sat.tools.xml_tools import ElementParser from wokkel import pubsub +from wokkel import rsm from feed import atom, date import uuid from time import time @@ -77,6 +78,10 @@ async=True) host.bridge.addMethod("mBSubscribeToMany", ".plugin", in_sign='sass', out_sign='s', method=self._mBSubscribeToMany) + host.bridge.addMethod("mBGetFromManyRTResult", ".plugin", in_sign='ss', out_sign='(ua(sssaa{ss}a{ss}))', + method=self._mBGetFromManyRTResult, async=True) + host.bridge.addMethod("mBGetFromMany", ".plugin", in_sign='sasia{ss}s', out_sign='s', + method=self._mBGetFromMany) ## plugin management methods ## @@ -329,15 +334,9 @@ @return: a deferred couple with the list of items and metadatas. """ - items, metadata = yield self.host.plugins["XEP-0060"].getItems(pub_jid, NS_MICROBLOG, max_items=max_items, profile_key=profile_key) - dlist_result = yield defer.DeferredList(map(self.item2mbdata, items), consumeErrors=True) - items_data = [] - for success, value in dlist_result: - if success: - items_data.append(value) - else: - log.warning(u"Error while parsing microblog data: {}".format(value.value)) - defer.returnValue((items_data, metadata)) + items_data = yield self._p.getItems(pub_jid, NS_MICROBLOG, max_items=max_items, profile_key=profile_key) + serialised = yield self._p.serItemsDataD(items_data, self.item2mbdata) + defer.returnValue(serialised) def parseCommentUrl(self, node_url): """Parse a XMPP URI @@ -468,3 +467,58 @@ client, node_data = self._getClientAndNodeData(publishers_type, publishers, profile_key) return self._p.subscribeToMany(node_data, client.jid.userhostJID(), profile_key=profile_key) + # get # + + def _mBGetFromManyRTResult(self, session_id, profile_key=C.PROF_KEY_DEFAULT): + """Get real-time results for mBGetFromMany 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 with + - service (unicode): pubsub service + - node (unicode): pubsub node + - failure (unicode): empty string in case of success, error message else + - items_data(tuple): data tuple as returned by [getLastMicroblogs] + @param profile_key: %(doc_profile_key)s + """ + def onSuccess(items_data): + """convert items elements to list of microblog data in items_data""" + d = self._p.serItemsDataD(items_data, self.item2mbdata) + d.addCallback(lambda serialised:('', serialised)) + return d + + profile = self.host.getClient(profile_key).profile + d = self._p.getRTResults(session_id, + on_success = onSuccess, + on_error = lambda failure: (unicode(failure.value), ([],{})), + 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 _mBGetFromMany(self, publishers_type, publishers, 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 + publishers_type, publishers = self._checkPublishers(publishers_type, publishers) + return self.mBGetFromMany(publishers_type, publishers, max_item, rsm.RSMRequest(**rsm_dict) if rsm_dict else None, profile_key) + + def mBGetFromMany(self, publishers_type, publishers, max_item=None, rsm_data=None, profile_key=C.PROF_KEY_NONE): + """Get the published microblogs for a list of groups or jids + + @param publishers_type (str): type of the list of publishers (one of "GROUP" or "JID" or "ALL") + @param publishers (list): list of publishers, according to publishers_type (list of groups or list of jids) + @param max_items (int): optional limit on the number of retrieved items. + @param rsm_data (rsm.RSMRequest): RSM request data, common to all publishers + @param profile_key: profile key + @return: a deferred dict with: + - key: publisher (unicode) + - value: couple (list[dict], dict) with: + - the microblogs data + - RSM response data + """ + client, node_data = self._getClientAndNodeData(publishers_type, publishers, profile_key) + return self._p.getFromMany(node_data, max_item, rsm_data, profile_key=profile_key)