peer: add an iterbatcher interface

peer: add an iterbatcher interface

This is very much like ordinary batch(), but it will let me add a mode
for batch where we have pathologically large requests which are then
handled streamily. This will be a significant improvement for things
like remotefilelog, which may want to request thousands of entities at
once.

--- a/mercurial/peer.py	Wed Mar 02 14:18:43 2016 -0500
+++ b/mercurial/peer.py	Tue Mar 01 18:39:25 2016 -0500
@@ -41,6 +41,14 @@
     def submit(self):
         raise NotImplementedError()
 
+class iterbatcher(batcher):
+
+    def submit(self):
+        raise NotImplementedError()
+
+    def results(self):
+        raise NotImplementedError()
+
 class localbatch(batcher):
     '''performs the queued calls directly'''
     def __init__(self, local):
@@ -50,6 +58,19 @@
         for name, args, opts, resref in self.calls:
             resref.set(getattr(self.local, name)(*args, **opts))
 
+class localiterbatcher(iterbatcher):
+    def __init__(self, local):
+        super(iterbatcher, self).__init__()
+        self.local = local
+
+    def submit(self):
+        # submit for a local iter batcher is a noop
+        pass
+
+    def results(self):
+        for name, args, opts, resref in self.calls:
+            yield getattr(self.local, name)(*args, **opts)
+
 def batchable(f):
     '''annotation for batchable methods
 
@@ -91,6 +112,14 @@
     def batch(self):
         return localbatch(self)
 
+    def iterbatch(self):
+        """Batch requests but allow iterating over the results.
+
+        This is to allow interleaving responses with things like
+        progress updates for clients.
+        """
+        return localiterbatcher(self)
+
     def capable(self, name):
         '''tell whether repo supports named capability.
         return False if not supported.

--- a/mercurial/wireproto.py	Wed Mar 02 14:18:43 2016 -0500
+++ b/mercurial/wireproto.py	Tue Mar 01 18:39:25 2016 -0500
@@ -114,6 +114,25 @@
             encresref.set(encres)
             resref.set(batchable.next())
 
+class remoteiterbatcher(peer.iterbatcher):
+    def __init__(self, remote):
+        super(remoteiterbatcher, self).__init__()
+        self._remote = remote
+
+    def submit(self):
+        """Break the batch request into many patch calls and pipeline them.
+
+        This is mostly valuable over http where request sizes can be
+        limited, but can be used in other places as well.
+        """
+        rb = self._remote.batch()
+        rb.calls = self.calls
+        rb.submit()
+
+    def results(self):
+        for name, args, opts, resref in self.calls:
+            yield resref.value
+
 # Forward a couple of names from peer to make wireproto interactions
 # slightly more sensible.
 batchable = peer.batchable
@@ -193,6 +212,9 @@
     def _submitone(self, op, args):
         return self._call(op, **args)
 
+    def iterbatch(self):
+        return remoteiterbatcher(self)
+
     @batchable
     def lookup(self, key):
         self.requirecap('lookup', _('look up remote revision'))