wireprotov1peer: update all rpcs to use the new batchable scheme

If desired, we could keep the future class and the function that
upgrades an old style rpc instead of a new style, for extensions.

Differential Revision: https://phab.mercurial-scm.org/D11212

hgext/fastannotate/protocol.py

@@ -140,12 +140,10 b' def peersetup(ui, peer):'
         def getannotate(self, path, lastnode=None):
             if not self.capable(b'getannotate'):
                 ui.warn(_(b'remote peer cannot provide annotate cache\n'))
-                yield None, None
+                return None, None
             else:
                 args = {b'path': path, b'lastnode': lastnode or b''}
-                f = wireprotov1peer.future()
-                yield args, f
-                yield _parseresponse(f.value)
+                return args, _parseresponse
 
     peer.__class__ = fastannotatepeer
 

hgext/infinitepush/__init__.py

@@ -431,18 +431,19 b' def localrepolistkeys(orig, self, namesp'
 @wireprotov1peer.batchable
 def listkeyspatterns(self, namespace, patterns):
     if not self.capable(b'pushkey'):
-        yield {}, None
-    f = wireprotov1peer.future()
+        return {}, None
     self.ui.debug(b'preparing listkeys for "%s"\n' % namespace)
-    yield {
+
+    def decode(d):
+        self.ui.debug(
+            b'received listkey for "%s": %i bytes\n' % (namespace, len(d))
+        )
+        return pushkey.decodekeys(d)
+
+    return {
         b'namespace': encoding.fromlocal(namespace),
         b'patterns': wireprototypes.encodelist(patterns),
-    }, f
-    d = f.value
-    self.ui.debug(
-        b'received listkey for "%s": %i bytes\n' % (namespace, len(d))
-    )
-    yield pushkey.decodekeys(d)
+    }, decode
 
 
 def _readbundlerevs(bundlerepo):

hgext/largefiles/proto.py

@@ -184,17 +184,18 b' def wirereposetup(ui, repo):'
 
         @wireprotov1peer.batchable
         def statlfile(self, sha):
-            f = wireprotov1peer.future()
+            def decode(d):
+                try:
+                    return int(d)
+                except (ValueError, urlerr.httperror):
+                    # If the server returns anything but an integer followed by a
+                    # newline, newline, it's not speaking our language; if we get
+                    # an HTTP error, we can't be sure the largefile is present;
+                    # either way, consider it missing.
+                    return 2
+
             result = {b'sha': sha}
-            yield result, f
-            try:
-                yield int(f.value)
-            except (ValueError, urlerr.httperror):
-                # If the server returns anything but an integer followed by a
-                # newline, newline, it's not speaking our language; if we get
-                # an HTTP error, we can't be sure the largefile is present;
-                # either way, consider it missing.
-                yield 2
+            return result, decode
 
     repo.__class__ = lfileswirerepository
 

hgext/remotefilelog/fileserverclient.py

@@ -63,12 +63,14 b' def peersetup(ui, peer):'
                 raise error.Abort(
                     b'configured remotefile server does not support getfile'
                 )
-            f = wireprotov1peer.future()
-            yield {b'file': file, b'node': node}, f
-            code, data = f.value.split(b'\0', 1)
-            if int(code):
-                raise error.LookupError(file, node, data)
-            yield data
+
+            def decode(d):
+                code, data = d.split(b'\0', 1)
+                if int(code):
+                    raise error.LookupError(file, node, data)
+                return data
+
+            return {b'file': file, b'node': node}, decode
 
         @wireprotov1peer.batchable
         def x_rfl_getflogheads(self, path):
@@ -77,10 +79,11 b' def peersetup(ui, peer):'
                     b'configured remotefile server does not '
                     b'support getflogheads'
                 )
-            f = wireprotov1peer.future()
-            yield {b'path': path}, f
-            heads = f.value.split(b'\n') if f.value else []
-            yield heads
+
+            def decode(d):
+                return d.split(b'\n') if d else []
+
+            return {b'path': path}, decode
 
         def _updatecallstreamopts(self, command, opts):
             if command != b'getbundle':

mercurial/wireprotov1peer.py

@@ -35,7 +35,7 b' from .utils import hashutil'
 urlreq = util.urlreq
 
 
