protocol: unify client unbundle support

- introduce _callpush helper
- factor out differences in result handling into helpers
- unify

mercurial/httprepo.py

@@ -1,215 +1,200
 # httprepo.py - HTTP repository proxy classes for mercurial
 #
 # Copyright 2005, 2006 Matt Mackall <[e-mail]>
 # Copyright 2006 Vadim Gelfer <[e-mail]>
 #
 # This software may be used and distributed according to the terms of the
 # GNU General Public License version 2 or any later version.
 
 from node import bin, hex, nullid
 from i18n import _
 import repo, changegroup, statichttprepo, error, url, util, wireproto
 import os, urllib, urllib2, urlparse, zlib, httplib
 import errno, socket
 import encoding
 
 def zgenerator(f):
     zd = zlib.decompressobj()
     try:
         for chunk in util.filechunkiter(f):
             yield zd.decompress(chunk)
     except httplib.HTTPException:
         raise IOError(None, _('connection ended unexpectedly'))
     yield zd.flush()
 
 class httprepository(wireproto.wirerepository):
     def __init__(self, ui, path):
         self.path = path
         self.caps = None
         self.handler = None
         scheme, netloc, urlpath, query, frag = urlparse.urlsplit(path)
         if query or frag:
             raise util.Abort(_('unsupported URL component: "%s"') %
                              (query or frag))
 
         # urllib cannot handle URLs with embedded user or passwd
         self._url, authinfo = url.getauthinfo(path)
 
         self.ui = ui
         self.ui.debug('using %s\n' % self._url)
 
         self.urlopener = url.opener(ui, authinfo)
 
     def __del__(self):
         for h in self.urlopener.handlers:
             h.close()
             if hasattr(h, "close_all"):
                 h.close_all()
 
     def url(self):
         return self.path
 
     # look up capabilities only when needed
 
     def get_caps(self):
         if self.caps is None:
             try:
                 self.caps = set(self._call('capabilities').split())
             except error.RepoError:
                 self.caps = set()
             self.ui.debug('capabilities: %s\n' %
                           (' '.join(self.caps or ['none'])))
         return self.caps
 
     capabilities = property(get_caps)
 
     def lock(self):
         raise util.Abort(_('operation not supported over http'))
 
     def _callstream(self, cmd, **args):
         data = args.pop('data', None)
         headers = args.pop('headers', {})
         self.ui.debug("sending %s command\n" % cmd)
         q = {"cmd": cmd}
         q.update(args)
         qs = '?%s' % urllib.urlencode(q)
         cu = "%s%s" % (self._url, qs)
         req = urllib2.Request(cu, data, headers)
         if data is not None:
             # len(data) is broken if data doesn't fit into Py_ssize_t
             # add the header ourself to avoid OverflowError
             size = data.__len__()
             self.ui.debug("sending %s bytes\n" % size)
             req.add_unredirected_header('Content-Length', '%d' % size)
         try:
             resp = self.urlopener.open(req)
         except urllib2.HTTPError, inst:
             if inst.code == 401:
                 raise util.Abort(_('authorization failed'))
             raise
         except httplib.HTTPException, inst:
             self.ui.debug('http error while sending %s command\n' % cmd)
             self.ui.traceback()
             raise IOError(None, inst)
         except IndexError:
             # this only happens with Python 2.3, later versions raise URLError
             raise util.Abort(_('http error, possibly caused by proxy setting'))
         # record the url we got redirected to
         resp_url = resp.geturl()
         if resp_url.endswith(qs):
             resp_url = resp_url[:-len(qs)]
         if self._url.rstrip('/') != resp_url.rstrip('/'):
             self.ui.status(_('real URL is %s\n') % resp_url)
         self._url = resp_url
         try:
             proto = resp.getheader('content-type')
         except AttributeError:
             proto = resp.headers['content-type']
 
         safeurl = url.hidepassword(self._url)
         # accept old "text/plain" and "application/hg-changegroup" for now
         if not (proto.startswith('application/mercurial-') or
                 proto.startswith('text/plain') or
                 proto.startswith('application/hg-changegroup')):
             self.ui.debug("requested URL: '%s'\n" % url.hidepassword(cu))
             raise error.RepoError(
                 _("'%s' does not appear to be an hg repository:\n"
                   "---%%<--- (%s)\n%s\n---%%<---\n")
                 % (safeurl, proto, resp.read()))
 
         if proto.startswith('application/mercurial-'):
             try:
                 version = proto.split('-', 1)[1]
                 version_info = tuple([int(n) for n in version.split('.')])
             except ValueError:
                 raise error.RepoError(_("'%s' sent a broken Content-Type "
                                         "header (%s)") % (safeurl, proto))
             if version_info > (0, 1):
                 raise error.RepoError(_("'%s' uses newer protocol %s") %
                                       (safeurl, version))
 
         return resp
 
     def _call(self, cmd, **args):
         fp = self._callstream(cmd, **args)
         try:
             return fp.read()
         finally:
             # if using keepalive, allow connection to be reused
             fp.close()
 
