This is an archive of the discontinued Mercurial Phabricator instance.

Differential D338

wireproto: use new peer interface
ClosedPublic

Authored by indygreg on Aug 11 2017, 12:20 AM.

Details

Reviewers
durin42
Group Reviewers
hg-reviewers
Commits
rHGdedab036215d: wireproto: use new peer interface
Summary

The wirepeer class provides concrete implementations of peer interface
methods for calling wire protocol commands. It makes sense for this
class to inherit from the peer abstract base class. So we change
that.

Since httppeer and sshpeer have already been converted to the new
interface, peerrepository is no longer adding any value. So it has
been removed. httppeer and sshpeer have been updated to reflect the
loss of peerrepository and the inheritance of the abstract base
class in wirepeer.

The code changes in wirepeer are reordering of methods to group
by interface.

Some Python code in tests was updated to reflect changed APIs.

.. api::

peer.peerrepository has been removed. Use repository.peer abstract
base class to represent a peer repository.

Diff Detail

Repository
rHG Mercurial
Lint
Lint Skipped
Unit
Unit Tests Skipped

Event Timeline

indygreg created this revision.Aug 11 2017, 12:20 AM
durin42 accepted this revision.Aug 11 2017, 2:32 PM
This revision is now accepted and ready to land.Aug 11 2017, 2:32 PM
Closed by commit rHGdedab036215d: wireproto: use new peer interface (authored by indygreg). · Explain WhyAug 15 2017, 1:51 PM
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

PathPackages
Mmercurial/httppeer.py (6 lines)
Mmercurial/peer.py (43 lines)
Mmercurial/sshpeer.py (6 lines)
Mmercurial/wireproto.py (145 lines)
Mtests/notcapable (4 lines)
Mtests/test-wireproto.py (21 lines)

Diff 763

mercurial/httppeer.py

from .i18n import _ from .i18n import _
from .node import nullid from .node import nullid
from . import ( from . import (
bundle2, bundle2,
error, error,
httpconnection, httpconnection,
pycompat, pycompat,
repository,
statichttprepo, statichttprepo,
url, url,
util, util,
wireproto, wireproto,
) )
httplib = util.httplib httplib = util.httplib
urlerr = util.urlerr urlerr = util.urlerr
raise error.PeerTransportError( raise error.PeerTransportError(
_('HTTP request error (%s)') % e, _('HTTP request error (%s)') % e,
hint=_('this may be an intermittent network failure; ' hint=_('this may be an intermittent network failure; '
'if the error persists, consider contacting the ' 'if the error persists, consider contacting the '
'network or server operator')) 'network or server operator'))
resp.__class__ = readerproxy resp.__class__ = readerproxy
class httppeer(wireproto.wirepeer, repository.legacypeer): class httppeer(wireproto.wirepeer):
def __init__(self, ui, path): def __init__(self, ui, path):
self._path = path self._path = path
self._caps = None self._caps = None
self._urlopener = None self._urlopener = None
self._requestbuilder = None self._requestbuilder = None
u = util.url(path) u = util.url(path)
if u.query or u.fragment: if u.query or u.fragment:
raise error.Abort(_('unsupported URL component: "%s"') % raise error.Abort(_('unsupported URL component: "%s"') %
(u.query or u.fragment)) (u.query or u.fragment))
# urllib cannot handle URLs with embedded user or passwd # urllib cannot handle URLs with embedded user or passwd
self._url, authinfo = u.authinfo() self._url, authinfo = u.authinfo()
self._ui = ui self._ui = ui
ui.debug('using %s\n' % self._url) ui.debug('using %s\n' % self._url)
self._urlopener = url.opener(ui, authinfo) self._urlopener = url.opener(ui, authinfo)
self._requestbuilder = urlreq.request self._requestbuilder = urlreq.request
# TODO remove once peerrepository isn't in inheritance.
self._capabilities = self.capabilities
def __del__(self): def __del__(self):
urlopener = getattr(self, '_urlopener', None) urlopener = getattr(self, '_urlopener', None)
if urlopener: if urlopener:
for h in urlopener.handlers: for h in urlopener.handlers:
h.close() h.close()
getattr(h, "close_all", lambda : None)() getattr(h, "close_all", lambda : None)()
# Begin of _basepeer interface. # Begin of _basepeer interface.

