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

src/plugins/plugin_misc_groupblog.py

@@ -137,7 +137,6 @@ class GroupBlog(object):
                               method=self.subscribeGroupBlog,
                               async=True)
 
-
         host.trigger.add("PubSubItemsReceived", self.pubSubItemsReceivedTrigger)
 
     ## plugin management methods ##

src/plugins/plugin_xep_0060.py

@@ -68,6 +68,8 @@ class XEP_0060(object):
         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 @@ 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=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 @@ class XEP_0060(object):
 
     ## 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 @@ class XEP_0060(object):
         #     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)

src/plugins/plugin_xep_0277.py

@@ -28,6 +28,7 @@ from sat.core import exceptions
 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 @@ class XEP_0277(object):
                               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 @@ class XEP_0277(object):
 
         @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 @@ class XEP_0277(object):
         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)