-    def _abort(self, exception):
-        raise exception
-
-    def _decompress(self, stream):
-        return util.chunkbuffer(zgenerator(stream))
-
-    def unbundle(self, cg, heads, source):
-        '''Send cg (a readable file-like object representing the
-        changegroup to push, typically a chunkbuffer object) to the
-        remote server as a bundle. Return an integer response code:
-        non-zero indicates a successful push (see
-        localrepository.addchangegroup()), and zero indicates either
-        error or nothing to push.'''
+    def _callpush(self, cmd, cg, **args):
         # have to stream bundle to a temp file because we do not have
         # http 1.1 chunked transfer.
 
         type = ""
         types = self.capable('unbundle')
         # servers older than d1b16a746db6 will send 'unbundle' as a
         # boolean capability
         try:
             types = types.split(',')
         except AttributeError:
             types = [""]
         if types:
             for x in types:
                 if x in changegroup.bundletypes:
                     type = x
                     break
 
         tempname = changegroup.writebundle(cg, None, type)
         fp = url.httpsendfile(tempname, "rb")
+        headers = {'Content-Type': 'application/mercurial-0.1'}
+
         try:
             try:
-                resp = self._call(
-                     'unbundle', data=fp,
-                     headers={'Content-Type': 'application/mercurial-0.1'},
-                     heads=' '.join(map(hex, heads)))
-                resp_code, output = resp.split('\n', 1)
-                try:
-                    ret = int(resp_code)
-                except ValueError, err:
-                    raise error.ResponseError(
-                            _('push failed (unexpected response):'), resp)
-                for l in output.splitlines(True):
-                    self.ui.status(_('remote: '), l)
-                return ret
+                r = self._call(cmd, data=fp, headers=headers, **args)
+                return r.split('\n', 1)
             except socket.error, err:
                 if err.args[0] in (errno.ECONNRESET, errno.EPIPE):
                     raise util.Abort(_('push failed: %s') % err.args[1])
                 raise util.Abort(err.args[1])
         finally:
             fp.close()
             os.unlink(tempname)
 
+    def _abort(self, exception):
+        raise exception
+
+    def _decompress(self, stream):
+        return util.chunkbuffer(zgenerator(stream))
+
 class httpsrepository(httprepository):
     def __init__(self, ui, path):
         if not url.has_https:
             raise util.Abort(_('Python support for SSL and HTTPS '
                                'is not installed'))
         httprepository.__init__(self, ui, path)
 
 def instance(ui, path, create):
     if create:
         raise util.Abort(_('cannot create new http repository'))
     try:
         if path.startswith('https:'):
             inst = httpsrepository(ui, path)
         else:
             inst = httprepository(ui, path)
         inst.between([(nullid, nullid)])
         return inst
     except error.RepoError:
         ui.note('(falling back to static-http)\n')
         return statichttprepo.instance(ui, "static-" + path, create)

mercurial/sshrepo.py