-def batchable_new_style(f):
+def batchable(f):
     """annotation for batchable methods
 
     Such methods must implement a coroutine as follows:
@@ -68,33 +68,6 b' def batchable_new_style(f):'
     return plain
 
 
-def batchable(f):
-    def upgraded(*args, **opts):
-        batchable = f(*args, **opts)
-        encoded_args_or_res, encoded_res_future = next(batchable)
-        if not encoded_res_future:
-            decode = None
-        else:
-
-            def decode(d):
-                encoded_res_future.set(d)
-                return next(batchable)
-
-        return encoded_args_or_res, decode
-
-    setattr(upgraded, '__name__', f.__name__)
-    return batchable_new_style(upgraded)
-
-
-class future(object):
-    '''placeholder for a value to be set later'''
-
-    def set(self, value):
-        if util.safehasattr(self, b'value'):
-            raise error.RepoError(b"future is already set")
-        self.value = value
-
-
 def encodebatchcmds(req):
     """Return a ``cmds`` argument value for the ``batch`` command."""
     escapearg = wireprototypes.escapebatcharg
@@ -372,87 +345,90 b' class wirepeer(repository.peer):'
     @batchable
     def lookup(self, key):
         self.requirecap(b'lookup', _(b'look up remote revision'))
-        f = future()
-        yield {b'key': encoding.fromlocal(key)}, f
-        d = f.value
-        success, data = d[:-1].split(b" ", 1)
-        if int(success):
-            yield bin(data)
-        else:
-            self._abort(error.RepoError(data))
+
+        def decode(d):
+            success, data = d[:-1].split(b" ", 1)
+            if int(success):
+                return bin(data)
+            else:
+                self._abort(error.RepoError(data))
+
+        return {b'key': encoding.fromlocal(key)}, decode
 
     @batchable
     def heads(self):
-        f = future()
-        yield {}, f
-        d = f.value
-        try:
-            yield wireprototypes.decodelist(d[:-1])
-        except ValueError:
-            self._abort(error.ResponseError(_(b"unexpected response:"), d))
+        def decode(d):
+            try:
+                return wireprototypes.decodelist(d[:-1])
+            except ValueError:
+                self._abort(error.ResponseError(_(b"unexpected response:"), d))
+
+        return {}, decode
 
     @batchable
     def known(self, nodes):
-        f = future()
-        yield {b'nodes': wireprototypes.encodelist(nodes)}, f
-        d = f.value
-        try:
-            yield [bool(int(b)) for b in pycompat.iterbytestr(d)]
-        except ValueError:
-            self._abort(error.ResponseError(_(b"unexpected response:"), d))
+        def decode(d):
+            try:
+                return [bool(int(b)) for b in pycompat.iterbytestr(d)]
+            except ValueError:
+                self._abort(error.ResponseError(_(b"unexpected response:"), d))
+
+        return {b'nodes': wireprototypes.encodelist(nodes)}, decode
 
     @batchable
     def branchmap(self):
-        f = future()
-        yield {}, f
-        d = f.value
-        try:
-            branchmap = {}
-            for branchpart in d.splitlines():
-                branchname, branchheads = branchpart.split(b' ', 1)
-                branchname = encoding.tolocal(urlreq.unquote(branchname))
-                branchheads = wireprototypes.decodelist(branchheads)
-                branchmap[branchname] = branchheads
-            yield branchmap
-        except TypeError:
-            self._abort(error.ResponseError(_(b"unexpected response:"), d))
+        def decode(d):
+            try:
+                branchmap = {}
+                for branchpart in d.splitlines():
+                    branchname, branchheads = branchpart.split(b' ', 1)
+                    branchname = encoding.tolocal(urlreq.unquote(branchname))
+                    branchheads = wireprototypes.decodelist(branchheads)
+                    branchmap[branchname] = branchheads
+                return branchmap
+            except TypeError:
+                self._abort(error.ResponseError(_(b"unexpected response:"), d))
+
+        return {}, decode
 
     @batchable
     def listkeys(self, namespace):
         if not self.capable(b'pushkey'):
-            yield {}, None
-        f = future()
+            return {}, None
         self.ui.debug(b'preparing listkeys for "%s"\n' % namespace)
-        yield {b'namespace': encoding.fromlocal(namespace)}, f
-        d = f.value
-        self.ui.debug(
-            b'received listkey for "%s": %i bytes\n' % (namespace, len(d))
-        )
-        yield pushkeymod.decodekeys(d)
+
+        def decode(d):
+            self.ui.debug(
+                b'received listkey for "%s": %i bytes\n' % (namespace, len(d))
+            )
+            return pushkeymod.decodekeys(d)
+
+        return {b'namespace': encoding.fromlocal(namespace)}, decode
 
     @batchable
     def pushkey(self, namespace, key, old, new):
         if not self.capable(b'pushkey'):
-            yield False, None
-        f = future()
+            return False, None
         self.ui.debug(b'preparing pushkey for "%s:%s"\n' % (namespace, key))
-        yield {
+
+        def decode(d):
+            d, output = d.split(b'\n', 1)
+            try:
+                d = bool(int(d))
+            except ValueError:
+                raise error.ResponseError(
+                    _(b'push failed (unexpected response):'), d
+                )
+            for l in output.splitlines(True):
+                self.ui.status(_(b'remote: '), l)
+            return d
+
+        return {
             b'namespace': encoding.fromlocal(namespace),
             b'key': encoding.fromlocal(key),
             b'old': encoding.fromlocal(old),
             b'new': encoding.fromlocal(new),
-        }, f
-        d = f.value
-        d, output = d.split(b'\n', 1)
-        try:
-            d = bool(int(d))
-        except ValueError:
-            raise error.ResponseError(
-                _(b'push failed (unexpected response):'), d
-            )
-        for l in output.splitlines(True):
-            self.ui.status(_(b'remote: '), l)
-        yield d
+        }, decode
 
     def stream_out(self):
         return self._callstream(b'stream_out')

tests/test-batching.py

@@ -214,14 +214,11 b' class remotething(thing):'
                 mangle(two),
             ),
         ]
-        encoded_res_future = wireprotov1peer.future()
-        yield encoded_args, encoded_res_future
-        yield unmangle(encoded_res_future.value)
+        return encoded_args, unmangle
 
     @wireprotov1peer.batchable
     def bar(self, b, a):
-        encresref = wireprotov1peer.future()
-        yield [
+        return [
             (
                 b'b',
                 mangle(b),
@@ -230,8 +227,7 b' class remotething(thing):'
                 b'a',
                 mangle(a),
             ),
-        ], encresref
-        yield unmangle(encresref.value)
+        ], unmangle
 
     # greet is coded directly. It therefore does not support batching. If it
     # does appear in a batch, the batch is split around greet, and the call to

tests/test-wireproto.py

@@ -75,9 +75,7 b' class clientpeer(wireprotov1peer.wirepee'
 
     @wireprotov1peer.batchable
     def greet(self, name):
-        f = wireprotov1peer.future()
-        yield {b'name': mangle(name)}, f
-        yield unmangle(f.value)
+        return {b'name': mangle(name)}, unmangle
 
 
 class serverrepo(object):