mercurial/peer.py

encargsorres, encresref = next(batchable) encargsorres, encresref = next(batchable)
if not encresref: if not encresref:
return encargsorres # a local result in this case return encargsorres # a local result in this case
self = args[0] self = args[0]
encresref.set(self._submitone(f.func_name, encargsorres)) encresref.set(self._submitone(f.func_name, encargsorres))
return next(batchable) return next(batchable)
setattr(plain, 'batchable', f) setattr(plain, 'batchable', f)
return plain return plain
class peerrepository(object):
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.
if boolean capability, return True.
if string capability, return string.'''
caps = self._capabilities()
if name in caps:
return True
name_eq = name + '='
for cap in caps:
if cap.startswith(name_eq):
return cap[len(name_eq):]
return False
def requirecap(self, name, purpose):
'''raise an exception if the given capability is not present'''
if not self.capable(name):
raise error.CapabilityError(
_('cannot %s; remote repository does not '
'support the %r capability') % (purpose, name))
def local(self):
'''return peer as a localrepo, or None'''
return None
def peer(self):
return self
def canpush(self):
return True
def close(self):
pass

mercurial/sshpeer.py

# sshpeer.py - ssh repository proxy class for mercurial # sshpeer.py - ssh repository proxy class for mercurial
# #
# Copyright 2005, 2006 Matt Mackall <mpm@selenic.com> # Copyright 2005, 2006 Matt Mackall <mpm@selenic.com>
# #
# This software may be used and distributed according to the terms of the # This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version. # GNU General Public License version 2 or any later version.
from __future__ import absolute_import from __future__ import absolute_import
import re import re
from .i18n import _ from .i18n import _
from . import ( from . import (
error, error,
pycompat, pycompat,
repository,
util, util,
wireproto, wireproto,
) )
def _serverquote(s): def _serverquote(s):
if not s: if not s:
return s return s
'''quote a string for the remote shell ... which we assume is sh''' '''quote a string for the remote shell ... which we assume is sh'''
return meth(data) return meth(data)
def close(self): def close(self):
return self._main.close() return self._main.close()
def flush(self): def flush(self):
return self._main.flush() return self._main.flush()
class sshpeer(wireproto.wirepeer, repository.legacypeer): class sshpeer(wireproto.wirepeer):
def __init__(self, ui, path, create=False): def __init__(self, ui, path, create=False):
self._url = path self._url = path
self._ui = ui self._ui = ui
self._pipeo = self._pipei = self._pipee = None self._pipeo = self._pipei = self._pipee = None
u = util.url(path, parsequery=False, parsefragment=False) u = util.url(path, parsequery=False, parsefragment=False)
if u.scheme != 'ssh' or not u.host or u.path is None: if u.scheme != 'ssh' or not u.host or u.path is None:
self._abort(error.RepoError(_("couldn't parse location %s") % path)) self._abort(error.RepoError(_("couldn't parse location %s") % path))
(_serverquote(remotecmd), _serverquote(self._path)))) (_serverquote(remotecmd), _serverquote(self._path))))
ui.debug('running %s\n' % cmd) ui.debug('running %s\n' % cmd)
res = ui.system(cmd, blockedtag='sshpeer') res = ui.system(cmd, blockedtag='sshpeer')
if res != 0: if res != 0:
self._abort(error.RepoError(_("could not create remote repo"))) self._abort(error.RepoError(_("could not create remote repo")))
self._validaterepo(sshcmd, args, remotecmd) self._validaterepo(sshcmd, args, remotecmd)
# TODO remove this alias once peerrepository inheritance is removed.
self._capabilities = self.capabilities
# Begin of _basepeer interface. # Begin of _basepeer interface.
@util.propertycache @util.propertycache
def ui(self): def ui(self):
return self._ui return self._ui
def url(self): def url(self):
return self._url return self._url

mercurial/wireproto.py

bundle2, bundle2,
changegroup as changegroupmod, changegroup as changegroupmod,
encoding, encoding,
error, error,
exchange, exchange,
peer, peer,
pushkey as pushkeymod, pushkey as pushkeymod,
pycompat, pycompat,
repository,
streamclone, streamclone,
util, util,
) )
urlerr = util.urlerr urlerr = util.urlerr
urlreq = util.urlreq urlreq = util.urlreq
bundle2requiredmain = _('incompatible Mercurial client; bundle2 required') bundle2requiredmain = _('incompatible Mercurial client; bundle2 required')
'obsmarkers': 'boolean', 'obsmarkers': 'boolean',
'bundlecaps': 'scsv', 'bundlecaps': 'scsv',
'listkeys': 'csv', 'listkeys': 'csv',
'cg': 'boolean', 'cg': 'boolean',
'cbattempted': 'boolean'} 'cbattempted': 'boolean'}
# client side # client side
class wirepeer(peer.peerrepository): class wirepeer(repository.legacypeer):
"""Client-side interface for communicating with a peer repository. """Client-side interface for communicating with a peer repository.
Methods commonly call wire protocol commands of the same name. Methods commonly call wire protocol commands of the same name.
See also httppeer.py and sshpeer.py for protocol-specific See also httppeer.py and sshpeer.py for protocol-specific
implementations of this interface. implementations of this interface.
""" """
def _submitbatch(self, req): # Begin of basewirepeer interface.
"""run batch request <req> on the server
Returns an iterator of the raw responses from the server.
"""
rsp = self._callstream("batch", cmds=encodebatchcmds(req))
chunk = rsp.read(1024)
work = [chunk]
while chunk:
while ';' not in chunk and chunk:
chunk = rsp.read(1024)
work.append(chunk)
merged = ''.join(work)
while ';' in merged:
one, merged = merged.split(';', 1)
yield unescapearg(one)
chunk = rsp.read(1024)
work = [merged, chunk]
yield unescapearg(''.join(work))
def _submitone(self, op, args):
return self._call(op, **args)
def iterbatch(self): def iterbatch(self):
return remoteiterbatcher(self) return remoteiterbatcher(self)
@batchable @batchable
def lookup(self, key): def lookup(self, key):
self.requirecap('lookup', _('look up remote revision')) self.requirecap('lookup', _('look up remote revision'))
f = future() f = future()
branchname, branchheads = branchpart.split(' ', 1) branchname, branchheads = branchpart.split(' ', 1)
branchname = encoding.tolocal(urlreq.unquote(branchname)) branchname = encoding.tolocal(urlreq.unquote(branchname))
branchheads = decodelist(branchheads) branchheads = decodelist(branchheads)
branchmap[branchname] = branchheads branchmap[branchname] = branchheads
yield branchmap yield branchmap
except TypeError: except TypeError:
self._abort(error.ResponseError(_("unexpected response:"), d)) self._abort(error.ResponseError(_("unexpected response:"), d))
def branches(self, nodes): @batchable
n = encodelist(nodes) def listkeys(self, namespace):
d = self._call("branches", nodes=n) if not self.capable('pushkey'):
try: yield {}, None
br = [tuple(decodelist(b)) for b in d.splitlines()] f = future()
return br self.ui.debug('preparing listkeys for "%s"\n' % namespace)
except ValueError: yield {'namespace': encoding.fromlocal(namespace)}, f
self._abort(error.ResponseError(_("unexpected response:"), d)) d = f.value
self.ui.debug('received listkey for "%s": %i bytes\n'
def between(self, pairs): % (namespace, len(d)))
batch = 8 # avoid giant requests yield pushkeymod.decodekeys(d)
r = []
for i in xrange(0, len(pairs), batch):
n = " ".join([encodelist(p, '-') for p in pairs[i:i + batch]])
d = self._call("between", pairs=n)
try:
r.extend(l and decodelist(l) or [] for l in d.splitlines())
except ValueError:
self._abort(error.ResponseError(_("unexpected response:"), d))
return r
@batchable @batchable
def pushkey(self, namespace, key, old, new): def pushkey(self, namespace, key, old, new):
if not self.capable('pushkey'): if not self.capable('pushkey'):
yield False, None yield False, None
f = future() f = future()
self.ui.debug('preparing pushkey for "%s:%s"\n' % (namespace, key)) self.ui.debug('preparing pushkey for "%s:%s"\n' % (namespace, key))
yield {'namespace': encoding.fromlocal(namespace), yield {'namespace': encoding.fromlocal(namespace),
'key': encoding.fromlocal(key), 'key': encoding.fromlocal(key),
'old': encoding.fromlocal(old), 'old': encoding.fromlocal(old),
'new': encoding.fromlocal(new)}, f 'new': encoding.fromlocal(new)}, f
d = f.value d = f.value
d, output = d.split('\n', 1) d, output = d.split('\n', 1)
try: try:
d = bool(int(d)) d = bool(int(d))
except ValueError: except ValueError:
raise error.ResponseError( raise error.ResponseError(
_('push failed (unexpected response):'), d) _('push failed (unexpected response):'), d)
for l in output.splitlines(True): for l in output.splitlines(True):
self.ui.status(_('remote: '), l) self.ui.status(_('remote: '), l)
yield d yield d
@batchable
def listkeys(self, namespace):
if not self.capable('pushkey'):
yield {}, None
f = future()
self.ui.debug('preparing listkeys for "%s"\n' % namespace)
yield {'namespace': encoding.fromlocal(namespace)}, f
d = f.value
self.ui.debug('received listkey for "%s": %i bytes\n'
% (namespace, len(d)))
yield pushkeymod.decodekeys(d)
def stream_out(self): def stream_out(self):
return self._callstream('stream_out') return self._callstream('stream_out')
def changegroup(self, nodes, kind):
n = encodelist(nodes)
f = self._callcompressable("changegroup", roots=n)
return changegroupmod.cg1unpacker(f, 'UN')
def changegroupsubset(self, bases, heads, kind):
self.requirecap('changegroupsubset', _('look up remote changes'))
bases = encodelist(bases)
heads = encodelist(heads)
f = self._callcompressable("changegroupsubset",
bases=bases, heads=heads)
return changegroupmod.cg1unpacker(f, 'UN')
def getbundle(self, source, **kwargs): def getbundle(self, source, **kwargs):
self.requirecap('getbundle', _('look up remote changes')) self.requirecap('getbundle', _('look up remote changes'))
opts = {} opts = {}
bundlecaps = kwargs.get('bundlecaps') bundlecaps = kwargs.get('bundlecaps')
if bundlecaps is not None: if bundlecaps is not None:
kwargs['bundlecaps'] = sorted(bundlecaps) kwargs['bundlecaps'] = sorted(bundlecaps)
else: else:
bundlecaps = () # kwargs could have it to None bundlecaps = () # kwargs could have it to None
for l in output.splitlines(True): for l in output.splitlines(True):
self.ui.status(_('remote: '), l) self.ui.status(_('remote: '), l)
else: else:
# bundle2 push. Send a stream, fetch a stream. # bundle2 push. Send a stream, fetch a stream.
stream = self._calltwowaystream('unbundle', cg, heads=heads) stream = self._calltwowaystream('unbundle', cg, heads=heads)
ret = bundle2.getunbundler(self.ui, stream) ret = bundle2.getunbundler(self.ui, stream)
return ret return ret
# End of basewirepeer interface.
# Begin of baselegacywirepeer interface.
def branches(self, nodes):
n = encodelist(nodes)
d = self._call("branches", nodes=n)
try:
br = [tuple(decodelist(b)) for b in d.splitlines()]
return br
except ValueError:
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([encodelist(p, '-') for p in pairs[i:i + batch]])
d = self._call("between", pairs=n)
try:
r.extend(l and decodelist(l) or [] for l in d.splitlines())
except ValueError:
self._abort(error.ResponseError(_("unexpected response:"), d))
return r
def changegroup(self, nodes, kind):
n = encodelist(nodes)
f = self._callcompressable("changegroup", roots=n)
return changegroupmod.cg1unpacker(f, 'UN')
def changegroupsubset(self, bases, heads, kind):
self.requirecap('changegroupsubset', _('look up remote changes'))
bases = encodelist(bases)
heads = encodelist(heads)
f = self._callcompressable("changegroupsubset",
bases=bases, heads=heads)
return changegroupmod.cg1unpacker(f, 'UN')
# End of baselegacywirepeer interface.
def _submitbatch(self, req):
"""run batch request <req> on the server
Returns an iterator of the raw responses from the server.
"""
rsp = self._callstream("batch", cmds=encodebatchcmds(req))
chunk = rsp.read(1024)
work = [chunk]
while chunk:
while ';' not in chunk and chunk:
chunk = rsp.read(1024)
work.append(chunk)
merged = ''.join(work)
while ';' in merged:
one, merged = merged.split(';', 1)
yield unescapearg(one)
chunk = rsp.read(1024)
work = [merged, chunk]
yield unescapearg(''.join(work))
def _submitone(self, op, args):
return self._call(op, **args)
def debugwireargs(self, one, two, three=None, four=None, five=None): def debugwireargs(self, one, two, three=None, four=None, five=None):
# don't pass optional arguments left at their default value # don't pass optional arguments left at their default value
opts = {} opts = {}
if three is not None: if three is not None:
opts['three'] = three opts['three'] = three
if four is not None: if four is not None:
opts['four'] = four opts['four'] = four
return self._call('debugwireargs', one=one, two=two, **opts) return self._call('debugwireargs', one=one, two=two, **opts)

tests/notcapable

# Disable the $CAP wire protocol capability. # Disable the $CAP wire protocol capability.
if test -z "$CAP" if test -z "$CAP"
then then
echo "CAP environment variable not set." echo "CAP environment variable not set."
fi fi
cat > notcapable-$CAP.py << EOF cat > notcapable-$CAP.py << EOF
from mercurial import extensions, peer, localrepo from mercurial import extensions, localrepo, repository
def extsetup(): def extsetup():
extensions.wrapfunction(peer.peerrepository, 'capable', wrapcapable) extensions.wrapfunction(repository.peer, 'capable', wrapcapable)
extensions.wrapfunction(localrepo.localrepository, 'peer', wrappeer) extensions.wrapfunction(localrepo.localrepository, 'peer', wrappeer)
def wrapcapable(orig, self, name, *args, **kwargs): def wrapcapable(orig, self, name, *args, **kwargs):
if name in '$CAP'.split(' '): if name in '$CAP'.split(' '):
return False return False
return orig(self, name, *args, **kwargs) return orig(self, name, *args, **kwargs)
def wrappeer(orig, self): def wrappeer(orig, self):
# Since we're disabling some newer features, we need to make sure local # Since we're disabling some newer features, we need to make sure local
# repos add in the legacy features again. # repos add in the legacy features again.
return localrepo.locallegacypeer(self) return localrepo.locallegacypeer(self)
EOF EOF
echo '[extensions]' >> $HGRCPATH echo '[extensions]' >> $HGRCPATH
echo "notcapable-$CAP = `pwd`/notcapable-$CAP.py" >> $HGRCPATH echo "notcapable-$CAP = `pwd`/notcapable-$CAP.py" >> $HGRCPATH

tests/test-wireproto.py

args.setdefault('*', {}) args.setdefault('*', {})
names = spec.split() names = spec.split()
return [args[n] for n in names] return [args[n] for n in names]
class clientpeer(wireproto.wirepeer): class clientpeer(wireproto.wirepeer):
def __init__(self, serverrepo): def __init__(self, serverrepo):
self.serverrepo = serverrepo self.serverrepo = serverrepo
def _capabilities(self): @property
def ui(self):
return self.serverrepo.ui
def url(self):
return 'test'
def local(self):
return None
def peer(self):
return self
def canpush(self):
return True
def close(self):
pass
def capabilities(self):
return ['batch'] return ['batch']
def _call(self, cmd, **args): def _call(self, cmd, **args):
return wireproto.dispatch(self.serverrepo, proto(args), cmd) return wireproto.dispatch(self.serverrepo, proto(args), cmd)
def _callstream(self, cmd, **args): def _callstream(self, cmd, **args):
return stringio(self._call(cmd, **args)) return stringio(self._call(cmd, **args))