@@ -1,212 +1,198
 # sshrepo.py - ssh repository proxy class for mercurial
 #
 # Copyright 2005, 2006 Matt Mackall <[e-mail]>
 #
 # This software may be used and distributed according to the terms of the
 # GNU General Public License version 2 or any later version.
 
 from node import bin, hex
 from i18n import _
 import repo, util, error, encoding, wireproto
 import re, urllib
 
 class remotelock(object):
     def __init__(self, repo):
         self.repo = repo
     def release(self):
         self.repo.unlock()
         self.repo = None
     def __del__(self):
         if self.repo:
             self.release()
 
 class sshrepository(wireproto.wirerepository):
     def __init__(self, ui, path, create=0):
         self._url = path
         self.ui = ui
 
         m = re.match(r'^ssh://(([^@]+)@)?([^:/]+)(:(\d+))?(/(.*))?$', path)
         if not m:
             self._abort(error.RepoError(_("couldn't parse location %s") % path))
 
         self.user = m.group(2)
         self.host = m.group(3)
         self.port = m.group(5)
         self.path = m.group(7) or "."
 
         sshcmd = self.ui.config("ui", "ssh", "ssh")
         remotecmd = self.ui.config("ui", "remotecmd", "hg")
 
         args = util.sshargs(sshcmd, self.host, self.user, self.port)
 
         if create:
             cmd = '%s %s "%s init %s"'
             cmd = cmd % (sshcmd, args, remotecmd, self.path)
 
             ui.note(_('running %s\n') % cmd)
             res = util.system(cmd)
             if res != 0:
                 self._abort(error.RepoError(_("could not create remote repo")))
 
         self.validate_repo(ui, sshcmd, args, remotecmd)
 
     def url(self):
         return self._url
 
     def validate_repo(self, ui, sshcmd, args, remotecmd):
         # cleanup up previous run
         self.cleanup()
 
         cmd = '%s %s "%s -R %s serve --stdio"'
         cmd = cmd % (sshcmd, args, remotecmd, self.path)
 
         cmd = util.quotecommand(cmd)
         ui.note(_('running %s\n') % cmd)
         self.pipeo, self.pipei, self.pipee = util.popen3(cmd)
 
         # skip any noise generated by remote shell
         self._callstream("hello")
         r = self._callstream("between", pairs=("%s-%s" % ("0"*40, "0"*40)))
         lines = ["", "dummy"]
         max_noise = 500
         while lines[-1] and max_noise:
             l = r.readline()
             self.readerr()
             if lines[-1] == "1\n" and l == "\n":
                 break
             if l:
                 ui.debug("remote: ", l)
             lines.append(l)
             max_noise -= 1
         else:
             self._abort(error.RepoError(_("no suitable response from remote hg")))
 
         self.capabilities = set()
         for l in reversed(lines):
             if l.startswith("capabilities:"):
                 self.capabilities.update(l[:-1].split(":")[1].split())
                 break
 
     def readerr(self):
         while 1:
             size = util.fstat(self.pipee).st_size
             if size == 0:
                 break
             l = self.pipee.readline()
             if not l:
                 break
             self.ui.status(_("remote: "), l)
 
     def _abort(self, exception):
         self.cleanup()
         raise exception
 
     def cleanup(self):
         try:
             self.pipeo.close()
             self.pipei.close()
             # read the error descriptor until EOF
             for l in self.pipee:
                 self.ui.status(_("remote: "), l)
             self.pipee.close()
         except:
             pass
 
     __del__ = cleanup
 
     def _callstream(self, cmd, **args):
         self.ui.debug("sending %s command\n" % cmd)
         self.pipeo.write("%s\n" % cmd)
         for k, v in sorted(args.iteritems()):
             self.pipeo.write("%s %d\n" % (k, len(v)))
             self.pipeo.write(v)
         self.pipeo.flush()
 
         return self.pipei
 
     def _call(self, cmd, **args):
         self._callstream(cmd, **args)
         return self._recv()
 
+    def _callpush(self, cmd, fp, **args):
+        r = self._call(cmd, **args)
+        if r:
+            return '', r
+        while 1:
+            d = fp.read(4096)
+            if not d:
+                break
+            self._send(d)
+        self._send("", flush=True)
+        r = self._recv()
+        if r:
+            return '', r
+        return self._recv(), ''
+
     def _decompress(self, stream):
         return stream
 
     def _recv(self):
         l = self.pipei.readline()
         self.readerr()
         try:
             l = int(l)
         except:
             self._abort(error.ResponseError(_("unexpected response:"), l))
         return self.pipei.read(l)
 
     def _send(self, data, flush=False):
         self.pipeo.write("%d\n" % len(data))
         if data:
             self.pipeo.write(data)
         if flush:
             self.pipeo.flush()
         self.readerr()
 
     def lock(self):
         self._call("lock")
         return remotelock(self)
 
     def unlock(self):
         self._call("unlock")
 
-    def unbundle(self, cg, heads, source):
-        '''Send cg (a readable file-like object representing the
-        changegroup to push, typically a chunkbuffer object) to the
-        remote server as a bundle. Return an integer indicating the
-        result of the push (see localrepository.addchangegroup()).'''
-        d = self._call("unbundle", heads=' '.join(map(hex, heads)))
-        if d:
-            # remote may send "unsynced changes"
-            self._abort(error.RepoError(_("push refused: %s") % d))
-
-        while 1:
-            d = cg.read(4096)
-            if not d:
-                break
-            self._send(d)
-
-        self._send("", flush=True)
-
-        r = self._recv()
-        if r:
-            # remote may send "unsynced changes"
-            self._abort(error.RepoError(_("push failed: %s") % r))
-
-        r = self._recv()
-        try:
-            return int(r)
-        except:
-            self._abort(error.ResponseError(_("unexpected response:"), r))
-
     def addchangegroup(self, cg, source, url):
         '''Send a changegroup to the remote server.  Return an integer
         similar to unbundle(). DEPRECATED, since it requires locking the
         remote.'''
         d = self._call("addchangegroup")
         if d:
             self._abort(error.RepoError(_("push refused: %s") % d))
         while 1:
             d = cg.read(4096)
             if not d:
                 break
             self.pipeo.write(d)
             self.readerr()
 
         self.pipeo.flush()
 
         self.readerr()
         r = self._recv()
         if not r:
             return 1
         try:
             return int(r)
         except:
             self._abort(error.ResponseError(_("unexpected response:"), r))
 
 instance = sshrepository

mercurial/wireproto.py

@@ -1,192 +1,212
 # wireproto.py - generic wire protocol support functions
 #
 # Copyright 2005-2010 Matt Mackall <[e-mail]>
 #
 # This software may be used and distributed according to the terms of the
 # GNU General Public License version 2 or any later version.
 
 from i18n import _
 from node import bin, hex
 import urllib
 import streamclone, repo, error, encoding
 import pushkey as pushkey_
 
 # client side
 
 class wirerepository(repo.repository):
     def lookup(self, key):
         self.requirecap('lookup', _('look up remote revision'))
         d = self._call("lookup", key=key)
         success, data = d[:-1].split(" ", 1)
         if int(success):
             return bin(data)
         self._abort(error.RepoError(data))
 
     def heads(self):
         d = self._call("heads")
         try:
             return map(bin, d[:-1].split(" "))
         except:
             self.abort(error.ResponseError(_("unexpected response:"), d))
 
     def branchmap(self):
         d = self._call("branchmap")
         try:
             branchmap = {}
             for branchpart in d.splitlines():
                 branchheads = branchpart.split(' ')
                 branchname = urllib.unquote(branchheads[0])
                 # Earlier servers (1.3.x) send branch names in (their) local
                 # charset. The best we can do is assume it's identical to our
                 # own local charset, in case it's not utf-8.
                 try:
                     branchname.decode('utf-8')
                 except UnicodeDecodeError:
                     branchname = encoding.fromlocal(branchname)
                 branchheads = [bin(x) for x in branchheads[1:]]
                 branchmap[branchname] = branchheads
             return branchmap
         except TypeError:
             self._abort(error.ResponseError(_("unexpected response:"), d))
 
     def branches(self, nodes):
         n = " ".join(map(hex, nodes))
         d = self._call("branches", nodes=n)
         try:
             br = [tuple(map(bin, b.split(" "))) for b in d.splitlines()]
             return br
         except:
             self._abort(error.ResponseError(_("unexpected response:"), d))
 
     def between(self, pairs):
         batch = 8 # avoid giant requests
         r = []
         for i in xrange(0, len(pairs), batch):
             n = " ".join(["-".join(map(hex, p)) for p in pairs[i:i + batch]])
             d = self._call("between", pairs=n)
             try:
                 r += [l and map(bin, l.split(" ")) or []
                       for l in d.splitlines()]
             except:
                 self._abort(error.ResponseError(_("unexpected response:"), d))
         return r
 
     def pushkey(self, namespace, key, old, new):
         if not self.capable('pushkey'):
             return False
         d = self._call("pushkey",
                       namespace=namespace, key=key, old=old, new=new)
         return bool(int(d))
 
     def listkeys(self, namespace):
         if not self.capable('pushkey'):
             return {}
         d = self._call("listkeys", namespace=namespace)
         r = {}
         for l in d.splitlines():
             k, v = l.split('\t')
             r[k.decode('string-escape')] = v.decode('string-escape')
         return r
 
     def stream_out(self):
         return self._callstream('stream_out')
 
     def changegroup(self, nodes, kind):
         n = " ".join(map(hex, nodes))
         f = self._callstream("changegroup", roots=n)
         return self._decompress(f)
 
     def changegroupsubset(self, bases, heads, kind):
         self.requirecap('changegroupsubset', _('look up remote changes'))
         bases = " ".join(map(hex, bases))
         heads = " ".join(map(hex, heads))
         return self._decompress(self._callstream("changegroupsubset",
                                                  bases=bases, heads=heads))
 
+    def unbundle(self, cg, heads, source):
+        '''Send cg (a readable file-like object representing the
+        changegroup to push, typically a chunkbuffer object) to the
+        remote server as a bundle. Return an integer indicating the
+        result of the push (see localrepository.addchangegroup()).'''
+
+        ret, output = self._callpush("unbundle", cg, heads=' '.join(map(hex, heads)))
+        if ret == "":
+            raise error.ResponseError(
+                _('push failed:'), output)
+        try:
+            ret = int(ret)
+        except ValueError, err:
+            raise error.ResponseError(
+                _('push failed (unexpected response):'), ret)
+
+        for l in output.splitlines(True):
+            self.ui.status(_('remote: '), l)
+        return ret
+
 # server side
 
 def dispatch(repo, proto, command):
     if command not in commands:
         return False
     func, spec = commands[command]
     args = proto.getargs(spec)
     r = func(repo, proto, *args)
     if r != None:
         proto.respond(r)
     return True
 
 def between(repo, proto, pairs):
     pairs = [map(bin, p.split("-")) for p in pairs.split(" ")]
     r = []
     for b in repo.between(pairs):
         r.append(" ".join(map(hex, b)) + "\n")
     return "".join(r)
 
 def branchmap(repo, proto):
     branchmap = repo.branchmap()
     heads = []
     for branch, nodes in branchmap.iteritems():
         branchname = urllib.quote(branch)
         branchnodes = [hex(node) for node in nodes]
         heads.append('%s %s' % (branchname, ' '.join(branchnodes)))
     return '\n'.join(heads)
 
 def branches(repo, proto, nodes):
     nodes = map(bin, nodes.split(" "))
     r = []
     for b in repo.branches(nodes):
         r.append(" ".join(map(hex, b)) + "\n")
     return "".join(r)
 
 def changegroup(repo, proto, roots):
     nodes = map(bin, roots.split(" "))
     cg = repo.changegroup(nodes, 'serve')
     proto.sendchangegroup(cg)
 
 def changegroupsubset(repo, proto, bases, heads):
     bases = [bin(n) for n in bases.split(' ')]
     heads = [bin(n) for n in heads.split(' ')]
     cg = repo.changegroupsubset(bases, heads, 'serve')
     proto.sendchangegroup(cg)
 
 def heads(repo, proto):
     h = repo.heads()
     return " ".join(map(hex, h)) + "\n"
 
 def listkeys(repo, proto, namespace):
     d = pushkey_.list(repo, namespace).items()
     t = '\n'.join(['%s\t%s' % (k.encode('string-escape'),
                                v.encode('string-escape')) for k, v in d])
     return t
 
 def lookup(repo, proto, key):
     try:
         r = hex(repo.lookup(key))
         success = 1
     except Exception, inst:
         r = str(inst)
         success = 0
     return "%s %s\n" % (success, r)
 
 def pushkey(repo, proto, namespace, key, old, new):
     r = pushkey_.push(repo, namespace, key, old, new)
     return '%s\n' % int(r)
 
 def stream(repo, proto):
     try:
         proto.sendstream(streamclone.stream_out(repo))
     except streamclone.StreamException, inst:
         return str(inst)
 
 commands = {
     'between': (between, 'pairs'),
     'branchmap': (branchmap, ''),
     'branches': (branches, 'nodes'),
     'changegroup': (changegroup, 'roots'),
     'changegroupsubset': (changegroupsubset, 'bases heads'),
     'heads': (heads, ''),
     'listkeys': (listkeys, 'namespace'),
     'lookup': (lookup, 'key'),
     'pushkey': (pushkey, 'namespace key old new'),
     'stream_out': (stream, ''),
 }