wireproto: move version 1 peer functionality to standalone module (API)

wireproto.py contains code for both the client and the server. There
*should* be a somewhat strong separation between the two.

This commit extracts the client-side code from wireproto.py into a new
module - wireprotov1peer.

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

hgext/infinitepush/__init__.py

@@ -1,1187 +1,1188 b''
 # Infinite push
 #
 # Copyright 2016 Facebook, Inc.
 #
 # This software may be used and distributed according to the terms of the
 # GNU General Public License version 2 or any later version.
 """ store some pushes in a remote blob store on the server (EXPERIMENTAL)
 
     [infinitepush]
     # Server-side and client-side option. Pattern of the infinitepush bookmark
     branchpattern = PATTERN
 
     # Server or client
     server = False
 
     # Server-side option. Possible values: 'disk' or 'sql'. Fails if not set
     indextype = disk
 
     # Server-side option. Used only if indextype=sql.
     # Format: 'IP:PORT:DB_NAME:USER:PASSWORD'
     sqlhost = IP:PORT:DB_NAME:USER:PASSWORD
 
     # Server-side option. Used only if indextype=disk.
     # Filesystem path to the index store
     indexpath = PATH
 
     # Server-side option. Possible values: 'disk' or 'external'
     # Fails if not set
     storetype = disk
 
     # Server-side option.
     # Path to the binary that will save bundle to the bundlestore
     # Formatted cmd line will be passed to it (see `put_args`)
     put_binary = put
 
     # Serser-side option. Used only if storetype=external.
     # Format cmd-line string for put binary. Placeholder: {filename}
     put_args = {filename}
 
     # Server-side option.
     # Path to the binary that get bundle from the bundlestore.
     # Formatted cmd line will be passed to it (see `get_args`)
     get_binary = get
 
     # Serser-side option. Used only if storetype=external.
     # Format cmd-line string for get binary. Placeholders: {filename} {handle}
     get_args = {filename} {handle}
 
     # Server-side option
     logfile = FIlE
 
     # Server-side option
     loglevel = DEBUG
 
     # Server-side option. Used only if indextype=sql.
     # Sets mysql wait_timeout option.
     waittimeout = 300
 
     # Server-side option. Used only if indextype=sql.
     # Sets mysql innodb_lock_wait_timeout option.
     locktimeout = 120
 
     # Server-side option. Used only if indextype=sql.
     # Name of the repository
     reponame = ''
 
     # Client-side option. Used by --list-remote option. List of remote scratch
     # patterns to list if no patterns are specified.
     defaultremotepatterns = ['*']
 
     # Instructs infinitepush to forward all received bundle2 parts to the
     # bundle for storage. Defaults to False.
     storeallparts = True
 
     # routes each incoming push to the bundlestore. defaults to False
     pushtobundlestore = True
 
     [remotenames]
     # Client-side option
     # This option should be set only if remotenames extension is enabled.
     # Whether remote bookmarks are tracked by remotenames extension.
     bookmarks = True
 """
 
 from __future__ import absolute_import
 
 import collections
 import contextlib
 import errno
 import functools
 import logging
 import os
 import random
 import re
 import socket
 import subprocess
 import tempfile
 import time
 
 from mercurial.node import (
     bin,
     hex,
 )
 
 from mercurial.i18n import _
 
 from mercurial.utils import (
     procutil,
     stringutil,
 )
 
 from mercurial import (
     bundle2,
     changegroup,
     commands,
     discovery,
     encoding,
     error,
     exchange,
     extensions,
     hg,
     localrepo,
     peer,
     phases,
     pushkey,
     pycompat,
     registrar,
     util,
     wireproto,
     wireprototypes,
+    wireprotov1peer,
 )
 
 from . import (
     bundleparts,
     common,
 )
 
 # Note for extension authors: ONLY specify testedwith = 'ships-with-hg-core' for
 # extensions which SHIP WITH MERCURIAL. Non-mainline extensions should
 # be specifying the version(s) of Mercurial they are tested with, or
 # leave the attribute unspecified.
 testedwith = 'ships-with-hg-core'
 
 configtable = {}
 configitem = registrar.configitem(configtable)
 
 configitem('infinitepush', 'server',
     default=False,
 )
 configitem('infinitepush', 'storetype',
     default='',
 )
 configitem('infinitepush', 'indextype',
     default='',
 )
 configitem('infinitepush', 'indexpath',
     default='',
 )
 configitem('infinitepush', 'storeallparts',
     default=False,
 )
 configitem('infinitepush', 'reponame',
     default='',
 )
 configitem('scratchbranch', 'storepath',
     default='',
 )
 configitem('infinitepush', 'branchpattern',
     default='',
 )
 configitem('infinitepush', 'pushtobundlestore',
     default=False,
 )
 configitem('experimental', 'server-bundlestore-bookmark',
     default='',
 )
 configitem('experimental', 'infinitepush-scratchpush',
     default=False,
 )
 
 experimental = 'experimental'
 configbookmark = 'server-bundlestore-bookmark'
 configscratchpush = 'infinitepush-scratchpush'
 
 scratchbranchparttype = bundleparts.scratchbranchparttype
 revsetpredicate = registrar.revsetpredicate()
 templatekeyword = registrar.templatekeyword()
 _scratchbranchmatcher = lambda x: False
 _maybehash = re.compile(r'^[a-f0-9]+$').search
 
 def _buildexternalbundlestore(ui):
     put_args = ui.configlist('infinitepush', 'put_args', [])
     put_binary = ui.config('infinitepush', 'put_binary')
     if not put_binary:
         raise error.Abort('put binary is not specified')
     get_args = ui.configlist('infinitepush', 'get_args', [])
     get_binary = ui.config('infinitepush', 'get_binary')
     if not get_binary:
         raise error.Abort('get binary is not specified')
     from . import store
     return store.externalbundlestore(put_binary, put_args, get_binary, get_args)
 
 def _buildsqlindex(ui):
     sqlhost = ui.config('infinitepush', 'sqlhost')
     if not sqlhost:
         raise error.Abort(_('please set infinitepush.sqlhost'))
     host, port, db, user, password = sqlhost.split(':')
     reponame = ui.config('infinitepush', 'reponame')
     if not reponame:
         raise error.Abort(_('please set infinitepush.reponame'))
 
     logfile = ui.config('infinitepush', 'logfile', '')
     waittimeout = ui.configint('infinitepush', 'waittimeout', 300)
     locktimeout = ui.configint('infinitepush', 'locktimeout', 120)
     from . import sqlindexapi
     return sqlindexapi.sqlindexapi(
         reponame, host, port, db, user, password,
         logfile, _getloglevel(ui), waittimeout=waittimeout,
         locktimeout=locktimeout)
 
 def _getloglevel(ui):
     loglevel = ui.config('infinitepush', 'loglevel', 'DEBUG')
     numeric_loglevel = getattr(logging, loglevel.upper(), None)
     if not isinstance(numeric_loglevel, int):
         raise error.Abort(_('invalid log level %s') % loglevel)
     return numeric_loglevel
 
 def _tryhoist(ui, remotebookmark):
     '''returns a bookmarks with hoisted part removed
 
     Remotenames extension has a 'hoist' config that allows to use remote
     bookmarks without specifying remote path. For example, 'hg update master'
     works as well as 'hg update remote/master'. We want to allow the same in
     infinitepush.
     '''
 
     if common.isremotebooksenabled(ui):
         hoist = ui.config('remotenames', 'hoistedpeer') + '/'
         if remotebookmark.startswith(hoist):
             return remotebookmark[len(hoist):]
     return remotebookmark
 
 class bundlestore(object):
     def __init__(self, repo):
         self._repo = repo
         storetype = self._repo.ui.config('infinitepush', 'storetype')
         if storetype == 'disk':
             from . import store
             self.store = store.filebundlestore(self._repo.ui, self._repo)
         elif storetype == 'external':
             self.store = _buildexternalbundlestore(self._repo.ui)
         else:
             raise error.Abort(
                 _('unknown infinitepush store type specified %s') % storetype)
 
         indextype = self._repo.ui.config('infinitepush', 'indextype')
         if indextype == 'disk':
             from . import fileindexapi
             self.index = fileindexapi.fileindexapi(self._repo)
         elif indextype == 'sql':
             self.index = _buildsqlindex(self._repo.ui)
         else:
             raise error.Abort(
                 _('unknown infinitepush index type specified %s') % indextype)
 
 def _isserver(ui):
     return ui.configbool('infinitepush', 'server')
 
 def reposetup(ui, repo):
     if _isserver(ui) and repo.local():
         repo.bundlestore = bundlestore(repo)
 
 def extsetup(ui):
     commonsetup(ui)
     if _isserver(ui):
         serverextsetup(ui)
     else:
         clientextsetup(ui)
 
 def commonsetup(ui):
     wireproto.commands['listkeyspatterns'] = (
         wireprotolistkeyspatterns, 'namespace patterns')
     scratchbranchpat = ui.config('infinitepush', 'branchpattern')
     if scratchbranchpat:
         global _scratchbranchmatcher
         kind, pat, _scratchbranchmatcher = \
                 stringutil.stringmatcher(scratchbranchpat)
 
 def serverextsetup(ui):
     origpushkeyhandler = bundle2.parthandlermapping['pushkey']
 
     def newpushkeyhandler(*args, **kwargs):
         bundle2pushkey(origpushkeyhandler, *args, **kwargs)
     newpushkeyhandler.params = origpushkeyhandler.params
     bundle2.parthandlermapping['pushkey'] = newpushkeyhandler
 
     orighandlephasehandler = bundle2.parthandlermapping['phase-heads']
     newphaseheadshandler = lambda *args, **kwargs: \
         bundle2handlephases(orighandlephasehandler, *args, **kwargs)
     newphaseheadshandler.params = orighandlephasehandler.params
     bundle2.parthandlermapping['phase-heads'] = newphaseheadshandler
 
     extensions.wrapfunction(localrepo.localrepository, 'listkeys',
                             localrepolistkeys)
     wireproto.commands['lookup'] = (
         _lookupwrap(wireproto.commands['lookup'][0]), 'key')
     extensions.wrapfunction(exchange, 'getbundlechunks', getbundlechunks)
 
     extensions.wrapfunction(bundle2, 'processparts', processparts)
 
 def clientextsetup(ui):
     entry = extensions.wrapcommand(commands.table, 'push', _push)
 
     entry[1].append(
         ('', 'bundle-store', None,
          _('force push to go to bundle store (EXPERIMENTAL)')))
 
     extensions.wrapcommand(commands.table, 'pull', _pull)
 
     extensions.wrapfunction(discovery, 'checkheads', _checkheads)
 
-    wireproto.wirepeer.listkeyspatterns = listkeyspatterns
+    wireprotov1peer.wirepeer.listkeyspatterns = listkeyspatterns
 
     partorder = exchange.b2partsgenorder
     index = partorder.index('changeset')
     partorder.insert(
         index, partorder.pop(partorder.index(scratchbranchparttype)))
 
 def _checkheads(orig, pushop):
     if pushop.ui.configbool(experimental, configscratchpush, False):
         return
     return orig(pushop)
 
 def wireprotolistkeyspatterns(repo, proto, namespace, patterns):
     patterns = wireprototypes.decodelist(patterns)
     d = repo.listkeys(encoding.tolocal(namespace), patterns).iteritems()
     return pushkey.encodekeys(d)
 
 def localrepolistkeys(orig, self, namespace, patterns=None):
     if namespace == 'bookmarks' and patterns:
         index = self.bundlestore.index
         results = {}
         bookmarks = orig(self, namespace)
         for pattern in patterns:
             results.update(index.getbookmarks(pattern))
             if pattern.endswith('*'):
                 pattern = 're:^' + pattern[:-1] + '.*'
             kind, pat, matcher = stringutil.stringmatcher(pattern)
             for bookmark, node in bookmarks.iteritems():
                 if matcher(bookmark):
                     results[bookmark] = node
         return results
     else:
         return orig(self, namespace)
 
 @peer.batchable
 def listkeyspatterns(self, namespace, patterns):
     if not self.capable('pushkey'):
         yield {}, None
     f = peer.future()
     self.ui.debug('preparing listkeys for "%s" with pattern "%s"\n' %
                   (namespace, patterns))
     yield {
         'namespace': encoding.fromlocal(namespace),
         'patterns': wireprototypes.encodelist(patterns)
     }, f
     d = f.value
     self.ui.debug('received listkey for "%s": %i bytes\n'
                   % (namespace, len(d)))
     yield pushkey.decodekeys(d)
 
 def _readbundlerevs(bundlerepo):
     return list(bundlerepo.revs('bundle()'))
 
 def _includefilelogstobundle(bundlecaps, bundlerepo, bundlerevs, ui):
     '''Tells remotefilelog to include all changed files to the changegroup
 
     By default remotefilelog doesn't include file content to the changegroup.
     But we need to include it if we are fetching from bundlestore.
     '''
     changedfiles = set()
     cl = bundlerepo.changelog
     for r in bundlerevs:
         # [3] means changed files
         changedfiles.update(cl.read(r)[3])
     if not changedfiles:
         return bundlecaps
 
     changedfiles = '\0'.join(changedfiles)
     newcaps = []
     appended = False
     for cap in (bundlecaps or []):
         if cap.startswith('excludepattern='):
             newcaps.append('\0'.join((cap, changedfiles)))
             appended = True
         else:
             newcaps.append(cap)
     if not appended:
         # Not found excludepattern cap. Just append it
         newcaps.append('excludepattern=' + changedfiles)
 
     return newcaps
 
 def _rebundle(bundlerepo, bundleroots, unknownhead):
     '''
     Bundle may include more revision then user requested. For example,
     if user asks for revision but bundle also consists its descendants.
     This function will filter out all revision that user is not requested.
     '''
     parts = []
 
     version = '02'
     outgoing = discovery.outgoing(bundlerepo, commonheads=bundleroots,
                                   missingheads=[unknownhead])
     cgstream = changegroup.makestream(bundlerepo, outgoing, version, 'pull')
     cgstream = util.chunkbuffer(cgstream).read()
     cgpart = bundle2.bundlepart('changegroup', data=cgstream)
     cgpart.addparam('version', version)
     parts.append(cgpart)
 
     return parts
 
 def _getbundleroots(oldrepo, bundlerepo, bundlerevs):
     cl = bundlerepo.changelog
     bundleroots = []
     for rev in bundlerevs:
         node = cl.node(rev)
         parents = cl.parents(node)
         for parent in parents:
             # include all revs that exist in the main repo
             # to make sure that bundle may apply client-side
             if parent in oldrepo:
                 bundleroots.append(parent)
     return bundleroots
 
 def _needsrebundling(head, bundlerepo):
     bundleheads = list(bundlerepo.revs('heads(bundle())'))
     return not (len(bundleheads) == 1 and
                 bundlerepo[bundleheads[0]].node() == head)
 
 def _generateoutputparts(head, bundlerepo, bundleroots, bundlefile):
     '''generates bundle that will be send to the user
 
     returns tuple with raw bundle string and bundle type
     '''
     parts = []
     if not _needsrebundling(head, bundlerepo):
         with util.posixfile(bundlefile, "rb") as f:
             unbundler = exchange.readbundle(bundlerepo.ui, f, bundlefile)
             if isinstance(unbundler, changegroup.cg1unpacker):
                 part = bundle2.bundlepart('changegroup',
                                           data=unbundler._stream.read())
                 part.addparam('version', '01')
                 parts.append(part)
             elif isinstance(unbundler, bundle2.unbundle20):
                 haschangegroup = False
                 for part in unbundler.iterparts():
                     if part.type == 'changegroup':
                         haschangegroup = True
                     newpart = bundle2.bundlepart(part.type, data=part.read())
                     for key, value in part.params.iteritems():
                         newpart.addparam(key, value)
                     parts.append(newpart)
 
                 if not haschangegroup:
                     raise error.Abort(
                         'unexpected bundle without changegroup part, ' +
                         'head: %s' % hex(head),
                         hint='report to administrator')
             else:
                 raise error.Abort('unknown bundle type')
     else:
         parts = _rebundle(bundlerepo, bundleroots, head)
 
     return parts
 
 def getbundlechunks(orig, repo, source, heads=None, bundlecaps=None, **kwargs):
     heads = heads or []
     # newheads are parents of roots of scratch bundles that were requested
     newphases = {}
     scratchbundles = []
     newheads = []
     scratchheads = []
     nodestobundle = {}
     allbundlestocleanup = []
     try:
         for head in heads:
             if head not in repo.changelog.nodemap:
                 if head not in nodestobundle:
                     newbundlefile = common.downloadbundle(repo, head)
                     bundlepath = "bundle:%s+%s" % (repo.root, newbundlefile)
                     bundlerepo = hg.repository(repo.ui, bundlepath)
 
                     allbundlestocleanup.append((bundlerepo, newbundlefile))
                     bundlerevs = set(_readbundlerevs(bundlerepo))
                     bundlecaps = _includefilelogstobundle(
                         bundlecaps, bundlerepo, bundlerevs, repo.ui)
                     cl = bundlerepo.changelog
                     bundleroots = _getbundleroots(repo, bundlerepo, bundlerevs)
                     for rev in bundlerevs:
                         node = cl.node(rev)
                         newphases[hex(node)] = str(phases.draft)
                         nodestobundle[node] = (bundlerepo, bundleroots,
                                                newbundlefile)
 
                 scratchbundles.append(
                     _generateoutputparts(head, *nodestobundle[head]))
                 newheads.extend(bundleroots)
                 scratchheads.append(head)
     finally:
         for bundlerepo, bundlefile in allbundlestocleanup:
             bundlerepo.close()
             try:
                 os.unlink(bundlefile)
             except (IOError, OSError):
                 # if we can't cleanup the file then just ignore the error,
                 # no need to fail
                 pass
 
     pullfrombundlestore = bool(scratchbundles)
     wrappedchangegrouppart = False
     wrappedlistkeys = False
     oldchangegrouppart = exchange.getbundle2partsmapping['changegroup']
     try:
         def _changegrouppart(bundler, *args, **kwargs):
             # Order is important here. First add non-scratch part
             # and only then add parts with scratch bundles because
             # non-scratch part contains parents of roots of scratch bundles.
             result = oldchangegrouppart(bundler, *args, **kwargs)
             for bundle in scratchbundles:
                 for part in bundle:
                     bundler.addpart(part)
             return result
 
         exchange.getbundle2partsmapping['changegroup'] = _changegrouppart
         wrappedchangegrouppart = True
 
         def _listkeys(orig, self, namespace):
             origvalues = orig(self, namespace)
             if namespace == 'phases' and pullfrombundlestore:
                 if origvalues.get('publishing') == 'True':
                     # Make repo non-publishing to preserve draft phase
                     del origvalues['publishing']
                 origvalues.update(newphases)
             return origvalues
 
         extensions.wrapfunction(localrepo.localrepository, 'listkeys',
                                 _listkeys)
         wrappedlistkeys = True
         heads = list((set(newheads) | set(heads)) - set(scratchheads))
         result = orig(repo, source, heads=heads,
                       bundlecaps=bundlecaps, **kwargs)
     finally:
         if wrappedchangegrouppart:
             exchange.getbundle2partsmapping['changegroup'] = oldchangegrouppart
         if wrappedlistkeys:
             extensions.unwrapfunction(localrepo.localrepository, 'listkeys',
                                       _listkeys)
     return result
 
 def _lookupwrap(orig):
     def _lookup(repo, proto, key):
         localkey = encoding.tolocal(key)
 
         if isinstance(localkey, str) and _scratchbranchmatcher(localkey):
             scratchnode = repo.bundlestore.index.getnode(localkey)
             if scratchnode:
                 return "%s %s\n" % (1, scratchnode)
             else:
                 return "%s %s\n" % (0, 'scratch branch %s not found' % localkey)
         else:
             try:
                 r = hex(repo.lookup(localkey))
                 return "%s %s\n" % (1, r)
             except Exception as inst:
                 if repo.bundlestore.index.getbundle(localkey):
                     return "%s %s\n" % (1, localkey)
                 else:
                     r = str(inst)
                     return "%s %s\n" % (0, r)
     return _lookup
 
 def _pull(orig, ui, repo, source="default", **opts):
     opts = pycompat.byteskwargs(opts)
     # Copy paste from `pull` command
     source, branches = hg.parseurl(ui.expandpath(source), opts.get('branch'))
 
     scratchbookmarks = {}
     unfi = repo.unfiltered()
     unknownnodes = []
     for rev in opts.get('rev', []):
         if rev not in unfi:
             unknownnodes.append(rev)
     if opts.get('bookmark'):
         bookmarks = []
         revs = opts.get('rev') or []
         for bookmark in opts.get('bookmark'):
             if _scratchbranchmatcher(bookmark):
                 # rev is not known yet
                 # it will be fetched with listkeyspatterns next
                 scratchbookmarks[bookmark] = 'REVTOFETCH'
             else:
                 bookmarks.append(bookmark)
 
         if scratchbookmarks:
             other = hg.peer(repo, opts, source)
             fetchedbookmarks = other.listkeyspatterns(
                 'bookmarks', patterns=scratchbookmarks)
             for bookmark in scratchbookmarks:
                 if bookmark not in fetchedbookmarks:
                     raise error.Abort('remote bookmark %s not found!' %
                                       bookmark)
                 scratchbookmarks[bookmark] = fetchedbookmarks[bookmark]
                 revs.append(fetchedbookmarks[bookmark])
         opts['bookmark'] = bookmarks
         opts['rev'] = revs
 
     if scratchbookmarks or unknownnodes:
         # Set anyincoming to True
         extensions.wrapfunction(discovery, 'findcommonincoming',
                                 _findcommonincoming)
     try:
         # Remote scratch bookmarks will be deleted because remotenames doesn't
         # know about them. Let's save it before pull and restore after
         remotescratchbookmarks = _readscratchremotebookmarks(ui, repo, source)
         result = orig(ui, repo, source, **pycompat.strkwargs(opts))
         # TODO(stash): race condition is possible
         # if scratch bookmarks was updated right after orig.
         # But that's unlikely and shouldn't be harmful.
         if common.isremotebooksenabled(ui):
             remotescratchbookmarks.update(scratchbookmarks)
             _saveremotebookmarks(repo, remotescratchbookmarks, source)
         else:
             _savelocalbookmarks(repo, scratchbookmarks)
         return result
     finally:
         if scratchbookmarks:
             extensions.unwrapfunction(discovery, 'findcommonincoming')
 
 def _readscratchremotebookmarks(ui, repo, other):
     if common.isremotebooksenabled(ui):
         remotenamesext = extensions.find('remotenames')
         remotepath = remotenamesext.activepath(repo.ui, other)
         result = {}
         # Let's refresh remotenames to make sure we have it up to date
         # Seems that `repo.names['remotebookmarks']` may return stale bookmarks
         # and it results in deleting scratch bookmarks. Our best guess how to
         # fix it is to use `clearnames()`
         repo._remotenames.clearnames()
         for remotebookmark in repo.names['remotebookmarks'].listnames(repo):
             path, bookname = remotenamesext.splitremotename(remotebookmark)
             if path == remotepath and _scratchbranchmatcher(bookname):
                 nodes = repo.names['remotebookmarks'].nodes(repo,
                                                             remotebookmark)
                 if nodes:
                     result[bookname] = hex(nodes[0])
         return result
     else:
         return {}
 
 def _saveremotebookmarks(repo, newbookmarks, remote):
     remotenamesext = extensions.find('remotenames')
     remotepath = remotenamesext.activepath(repo.ui, remote)
     branches = collections.defaultdict(list)
     bookmarks = {}
     remotenames = remotenamesext.readremotenames(repo)
     for hexnode, nametype, remote, rname in remotenames:
         if remote != remotepath:
             continue
         if nametype == 'bookmarks':
             if rname in newbookmarks:
                 # It's possible if we have a normal bookmark that matches
                 # scratch branch pattern. In this case just use the current
                 # bookmark node
                 del newbookmarks[rname]
             bookmarks[rname] = hexnode
         elif nametype == 'branches':
             # saveremotenames expects 20 byte binary nodes for branches
             branches[rname].append(bin(hexnode))
 
     for bookmark, hexnode in newbookmarks.iteritems():
         bookmarks[bookmark] = hexnode
     remotenamesext.saveremotenames(repo, remotepath, branches, bookmarks)
 
 def _savelocalbookmarks(repo, bookmarks):
     if not bookmarks:
         return
     with repo.wlock(), repo.lock(), repo.transaction('bookmark') as tr:
         changes = []
         for scratchbook, node in bookmarks.iteritems():
             changectx = repo[node]
             changes.append((scratchbook, changectx.node()))
         repo._bookmarks.applychanges(repo, tr, changes)
 
 def _findcommonincoming(orig, *args, **kwargs):
     common, inc, remoteheads = orig(*args, **kwargs)
     return common, True, remoteheads
 
 def _push(orig, ui, repo, dest=None, *args, **opts):
 
     bookmark = opts.get(r'bookmark')
     # we only support pushing one infinitepush bookmark at once
     if len(bookmark) == 1:
         bookmark = bookmark[0]
     else:
         bookmark = ''
 
     oldphasemove = None
     overrides = {(experimental, configbookmark): bookmark}
 
     with ui.configoverride(overrides, 'infinitepush'):
         scratchpush = opts.get('bundle_store')
         if _scratchbranchmatcher(bookmark):
             scratchpush = True
             # bundle2 can be sent back after push (for example, bundle2
             # containing `pushkey` part to update bookmarks)
             ui.setconfig(experimental, 'bundle2.pushback', True)
 
         if scratchpush:
             # this is an infinitepush, we don't want the bookmark to be applied
             # rather that should be stored in the bundlestore
             opts[r'bookmark'] = []
             ui.setconfig(experimental, configscratchpush, True)
             oldphasemove = extensions.wrapfunction(exchange,
                                                    '_localphasemove',
                                                    _phasemove)
         # Copy-paste from `push` command
         path = ui.paths.getpath(dest, default=('default-push', 'default'))
         if not path:
             raise error.Abort(_('default repository not configured!'),
                              hint=_("see 'hg help config.paths'"))
         destpath = path.pushloc or path.loc
         # Remote scratch bookmarks will be deleted because remotenames doesn't
         # know about them. Let's save it before push and restore after
         remotescratchbookmarks = _readscratchremotebookmarks(ui, repo, destpath)
         result = orig(ui, repo, dest, *args, **opts)
         if common.isremotebooksenabled(ui):
             if bookmark and scratchpush:
                 other = hg.peer(repo, opts, destpath)
                 fetchedbookmarks = other.listkeyspatterns('bookmarks',
                                                           patterns=[bookmark])
                 remotescratchbookmarks.update(fetchedbookmarks)
             _saveremotebookmarks(repo, remotescratchbookmarks, destpath)
     if oldphasemove:
         exchange._localphasemove = oldphasemove
     return result
 
 def _deleteinfinitepushbookmarks(ui, repo, path, names):
     """Prune remote names by removing the bookmarks we don't want anymore,
     then writing the result back to disk
     """
     remotenamesext = extensions.find('remotenames')
 
     # remotename format is:
     # (node, nametype ("branches" or "bookmarks"), remote, name)
     nametype_idx = 1
     remote_idx = 2
     name_idx = 3
     remotenames = [remotename for remotename in \
                    remotenamesext.readremotenames(repo) \
                    if remotename[remote_idx] == path]
     remote_bm_names = [remotename[name_idx] for remotename in \
                        remotenames if remotename[nametype_idx] == "bookmarks"]
 
     for name in names:
         if name not in remote_bm_names:
             raise error.Abort(_("infinitepush bookmark '{}' does not exist "
                                 "in path '{}'").format(name, path))
 
     bookmarks = {}
     branches = collections.defaultdict(list)
     for node, nametype, remote, name in remotenames:
         if nametype == "bookmarks" and name not in names:
             bookmarks[name] = node
         elif nametype == "branches":
             # saveremotenames wants binary nodes for branches
             branches[name].append(bin(node))
 
     remotenamesext.saveremotenames(repo, path, branches, bookmarks)
 
 def _phasemove(orig, pushop, nodes, phase=phases.public):
     """prevent commits from being marked public
 
     Since these are going to a scratch branch, they aren't really being
     published."""
 
     if phase != phases.public:
         orig(pushop, nodes, phase)
 
 @exchange.b2partsgenerator(scratchbranchparttype)
 def partgen(pushop, bundler):
     bookmark = pushop.ui.config(experimental, configbookmark)
     scratchpush = pushop.ui.configbool(experimental, configscratchpush)
     if 'changesets' in pushop.stepsdone or not scratchpush:
         return
 
     if scratchbranchparttype not in bundle2.bundle2caps(pushop.remote):
         return
 
     pushop.stepsdone.add('changesets')
     if not pushop.outgoing.missing:
         pushop.ui.status(_('no changes found\n'))
         pushop.cgresult = 0
         return
 
     # This parameter tells the server that the following bundle is an
     # infinitepush. This let's it switch the part processing to our infinitepush
     # code path.
     bundler.addparam("infinitepush", "True")
 
     scratchparts = bundleparts.getscratchbranchparts(pushop.repo,
                                                      pushop.remote,
                                                      pushop.outgoing,
                                                      pushop.ui,
                                                      bookmark)
 
     for scratchpart in scratchparts:
         bundler.addpart(scratchpart)
 
     def handlereply(op):
         # server either succeeds or aborts; no code to read
         pushop.cgresult = 1
 
     return handlereply
 
 bundle2.capabilities[bundleparts.scratchbranchparttype] = ()
 
 def _getrevs(bundle, oldnode, force, bookmark):
     'extracts and validates the revs to be imported'
     revs = [bundle[r] for r in bundle.revs('sort(bundle())')]
 
     # new bookmark
     if oldnode is None:
         return revs
 
     # Fast forward update
     if oldnode in bundle and list(bundle.set('bundle() & %s::', oldnode)):
         return revs
 
     return revs
 
 @contextlib.contextmanager
 def logservicecall(logger, service, **kwargs):
     start = time.time()
     logger(service, eventtype='start', **kwargs)
     try:
         yield
         logger(service, eventtype='success',
                elapsedms=(time.time() - start) * 1000, **kwargs)
     except Exception as e:
         logger(service, eventtype='failure',
                elapsedms=(time.time() - start) * 1000, errormsg=str(e),
                **kwargs)
         raise
 
 def _getorcreateinfinitepushlogger(op):
     logger = op.records['infinitepushlogger']
     if not logger:
         ui = op.repo.ui
         try:
             username = procutil.getuser()
         except Exception:
             username = 'unknown'
         # Generate random request id to be able to find all logged entries
         # for the same request. Since requestid is pseudo-generated it may
         # not be unique, but we assume that (hostname, username, requestid)
         # is unique.
         random.seed()
         requestid = random.randint(0, 2000000000)
         hostname = socket.gethostname()
         logger = functools.partial(ui.log, 'infinitepush', user=username,
                                    requestid=requestid, hostname=hostname,
                                    reponame=ui.config('infinitepush',
                                                       'reponame'))
         op.records.add('infinitepushlogger', logger)
     else:
         logger = logger[0]
     return logger
 
 def storetobundlestore(orig, repo, op, unbundler):
     """stores the incoming bundle coming from push command to the bundlestore
     instead of applying on the revlogs"""
 
     repo.ui.status(_("storing changesets on the bundlestore\n"))
     bundler = bundle2.bundle20(repo.ui)
 
     # processing each part and storing it in bundler
     with bundle2.partiterator(repo, op, unbundler) as parts:
         for part in parts:
             bundlepart = None
             if part.type == 'replycaps':
                 # This configures the current operation to allow reply parts.
                 bundle2._processpart(op, part)
             else:
                 bundlepart = bundle2.bundlepart(part.type, data=part.read())
                 for key, value in part.params.iteritems():
                     bundlepart.addparam(key, value)
 
                 # Certain parts require a response
                 if part.type in ('pushkey', 'changegroup'):
                     if op.reply is not None:
                         rpart = op.reply.newpart('reply:%s' % part.type)
                         rpart.addparam('in-reply-to', str(part.id),
                                        mandatory=False)
                         rpart.addparam('return', '1', mandatory=False)
 
             op.records.add(part.type, {
                 'return': 1,
             })
             if bundlepart:
                 bundler.addpart(bundlepart)
 
     # storing the bundle in the bundlestore
     buf = util.chunkbuffer(bundler.getchunks())
     fd, bundlefile = tempfile.mkstemp()
     try:
         try:
             fp = os.fdopen(fd, r'wb')
             fp.write(buf.read())
         finally:
             fp.close()
         storebundle(op, {}, bundlefile)
     finally:
         try:
             os.unlink(bundlefile)
         except Exception:
             # we would rather see the original exception
             pass
 
 def processparts(orig, repo, op, unbundler):
 
     # make sure we don't wrap processparts in case of `hg unbundle`
     if op.source == 'unbundle':
         return orig(repo, op, unbundler)
 
     # this server routes each push to bundle store
     if repo.ui.configbool('infinitepush', 'pushtobundlestore'):
         return storetobundlestore(orig, repo, op, unbundler)
 
     if unbundler.params.get('infinitepush') != 'True':
         return orig(repo, op, unbundler)
 
     handleallparts = repo.ui.configbool('infinitepush', 'storeallparts')
 
     bundler = bundle2.bundle20(repo.ui)
     cgparams = None
     with bundle2.partiterator(repo, op, unbundler) as parts:
         for part in parts:
             bundlepart = None
             if part.type == 'replycaps':
                 # This configures the current operation to allow reply parts.
                 bundle2._processpart(op, part)
             elif part.type == bundleparts.scratchbranchparttype:
                 # Scratch branch parts need to be converted to normal
                 # changegroup parts, and the extra parameters stored for later
                 # when we upload to the store. Eventually those parameters will
                 # be put on the actual bundle instead of this part, then we can
                 # send a vanilla changegroup instead of the scratchbranch part.
                 cgversion = part.params.get('cgversion', '01')
                 bundlepart = bundle2.bundlepart('changegroup', data=part.read())
                 bundlepart.addparam('version', cgversion)
                 cgparams = part.params
 
                 # If we're not dumping all parts into the new bundle, we need to
                 # alert the future pushkey and phase-heads handler to skip
                 # the part.
                 if not handleallparts:
                     op.records.add(scratchbranchparttype + '_skippushkey', True)
                     op.records.add(scratchbranchparttype + '_skipphaseheads',
                                    True)
             else:
                 if handleallparts:
                     # Ideally we would not process any parts, and instead just
                     # forward them to the bundle for storage, but since this
                     # differs from previous behavior, we need to put it behind a
                     # config flag for incremental rollout.
                     bundlepart = bundle2.bundlepart(part.type, data=part.read())
                     for key, value in part.params.iteritems():
                         bundlepart.addparam(key, value)
 
                     # Certain parts require a response
                     if part.type == 'pushkey':
                         if op.reply is not None:
                             rpart = op.reply.newpart('reply:pushkey')
                             rpart.addparam('in-reply-to', str(part.id),
                                            mandatory=False)
                             rpart.addparam('return', '1', mandatory=False)
                 else:
                     bundle2._processpart(op, part)
 
             if handleallparts:
                 op.records.add(part.type, {
                     'return': 1,
                 })
             if bundlepart:
                 bundler.addpart(bundlepart)
 
     # If commits were sent, store them
     if cgparams:
         buf = util.chunkbuffer(bundler.getchunks())
         fd, bundlefile = tempfile.mkstemp()
         try:
             try:
                 fp = os.fdopen(fd, r'wb')
                 fp.write(buf.read())
             finally:
                 fp.close()
             storebundle(op, cgparams, bundlefile)
         finally:
             try:
                 os.unlink(bundlefile)
             except Exception:
                 # we would rather see the original exception
                 pass
 
 def storebundle(op, params, bundlefile):
     log = _getorcreateinfinitepushlogger(op)
     parthandlerstart = time.time()
     log(scratchbranchparttype, eventtype='start')
     index = op.repo.bundlestore.index
     store = op.repo.bundlestore.store
     op.records.add(scratchbranchparttype + '_skippushkey', True)
 
     bundle = None
     try:  # guards bundle
         bundlepath = "bundle:%s+%s" % (op.repo.root, bundlefile)
         bundle = hg.repository(op.repo.ui, bundlepath)
 
         bookmark = params.get('bookmark')
         bookprevnode = params.get('bookprevnode', '')
         force = params.get('force')
 
         if bookmark:
             oldnode = index.getnode(bookmark)
         else:
             oldnode = None
         bundleheads = bundle.revs('heads(bundle())')
         if bookmark and len(bundleheads) > 1:
             raise error.Abort(
                 _('cannot push more than one head to a scratch branch'))
 
         revs = _getrevs(bundle, oldnode, force, bookmark)
 
         # Notify the user of what is being pushed
         plural = 's' if len(revs) > 1 else ''
         op.repo.ui.warn(_("pushing %d commit%s:\n") % (len(revs), plural))
         maxoutput = 10
         for i in range(0, min(len(revs), maxoutput)):
             firstline = bundle[revs[i]].description().split('\n')[0][:50]
             op.repo.ui.warn(("    %s  %s\n") % (revs[i], firstline))
 
         if len(revs) > maxoutput + 1:
             op.repo.ui.warn(("    ...\n"))
             firstline = bundle[revs[-1]].description().split('\n')[0][:50]
             op.repo.ui.warn(("    %s  %s\n") % (revs[-1], firstline))
 
         nodesctx = [bundle[rev] for rev in revs]
         inindex = lambda rev: bool(index.getbundle(bundle[rev].hex()))
         if bundleheads:
             newheadscount = sum(not inindex(rev) for rev in bundleheads)
         else:
             newheadscount = 0
         # If there's a bookmark specified, there should be only one head,
         # so we choose the last node, which will be that head.
         # If a bug or malicious client allows there to be a bookmark
         # with multiple heads, we will place the bookmark on the last head.
         bookmarknode = nodesctx[-1].hex() if nodesctx else None
         key = None
         if newheadscount:
             with open(bundlefile, 'r') as f:
                 bundledata = f.read()
                 with logservicecall(log, 'bundlestore',
                                     bundlesize=len(bundledata)):
                     bundlesizelimit = 100 * 1024 * 1024  # 100 MB
                     if len(bundledata) > bundlesizelimit:
                         error_msg = ('bundle is too big: %d bytes. ' +
                                      'max allowed size is 100 MB')
                         raise error.Abort(error_msg % (len(bundledata),))
                     key = store.write(bundledata)
 
         with logservicecall(log, 'index', newheadscount=newheadscount), index:
             if key:
                 index.addbundle(key, nodesctx)
             if bookmark:
                 index.addbookmark(bookmark, bookmarknode)
                 _maybeaddpushbackpart(op, bookmark, bookmarknode,
                                       bookprevnode, params)
         log(scratchbranchparttype, eventtype='success',
             elapsedms=(time.time() - parthandlerstart) * 1000)
 
     except Exception as e:
         log(scratchbranchparttype, eventtype='failure',
             elapsedms=(time.time() - parthandlerstart) * 1000,
             errormsg=str(e))
         raise
     finally:
         if bundle:
             bundle.close()
 
 @bundle2.parthandler(scratchbranchparttype,
                      ('bookmark', 'bookprevnode', 'force',
                       'pushbackbookmarks', 'cgversion'))
 def bundle2scratchbranch(op, part):
     '''unbundle a bundle2 part containing a changegroup to store'''
 
     bundler = bundle2.bundle20(op.repo.ui)
     cgversion = part.params.get('cgversion', '01')
     cgpart = bundle2.bundlepart('changegroup', data=part.read())
     cgpart.addparam('version', cgversion)
     bundler.addpart(cgpart)
     buf = util.chunkbuffer(bundler.getchunks())
 
     fd, bundlefile = tempfile.mkstemp()
     try:
         try:
             fp = os.fdopen(fd, r'wb')
             fp.write(buf.read())
         finally:
             fp.close()
         storebundle(op, part.params, bundlefile)
     finally:
         try:
             os.unlink(bundlefile)
         except OSError as e:
             if e.errno != errno.ENOENT:
                 raise
 
     return 1
 
 def _maybeaddpushbackpart(op, bookmark, newnode, oldnode, params):
     if params.get('pushbackbookmarks'):
         if op.reply and 'pushback' in op.reply.capabilities:
             params = {
                 'namespace': 'bookmarks',
                 'key': bookmark,
                 'new': newnode,
                 'old': oldnode,
             }
             op.reply.newpart('pushkey', mandatoryparams=params.iteritems())
 
 def bundle2pushkey(orig, op, part):
     '''Wrapper of bundle2.handlepushkey()
 
     The only goal is to skip calling the original function if flag is set.
     It's set if infinitepush push is happening.
     '''
     if op.records[scratchbranchparttype + '_skippushkey']:
         if op.reply is not None:
             rpart = op.reply.newpart('reply:pushkey')
             rpart.addparam('in-reply-to', str(part.id), mandatory=False)
             rpart.addparam('return', '1', mandatory=False)
         return 1
 
     return orig(op, part)
 
 def bundle2handlephases(orig, op, part):
     '''Wrapper of bundle2.handlephases()
 
     The only goal is to skip calling the original function if flag is set.
     It's set if infinitepush push is happening.
     '''
 
     if op.records[scratchbranchparttype + '_skipphaseheads']:
         return
 
     return orig(op, part)
 
 def _asyncsavemetadata(root, nodes):
     '''starts a separate process that fills metadata for the nodes
 
     This function creates a separate process and doesn't wait for it's
     completion. This was done to avoid slowing down pushes
     '''
 
     maxnodes = 50
     if len(nodes) > maxnodes:
         return
     nodesargs = []
     for node in nodes:
         nodesargs.append('--node')
         nodesargs.append(node)
     with open(os.devnull, 'w+b') as devnull:
         cmdline = [util.hgexecutable(), 'debugfillinfinitepushmetadata',
                    '-R', root] + nodesargs
         # Process will run in background. We don't care about the return code
         subprocess.Popen(cmdline, close_fds=True, shell=False,
                          stdin=devnull, stdout=devnull, stderr=devnull)

hgext/largefiles/proto.py

@@ -1,193 +1,193 b''
 # Copyright 2011 Fog Creek Software
 #
 # This software may be used and distributed according to the terms of the
 # GNU General Public License version 2 or any later version.
 from __future__ import absolute_import
 
 import os
 import re
 
 from mercurial.i18n import _
 
 from mercurial import (
     error,
     httppeer,
     util,
-    wireproto,
     wireprototypes,
+    wireprotov1peer,
 )
 
 from . import (
     lfutil,
 )
 
 urlerr = util.urlerr
 urlreq = util.urlreq
 
 LARGEFILES_REQUIRED_MSG = ('\nThis repository uses the largefiles extension.'
                            '\n\nPlease enable it in your Mercurial config '
                            'file.\n')
 
 # these will all be replaced by largefiles.uisetup
 ssholdcallstream = None
 httpoldcallstream = None
 
 def putlfile(repo, proto, sha):
     '''Server command for putting a largefile into a repository's local store
     and into the user cache.'''
     with proto.mayberedirectstdio() as output:
         path = lfutil.storepath(repo, sha)
         util.makedirs(os.path.dirname(path))
         tmpfp = util.atomictempfile(path, createmode=repo.store.createmode)
 
         try:
             for p in proto.getpayload():
                 tmpfp.write(p)
             tmpfp._fp.seek(0)
             if sha != lfutil.hexsha1(tmpfp._fp):
                 raise IOError(0, _('largefile contents do not match hash'))
             tmpfp.close()
             lfutil.linktousercache(repo, sha)
         except IOError as e:
             repo.ui.warn(_('largefiles: failed to put %s into store: %s\n') %
                          (sha, e.strerror))
             return wireprototypes.pushres(
                 1, output.getvalue() if output else '')
         finally:
             tmpfp.discard()
 
     return wireprototypes.pushres(0, output.getvalue() if output else '')
 
 def getlfile(repo, proto, sha):
     '''Server command for retrieving a largefile from the repository-local
     cache or user cache.'''
     filename = lfutil.findfile(repo, sha)
     if not filename:
         raise error.Abort(_('requested largefile %s not present in cache')
                           % sha)
     f = open(filename, 'rb')
     length = os.fstat(f.fileno())[6]
 
     # Since we can't set an HTTP content-length header here, and
     # Mercurial core provides no way to give the length of a streamres
     # (and reading the entire file into RAM would be ill-advised), we
     # just send the length on the first line of the response, like the
     # ssh proto does for string responses.
     def generator():
         yield '%d\n' % length
         for chunk in util.filechunkiter(f):
             yield chunk
     return wireprototypes.streamreslegacy(gen=generator())
 
 def statlfile(repo, proto, sha):
     '''Server command for checking if a largefile is present - returns '2\n' if
     the largefile is missing, '0\n' if it seems to be in good condition.
 
     The value 1 is reserved for mismatched checksum, but that is too expensive
     to be verified on every stat and must be caught be running 'hg verify'
     server side.'''
     filename = lfutil.findfile(repo, sha)
     if not filename:
         return wireprototypes.bytesresponse('2\n')
     return wireprototypes.bytesresponse('0\n')
 
 def wirereposetup(ui, repo):
     class lfileswirerepository(repo.__class__):
         def putlfile(self, sha, fd):
             # unfortunately, httprepository._callpush tries to convert its
             # input file-like into a bundle before sending it, so we can't use
             # it ...
             if issubclass(self.__class__, httppeer.httppeer):
                 res = self._call('putlfile', data=fd, sha=sha,
                     headers={r'content-type': r'application/mercurial-0.1'})
                 try:
                     d, output = res.split('\n', 1)
                     for l in output.splitlines(True):
                         self.ui.warn(_('remote: '), l) # assume l ends with \n
                     return int(d)
                 except ValueError:
                     self.ui.warn(_('unexpected putlfile response: %r\n') % res)
                     return 1
             # ... but we can't use sshrepository._call because the data=
             # argument won't get sent, and _callpush does exactly what we want
             # in this case: send the data straight through
             else:
                 try:
                     ret, output = self._callpush("putlfile", fd, sha=sha)
                     if ret == "":
                         raise error.ResponseError(_('putlfile failed:'),
                                 output)
                     return int(ret)
                 except IOError:
                     return 1
                 except ValueError:
                     raise error.ResponseError(
                         _('putlfile failed (unexpected response):'), ret)
 
         def getlfile(self, sha):
             """returns an iterable with the chunks of the file with sha sha"""
             stream = self._callstream("getlfile", sha=sha)
             length = stream.readline()
             try:
                 length = int(length)
             except ValueError:
                 self._abort(error.ResponseError(_("unexpected response:"),
                                                 length))
 
             # SSH streams will block if reading more than length
             for chunk in util.filechunkiter(stream, limit=length):
                 yield chunk
             # HTTP streams must hit the end to process the last empty
             # chunk of Chunked-Encoding so the connection can be reused.
             if issubclass(self.__class__, httppeer.httppeer):
                 chunk = stream.read(1)
                 if chunk:
                     self._abort(error.ResponseError(_("unexpected response:"),
                                                     chunk))
 
-        @wireproto.batchable
+        @wireprotov1peer.batchable
         def statlfile(self, sha):
-            f = wireproto.future()
+            f = wireprotov1peer.future()
             result = {'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
 
     repo.__class__ = lfileswirerepository
 
 # advertise the largefiles=serve capability
 def _capabilities(orig, repo, proto):
     '''announce largefile server capability'''
     caps = orig(repo, proto)
     caps.append('largefiles=serve')
     return caps
 
 def heads(orig, repo, proto):
     '''Wrap server command - largefile capable clients will know to call
     lheads instead'''
     if lfutil.islfilesrepo(repo):
         return wireprototypes.ooberror(LARGEFILES_REQUIRED_MSG)
 
     return orig(repo, proto)
 
 def sshrepocallstream(self, cmd, **args):
     if cmd == 'heads' and self.capable('largefiles'):
         cmd = 'lheads'
     if cmd == 'batch' and self.capable('largefiles'):
         args[r'cmds'] = args[r'cmds'].replace('heads ', 'lheads ')
     return ssholdcallstream(self, cmd, **args)
 
 headsre = re.compile(br'(^|;)heads\b')
 
 def httprepocallstream(self, cmd, **args):
     if cmd == 'heads' and self.capable('largefiles'):
         cmd = 'lheads'
     if cmd == 'batch' and self.capable('largefiles'):
         args[r'cmds'] = headsre.sub('lheads', args[r'cmds'])
     return httpoldcallstream(self, cmd, **args)

mercurial/httppeer.py

@@ -1,816 +1,817 b''
 # httppeer.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 __future__ import absolute_import
 
 import errno
 import io
 import os
 import socket
 import struct
 import tempfile
 
 from .i18n import _
 from .thirdparty import (
     cbor,
 )
 from .thirdparty.zope import (
     interface as zi,
 )
 from . import (
     bundle2,
     error,
     httpconnection,
     pycompat,
     repository,
     statichttprepo,
     url as urlmod,
     util,
     wireproto,
     wireprotoframing,
     wireprototypes,
+    wireprotov1peer,
     wireprotov2server,
 )
 
 httplib = util.httplib
 urlerr = util.urlerr
 urlreq = util.urlreq
 
 def encodevalueinheaders(value, header, limit):
     """Encode a string value into multiple HTTP headers.
 
     ``value`` will be encoded into 1 or more HTTP headers with the names
     ``header-<N>`` where ``<N>`` is an integer starting at 1. Each header
     name + value will be at most ``limit`` bytes long.
 
     Returns an iterable of 2-tuples consisting of header names and
     values as native strings.
     """
     # HTTP Headers are ASCII. Python 3 requires them to be unicodes,
     # not bytes. This function always takes bytes in as arguments.
     fmt = pycompat.strurl(header) + r'-%s'
     # Note: it is *NOT* a bug that the last bit here is a bytestring
     # and not a unicode: we're just getting the encoded length anyway,
     # and using an r-string to make it portable between Python 2 and 3
     # doesn't work because then the \r is a literal backslash-r
     # instead of a carriage return.
     valuelen = limit - len(fmt % r'000') - len(': \r\n')
     result = []
 
     n = 0
     for i in xrange(0, len(value), valuelen):
         n += 1
         result.append((fmt % str(n), pycompat.strurl(value[i:i + valuelen])))
 
     return result
 
 def _wraphttpresponse(resp):
     """Wrap an HTTPResponse with common error handlers.
 
     This ensures that any I/O from any consumer raises the appropriate
     error and messaging.
     """
     origread = resp.read
 
     class readerproxy(resp.__class__):
         def read(self, size=None):
             try:
                 return origread(size)
             except httplib.IncompleteRead as e:
                 # e.expected is an integer if length known or None otherwise.
                 if e.expected:
                     msg = _('HTTP request error (incomplete response; '
                             'expected %d bytes got %d)') % (e.expected,
                                                            len(e.partial))
                 else:
                     msg = _('HTTP request error (incomplete response)')
 
                 raise error.PeerTransportError(
                     msg,
                     hint=_('this may be an intermittent network failure; '
                            'if the error persists, consider contacting the '
                            'network or server operator'))
             except httplib.HTTPException as e:
                 raise error.PeerTransportError(
                     _('HTTP request error (%s)') % e,
                     hint=_('this may be an intermittent network failure; '
                            'if the error persists, consider contacting the '
                            'network or server operator'))
 
     resp.__class__ = readerproxy
 
 class _multifile(object):
     def __init__(self, *fileobjs):
         for f in fileobjs:
             if not util.safehasattr(f, 'length'):
                 raise ValueError(
                     '_multifile only supports file objects that '
                     'have a length but this one does not:', type(f), f)
         self._fileobjs = fileobjs
         self._index = 0
 
     @property
     def length(self):
         return sum(f.length for f in self._fileobjs)
 
     def read(self, amt=None):
         if amt <= 0:
             return ''.join(f.read() for f in self._fileobjs)
         parts = []
         while amt and self._index < len(self._fileobjs):
             parts.append(self._fileobjs[self._index].read(amt))
             got = len(parts[-1])
             if got < amt:
                 self._index += 1
             amt -= got
         return ''.join(parts)
 
     def seek(self, offset, whence=os.SEEK_SET):
         if whence != os.SEEK_SET:
             raise NotImplementedError(
                 '_multifile does not support anything other'
                 ' than os.SEEK_SET for whence on seek()')
         if offset != 0:
             raise NotImplementedError(
                 '_multifile only supports seeking to start, but that '
                 'could be fixed if you need it')
         for f in self._fileobjs:
             f.seek(0)
         self._index = 0
 
 def makev1commandrequest(ui, requestbuilder, caps, capablefn,
                          repobaseurl, cmd, args):
     """Make an HTTP request to run a command for a version 1 client.
 
     ``caps`` is a set of known server capabilities. The value may be
     None if capabilities are not yet known.
 
     ``capablefn`` is a function to evaluate a capability.
 
     ``cmd``, ``args``, and ``data`` define the command, its arguments, and
     raw data to pass to it.
     """
     if cmd == 'pushkey':
         args['data'] = ''
     data = args.pop('data', None)
     headers = args.pop('headers', {})
 
     ui.debug("sending %s command\n" % cmd)
     q = [('cmd', cmd)]
     headersize = 0
     # Important: don't use self.capable() here or else you end up
     # with infinite recursion when trying to look up capabilities
     # for the first time.
     postargsok = caps is not None and 'httppostargs' in caps
 
     # Send arguments via POST.
     if postargsok and args:
         strargs = urlreq.urlencode(sorted(args.items()))
         if not data:
             data = strargs
         else:
             if isinstance(data, bytes):
                 i = io.BytesIO(data)
                 i.length = len(data)
                 data = i
             argsio = io.BytesIO(strargs)
             argsio.length = len(strargs)
             data = _multifile(argsio, data)
         headers[r'X-HgArgs-Post'] = len(strargs)
     elif args:
         # Calling self.capable() can infinite loop if we are calling
         # "capabilities". But that command should never accept wire
         # protocol arguments. So this should never happen.
         assert cmd != 'capabilities'
         httpheader = capablefn('httpheader')
         if httpheader:
             headersize = int(httpheader.split(',', 1)[0])
 
         # Send arguments via HTTP headers.
         if headersize > 0:
             # The headers can typically carry more data than the URL.
             encargs = urlreq.urlencode(sorted(args.items()))
             for header, value in encodevalueinheaders(encargs, 'X-HgArg',
                                                       headersize):
                 headers[header] = value
         # Send arguments via query string (Mercurial <1.9).
         else:
             q += sorted(args.items())
 
     qs = '?%s' % urlreq.urlencode(q)
     cu = "%s%s" % (repobaseurl, qs)
     size = 0
     if util.safehasattr(data, 'length'):
         size = data.length
     elif data is not None:
         size = len(data)
     if data is not None and r'Content-Type' not in headers:
         headers[r'Content-Type'] = r'application/mercurial-0.1'
 
     # Tell the server we accept application/mercurial-0.2 and multiple
     # compression formats if the server is capable of emitting those
     # payloads.
     # Note: Keep this set empty by default, as client advertisement of
     # protocol parameters should only occur after the handshake.
     protoparams = set()
 
     mediatypes = set()
     if caps is not None:
         mt = capablefn('httpmediatype')
         if mt:
             protoparams.add('0.1')
             mediatypes = set(mt.split(','))
 
         protoparams.add('partial-pull')
 
     if '0.2tx' in mediatypes:
         protoparams.add('0.2')
 
     if '0.2tx' in mediatypes and capablefn('compression'):
         # We /could/ compare supported compression formats and prune
         # non-mutually supported or error if nothing is mutually supported.
         # For now, send the full list to the server and have it error.
         comps = [e.wireprotosupport().name for e in
                  util.compengines.supportedwireengines(util.CLIENTROLE)]
         protoparams.add('comp=%s' % ','.join(comps))
 
     if protoparams:
         protoheaders = encodevalueinheaders(' '.join(sorted(protoparams)),
                                             'X-HgProto',
                                             headersize or 1024)
         for header, value in protoheaders:
             headers[header] = value
 
     varyheaders = []
     for header in headers:
         if header.lower().startswith(r'x-hg'):
             varyheaders.append(header)
 
     if varyheaders:
         headers[r'Vary'] = r','.join(sorted(varyheaders))
 
     req = requestbuilder(pycompat.strurl(cu), data, headers)
 
     if data is not None:
         ui.debug("sending %d bytes\n" % size)
         req.add_unredirected_header(r'Content-Length', r'%d' % size)
 
     return req, cu, qs
 
 def sendrequest(ui, opener, req):
     """Send a prepared HTTP request.
 
     Returns the response object.
     """
     if (ui.debugflag
         and ui.configbool('devel', 'debug.peer-request')):
         dbg = ui.debug
         line = 'devel-peer-request: %s\n'
         dbg(line % '%s %s' % (req.get_method(), req.get_full_url()))
         hgargssize = None
 
         for header, value in sorted(req.header_items()):
             if header.startswith('X-hgarg-'):
                 if hgargssize is None:
                     hgargssize = 0
                 hgargssize += len(value)
             else:
                 dbg(line % '  %s %s' % (header, value))
 
         if hgargssize is not None:
             dbg(line % '  %d bytes of commands arguments in headers'
                 % hgargssize)
 
         if req.has_data():
             data = req.get_data()
             length = getattr(data, 'length', None)
             if length is None:
                 length = len(data)
             dbg(line % '  %d bytes of data' % length)
 
         start = util.timer()
 
     try:
         res = opener.open(req)
     except urlerr.httperror as inst:
         if inst.code == 401:
             raise error.Abort(_('authorization failed'))
         raise
     except httplib.HTTPException as inst:
         ui.debug('http error requesting %s\n' %
                  util.hidepassword(req.get_full_url()))
         ui.traceback()
         raise IOError(None, inst)
     finally:
         if ui.configbool('devel', 'debug.peer-request'):
             dbg(line % '  finished in %.4f seconds (%s)'
                 % (util.timer() - start, res.code))
 
     # Insert error handlers for common I/O failures.
     _wraphttpresponse(res)
 
     return res
 
 def parsev1commandresponse(ui, baseurl, requrl, qs, resp, compressible,
                            allowcbor=False):
     # record the url we got redirected to
     respurl = pycompat.bytesurl(resp.geturl())
     if respurl.endswith(qs):
         respurl = respurl[:-len(qs)]
     if baseurl.rstrip('/') != respurl.rstrip('/'):
         if not ui.quiet:
             ui.warn(_('real URL is %s\n') % respurl)
 
     try:
         proto = pycompat.bytesurl(resp.getheader(r'content-type', r''))
     except AttributeError:
         proto = pycompat.bytesurl(resp.headers.get(r'content-type', r''))
 
     safeurl = util.hidepassword(baseurl)
     if proto.startswith('application/hg-error'):
         raise error.OutOfBandError(resp.read())
 
     # Pre 1.0 versions of Mercurial used text/plain and
     # application/hg-changegroup. We don't support such old servers.
     if not proto.startswith('application/mercurial-'):
         ui.debug("requested URL: '%s'\n" % util.hidepassword(requrl))
         raise error.RepoError(
             _("'%s' does not appear to be an hg repository:\n"
               "---%%<--- (%s)\n%s\n---%%<---\n")
             % (safeurl, proto or 'no content-type', resp.read(1024)))
 
     try:
         subtype = proto.split('-', 1)[1]
 
         # Unless we end up supporting CBOR in the legacy wire protocol,
         # this should ONLY be encountered for the initial capabilities
         # request during handshake.
         if subtype == 'cbor':
             if allowcbor:
                 return respurl, proto, resp
             else:
                 raise error.RepoError(_('unexpected CBOR response from '
                                         'server'))
 
         version_info = tuple([int(n) for n in subtype.split('.')])
     except ValueError:
         raise error.RepoError(_("'%s' sent a broken Content-Type "
                                 "header (%s)") % (safeurl, proto))
 
     # TODO consider switching to a decompression reader that uses
     # generators.
     if version_info == (0, 1):
         if compressible:
             resp = util.compengines['zlib'].decompressorreader(resp)
 
     elif version_info == (0, 2):
         # application/mercurial-0.2 always identifies the compression
         # engine in the payload header.
         elen = struct.unpack('B', resp.read(1))[0]
         ename = resp.read(elen)
         engine = util.compengines.forwiretype(ename)
 
         resp = engine.decompressorreader(resp)
     else:
         raise error.RepoError(_("'%s' uses newer protocol %s") %
                               (safeurl, subtype))
 
     return respurl, proto, resp
 
-class httppeer(wireproto.wirepeer):
+class httppeer(wireprotov1peer.wirepeer):
     def __init__(self, ui, path, url, opener, requestbuilder, caps):
         self.ui = ui
         self._path = path
         self._url = url
         self._caps = caps
         self._urlopener = opener
         self._requestbuilder = requestbuilder
 
     def __del__(self):
         for h in self._urlopener.handlers:
             h.close()
             getattr(h, "close_all", lambda: None)()
 
     # Begin of ipeerconnection interface.
 
     def url(self):
         return self._path
 
     def local(self):
         return None
 
     def peer(self):
         return self
 
     def canpush(self):
         return True
 
     def close(self):
         pass
 
     # End of ipeerconnection interface.
 
     # Begin of ipeercommands interface.
 
     def capabilities(self):
         return self._caps
 
     # End of ipeercommands interface.
 
     # look up capabilities only when needed
 
     def _callstream(self, cmd, _compressible=False, **args):
         args = pycompat.byteskwargs(args)
 
         req, cu, qs = makev1commandrequest(self.ui, self._requestbuilder,
                                            self._caps, self.capable,
                                            self._url, cmd, args)
 
         resp = sendrequest(self.ui, self._urlopener, req)
 
         self._url, ct, resp = parsev1commandresponse(self.ui, self._url, cu, qs,
                                                      resp, _compressible)
 
         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 _callpush(self, cmd, cg, **args):
         # have to stream bundle to a temp file because we do not have
         # http 1.1 chunked transfer.
 
         types = self.capable('unbundle')
         try:
             types = types.split(',')
         except AttributeError:
             # servers older than d1b16a746db6 will send 'unbundle' as a
             # boolean capability. They only support headerless/uncompressed
             # bundles.
             types = [""]
         for x in types:
             if x in bundle2.bundletypes:
                 type = x
                 break
 
         tempname = bundle2.writebundle(self.ui, cg, None, type)
         fp = httpconnection.httpsendfile(self.ui, tempname, "rb")
         headers = {r'Content-Type': r'application/mercurial-0.1'}
 
         try:
             r = self._call(cmd, data=fp, headers=headers, **args)
             vals = r.split('\n', 1)
             if len(vals) < 2:
                 raise error.ResponseError(_("unexpected response:"), r)
             return vals
         except urlerr.httperror:
             # Catch and re-raise these so we don't try and treat them
             # like generic socket errors. They lack any values in
             # .args on Python 3 which breaks our socket.error block.
             raise
         except socket.error as err:
             if err.args[0] in (errno.ECONNRESET, errno.EPIPE):
                 raise error.Abort(_('push failed: %s') % err.args[1])
             raise error.Abort(err.args[1])
         finally:
             fp.close()
             os.unlink(tempname)
 
     def _calltwowaystream(self, cmd, fp, **args):
         fh = None
         fp_ = None
         filename = None
         try:
             # dump bundle to disk
             fd, filename = tempfile.mkstemp(prefix="hg-bundle-", suffix=".hg")
             fh = os.fdopen(fd, r"wb")
             d = fp.read(4096)
             while d:
                 fh.write(d)
                 d = fp.read(4096)
             fh.close()
             # start http push
             fp_ = httpconnection.httpsendfile(self.ui, filename, "rb")
             headers = {r'Content-Type': r'application/mercurial-0.1'}
             return self._callstream(cmd, data=fp_, headers=headers, **args)
         finally:
             if fp_ is not None:
                 fp_.close()
             if fh is not None:
                 fh.close()
                 os.unlink(filename)
 
     def _callcompressable(self, cmd, **args):
         return self._callstream(cmd, _compressible=True, **args)
 
     def _abort(self, exception):
         raise exception
 
 # TODO implement interface for version 2 peers
 @zi.implementer(repository.ipeerconnection, repository.ipeercapabilities)
 class httpv2peer(object):
     def __init__(self, ui, repourl, apipath, opener, requestbuilder,
                  apidescriptor):
         self.ui = ui
 
         if repourl.endswith('/'):
             repourl = repourl[:-1]
 
         self._url = repourl
         self._apipath = apipath
         self._opener = opener
         self._requestbuilder = requestbuilder
         self._descriptor = apidescriptor
 
     # Start of ipeerconnection.
 
     def url(self):
         return self._url
 
     def local(self):
         return None
 
     def peer(self):
         return self
 
     def canpush(self):
         # TODO change once implemented.
         return False
 
     def close(self):
         pass
 
     # End of ipeerconnection.
 
     # Start of ipeercapabilities.
 
     def capable(self, name):
         # The capabilities used internally historically map to capabilities
         # advertised from the "capabilities" wire protocol command. However,
         # version 2 of that command works differently.
 
         # Maps to commands that are available.
         if name in ('branchmap', 'getbundle', 'known', 'lookup', 'pushkey'):
             return True
 
         # Other concepts.
         if name in ('bundle2',):
             return True
 
         return False
 
     def requirecap(self, name, purpose):
         if self.capable(name):
             return
 
         raise error.CapabilityError(
             _('cannot %s; client or remote repository does not support the %r '
               'capability') % (purpose, name))
 
     # End of ipeercapabilities.
 
     # TODO require to be part of a batched primitive, use futures.
     def _call(self, name, **args):
         """Call a wire protocol command with arguments."""
 
         # Having this early has a side-effect of importing wireprotov2server,
         # which has the side-effect of ensuring commands are registered.
 
         # TODO modify user-agent to reflect v2.
         headers = {
             r'Accept': wireprotov2server.FRAMINGTYPE,
             r'Content-Type': wireprotov2server.FRAMINGTYPE,
         }
 
         # TODO permissions should come from capabilities results.
         permission = wireproto.commandsv2[name].permission
         if permission not in ('push', 'pull'):
             raise error.ProgrammingError('unknown permission type: %s' %
                                          permission)
 
         permission = {
             'push': 'rw',
             'pull': 'ro',
         }[permission]
 
         url = '%s/%s/%s/%s' % (self._url, self._apipath, permission, name)
 
         # TODO this should be part of a generic peer for the frame-based
         # protocol.
         reactor = wireprotoframing.clientreactor(hasmultiplesend=False,
                                                  buffersends=True)
 
         request, action, meta = reactor.callcommand(name, args)
         assert action == 'noop'
 
         action, meta = reactor.flushcommands()
         assert action == 'sendframes'
 
         body = b''.join(map(bytes, meta['framegen']))
         req = self._requestbuilder(pycompat.strurl(url), body, headers)
         req.add_unredirected_header(r'Content-Length', r'%d' % len(body))
 
         # TODO unify this code with httppeer.
         try:
             res = self._opener.open(req)
         except urlerr.httperror as e:
             if e.code == 401:
                 raise error.Abort(_('authorization failed'))
 
             raise
         except httplib.HTTPException as e:
             self.ui.traceback()
             raise IOError(None, e)
 
         # TODO validate response type, wrap response to handle I/O errors.
         # TODO more robust frame receiver.
         results = []
 
         while True:
             frame = wireprotoframing.readframe(res)
             if frame is None:
                 break
 
             self.ui.note(_('received %r\n') % frame)
 
             action, meta = reactor.onframerecv(frame)
 
             if action == 'responsedata':
                 if meta['cbor']:
                     payload = util.bytesio(meta['data'])
 
                     decoder = cbor.CBORDecoder(payload)
                     while payload.tell() + 1 < len(meta['data']):
                         results.append(decoder.decode())
                 else:
                     results.append(meta['data'])
             else:
                 error.ProgrammingError('unhandled action: %s' % action)
 
         return results
 
 # Registry of API service names to metadata about peers that handle it.
 #
 # The following keys are meaningful:
 #
 # init
 #    Callable receiving (ui, repourl, servicepath, opener, requestbuilder,
 #                        apidescriptor) to create a peer.
 #
 # priority
 #    Integer priority for the service. If we could choose from multiple
 #    services, we choose the one with the highest priority.
 API_PEERS = {
     wireprototypes.HTTPV2: {
         'init': httpv2peer,
         'priority': 50,
     },
 }
 
 def performhandshake(ui, url, opener, requestbuilder):
     # The handshake is a request to the capabilities command.
 
     caps = None
     def capable(x):
         raise error.ProgrammingError('should not be called')
 
     args = {}
 
     # The client advertises support for newer protocols by adding an
     # X-HgUpgrade-* header with a list of supported APIs and an
     # X-HgProto-* header advertising which serializing formats it supports.
     # We only support the HTTP version 2 transport and CBOR responses for
     # now.
     advertisev2 = ui.configbool('experimental', 'httppeer.advertise-v2')
 
     if advertisev2:
         args['headers'] = {
             r'X-HgProto-1': r'cbor',
         }
 
         args['headers'].update(
             encodevalueinheaders(' '.join(sorted(API_PEERS)),
                                  'X-HgUpgrade',
                                  # We don't know the header limit this early.
                                  # So make it small.
                                  1024))
 
     req, requrl, qs = makev1commandrequest(ui, requestbuilder, caps,
                                            capable, url, 'capabilities',
                                            args)
 
     resp = sendrequest(ui, opener, req)
 
     respurl, ct, resp = parsev1commandresponse(ui, url, requrl, qs, resp,
                                                compressible=False,
                                                allowcbor=advertisev2)
 
     try:
         rawdata = resp.read()
     finally:
         resp.close()
 
     if not ct.startswith('application/mercurial-'):
         raise error.ProgrammingError('unexpected content-type: %s' % ct)
 
     if advertisev2:
         if ct == 'application/mercurial-cbor':
             try:
                 info = cbor.loads(rawdata)
             except cbor.CBORDecodeError:
                 raise error.Abort(_('error decoding CBOR from remote server'),
                                   hint=_('try again and consider contacting '
                                          'the server operator'))
 
         # We got a legacy response. That's fine.
         elif ct in ('application/mercurial-0.1', 'application/mercurial-0.2'):
             info = {
                 'v1capabilities': set(rawdata.split())
             }
 
         else:
             raise error.RepoError(
                 _('unexpected response type from server: %s') % ct)
     else:
         info = {
             'v1capabilities': set(rawdata.split())
         }
 
     return respurl, info
 
 def makepeer(ui, path, opener=None, requestbuilder=urlreq.request):
     """Construct an appropriate HTTP peer instance.
 
     ``opener`` is an ``url.opener`` that should be used to establish
     connections, perform HTTP requests.
 
     ``requestbuilder`` is the type used for constructing HTTP requests.
     It exists as an argument so extensions can override the default.
     """
     u = util.url(path)
     if u.query or u.fragment:
         raise error.Abort(_('unsupported URL component: "%s"') %
                           (u.query or u.fragment))
 
     # urllib cannot handle URLs with embedded user or passwd.
     url, authinfo = u.authinfo()
     ui.debug('using %s\n' % url)
 
     opener = opener or urlmod.opener(ui, authinfo)
 
     respurl, info = performhandshake(ui, url, opener, requestbuilder)
 
     # Given the intersection of APIs that both we and the server support,
     # sort by their advertised priority and pick the first one.
     #
     # TODO consider making this request-based and interface driven. For
     # example, the caller could say "I want a peer that does X." It's quite
     # possible that not all peers would do that. Since we know the service
     # capabilities, we could filter out services not meeting the
     # requirements. Possibly by consulting the interfaces defined by the
     # peer type.
     apipeerchoices = set(info.get('apis', {}).keys()) & set(API_PEERS.keys())
 
     preferredchoices = sorted(apipeerchoices,
                               key=lambda x: API_PEERS[x]['priority'],
                               reverse=True)
 
     for service in preferredchoices:
         apipath = '%s/%s' % (info['apibase'].rstrip('/'), service)
 
         return API_PEERS[service]['init'](ui, respurl, apipath, opener,
                                           requestbuilder,
                                           info['apis'][service])
 
     # Failed to construct an API peer. Fall back to legacy.
     return httppeer(ui, path, respurl, opener, requestbuilder,
                     info['v1capabilities'])
 
 def instance(ui, path, create):
     if create:
         raise error.Abort(_('cannot create new http repository'))
     try:
         if path.startswith('https:') and not urlmod.has_https:
             raise error.Abort(_('Python support for SSL and HTTPS '
                                 'is not installed'))
 
         inst = makepeer(ui, path)
 
         return inst
     except error.RepoError as httpexception:
         try:
             r = statichttprepo.instance(ui, "static-" + path, create)
             ui.note(_('(falling back to static-http)\n'))
             return r
         except error.RepoError:
             raise httpexception # use the original http RepoError instead

mercurial/sshpeer.py

@@ -1,635 +1,636 b''
 # sshpeer.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 __future__ import absolute_import
 
 import re
 import uuid
 
 from .i18n import _
 from . import (
     error,
     pycompat,
     util,
     wireproto,
     wireprotoserver,
     wireprototypes,
+    wireprotov1peer,
 )
 from .utils import (
     procutil,
 )
 
 def _serverquote(s):
     """quote a string for the remote shell ... which we assume is sh"""
     if not s:
         return s
     if re.match('[a-zA-Z0-9@%_+=:,./-]*$', s):
         return s
     return "'%s'" % s.replace("'", "'\\''")
 
 def _forwardoutput(ui, pipe):
     """display all data currently available on pipe as remote output.
 
     This is non blocking."""
     if pipe:
         s = procutil.readpipe(pipe)
         if s:
             for l in s.splitlines():
                 ui.status(_("remote: "), l, '\n')
 
 class doublepipe(object):
     """Operate a side-channel pipe in addition of a main one
 
     The side-channel pipe contains server output to be forwarded to the user
     input. The double pipe will behave as the "main" pipe, but will ensure the
     content of the "side" pipe is properly processed while we wait for blocking
     call on the "main" pipe.
 
     If large amounts of data are read from "main", the forward will cease after
     the first bytes start to appear. This simplifies the implementation
     without affecting actual output of sshpeer too much as we rarely issue
     large read for data not yet emitted by the server.
 
     The main pipe is expected to be a 'bufferedinputpipe' from the util module
     that handle all the os specific bits. This class lives in this module
     because it focus on behavior specific to the ssh protocol."""
 
     def __init__(self, ui, main, side):
         self._ui = ui
         self._main = main
         self._side = side
 
     def _wait(self):
         """wait until some data are available on main or side
 
         return a pair of boolean (ismainready, issideready)
 
         (This will only wait for data if the setup is supported by `util.poll`)
         """
         if (isinstance(self._main, util.bufferedinputpipe) and
             self._main.hasbuffer):
             # Main has data. Assume side is worth poking at.
             return True, True
 
         fds = [self._main.fileno(), self._side.fileno()]
         try:
             act = util.poll(fds)
         except NotImplementedError:
             # non supported yet case, assume all have data.
             act = fds
         return (self._main.fileno() in act, self._side.fileno() in act)
 
     def write(self, data):
         return self._call('write', data)
 
     def read(self, size):
         r = self._call('read', size)
         if size != 0 and not r:
             # We've observed a condition that indicates the
             # stdout closed unexpectedly. Check stderr one
             # more time and snag anything that's there before
             # letting anyone know the main part of the pipe
             # closed prematurely.
             _forwardoutput(self._ui, self._side)
         return r
 
     def readline(self):
         return self._call('readline')
 
     def _call(self, methname, data=None):
         """call <methname> on "main", forward output of "side" while blocking
         """
         # data can be '' or 0
         if (data is not None and not data) or self._main.closed:
             _forwardoutput(self._ui, self._side)
             return ''
         while True:
             mainready, sideready = self._wait()
             if sideready:
                 _forwardoutput(self._ui, self._side)
             if mainready:
                 meth = getattr(self._main, methname)
                 if data is None:
                     return meth()
                 else:
                     return meth(data)
 
     def close(self):
         return self._main.close()
 
     def flush(self):
         return self._main.flush()
 
 def _cleanuppipes(ui, pipei, pipeo, pipee):
     """Clean up pipes used by an SSH connection."""
     if pipeo:
         pipeo.close()
     if pipei:
         pipei.close()
 
     if pipee:
         # Try to read from the err descriptor until EOF.
         try:
             for l in pipee:
                 ui.status(_('remote: '), l)
         except (IOError, ValueError):
             pass
 
         pipee.close()
 
 def _makeconnection(ui, sshcmd, args, remotecmd, path, sshenv=None):
     """Create an SSH connection to a server.
 
     Returns a tuple of (process, stdin, stdout, stderr) for the
     spawned process.
     """
     cmd = '%s %s %s' % (
         sshcmd,
         args,
         procutil.shellquote('%s -R %s serve --stdio' % (
             _serverquote(remotecmd), _serverquote(path))))
 
     ui.debug('running %s\n' % cmd)
     cmd = procutil.quotecommand(cmd)
 
     # no buffer allow the use of 'select'
     # feel free to remove buffering and select usage when we ultimately
     # move to threading.
     stdin, stdout, stderr, proc = procutil.popen4(cmd, bufsize=0, env=sshenv)
 
     return proc, stdin, stdout, stderr
 
 def _clientcapabilities():
     """Return list of capabilities of this client.
 
     Returns a list of capabilities that are supported by this client.
     """
     protoparams = {'partial-pull'}
     comps = [e.wireprotosupport().name for e in
              util.compengines.supportedwireengines(util.CLIENTROLE)]
     protoparams.add('comp=%s' % ','.join(comps))
     return protoparams
 
 def _performhandshake(ui, stdin, stdout, stderr):
     def badresponse():
         # Flush any output on stderr.
         _forwardoutput(ui, stderr)
 
         msg = _('no suitable response from remote hg')
         hint = ui.config('ui', 'ssherrorhint')
         raise error.RepoError(msg, hint=hint)
 
     # The handshake consists of sending wire protocol commands in reverse
     # order of protocol implementation and then sniffing for a response
     # to one of them.
     #
     # Those commands (from oldest to newest) are:
     #
     # ``between``
     #   Asks for the set of revisions between a pair of revisions. Command
     #   present in all Mercurial server implementations.
     #
     # ``hello``
     #   Instructs the server to advertise its capabilities. Introduced in
     #   Mercurial 0.9.1.
     #
     # ``upgrade``
     #   Requests upgrade from default transport protocol version 1 to
     #   a newer version. Introduced in Mercurial 4.6 as an experimental
     #   feature.
     #
     # The ``between`` command is issued with a request for the null
     # range. If the remote is a Mercurial server, this request will
     # generate a specific response: ``1\n\n``. This represents the
     # wire protocol encoded value for ``\n``. We look for ``1\n\n``
     # in the output stream and know this is the response to ``between``
     # and we're at the end of our handshake reply.
     #
     # The response to the ``hello`` command will be a line with the
     # length of the value returned by that command followed by that
     # value. If the server doesn't support ``hello`` (which should be
     # rare), that line will be ``0\n``. Otherwise, the value will contain
     # RFC 822 like lines. Of these, the ``capabilities:`` line contains
     # the capabilities of the server.
     #
     # The ``upgrade`` command isn't really a command in the traditional
     # sense of version 1 of the transport because it isn't using the
     # proper mechanism for formatting insteads: instead, it just encodes
     # arguments on the line, delimited by spaces.
     #
     # The ``upgrade`` line looks like ``upgrade <token> <capabilities>``.
     # If the server doesn't support protocol upgrades, it will reply to
     # this line with ``0\n``. Otherwise, it emits an
     # ``upgraded <token> <protocol>`` line to both stdout and stderr.
     # Content immediately following this line describes additional
     # protocol and server state.
     #
     # In addition to the responses to our command requests, the server
     # may emit "banner" output on stdout. SSH servers are allowed to
     # print messages to stdout on login. Issuing commands on connection
     # allows us to flush this banner output from the server by scanning
     # for output to our well-known ``between`` command. Of course, if
     # the banner contains ``1\n\n``, this will throw off our detection.
 
     requestlog = ui.configbool('devel', 'debug.peer-request')
 
     # Generate a random token to help identify responses to version 2
     # upgrade request.
     token = pycompat.sysbytes(str(uuid.uuid4()))
     upgradecaps = [
         ('proto', wireprotoserver.SSHV2),
     ]
     upgradecaps = util.urlreq.urlencode(upgradecaps)
 
     try:
         pairsarg = '%s-%s' % ('0' * 40, '0' * 40)
         handshake = [
             'hello\n',
             'between\n',
             'pairs %d\n' % len(pairsarg),
             pairsarg,
         ]
 
         # Request upgrade to version 2 if configured.
         if ui.configbool('experimental', 'sshpeer.advertise-v2'):
             ui.debug('sending upgrade request: %s %s\n' % (token, upgradecaps))
             handshake.insert(0, 'upgrade %s %s\n' % (token, upgradecaps))
 
         if requestlog:
             ui.debug('devel-peer-request: hello\n')
         ui.debug('sending hello command\n')
         if requestlog:
             ui.debug('devel-peer-request: between\n')
             ui.debug('devel-peer-request:   pairs: %d bytes\n' % len(pairsarg))
         ui.debug('sending between command\n')
 
         stdin.write(''.join(handshake))
         stdin.flush()
     except IOError:
         badresponse()
 
     # Assume version 1 of wire protocol by default.
     protoname = wireprototypes.SSHV1
     reupgraded = re.compile(b'^upgraded %s (.*)$' % re.escape(token))
 
     lines = ['', 'dummy']
     max_noise = 500
     while lines[-1] and max_noise:
         try:
             l = stdout.readline()
             _forwardoutput(ui, stderr)
 
             # Look for reply to protocol upgrade request. It has a token
             # in it, so there should be no false positives.
             m = reupgraded.match(l)
             if m:
                 protoname = m.group(1)
                 ui.debug('protocol upgraded to %s\n' % protoname)
                 # If an upgrade was handled, the ``hello`` and ``between``
                 # requests are ignored. The next output belongs to the
                 # protocol, so stop scanning lines.
                 break
 
             # Otherwise it could be a banner, ``0\n`` response if server
             # doesn't support upgrade.
 
             if lines[-1] == '1\n' and l == '\n':
                 break
             if l:
                 ui.debug('remote: ', l)
             lines.append(l)
             max_noise -= 1
         except IOError:
             badresponse()
     else:
         badresponse()
 
     caps = set()
 
     # For version 1, we should see a ``capabilities`` line in response to the
     # ``hello`` command.
     if protoname == wireprototypes.SSHV1:
         for l in reversed(lines):
             # Look for response to ``hello`` command. Scan from the back so
             # we don't misinterpret banner output as the command reply.
             if l.startswith('capabilities:'):
                 caps.update(l[:-1].split(':')[1].split())
                 break
     elif protoname == wireprotoserver.SSHV2:
         # We see a line with number of bytes to follow and then a value
         # looking like ``capabilities: *``.
         line = stdout.readline()
         try:
             valuelen = int(line)
         except ValueError:
             badresponse()
 
         capsline = stdout.read(valuelen)
         if not capsline.startswith('capabilities: '):
             badresponse()
 
         ui.debug('remote: %s\n' % capsline)
 
         caps.update(capsline.split(':')[1].split())
         # Trailing newline.
         stdout.read(1)
 
     # Error if we couldn't find capabilities, this means:
     #
     # 1. Remote isn't a Mercurial server
     # 2. Remote is a <0.9.1 Mercurial server
     # 3. Remote is a future Mercurial server that dropped ``hello``
     #    and other attempted handshake mechanisms.
     if not caps:
         badresponse()
 
     # Flush any output on stderr before proceeding.
     _forwardoutput(ui, stderr)
 
     return protoname, caps
 
-class sshv1peer(wireproto.wirepeer):
+class sshv1peer(wireprotov1peer.wirepeer):
     def __init__(self, ui, url, proc, stdin, stdout, stderr, caps,
                  autoreadstderr=True):
         """Create a peer from an existing SSH connection.
 
         ``proc`` is a handle on the underlying SSH process.
         ``stdin``, ``stdout``, and ``stderr`` are handles on the stdio
         pipes for that process.
         ``caps`` is a set of capabilities supported by the remote.
         ``autoreadstderr`` denotes whether to automatically read from
         stderr and to forward its output.
         """
         self._url = url
         self.ui = ui
         # self._subprocess is unused. Keeping a handle on the process
         # holds a reference and prevents it from being garbage collected.
         self._subprocess = proc
 
         # And we hook up our "doublepipe" wrapper to allow querying
         # stderr any time we perform I/O.
         if autoreadstderr:
             stdout = doublepipe(ui, util.bufferedinputpipe(stdout), stderr)
             stdin = doublepipe(ui, stdin, stderr)
 
         self._pipeo = stdin
         self._pipei = stdout
         self._pipee = stderr
         self._caps = caps
         self._autoreadstderr = autoreadstderr
 
     # Commands that have a "framed" response where the first line of the
     # response contains the length of that response.
     _FRAMED_COMMANDS = {
         'batch',
     }
 
     # Begin of ipeerconnection interface.
 
     def url(self):
         return self._url
 
     def local(self):
         return None
 
     def peer(self):
         return self
 
     def canpush(self):
         return True
 
     def close(self):
         pass
 
     # End of ipeerconnection interface.
 
     # Begin of ipeercommands interface.
 
     def capabilities(self):
         return self._caps
 
     # End of ipeercommands interface.
 
     def _readerr(self):
         _forwardoutput(self.ui, self._pipee)
 
     def _abort(self, exception):
         self._cleanup()
         raise exception
 
     def _cleanup(self):
         _cleanuppipes(self.ui, self._pipei, self._pipeo, self._pipee)
 
     __del__ = _cleanup
 
     def _sendrequest(self, cmd, args, framed=False):
         if (self.ui.debugflag
             and self.ui.configbool('devel', 'debug.peer-request')):
             dbg = self.ui.debug
             line = 'devel-peer-request: %s\n'
             dbg(line % cmd)
             for key, value in sorted(args.items()):
                 if not isinstance(value, dict):
                     dbg(line % '  %s: %d bytes' % (key, len(value)))
                 else:
                     for dk, dv in sorted(value.items()):
                         dbg(line % '  %s-%s: %d' % (key, dk, len(dv)))
         self.ui.debug("sending %s command\n" % cmd)
         self._pipeo.write("%s\n" % cmd)
         _func, names = wireproto.commands[cmd]
         keys = names.split()
         wireargs = {}
         for k in keys:
             if k == '*':
                 wireargs['*'] = args
                 break
             else:
                 wireargs[k] = args[k]
                 del args[k]
         for k, v in sorted(wireargs.iteritems()):
             self._pipeo.write("%s %d\n" % (k, len(v)))
             if isinstance(v, dict):
                 for dk, dv in v.iteritems():
                     self._pipeo.write("%s %d\n" % (dk, len(dv)))
                     self._pipeo.write(dv)
             else:
                 self._pipeo.write(v)
         self._pipeo.flush()
 
         # We know exactly how many bytes are in the response. So return a proxy
         # around the raw output stream that allows reading exactly this many
         # bytes. Callers then can read() without fear of overrunning the
         # response.
         if framed:
             amount = self._getamount()
             return util.cappedreader(self._pipei, amount)
 
         return self._pipei
 
     def _callstream(self, cmd, **args):
         args = pycompat.byteskwargs(args)
         return self._sendrequest(cmd, args, framed=cmd in self._FRAMED_COMMANDS)
 
     def _callcompressable(self, cmd, **args):
         args = pycompat.byteskwargs(args)
         return self._sendrequest(cmd, args, framed=cmd in self._FRAMED_COMMANDS)
 
     def _call(self, cmd, **args):
         args = pycompat.byteskwargs(args)
         return self._sendrequest(cmd, args, framed=True).read()
 
     def _callpush(self, cmd, fp, **args):
         # The server responds with an empty frame if the client should
         # continue submitting the payload.
         r = self._call(cmd, **args)
         if r:
             return '', r
 
         # The payload consists of frames with content followed by an empty
         # frame.
         for d in iter(lambda: fp.read(4096), ''):
             self._writeframed(d)
         self._writeframed("", flush=True)
 
         # In case of success, there is an empty frame and a frame containing
         # the integer result (as a string).
         # In case of error, there is a non-empty frame containing the error.
         r = self._readframed()
         if r:
             return '', r
         return self._readframed(), ''
 
     def _calltwowaystream(self, cmd, fp, **args):
         # The server responds with an empty frame if the client should
         # continue submitting the payload.
         r = self._call(cmd, **args)
         if r:
             # XXX needs to be made better
             raise error.Abort(_('unexpected remote reply: %s') % r)
 
         # The payload consists of frames with content followed by an empty
         # frame.
         for d in iter(lambda: fp.read(4096), ''):
             self._writeframed(d)
         self._writeframed("", flush=True)
 
         return self._pipei
 
     def _getamount(self):
         l = self._pipei.readline()
         if l == '\n':
             if self._autoreadstderr:
                 self._readerr()
             msg = _('check previous remote output')
             self._abort(error.OutOfBandError(hint=msg))
         if self._autoreadstderr:
             self._readerr()
         try:
             return int(l)
         except ValueError:
             self._abort(error.ResponseError(_("unexpected response:"), l))
 
     def _readframed(self):
         size = self._getamount()
         if not size:
             return b''
 
         return self._pipei.read(size)
 
     def _writeframed(self, data, flush=False):
         self._pipeo.write("%d\n" % len(data))
         if data:
             self._pipeo.write(data)
         if flush:
             self._pipeo.flush()
         if self._autoreadstderr:
             self._readerr()
 
 class sshv2peer(sshv1peer):
     """A peer that speakers version 2 of the transport protocol."""
     # Currently version 2 is identical to version 1 post handshake.
     # And handshake is performed before the peer is instantiated. So
     # we need no custom code.
 
 def makepeer(ui, path, proc, stdin, stdout, stderr, autoreadstderr=True):
     """Make a peer instance from existing pipes.
 
     ``path`` and ``proc`` are stored on the eventual peer instance and may
     not be used for anything meaningful.
 
     ``stdin``, ``stdout``, and ``stderr`` are the pipes connected to the
     SSH server's stdio handles.
 
     This function is factored out to allow creating peers that don't
     actually spawn a new process. It is useful for starting SSH protocol
     servers and clients via non-standard means, which can be useful for
     testing.
     """
     try:
         protoname, caps = _performhandshake(ui, stdin, stdout, stderr)
     except Exception:
         _cleanuppipes(ui, stdout, stdin, stderr)
         raise
 
     if protoname == wireprototypes.SSHV1:
         return sshv1peer(ui, path, proc, stdin, stdout, stderr, caps,
                          autoreadstderr=autoreadstderr)
     elif protoname == wireprototypes.SSHV2:
         return sshv2peer(ui, path, proc, stdin, stdout, stderr, caps,
                          autoreadstderr=autoreadstderr)
     else:
         _cleanuppipes(ui, stdout, stdin, stderr)
         raise error.RepoError(_('unknown version of SSH protocol: %s') %
                               protoname)
 
 def instance(ui, path, create):
     """Create an SSH peer.
 
-    The returned object conforms to the ``wireproto.wirepeer`` interface.
+    The returned object conforms to the ``wireprotov1peer.wirepeer`` interface.
     """
     u = util.url(path, parsequery=False, parsefragment=False)
     if u.scheme != 'ssh' or not u.host or u.path is None:
         raise error.RepoError(_("couldn't parse location %s") % path)
 
     util.checksafessh(path)
 
     if u.passwd is not None:
         raise error.RepoError(_('password in URL not supported'))
 
     sshcmd = ui.config('ui', 'ssh')
     remotecmd = ui.config('ui', 'remotecmd')
     sshaddenv = dict(ui.configitems('sshenv'))
     sshenv = procutil.shellenviron(sshaddenv)
     remotepath = u.path or '.'
 
     args = procutil.sshargs(sshcmd, u.host, u.user, u.port)
 
     if create:
         cmd = '%s %s %s' % (sshcmd, args,
             procutil.shellquote('%s init %s' %
                 (_serverquote(remotecmd), _serverquote(remotepath))))
         ui.debug('running %s\n' % cmd)
         res = ui.system(cmd, blockedtag='sshpeer', environ=sshenv)
         if res != 0:
             raise error.RepoError(_('could not create remote repo'))
 
     proc, stdin, stdout, stderr = _makeconnection(ui, sshcmd, args, remotecmd,
                                                   remotepath, sshenv)
 
     peer = makepeer(ui, path, proc, stdin, stdout, stderr)
 
     # Finally, if supported by the server, notify it about our own
     # capabilities.
     if 'protocaps' in peer.capabilities():
         try:
             peer._call("protocaps",
                        caps=' '.join(sorted(_clientcapabilities())))
         except IOError:
             peer._cleanup()
             raise error.RepoError(_('capability exchange failed'))
 
     return peer

mercurial/wireproto.py

@@ -1,1265 +1,866 b''
 # 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 __future__ import absolute_import
 
-import hashlib
 import os
 import tempfile
 
 from .i18n import _
 from .node import (
-    bin,
     hex,
     nullid,
 )
 
 from . import (
     bundle2,
     changegroup as changegroupmod,
     discovery,
     encoding,
     error,
     exchange,
-    peer,
     pushkey as pushkeymod,
     pycompat,
-    repository,
     streamclone,
     util,
     wireprototypes,
 )
 
 from .utils import (
     procutil,
     stringutil,
 )
 
 urlerr = util.urlerr
 urlreq = util.urlreq
 
 bundle2requiredmain = _('incompatible Mercurial client; bundle2 required')
 bundle2requiredhint = _('see https://www.mercurial-scm.org/wiki/'
                         'IncompatibleClient')
 bundle2required = '%s\n(%s)\n' % (bundle2requiredmain, bundle2requiredhint)
 
-class remoteiterbatcher(peer.iterbatcher):
-    def __init__(self, remote):
-        super(remoteiterbatcher, self).__init__()
-        self._remote = remote
-
-    def __getattr__(self, name):
-        # Validate this method is batchable, since submit() only supports
-        # batchable methods.
-        fn = getattr(self._remote, name)
-        if not getattr(fn, 'batchable', None):
-            raise error.ProgrammingError('Attempted to batch a non-batchable '
-                                         'call to %r' % name)
-
-        return super(remoteiterbatcher, self).__getattr__(name)
-
-    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.
-        """
-        # 2-tuple of (command, arguments) that represents what will be
-        # sent over the wire.
-        requests = []
-
-        # 4-tuple of (command, final future, @batchable generator, remote
-        # future).
-        results = []
-
-        for command, args, opts, finalfuture in self.calls:
-            mtd = getattr(self._remote, command)
-            batchable = mtd.batchable(mtd.__self__, *args, **opts)
-
-            commandargs, fremote = next(batchable)
-            assert fremote
-            requests.append((command, commandargs))
-            results.append((command, finalfuture, batchable, fremote))
-
-        if requests:
-            self._resultiter = self._remote._submitbatch(requests)
-
-        self._results = results
-
-    def results(self):
-        for command, finalfuture, batchable, remotefuture in self._results:
-            # Get the raw result, set it in the remote future, feed it
-            # back into the @batchable generator so it can be decoded, and
-            # set the result on the final future to this value.
-            remoteresult = next(self._resultiter)
-            remotefuture.set(remoteresult)
-            finalfuture.set(next(batchable))
-
-            # Verify our @batchable generators only emit 2 values.
-            try:
-                next(batchable)
-            except StopIteration:
-                pass
-            else:
-                raise error.ProgrammingError('%s @batchable generator emitted '
-                                             'unexpected value count' % command)
-
-            yield finalfuture.value
-
-# Forward a couple of names from peer to make wireproto interactions
-# slightly more sensible.
-batchable = peer.batchable
-future = peer.future
-
-
-def encodebatchcmds(req):
-    """Return a ``cmds`` argument value for the ``batch`` command."""
-    escapearg = wireprototypes.escapebatcharg
-
-    cmds = []
-    for op, argsdict in req:
-        # Old servers didn't properly unescape argument names. So prevent
-        # the sending of argument names that may not be decoded properly by
-        # servers.
-        assert all(escapearg(k) == k for k in argsdict)
-
-        args = ','.join('%s=%s' % (escapearg(k), escapearg(v))
-                        for k, v in argsdict.iteritems())
-        cmds.append('%s %s' % (op, args))
-
-    return ';'.join(cmds)
-
 def clientcompressionsupport(proto):
     """Returns a list of compression methods supported by the client.
 
     Returns a list of the compression methods supported by the client
     according to the protocol capabilities. If no such capability has
     been announced, fallback to the default of zlib and uncompressed.
     """
     for cap in proto.getprotocaps():
         if cap.startswith('comp='):
             return cap[5:].split(',')
     return ['zlib', 'none']
 
-# client side
-
-class wirepeer(repository.legacypeer):
-    """Client-side interface for communicating with a peer repository.
-
-    Methods commonly call wire protocol commands of the same name.
-
-    See also httppeer.py and sshpeer.py for protocol-specific
-    implementations of this interface.
-    """
-    # Begin of ipeercommands interface.
-
-    def iterbatch(self):
-        return remoteiterbatcher(self)
-
-    @batchable
-    def lookup(self, key):
-        self.requirecap('lookup', _('look up remote revision'))
-        f = future()
-        yield {'key': encoding.fromlocal(key)}, f
-        d = f.value
-        success, data = d[:-1].split(" ", 1)
-        if int(success):
-            yield bin(data)
-        else:
-            self._abort(error.RepoError(data))
-
-    @batchable
-    def heads(self):
-        f = future()
-        yield {}, f
-        d = f.value
-        try:
-            yield wireprototypes.decodelist(d[:-1])
-        except ValueError:
-            self._abort(error.ResponseError(_("unexpected response:"), d))
-
-    @batchable
-    def known(self, nodes):
-        f = future()
-        yield {'nodes': wireprototypes.encodelist(nodes)}, f
-        d = f.value
-        try:
-            yield [bool(int(b)) for b in d]
-        except ValueError:
-            self._abort(error.ResponseError(_("unexpected response:"), d))
-
-    @batchable
-    def branchmap(self):
-        f = future()
-        yield {}, f
-        d = f.value
-        try:
-            branchmap = {}
-            for branchpart in d.splitlines():
-                branchname, branchheads = branchpart.split(' ', 1)
-                branchname = encoding.tolocal(urlreq.unquote(branchname))
-                branchheads = wireprototypes.decodelist(branchheads)
-                branchmap[branchname] = branchheads
-            yield branchmap
-        except TypeError:
-            self._abort(error.ResponseError(_("unexpected response:"), 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)
-
-    @batchable
-    def pushkey(self, namespace, key, old, new):
-        if not self.capable('pushkey'):
-            yield False, None
-        f = future()
-        self.ui.debug('preparing pushkey for "%s:%s"\n' % (namespace, key))
-        yield {'namespace': encoding.fromlocal(namespace),
-               'key': encoding.fromlocal(key),
-               'old': encoding.fromlocal(old),
-               'new': encoding.fromlocal(new)}, f
-        d = f.value
-        d, output = d.split('\n', 1)
-        try:
-            d = bool(int(d))
-        except ValueError:
-            raise error.ResponseError(
-                _('push failed (unexpected response):'), d)
-        for l in output.splitlines(True):
-            self.ui.status(_('remote: '), l)
-        yield d
-
-    def stream_out(self):
-        return self._callstream('stream_out')
-
-    def getbundle(self, source, **kwargs):
-        kwargs = pycompat.byteskwargs(kwargs)
-        self.requirecap('getbundle', _('look up remote changes'))
-        opts = {}
-        bundlecaps = kwargs.get('bundlecaps') or set()
-        for key, value in kwargs.iteritems():
-            if value is None:
-                continue
-            keytype = wireprototypes.GETBUNDLE_ARGUMENTS.get(key)
-            if keytype is None:
-                raise error.ProgrammingError(
-                    'Unexpectedly None keytype for key %s' % key)
-            elif keytype == 'nodes':
-                value = wireprototypes.encodelist(value)
-            elif keytype == 'csv':
-                value = ','.join(value)
-            elif keytype == 'scsv':
-                value = ','.join(sorted(value))
-            elif keytype == 'boolean':
-                value = '%i' % bool(value)
-            elif keytype != 'plain':
-                raise KeyError('unknown getbundle option type %s'
-                               % keytype)
-            opts[key] = value
-        f = self._callcompressable("getbundle", **pycompat.strkwargs(opts))
-        if any((cap.startswith('HG2') for cap in bundlecaps)):
-            return bundle2.getunbundler(self.ui, f)
-        else:
-            return changegroupmod.cg1unpacker(f, 'UN')
-
-    def unbundle(self, cg, heads, url):
-        '''Send cg (a readable file-like object representing the
-        changegroup to push, typically a chunkbuffer object) to the
-        remote server as a bundle.
-
-        When pushing a bundle10 stream, return an integer indicating the
-        result of the push (see changegroup.apply()).
-
-        When pushing a bundle20 stream, return a bundle20 stream.
-
-        `url` is the url the client thinks it's pushing to, which is
-        visible to hooks.
-        '''
-
-        if heads != ['force'] and self.capable('unbundlehash'):
-            heads = wireprototypes.encodelist(
-                ['hashed', hashlib.sha1(''.join(sorted(heads))).digest()])
-        else:
-            heads = wireprototypes.encodelist(heads)
-
-        if util.safehasattr(cg, 'deltaheader'):
-            # this a bundle10, do the old style call sequence
-            ret, output = self._callpush("unbundle", cg, heads=heads)
-            if ret == "":
-                raise error.ResponseError(
-                    _('push failed:'), output)
-            try:
-                ret = int(ret)
-            except ValueError:
-                raise error.ResponseError(
-                    _('push failed (unexpected response):'), ret)
-
-            for l in output.splitlines(True):
-                self.ui.status(_('remote: '), l)
-        else:
-            # bundle2 push. Send a stream, fetch a stream.
-            stream = self._calltwowaystream('unbundle', cg, heads=heads)
-            ret = bundle2.getunbundler(self.ui, stream)
-        return ret
-
-    # End of ipeercommands interface.
-
-    # Begin of ipeerlegacycommands interface.
-
-    def branches(self, nodes):
-        n = wireprototypes.encodelist(nodes)
-        d = self._call("branches", nodes=n)
-        try:
-            br = [tuple(wireprototypes.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([wireprototypes.encodelist(p, '-')
-                          for p in pairs[i:i + batch]])
-            d = self._call("between", pairs=n)
-            try:
-                r.extend(l and wireprototypes.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 = wireprototypes.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 = wireprototypes.encodelist(bases)
-        heads = wireprototypes.encodelist(heads)
-        f = self._callcompressable("changegroupsubset",
-                                   bases=bases, heads=heads)
-        return changegroupmod.cg1unpacker(f, 'UN')
-
-    # End of ipeerlegacycommands interface.
-
-    def _submitbatch(self, req):
-        """run batch request <req> on the server
-
-        Returns an iterator of the raw responses from the server.
-        """
-        ui = self.ui
-        if ui.debugflag and ui.configbool('devel', 'debug.peer-request'):
-            ui.debug('devel-peer-request: batched-content\n')
-            for op, args in req:
-                msg = 'devel-peer-request:    - %s (%d arguments)\n'
-                ui.debug(msg % (op, len(args)))
-
-        unescapearg = wireprototypes.unescapebatcharg
-
-        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, **pycompat.strkwargs(args))
-
-    def debugwireargs(self, one, two, three=None, four=None, five=None):
-        # don't pass optional arguments left at their default value
-        opts = {}
-        if three is not None:
-            opts[r'three'] = three
-        if four is not None:
-            opts[r'four'] = four
-        return self._call('debugwireargs', one=one, two=two, **opts)
-
-    def _call(self, cmd, **args):
-        """execute <cmd> on the server
-
-        The command is expected to return a simple string.
-
-        returns the server reply as a string."""
-        raise NotImplementedError()
-
-    def _callstream(self, cmd, **args):
-        """execute <cmd> on the server
-
-        The command is expected to return a stream. Note that if the
-        command doesn't return a stream, _callstream behaves
-        differently for ssh and http peers.
-
-        returns the server reply as a file like object.
-        """
-        raise NotImplementedError()
-
-    def _callcompressable(self, cmd, **args):
-        """execute <cmd> on the server
-
-        The command is expected to return a stream.
-
-        The stream may have been compressed in some implementations. This
-        function takes care of the decompression. This is the only difference
-        with _callstream.
-
-        returns the server reply as a file like object.
-        """
-        raise NotImplementedError()
-
-    def _callpush(self, cmd, fp, **args):
-        """execute a <cmd> on server
-
-        The command is expected to be related to a push. Push has a special
-        return method.
-
-        returns the server reply as a (ret, output) tuple. ret is either
-        empty (error) or a stringified int.
-        """
-        raise NotImplementedError()
-
-    def _calltwowaystream(self, cmd, fp, **args):
-        """execute <cmd> on server
-
-        The command will send a stream to the server and get a stream in reply.
-        """
-        raise NotImplementedError()
-
-    def _abort(self, exception):
-        """clearly abort the wire protocol connection and raise the exception
-        """
-        raise NotImplementedError()
-
-# server side
-
 # wire protocol command can either return a string or one of these classes.
 
 def getdispatchrepo(repo, proto, command):
     """Obtain the repo used for processing wire protocol commands.
 
     The intent of this function is to serve as a monkeypatch point for
     extensions that need commands to operate on different repo views under
     specialized circumstances.
     """
     return repo.filtered('served')
 
 def dispatch(repo, proto, command):
     repo = getdispatchrepo(repo, proto, command)
 
     transportversion = wireprototypes.TRANSPORTS[proto.name]['version']
     commandtable = commandsv2 if transportversion == 2 else commands
     func, spec = commandtable[command]
 
     args = proto.getargs(spec)
 
     # Version 1 protocols define arguments as a list. Version 2 uses a dict.
     if isinstance(args, list):
         return func(repo, proto, *args)
     elif isinstance(args, dict):
         return func(repo, proto, **args)
     else:
         raise error.ProgrammingError('unexpected type returned from '
                                      'proto.getargs(): %s' % type(args))
 
 def options(cmd, keys, others):
     opts = {}
     for k in keys:
         if k in others:
             opts[k] = others[k]
             del others[k]
     if others:
         procutil.stderr.write("warning: %s ignored unexpected arguments %s\n"
                               % (cmd, ",".join(others)))
     return opts
 
 def bundle1allowed(repo, action):
     """Whether a bundle1 operation is allowed from the server.
 
     Priority is:
 
     1. server.bundle1gd.<action> (if generaldelta active)
     2. server.bundle1.<action>
     3. server.bundle1gd (if generaldelta active)
     4. server.bundle1
     """
     ui = repo.ui
     gd = 'generaldelta' in repo.requirements
 
     if gd:
         v = ui.configbool('server', 'bundle1gd.%s' % action)
         if v is not None:
             return v
 
     v = ui.configbool('server', 'bundle1.%s' % action)
     if v is not None:
         return v
 
     if gd:
         v = ui.configbool('server', 'bundle1gd')
         if v is not None:
             return v
 
     return ui.configbool('server', 'bundle1')
 
 def supportedcompengines(ui, role):
     """Obtain the list of supported compression engines for a request."""
     assert role in (util.CLIENTROLE, util.SERVERROLE)
 
     compengines = util.compengines.supportedwireengines(role)
 
     # Allow config to override default list and ordering.
     if role == util.SERVERROLE:
         configengines = ui.configlist('server', 'compressionengines')
         config = 'server.compressionengines'
     else:
         # This is currently implemented mainly to facilitate testing. In most
         # cases, the server should be in charge of choosing a compression engine
         # because a server has the most to lose from a sub-optimal choice. (e.g.
         # CPU DoS due to an expensive engine or a network DoS due to poor
         # compression ratio).
         configengines = ui.configlist('experimental',
                                       'clientcompressionengines')
         config = 'experimental.clientcompressionengines'
 
     # No explicit config. Filter out the ones that aren't supposed to be
     # advertised and return default ordering.
     if not configengines:
         attr = 'serverpriority' if role == util.SERVERROLE else 'clientpriority'
         return [e for e in compengines
                 if getattr(e.wireprotosupport(), attr) > 0]
 
     # If compression engines are listed in the config, assume there is a good
     # reason for it (like server operators wanting to achieve specific
     # performance characteristics). So fail fast if the config references
     # unusable compression engines.
     validnames = set(e.name() for e in compengines)
     invalidnames = set(e for e in configengines if e not in validnames)
     if invalidnames:
         raise error.Abort(_('invalid compression engine defined in %s: %s') %
                           (config, ', '.join(sorted(invalidnames))))
 
     compengines = [e for e in compengines if e.name() in configengines]
     compengines = sorted(compengines,
                          key=lambda e: configengines.index(e.name()))
 
     if not compengines:
         raise error.Abort(_('%s config option does not specify any known '
                             'compression engines') % config,
                           hint=_('usable compression engines: %s') %
                           ', '.sorted(validnames))
 
     return compengines
 
 class commandentry(object):
     """Represents a declared wire protocol command."""
     def __init__(self, func, args='', transports=None,
                  permission='push'):
         self.func = func
         self.args = args
         self.transports = transports or set()
         self.permission = permission
 
     def _merge(self, func, args):
         """Merge this instance with an incoming 2-tuple.
 
         This is called when a caller using the old 2-tuple API attempts
         to replace an instance. The incoming values are merged with
         data not captured by the 2-tuple and a new instance containing
         the union of the two objects is returned.
         """
         return commandentry(func, args=args, transports=set(self.transports),
                             permission=self.permission)
 
     # Old code treats instances as 2-tuples. So expose that interface.
     def __iter__(self):
         yield self.func
         yield self.args
 
     def __getitem__(self, i):
         if i == 0:
             return self.func
         elif i == 1:
             return self.args
         else:
             raise IndexError('can only access elements 0 and 1')
 
 class commanddict(dict):
     """Container for registered wire protocol commands.
 
     It behaves like a dict. But __setitem__ is overwritten to allow silent
     coercion of values from 2-tuples for API compatibility.
     """
     def __setitem__(self, k, v):
         if isinstance(v, commandentry):
             pass
         # Cast 2-tuples to commandentry instances.
         elif isinstance(v, tuple):
             if len(v) != 2:
                 raise ValueError('command tuples must have exactly 2 elements')
 
             # It is common for extensions to wrap wire protocol commands via
             # e.g. ``wireproto.commands[x] = (newfn, args)``. Because callers
             # doing this aren't aware of the new API that uses objects to store
             # command entries, we automatically merge old state with new.
             if k in self:
                 v = self[k]._merge(v[0], v[1])
             else:
                 # Use default values from @wireprotocommand.
                 v = commandentry(v[0], args=v[1],
                                  transports=set(wireprototypes.TRANSPORTS),
                                  permission='push')
         else:
             raise ValueError('command entries must be commandentry instances '
                              'or 2-tuples')
 
         return super(commanddict, self).__setitem__(k, v)
 
     def commandavailable(self, command, proto):
         """Determine if a command is available for the requested protocol."""
         assert proto.name in wireprototypes.TRANSPORTS
 
         entry = self.get(command)
 
         if not entry:
             return False
 
         if proto.name not in entry.transports:
             return False
 
         return True
 
 # Constants specifying which transports a wire protocol command should be
 # available on. For use with @wireprotocommand.
 POLICY_V1_ONLY = 'v1-only'
 POLICY_V2_ONLY = 'v2-only'
 
 # For version 1 transports.
 commands = commanddict()
 
 # For version 2 transports.
 commandsv2 = commanddict()
 
 def wireprotocommand(name, args=None, transportpolicy=POLICY_V1_ONLY,
                      permission='push'):
     """Decorator to declare a wire protocol command.
 
     ``name`` is the name of the wire protocol command being provided.
 
     ``args`` defines the named arguments accepted by the command. It is
     ideally a dict mapping argument names to their types. For backwards
     compatibility, it can be a space-delimited list of argument names. For
     version 1 transports, ``*`` denotes a special value that says to accept
     all named arguments.
 
     ``transportpolicy`` is a POLICY_* constant denoting which transports
     this wire protocol command should be exposed to. By default, commands
     are exposed to all wire protocol transports.
 
     ``permission`` defines the permission type needed to run this command.
     Can be ``push`` or ``pull``. These roughly map to read-write and read-only,
     respectively. Default is to assume command requires ``push`` permissions
     because otherwise commands not declaring their permissions could modify
     a repository that is supposed to be read-only.
     """
     if transportpolicy == POLICY_V1_ONLY:
         transports = {k for k, v in wireprototypes.TRANSPORTS.items()
                       if v['version'] == 1}
         transportversion = 1
     elif transportpolicy == POLICY_V2_ONLY:
         transports = {k for k, v in wireprototypes.TRANSPORTS.items()
                       if v['version'] == 2}
         transportversion = 2
     else:
         raise error.ProgrammingError('invalid transport policy value: %s' %
                                      transportpolicy)
 
     # Because SSHv2 is a mirror of SSHv1, we allow "batch" commands through to
     # SSHv2.
     # TODO undo this hack when SSH is using the unified frame protocol.
     if name == b'batch':
         transports.add(wireprototypes.SSHV2)
 
     if permission not in ('push', 'pull'):
         raise error.ProgrammingError('invalid wire protocol permission; '
                                      'got %s; expected "push" or "pull"' %
                                      permission)
 
     if transportversion == 1:
         if args is None:
             args = ''
 
         if not isinstance(args, bytes):
             raise error.ProgrammingError('arguments for version 1 commands '
                                          'must be declared as bytes')
     elif transportversion == 2:
         if args is None:
             args = {}
 
         if not isinstance(args, dict):
             raise error.ProgrammingError('arguments for version 2 commands '
                                          'must be declared as dicts')
 
     def register(func):
         if transportversion == 1:
             if name in commands:
                 raise error.ProgrammingError('%s command already registered '
                                              'for version 1' % name)
             commands[name] = commandentry(func, args=args,
                                           transports=transports,
                                           permission=permission)
         elif transportversion == 2:
             if name in commandsv2:
                 raise error.ProgrammingError('%s command already registered '
                                              'for version 2' % name)
 
             commandsv2[name] = commandentry(func, args=args,
                                             transports=transports,
                                             permission=permission)
         else:
             raise error.ProgrammingError('unhandled transport version: %d' %
                                          transportversion)
 
         return func
     return register
 
 # TODO define a more appropriate permissions type to use for this.
 @wireprotocommand('batch', 'cmds *', permission='pull',
                   transportpolicy=POLICY_V1_ONLY)
 def batch(repo, proto, cmds, others):
     unescapearg = wireprototypes.unescapebatcharg
     repo = repo.filtered("served")
     res = []
     for pair in cmds.split(';'):
         op, args = pair.split(' ', 1)
         vals = {}
         for a in args.split(','):
             if a:
                 n, v = a.split('=')
                 vals[unescapearg(n)] = unescapearg(v)
         func, spec = commands[op]
 
         # Validate that client has permissions to perform this command.
         perm = commands[op].permission
         assert perm in ('push', 'pull')
         proto.checkperm(perm)
 
         if spec:
             keys = spec.split()
             data = {}
             for k in keys:
                 if k == '*':
                     star = {}
                     for key in vals.keys():
                         if key not in keys:
                             star[key] = vals[key]
                     data['*'] = star
                 else:
                     data[k] = vals[k]
             result = func(repo, proto, *[data[k] for k in keys])
         else:
             result = func(repo, proto)
         if isinstance(result, wireprototypes.ooberror):
             return result
 
         # For now, all batchable commands must return bytesresponse or
         # raw bytes (for backwards compatibility).
         assert isinstance(result, (wireprototypes.bytesresponse, bytes))
         if isinstance(result, wireprototypes.bytesresponse):
             result = result.data
         res.append(wireprototypes.escapebatcharg(result))
 
     return wireprototypes.bytesresponse(';'.join(res))
 
 @wireprotocommand('between', 'pairs', transportpolicy=POLICY_V1_ONLY,
                   permission='pull')
 def between(repo, proto, pairs):
     pairs = [wireprototypes.decodelist(p, '-') for p in pairs.split(" ")]
     r = []
     for b in repo.between(pairs):
         r.append(wireprototypes.encodelist(b) + "\n")
 
     return wireprototypes.bytesresponse(''.join(r))
 
 @wireprotocommand('branchmap', permission='pull',
                   transportpolicy=POLICY_V1_ONLY)
 def branchmap(repo, proto):
     branchmap = repo.branchmap()
     heads = []
     for branch, nodes in branchmap.iteritems():
         branchname = urlreq.quote(encoding.fromlocal(branch))
         branchnodes = wireprototypes.encodelist(nodes)
         heads.append('%s %s' % (branchname, branchnodes))
 
     return wireprototypes.bytesresponse('\n'.join(heads))
 
 @wireprotocommand('branches', 'nodes', transportpolicy=POLICY_V1_ONLY,
                   permission='pull')
 def branches(repo, proto, nodes):
     nodes = wireprototypes.decodelist(nodes)
     r = []
     for b in repo.branches(nodes):
         r.append(wireprototypes.encodelist(b) + "\n")
 
     return wireprototypes.bytesresponse(''.join(r))
 
 @wireprotocommand('clonebundles', '', permission='pull',
                   transportpolicy=POLICY_V1_ONLY)
 def clonebundles(repo, proto):
     """Server command for returning info for available bundles to seed clones.
 
     Clients will parse this response and determine what bundle to fetch.
 
     Extensions may wrap this command to filter or dynamically emit data
     depending on the request. e.g. you could advertise URLs for the closest
     data center given the client's IP address.
     """
     return wireprototypes.bytesresponse(
         repo.vfs.tryread('clonebundles.manifest'))
 
 wireprotocaps = ['lookup', 'branchmap', 'pushkey',
                  'known', 'getbundle', 'unbundlehash']
 
 def _capabilities(repo, proto):
     """return a list of capabilities for a repo
 
     This function exists to allow extensions to easily wrap capabilities
     computation
 
     - returns a lists: easy to alter
     - change done here will be propagated to both `capabilities` and `hello`
       command without any other action needed.
     """
     # copy to prevent modification of the global list
     caps = list(wireprotocaps)
 
     # Command of same name as capability isn't exposed to version 1 of
     # transports. So conditionally add it.
     if commands.commandavailable('changegroupsubset', proto):
         caps.append('changegroupsubset')
 
     if streamclone.allowservergeneration(repo):
         if repo.ui.configbool('server', 'preferuncompressed'):
             caps.append('stream-preferred')
         requiredformats = repo.requirements & repo.supportedformats
         # if our local revlogs are just revlogv1, add 'stream' cap
         if not requiredformats - {'revlogv1'}:
             caps.append('stream')
         # otherwise, add 'streamreqs' detailing our local revlog format
         else:
             caps.append('streamreqs=%s' % ','.join(sorted(requiredformats)))
     if repo.ui.configbool('experimental', 'bundle2-advertise'):
         capsblob = bundle2.encodecaps(bundle2.getrepocaps(repo, role='server'))
         caps.append('bundle2=' + urlreq.quote(capsblob))
     caps.append('unbundle=%s' % ','.join(bundle2.bundlepriority))
 
     return proto.addcapabilities(repo, caps)
 
 # If you are writing an extension and consider wrapping this function. Wrap
 # `_capabilities` instead.
 @wireprotocommand('capabilities', permission='pull',
                   transportpolicy=POLICY_V1_ONLY)
 def capabilities(repo, proto):
     caps = _capabilities(repo, proto)
     return wireprototypes.bytesresponse(' '.join(sorted(caps)))
 
 @wireprotocommand('changegroup', 'roots', transportpolicy=POLICY_V1_ONLY,
                   permission='pull')
 def changegroup(repo, proto, roots):
     nodes = wireprototypes.decodelist(roots)
     outgoing = discovery.outgoing(repo, missingroots=nodes,
                                   missingheads=repo.heads())
     cg = changegroupmod.makechangegroup(repo, outgoing, '01', 'serve')
     gen = iter(lambda: cg.read(32768), '')
     return wireprototypes.streamres(gen=gen)
 
 @wireprotocommand('changegroupsubset', 'bases heads',
                   transportpolicy=POLICY_V1_ONLY,
                   permission='pull')
 def changegroupsubset(repo, proto, bases, heads):
     bases = wireprototypes.decodelist(bases)
     heads = wireprototypes.decodelist(heads)
     outgoing = discovery.outgoing(repo, missingroots=bases,
                                   missingheads=heads)
     cg = changegroupmod.makechangegroup(repo, outgoing, '01', 'serve')
     gen = iter(lambda: cg.read(32768), '')
     return wireprototypes.streamres(gen=gen)
 
 @wireprotocommand('debugwireargs', 'one two *',
                   permission='pull', transportpolicy=POLICY_V1_ONLY)
 def debugwireargs(repo, proto, one, two, others):
     # only accept optional args from the known set
     opts = options('debugwireargs', ['three', 'four'], others)
     return wireprototypes.bytesresponse(repo.debugwireargs(
         one, two, **pycompat.strkwargs(opts)))
 
 def find_pullbundle(repo, proto, opts, clheads, heads, common):
     """Return a file object for the first matching pullbundle.
 
     Pullbundles are specified in .hg/pullbundles.manifest similar to
     clonebundles.
     For each entry, the bundle specification is checked for compatibility:
     - Client features vs the BUNDLESPEC.
     - Revisions shared with the clients vs base revisions of the bundle.
       A bundle can be applied only if all its base revisions are known by
       the client.
     - At least one leaf of the bundle's DAG is missing on the client.
     - Every leaf of the bundle's DAG is part of node set the client wants.
       E.g. do not send a bundle of all changes if the client wants only
       one specific branch of many.
     """
     def decodehexstring(s):
         return set([h.decode('hex') for h in s.split(';')])
 
     manifest = repo.vfs.tryread('pullbundles.manifest')
     if not manifest:
         return None
     res = exchange.parseclonebundlesmanifest(repo, manifest)
     res = exchange.filterclonebundleentries(repo, res)
     if not res:
         return None
     cl = repo.changelog
     heads_anc = cl.ancestors([cl.rev(rev) for rev in heads], inclusive=True)
     common_anc = cl.ancestors([cl.rev(rev) for rev in common], inclusive=True)
     compformats = clientcompressionsupport(proto)
     for entry in res:
         if 'COMPRESSION' in entry and entry['COMPRESSION'] not in compformats:
             continue
         # No test yet for VERSION, since V2 is supported by any client
         # that advertises partial pulls
         if 'heads' in entry:
             try:
                 bundle_heads = decodehexstring(entry['heads'])
             except TypeError:
                 # Bad heads entry
                 continue
             if bundle_heads.issubset(common):
                 continue # Nothing new
             if all(cl.rev(rev) in common_anc for rev in bundle_heads):
                 continue # Still nothing new
             if any(cl.rev(rev) not in heads_anc and
                    cl.rev(rev) not in common_anc for rev in bundle_heads):
                 continue
         if 'bases' in entry:
             try:
                 bundle_bases = decodehexstring(entry['bases'])
             except TypeError:
                 # Bad bases entry
                 continue
             if not all(cl.rev(rev) in common_anc for rev in bundle_bases):
                 continue
         path = entry['URL']
         repo.ui.debug('sending pullbundle "%s"\n' % path)
         try:
             return repo.vfs.open(path)
         except IOError:
             repo.ui.debug('pullbundle "%s" not accessible\n' % path)
             continue
     return None
 
 @wireprotocommand('getbundle', '*', permission='pull',
                   transportpolicy=POLICY_V1_ONLY)
 def getbundle(repo, proto, others):
     opts = options('getbundle', wireprototypes.GETBUNDLE_ARGUMENTS.keys(),
                    others)
     for k, v in opts.iteritems():
         keytype = wireprototypes.GETBUNDLE_ARGUMENTS[k]
         if keytype == 'nodes':
             opts[k] = wireprototypes.decodelist(v)
         elif keytype == 'csv':
             opts[k] = list(v.split(','))
         elif keytype == 'scsv':
             opts[k] = set(v.split(','))
         elif keytype == 'boolean':
             # Client should serialize False as '0', which is a non-empty string
             # so it evaluates as a True bool.
             if v == '0':
                 opts[k] = False
             else:
                 opts[k] = bool(v)
         elif keytype != 'plain':
             raise KeyError('unknown getbundle option type %s'
                            % keytype)
 
     if not bundle1allowed(repo, 'pull'):
         if not exchange.bundle2requested(opts.get('bundlecaps')):
             if proto.name == 'http-v1':
                 return wireprototypes.ooberror(bundle2required)
             raise error.Abort(bundle2requiredmain,
                               hint=bundle2requiredhint)
 
     prefercompressed = True
 
     try:
         clheads = set(repo.changelog.heads())
         heads = set(opts.get('heads', set()))
         common = set(opts.get('common', set()))
         common.discard(nullid)
         if (repo.ui.configbool('server', 'pullbundle') and
             'partial-pull' in proto.getprotocaps()):
             # Check if a pre-built bundle covers this request.
             bundle = find_pullbundle(repo, proto, opts, clheads, heads, common)
             if bundle:
                 return wireprototypes.streamres(gen=util.filechunkiter(bundle),
                                                 prefer_uncompressed=True)
 
         if repo.ui.configbool('server', 'disablefullbundle'):
             # Check to see if this is a full clone.
             changegroup = opts.get('cg', True)
             if changegroup and not common and clheads == heads:
                 raise error.Abort(
                     _('server has pull-based clones disabled'),
                     hint=_('remove --pull if specified or upgrade Mercurial'))
 
         info, chunks = exchange.getbundlechunks(repo, 'serve',
                                                 **pycompat.strkwargs(opts))
         prefercompressed = info.get('prefercompressed', True)
     except error.Abort as exc:
         # cleanly forward Abort error to the client
         if not exchange.bundle2requested(opts.get('bundlecaps')):
             if proto.name == 'http-v1':
                 return wireprototypes.ooberror(pycompat.bytestr(exc) + '\n')
             raise # cannot do better for bundle1 + ssh
         # bundle2 request expect a bundle2 reply
         bundler = bundle2.bundle20(repo.ui)
         manargs = [('message', pycompat.bytestr(exc))]
         advargs = []
         if exc.hint is not None:
             advargs.append(('hint', exc.hint))
         bundler.addpart(bundle2.bundlepart('error:abort',
                                            manargs, advargs))
         chunks = bundler.getchunks()
         prefercompressed = False
 
     return wireprototypes.streamres(
         gen=chunks, prefer_uncompressed=not prefercompressed)
 
 @wireprotocommand('heads', permission='pull', transportpolicy=POLICY_V1_ONLY)
 def heads(repo, proto):
     h = repo.heads()
     return wireprototypes.bytesresponse(wireprototypes.encodelist(h) + '\n')
 
 @wireprotocommand('hello', permission='pull', transportpolicy=POLICY_V1_ONLY)
 def hello(repo, proto):
     """Called as part of SSH handshake to obtain server info.
 
     Returns a list of lines describing interesting things about the
     server, in an RFC822-like format.
 
     Currently, the only one defined is ``capabilities``, which consists of a
     line of space separated tokens describing server abilities:
 
         capabilities: <token0> <token1> <token2>
     """
     caps = capabilities(repo, proto).data
     return wireprototypes.bytesresponse('capabilities: %s\n' % caps)
 
 @wireprotocommand('listkeys', 'namespace', permission='pull',
                   transportpolicy=POLICY_V1_ONLY)
 def listkeys(repo, proto, namespace):
     d = sorted(repo.listkeys(encoding.tolocal(namespace)).items())
     return wireprototypes.bytesresponse(pushkeymod.encodekeys(d))
 
 @wireprotocommand('lookup', 'key', permission='pull',
                   transportpolicy=POLICY_V1_ONLY)
 def lookup(repo, proto, key):
     try:
         k = encoding.tolocal(key)
         n = repo.lookup(k)
         r = hex(n)
         success = 1
     except Exception as inst:
         r = stringutil.forcebytestr(inst)
         success = 0
     return wireprototypes.bytesresponse('%d %s\n' % (success, r))
 
 @wireprotocommand('known', 'nodes *', permission='pull',
                   transportpolicy=POLICY_V1_ONLY)
 def known(repo, proto, nodes, others):
     v = ''.join(b and '1' or '0'
                 for b in repo.known(wireprototypes.decodelist(nodes)))
     return wireprototypes.bytesresponse(v)
 
 @wireprotocommand('protocaps', 'caps', permission='pull',
                   transportpolicy=POLICY_V1_ONLY)
 def protocaps(repo, proto, caps):
     if proto.name == wireprototypes.SSHV1:
         proto._protocaps = set(caps.split(' '))
     return wireprototypes.bytesresponse('OK')
 
 @wireprotocommand('pushkey', 'namespace key old new', permission='push',
                   transportpolicy=POLICY_V1_ONLY)
 def pushkey(repo, proto, namespace, key, old, new):
     # compatibility with pre-1.8 clients which were accidentally
     # sending raw binary nodes rather than utf-8-encoded hex
     if len(new) == 20 and stringutil.escapestr(new) != new:
         # looks like it could be a binary node
         try:
             new.decode('utf-8')
             new = encoding.tolocal(new) # but cleanly decodes as UTF-8
         except UnicodeDecodeError:
             pass # binary, leave unmodified
     else:
         new = encoding.tolocal(new) # normal path
 
     with proto.mayberedirectstdio() as output:
         r = repo.pushkey(encoding.tolocal(namespace), encoding.tolocal(key),
                          encoding.tolocal(old), new) or False
 
     output = output.getvalue() if output else ''
     return wireprototypes.bytesresponse('%d\n%s' % (int(r), output))
 
 @wireprotocommand('stream_out', permission='pull',
                   transportpolicy=POLICY_V1_ONLY)
 def stream(repo, proto):
     '''If the server supports streaming clone, it advertises the "stream"
     capability with a value representing the version and flags of the repo
     it is serving. Client checks to see if it understands the format.
     '''
     return wireprototypes.streamreslegacy(
         streamclone.generatev1wireproto(repo))
 
 @wireprotocommand('unbundle', 'heads', permission='push',
                   transportpolicy=POLICY_V1_ONLY)
 def unbundle(repo, proto, heads):
     their_heads = wireprototypes.decodelist(heads)
 
     with proto.mayberedirectstdio() as output:
         try:
             exchange.check_heads(repo, their_heads, 'preparing changes')
             cleanup = lambda: None
             try:
                 payload = proto.getpayload()
                 if repo.ui.configbool('server', 'streamunbundle'):
                     def cleanup():
                         # Ensure that the full payload is consumed, so
                         # that the connection doesn't contain trailing garbage.
                         for p in payload:
                             pass
                     fp = util.chunkbuffer(payload)
                 else:
                     # write bundle data to temporary file as it can be big
                     fp, tempname = None, None
                     def cleanup():
                         if fp:
                             fp.close()
                         if tempname:
                             os.unlink(tempname)
                     fd, tempname = tempfile.mkstemp(prefix='hg-unbundle-')
                     repo.ui.debug('redirecting incoming bundle to %s\n' %
                         tempname)
                     fp = os.fdopen(fd, pycompat.sysstr('wb+'))
                     r = 0
                     for p in payload:
                         fp.write(p)
                     fp.seek(0)
 
                 gen = exchange.readbundle(repo.ui, fp, None)
                 if (isinstance(gen, changegroupmod.cg1unpacker)
                     and not bundle1allowed(repo, 'push')):
                     if proto.name == 'http-v1':
                         # need to special case http because stderr do not get to
                         # the http client on failed push so we need to abuse
                         # some other error type to make sure the message get to
                         # the user.
                         return wireprototypes.ooberror(bundle2required)
                     raise error.Abort(bundle2requiredmain,
                                       hint=bundle2requiredhint)
 
                 r = exchange.unbundle(repo, gen, their_heads, 'serve',
                                       proto.client())
                 if util.safehasattr(r, 'addpart'):
                     # The return looks streamable, we are in the bundle2 case
                     # and should return a stream.
                     return wireprototypes.streamreslegacy(gen=r.getchunks())
                 return wireprototypes.pushres(
                     r, output.getvalue() if output else '')
 
             finally:
                 cleanup()
 
         except (error.BundleValueError, error.Abort, error.PushRaced) as exc:
             # handle non-bundle2 case first
             if not getattr(exc, 'duringunbundle2', False):
                 try:
                     raise
                 except error.Abort:
                     # The old code we moved used procutil.stderr directly.
                     # We did not change it to minimise code change.
                     # This need to be moved to something proper.
                     # Feel free to do it.
                     procutil.stderr.write("abort: %s\n" % exc)
                     if exc.hint is not None:
                         procutil.stderr.write("(%s)\n" % exc.hint)
                     procutil.stderr.flush()
                     return wireprototypes.pushres(
                         0, output.getvalue() if output else '')
                 except error.PushRaced:
                     return wireprototypes.pusherr(
                         pycompat.bytestr(exc),
                         output.getvalue() if output else '')
 
             bundler = bundle2.bundle20(repo.ui)
             for out in getattr(exc, '_bundle2salvagedoutput', ()):
                 bundler.addpart(out)
             try:
                 try:
                     raise
                 except error.PushkeyFailed as exc:
                     # check client caps
                     remotecaps = getattr(exc, '_replycaps', None)
                     if (remotecaps is not None
                             and 'pushkey' not in remotecaps.get('error', ())):
                         # no support remote side, fallback to Abort handler.
                         raise
                     part = bundler.newpart('error:pushkey')
                     part.addparam('in-reply-to', exc.partid)
                     if exc.namespace is not None:
                         part.addparam('namespace', exc.namespace,
                                       mandatory=False)
                     if exc.key is not None:
                         part.addparam('key', exc.key, mandatory=False)
                     if exc.new is not None:
                         part.addparam('new', exc.new, mandatory=False)
                     if exc.old is not None:
                         part.addparam('old', exc.old, mandatory=False)
                     if exc.ret is not None:
                         part.addparam('ret', exc.ret, mandatory=False)
             except error.BundleValueError as exc:
                 errpart = bundler.newpart('error:unsupportedcontent')
                 if exc.parttype is not None:
                     errpart.addparam('parttype', exc.parttype)
                 if exc.params:
                     errpart.addparam('params', '\0'.join(exc.params))
             except error.Abort as exc:
                 manargs = [('message', stringutil.forcebytestr(exc))]
                 advargs = []
                 if exc.hint is not None:
                     advargs.append(('hint', exc.hint))
                 bundler.addpart(bundle2.bundlepart('error:abort',
                                                    manargs, advargs))
             except error.PushRaced as exc:
                 bundler.newpart('error:pushraced',
                                 [('message', stringutil.forcebytestr(exc))])
             return wireprototypes.streamreslegacy(gen=bundler.getchunks())

mercurial/wireprotov1peer.py

@@ -1,1265 +1,420 b''
-# wireproto.py - generic wire protocol support functions
+# wireprotov1peer.py - Client-side functionality for wire protocol version 1.
 #
 # 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 __future__ import absolute_import
 
 import hashlib
-import os
-import tempfile
 
 from .i18n import _
 from .node import (
     bin,
-    hex,
-    nullid,
 )
 
 from . import (
     bundle2,
     changegroup as changegroupmod,
-    discovery,
     encoding,
     error,
-    exchange,
     peer,
     pushkey as pushkeymod,
     pycompat,
     repository,
-    streamclone,
     util,
     wireprototypes,
 )
 
-from .utils import (
-    procutil,
-    stringutil,
-)
-
-urlerr = util.urlerr
 urlreq = util.urlreq
 
-bundle2requiredmain = _('incompatible Mercurial client; bundle2 required')
-bundle2requiredhint = _('see https://www.mercurial-scm.org/wiki/'
-                        'IncompatibleClient')
-bundle2required = '%s\n(%s)\n' % (bundle2requiredmain, bundle2requiredhint)
-
 class remoteiterbatcher(peer.iterbatcher):
     def __init__(self, remote):
         super(remoteiterbatcher, self).__init__()
         self._remote = remote
 
     def __getattr__(self, name):
         # Validate this method is batchable, since submit() only supports
         # batchable methods.
         fn = getattr(self._remote, name)
         if not getattr(fn, 'batchable', None):
             raise error.ProgrammingError('Attempted to batch a non-batchable '
                                          'call to %r' % name)
 
         return super(remoteiterbatcher, self).__getattr__(name)
 
     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.
         """
         # 2-tuple of (command, arguments) that represents what will be
         # sent over the wire.
         requests = []
 
         # 4-tuple of (command, final future, @batchable generator, remote
         # future).
         results = []
 
         for command, args, opts, finalfuture in self.calls:
             mtd = getattr(self._remote, command)
             batchable = mtd.batchable(mtd.__self__, *args, **opts)
 
             commandargs, fremote = next(batchable)
             assert fremote
             requests.append((command, commandargs))
             results.append((command, finalfuture, batchable, fremote))
 
         if requests:
             self._resultiter = self._remote._submitbatch(requests)
 
         self._results = results
 
     def results(self):
         for command, finalfuture, batchable, remotefuture in self._results:
             # Get the raw result, set it in the remote future, feed it
             # back into the @batchable generator so it can be decoded, and
             # set the result on the final future to this value.
             remoteresult = next(self._resultiter)
             remotefuture.set(remoteresult)
             finalfuture.set(next(batchable))
 
             # Verify our @batchable generators only emit 2 values.
             try:
                 next(batchable)
             except StopIteration:
                 pass
             else:
                 raise error.ProgrammingError('%s @batchable generator emitted '
                                              'unexpected value count' % command)
 
             yield finalfuture.value
 
 # Forward a couple of names from peer to make wireproto interactions
 # slightly more sensible.
 batchable = peer.batchable
 future = peer.future
 
-
 def encodebatchcmds(req):
     """Return a ``cmds`` argument value for the ``batch`` command."""
     escapearg = wireprototypes.escapebatcharg
 
     cmds = []
     for op, argsdict in req:
         # Old servers didn't properly unescape argument names. So prevent
         # the sending of argument names that may not be decoded properly by
         # servers.
         assert all(escapearg(k) == k for k in argsdict)
 
         args = ','.join('%s=%s' % (escapearg(k), escapearg(v))
                         for k, v in argsdict.iteritems())
         cmds.append('%s %s' % (op, args))
 
     return ';'.join(cmds)
 
-def clientcompressionsupport(proto):
-    """Returns a list of compression methods supported by the client.
-
-    Returns a list of the compression methods supported by the client
-    according to the protocol capabilities. If no such capability has
-    been announced, fallback to the default of zlib and uncompressed.
-    """
-    for cap in proto.getprotocaps():
-        if cap.startswith('comp='):
-            return cap[5:].split(',')
-    return ['zlib', 'none']
-
-# client side
-
 class wirepeer(repository.legacypeer):
     """Client-side interface for communicating with a peer repository.
 
     Methods commonly call wire protocol commands of the same name.
 
     See also httppeer.py and sshpeer.py for protocol-specific
     implementations of this interface.
     """
     # Begin of ipeercommands interface.
 
     def iterbatch(self):
         return remoteiterbatcher(self)
 
     @batchable
     def lookup(self, key):
         self.requirecap('lookup', _('look up remote revision'))
         f = future()
         yield {'key': encoding.fromlocal(key)}, f
         d = f.value
         success, data = d[:-1].split(" ", 1)
         if int(success):
             yield bin(data)
         else:
             self._abort(error.RepoError(data))
 
     @batchable
     def heads(self):
         f = future()
         yield {}, f
         d = f.value
         try:
             yield wireprototypes.decodelist(d[:-1])
         except ValueError:
             self._abort(error.ResponseError(_("unexpected response:"), d))
 
     @batchable
     def known(self, nodes):
         f = future()
         yield {'nodes': wireprototypes.encodelist(nodes)}, f
         d = f.value
         try:
             yield [bool(int(b)) for b in d]
         except ValueError:
             self._abort(error.ResponseError(_("unexpected response:"), d))
 
     @batchable
     def branchmap(self):
         f = future()
         yield {}, f
         d = f.value
         try:
             branchmap = {}
             for branchpart in d.splitlines():
                 branchname, branchheads = branchpart.split(' ', 1)
                 branchname = encoding.tolocal(urlreq.unquote(branchname))
                 branchheads = wireprototypes.decodelist(branchheads)
                 branchmap[branchname] = branchheads
             yield branchmap
         except TypeError:
             self._abort(error.ResponseError(_("unexpected response:"), 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)
 
     @batchable
     def pushkey(self, namespace, key, old, new):
         if not self.capable('pushkey'):
             yield False, None
         f = future()
         self.ui.debug('preparing pushkey for "%s:%s"\n' % (namespace, key))
         yield {'namespace': encoding.fromlocal(namespace),
                'key': encoding.fromlocal(key),
                'old': encoding.fromlocal(old),
                'new': encoding.fromlocal(new)}, f
         d = f.value
         d, output = d.split('\n', 1)
         try:
             d = bool(int(d))
         except ValueError:
             raise error.ResponseError(
                 _('push failed (unexpected response):'), d)
         for l in output.splitlines(True):
             self.ui.status(_('remote: '), l)
         yield d
 
     def stream_out(self):
         return self._callstream('stream_out')
 
     def getbundle(self, source, **kwargs):
         kwargs = pycompat.byteskwargs(kwargs)
         self.requirecap('getbundle', _('look up remote changes'))
         opts = {}
         bundlecaps = kwargs.get('bundlecaps') or set()
         for key, value in kwargs.iteritems():
             if value is None:
                 continue
             keytype = wireprototypes.GETBUNDLE_ARGUMENTS.get(key)
             if keytype is None:
                 raise error.ProgrammingError(
                     'Unexpectedly None keytype for key %s' % key)
             elif keytype == 'nodes':
                 value = wireprototypes.encodelist(value)
             elif keytype == 'csv':
                 value = ','.join(value)
             elif keytype == 'scsv':
                 value = ','.join(sorted(value))
             elif keytype == 'boolean':
                 value = '%i' % bool(value)
             elif keytype != 'plain':
                 raise KeyError('unknown getbundle option type %s'
                                % keytype)
             opts[key] = value
         f = self._callcompressable("getbundle", **pycompat.strkwargs(opts))
         if any((cap.startswith('HG2') for cap in bundlecaps)):
             return bundle2.getunbundler(self.ui, f)
         else:
             return changegroupmod.cg1unpacker(f, 'UN')
 
     def unbundle(self, cg, heads, url):
         '''Send cg (a readable file-like object representing the
         changegroup to push, typically a chunkbuffer object) to the
         remote server as a bundle.
 
         When pushing a bundle10 stream, return an integer indicating the
         result of the push (see changegroup.apply()).
 
         When pushing a bundle20 stream, return a bundle20 stream.
 
         `url` is the url the client thinks it's pushing to, which is
         visible to hooks.
         '''
 
         if heads != ['force'] and self.capable('unbundlehash'):
             heads = wireprototypes.encodelist(
                 ['hashed', hashlib.sha1(''.join(sorted(heads))).digest()])
         else:
             heads = wireprototypes.encodelist(heads)
 
         if util.safehasattr(cg, 'deltaheader'):
             # this a bundle10, do the old style call sequence
             ret, output = self._callpush("unbundle", cg, heads=heads)
             if ret == "":
                 raise error.ResponseError(
                     _('push failed:'), output)
             try:
                 ret = int(ret)
             except ValueError:
                 raise error.ResponseError(
                     _('push failed (unexpected response):'), ret)
 
             for l in output.splitlines(True):
                 self.ui.status(_('remote: '), l)
         else:
             # bundle2 push. Send a stream, fetch a stream.
             stream = self._calltwowaystream('unbundle', cg, heads=heads)
             ret = bundle2.getunbundler(self.ui, stream)
         return ret
 
     # End of ipeercommands interface.
 
     # Begin of ipeerlegacycommands interface.
 
     def branches(self, nodes):
         n = wireprototypes.encodelist(nodes)
         d = self._call("branches", nodes=n)
         try:
             br = [tuple(wireprototypes.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([wireprototypes.encodelist(p, '-')
                           for p in pairs[i:i + batch]])
             d = self._call("between", pairs=n)
             try:
                 r.extend(l and wireprototypes.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 = wireprototypes.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 = wireprototypes.encodelist(bases)
         heads = wireprototypes.encodelist(heads)
         f = self._callcompressable("changegroupsubset",
                                    bases=bases, heads=heads)
         return changegroupmod.cg1unpacker(f, 'UN')
 
     # End of ipeerlegacycommands interface.
 
     def _submitbatch(self, req):
         """run batch request <req> on the server
 
         Returns an iterator of the raw responses from the server.
         """
         ui = self.ui
         if ui.debugflag and ui.configbool('devel', 'debug.peer-request'):
             ui.debug('devel-peer-request: batched-content\n')
             for op, args in req:
                 msg = 'devel-peer-request:    - %s (%d arguments)\n'
                 ui.debug(msg % (op, len(args)))
 
         unescapearg = wireprototypes.unescapebatcharg
 
         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, **pycompat.strkwargs(args))
 
     def debugwireargs(self, one, two, three=None, four=None, five=None):
         # don't pass optional arguments left at their default value
         opts = {}
         if three is not None:
             opts[r'three'] = three
         if four is not None:
             opts[r'four'] = four
         return self._call('debugwireargs', one=one, two=two, **opts)
 
     def _call(self, cmd, **args):
         """execute <cmd> on the server
 
         The command is expected to return a simple string.
 
         returns the server reply as a string."""
         raise NotImplementedError()
 
     def _callstream(self, cmd, **args):
         """execute <cmd> on the server
 
         The command is expected to return a stream. Note that if the
         command doesn't return a stream, _callstream behaves
         differently for ssh and http peers.
 
         returns the server reply as a file like object.
         """
         raise NotImplementedError()
 
     def _callcompressable(self, cmd, **args):
         """execute <cmd> on the server
 
         The command is expected to return a stream.
 
         The stream may have been compressed in some implementations. This
         function takes care of the decompression. This is the only difference
         with _callstream.
 
         returns the server reply as a file like object.
         """
         raise NotImplementedError()
 
     def _callpush(self, cmd, fp, **args):
         """execute a <cmd> on server
 
         The command is expected to be related to a push. Push has a special
         return method.
 
         returns the server reply as a (ret, output) tuple. ret is either
         empty (error) or a stringified int.
         """
         raise NotImplementedError()
 
     def _calltwowaystream(self, cmd, fp, **args):
         """execute <cmd> on server
 
         The command will send a stream to the server and get a stream in reply.
         """
         raise NotImplementedError()
 
     def _abort(self, exception):
         """clearly abort the wire protocol connection and raise the exception
         """
         raise NotImplementedError()
-
-# server side
-
-# wire protocol command can either return a string or one of these classes.
-
-def getdispatchrepo(repo, proto, command):
-    """Obtain the repo used for processing wire protocol commands.
-
-    The intent of this function is to serve as a monkeypatch point for
-    extensions that need commands to operate on different repo views under
-    specialized circumstances.
-    """
-    return repo.filtered('served')
-
-def dispatch(repo, proto, command):
-    repo = getdispatchrepo(repo, proto, command)
-
-    transportversion = wireprototypes.TRANSPORTS[proto.name]['version']
-    commandtable = commandsv2 if transportversion == 2 else commands
-    func, spec = commandtable[command]
-
-    args = proto.getargs(spec)
-
-    # Version 1 protocols define arguments as a list. Version 2 uses a dict.
-    if isinstance(args, list):
-        return func(repo, proto, *args)
-    elif isinstance(args, dict):
-        return func(repo, proto, **args)
-    else:
-        raise error.ProgrammingError('unexpected type returned from '
-                                     'proto.getargs(): %s' % type(args))
-
-def options(cmd, keys, others):
-    opts = {}
-    for k in keys:
-        if k in others:
-            opts[k] = others[k]
-            del others[k]
-    if others:
-        procutil.stderr.write("warning: %s ignored unexpected arguments %s\n"
-                              % (cmd, ",".join(others)))
-    return opts
-
-def bundle1allowed(repo, action):
-    """Whether a bundle1 operation is allowed from the server.
-
-    Priority is:
-
-    1. server.bundle1gd.<action> (if generaldelta active)
-    2. server.bundle1.<action>
-    3. server.bundle1gd (if generaldelta active)
-    4. server.bundle1
-    """
-    ui = repo.ui
-    gd = 'generaldelta' in repo.requirements
-
-    if gd:
-        v = ui.configbool('server', 'bundle1gd.%s' % action)
-        if v is not None:
-            return v
-
-    v = ui.configbool('server', 'bundle1.%s' % action)
-    if v is not None:
-        return v
-
-    if gd:
-        v = ui.configbool('server', 'bundle1gd')
-        if v is not None:
-            return v
-
-    return ui.configbool('server', 'bundle1')
-
-def supportedcompengines(ui, role):
-    """Obtain the list of supported compression engines for a request."""
-    assert role in (util.CLIENTROLE, util.SERVERROLE)
-
-    compengines = util.compengines.supportedwireengines(role)
-
-    # Allow config to override default list and ordering.
-    if role == util.SERVERROLE:
-        configengines = ui.configlist('server', 'compressionengines')
-        config = 'server.compressionengines'
-    else:
-        # This is currently implemented mainly to facilitate testing. In most
-        # cases, the server should be in charge of choosing a compression engine
-        # because a server has the most to lose from a sub-optimal choice. (e.g.
-        # CPU DoS due to an expensive engine or a network DoS due to poor
-        # compression ratio).
-        configengines = ui.configlist('experimental',
-                                      'clientcompressionengines')
-        config = 'experimental.clientcompressionengines'
-
-    # No explicit config. Filter out the ones that aren't supposed to be
-    # advertised and return default ordering.
-    if not configengines:
-        attr = 'serverpriority' if role == util.SERVERROLE else 'clientpriority'
-        return [e for e in compengines
-                if getattr(e.wireprotosupport(), attr) > 0]
-
-    # If compression engines are listed in the config, assume there is a good
-    # reason for it (like server operators wanting to achieve specific
-    # performance characteristics). So fail fast if the config references
-    # unusable compression engines.
-    validnames = set(e.name() for e in compengines)
-    invalidnames = set(e for e in configengines if e not in validnames)
-    if invalidnames:
-        raise error.Abort(_('invalid compression engine defined in %s: %s') %
-                          (config, ', '.join(sorted(invalidnames))))
-
-    compengines = [e for e in compengines if e.name() in configengines]
-    compengines = sorted(compengines,
-                         key=lambda e: configengines.index(e.name()))
-
-    if not compengines:
-        raise error.Abort(_('%s config option does not specify any known '
-                            'compression engines') % config,
-                          hint=_('usable compression engines: %s') %
-                          ', '.sorted(validnames))
-
-    return compengines
-
-class commandentry(object):
-    """Represents a declared wire protocol command."""
-    def __init__(self, func, args='', transports=None,
-                 permission='push'):
-        self.func = func
-        self.args = args
-        self.transports = transports or set()
-        self.permission = permission
-
-    def _merge(self, func, args):
-        """Merge this instance with an incoming 2-tuple.
-
-        This is called when a caller using the old 2-tuple API attempts
-        to replace an instance. The incoming values are merged with
-        data not captured by the 2-tuple and a new instance containing
-        the union of the two objects is returned.
-        """
-        return commandentry(func, args=args, transports=set(self.transports),
-                            permission=self.permission)
-
-    # Old code treats instances as 2-tuples. So expose that interface.
-    def __iter__(self):
-        yield self.func
-        yield self.args
-
-    def __getitem__(self, i):
-        if i == 0:
-            return self.func
-        elif i == 1:
-            return self.args
-        else:
-            raise IndexError('can only access elements 0 and 1')
-
-class commanddict(dict):
-    """Container for registered wire protocol commands.
-
-    It behaves like a dict. But __setitem__ is overwritten to allow silent
-    coercion of values from 2-tuples for API compatibility.
-    """
-    def __setitem__(self, k, v):
-        if isinstance(v, commandentry):
-            pass
-        # Cast 2-tuples to commandentry instances.
-        elif isinstance(v, tuple):
-            if len(v) != 2:
-                raise ValueError('command tuples must have exactly 2 elements')
-
-            # It is common for extensions to wrap wire protocol commands via
-            # e.g. ``wireproto.commands[x] = (newfn, args)``. Because callers
-            # doing this aren't aware of the new API that uses objects to store
-            # command entries, we automatically merge old state with new.
-            if k in self:
-                v = self[k]._merge(v[0], v[1])
-            else:
-                # Use default values from @wireprotocommand.
-                v = commandentry(v[0], args=v[1],
-                                 transports=set(wireprototypes.TRANSPORTS),
-                                 permission='push')
-        else:
-            raise ValueError('command entries must be commandentry instances '
-                             'or 2-tuples')
-
-        return super(commanddict, self).__setitem__(k, v)
-
-    def commandavailable(self, command, proto):
-        """Determine if a command is available for the requested protocol."""
-        assert proto.name in wireprototypes.TRANSPORTS
-
-        entry = self.get(command)
-
-        if not entry:
-            return False
-
-        if proto.name not in entry.transports:
-            return False
-
-        return True
-
-# Constants specifying which transports a wire protocol command should be
-# available on. For use with @wireprotocommand.
-POLICY_V1_ONLY = 'v1-only'
-POLICY_V2_ONLY = 'v2-only'
-
-# For version 1 transports.
-commands = commanddict()
-
-# For version 2 transports.
-commandsv2 = commanddict()
-
-def wireprotocommand(name, args=None, transportpolicy=POLICY_V1_ONLY,
-                     permission='push'):
-    """Decorator to declare a wire protocol command.
-
-    ``name`` is the name of the wire protocol command being provided.
-
-    ``args`` defines the named arguments accepted by the command. It is
-    ideally a dict mapping argument names to their types. For backwards
-    compatibility, it can be a space-delimited list of argument names. For
-    version 1 transports, ``*`` denotes a special value that says to accept
-    all named arguments.
-
-    ``transportpolicy`` is a POLICY_* constant denoting which transports
-    this wire protocol command should be exposed to. By default, commands
-    are exposed to all wire protocol transports.
-
-    ``permission`` defines the permission type needed to run this command.
-    Can be ``push`` or ``pull``. These roughly map to read-write and read-only,
-    respectively. Default is to assume command requires ``push`` permissions
-    because otherwise commands not declaring their permissions could modify
-    a repository that is supposed to be read-only.
-    """
-    if transportpolicy == POLICY_V1_ONLY:
-        transports = {k for k, v in wireprototypes.TRANSPORTS.items()
-                      if v['version'] == 1}
-        transportversion = 1
-    elif transportpolicy == POLICY_V2_ONLY:
-        transports = {k for k, v in wireprototypes.TRANSPORTS.items()
-                      if v['version'] == 2}
-        transportversion = 2
-    else:
-        raise error.ProgrammingError('invalid transport policy value: %s' %
-                                     transportpolicy)
-
-    # Because SSHv2 is a mirror of SSHv1, we allow "batch" commands through to
-    # SSHv2.
-    # TODO undo this hack when SSH is using the unified frame protocol.
-    if name == b'batch':
-        transports.add(wireprototypes.SSHV2)
-
-    if permission not in ('push', 'pull'):
-        raise error.ProgrammingError('invalid wire protocol permission; '
-                                     'got %s; expected "push" or "pull"' %
-                                     permission)
-
-    if transportversion == 1:
-        if args is None:
-            args = ''
-
-        if not isinstance(args, bytes):
-            raise error.ProgrammingError('arguments for version 1 commands '
-                                         'must be declared as bytes')
-    elif transportversion == 2:
-        if args is None:
-            args = {}
-
-        if not isinstance(args, dict):
-            raise error.ProgrammingError('arguments for version 2 commands '
-                                         'must be declared as dicts')
-
-    def register(func):
-        if transportversion == 1:
-            if name in commands:
-                raise error.ProgrammingError('%s command already registered '
-                                             'for version 1' % name)
-            commands[name] = commandentry(func, args=args,
-                                          transports=transports,
-                                          permission=permission)
-        elif transportversion == 2:
-            if name in commandsv2:
-                raise error.ProgrammingError('%s command already registered '
-                                             'for version 2' % name)
-
-            commandsv2[name] = commandentry(func, args=args,
-                                            transports=transports,
-                                            permission=permission)
-        else:
-            raise error.ProgrammingError('unhandled transport version: %d' %
-                                         transportversion)
-
-        return func
-    return register
-
-# TODO define a more appropriate permissions type to use for this.
-@wireprotocommand('batch', 'cmds *', permission='pull',
-                  transportpolicy=POLICY_V1_ONLY)
-def batch(repo, proto, cmds, others):
-    unescapearg = wireprototypes.unescapebatcharg
-    repo = repo.filtered("served")
-    res = []
-    for pair in cmds.split(';'):
-        op, args = pair.split(' ', 1)
-        vals = {}
-        for a in args.split(','):
-            if a:
-                n, v = a.split('=')
-                vals[unescapearg(n)] = unescapearg(v)
-        func, spec = commands[op]
-
-        # Validate that client has permissions to perform this command.
-        perm = commands[op].permission
-        assert perm in ('push', 'pull')
-        proto.checkperm(perm)
-
-        if spec:
-            keys = spec.split()
-            data = {}
-            for k in keys:
-                if k == '*':
-                    star = {}
-                    for key in vals.keys():
-                        if key not in keys:
-                            star[key] = vals[key]
-                    data['*'] = star
-                else:
-                    data[k] = vals[k]
-            result = func(repo, proto, *[data[k] for k in keys])
-        else:
-            result = func(repo, proto)
-        if isinstance(result, wireprototypes.ooberror):
-            return result
-
-        # For now, all batchable commands must return bytesresponse or
-        # raw bytes (for backwards compatibility).
-        assert isinstance(result, (wireprototypes.bytesresponse, bytes))
-        if isinstance(result, wireprototypes.bytesresponse):
-            result = result.data
-        res.append(wireprototypes.escapebatcharg(result))
-
-    return wireprototypes.bytesresponse(';'.join(res))
-
-@wireprotocommand('between', 'pairs', transportpolicy=POLICY_V1_ONLY,
-                  permission='pull')
-def between(repo, proto, pairs):
-    pairs = [wireprototypes.decodelist(p, '-') for p in pairs.split(" ")]
-    r = []
-    for b in repo.between(pairs):
-        r.append(wireprototypes.encodelist(b) + "\n")
-
-    return wireprototypes.bytesresponse(''.join(r))
-
-@wireprotocommand('branchmap', permission='pull',
-                  transportpolicy=POLICY_V1_ONLY)
-def branchmap(repo, proto):
-    branchmap = repo.branchmap()
-    heads = []
-    for branch, nodes in branchmap.iteritems():
-        branchname = urlreq.quote(encoding.fromlocal(branch))
-        branchnodes = wireprototypes.encodelist(nodes)
-        heads.append('%s %s' % (branchname, branchnodes))
-
-    return wireprototypes.bytesresponse('\n'.join(heads))
-
-@wireprotocommand('branches', 'nodes', transportpolicy=POLICY_V1_ONLY,
-                  permission='pull')
-def branches(repo, proto, nodes):
-    nodes = wireprototypes.decodelist(nodes)
-    r = []
-    for b in repo.branches(nodes):
-        r.append(wireprototypes.encodelist(b) + "\n")
-
-    return wireprototypes.bytesresponse(''.join(r))
-
-@wireprotocommand('clonebundles', '', permission='pull',
-                  transportpolicy=POLICY_V1_ONLY)
-def clonebundles(repo, proto):
-    """Server command for returning info for available bundles to seed clones.
-
-    Clients will parse this response and determine what bundle to fetch.
-
-    Extensions may wrap this command to filter or dynamically emit data
-    depending on the request. e.g. you could advertise URLs for the closest
-    data center given the client's IP address.
-    """
-    return wireprototypes.bytesresponse(
-        repo.vfs.tryread('clonebundles.manifest'))
-
-wireprotocaps = ['lookup', 'branchmap', 'pushkey',
-                 'known', 'getbundle', 'unbundlehash']
-
-def _capabilities(repo, proto):
-    """return a list of capabilities for a repo
-
-    This function exists to allow extensions to easily wrap capabilities
-    computation
-
-    - returns a lists: easy to alter
-    - change done here will be propagated to both `capabilities` and `hello`
-      command without any other action needed.
-    """
-    # copy to prevent modification of the global list
-    caps = list(wireprotocaps)
-
-    # Command of same name as capability isn't exposed to version 1 of
-    # transports. So conditionally add it.
-    if commands.commandavailable('changegroupsubset', proto):
-        caps.append('changegroupsubset')
-
-    if streamclone.allowservergeneration(repo):
-        if repo.ui.configbool('server', 'preferuncompressed'):
-            caps.append('stream-preferred')
-        requiredformats = repo.requirements & repo.supportedformats
-        # if our local revlogs are just revlogv1, add 'stream' cap
-        if not requiredformats - {'revlogv1'}:
-            caps.append('stream')
-        # otherwise, add 'streamreqs' detailing our local revlog format
-        else:
-            caps.append('streamreqs=%s' % ','.join(sorted(requiredformats)))
-    if repo.ui.configbool('experimental', 'bundle2-advertise'):
-        capsblob = bundle2.encodecaps(bundle2.getrepocaps(repo, role='server'))
-        caps.append('bundle2=' + urlreq.quote(capsblob))
-    caps.append('unbundle=%s' % ','.join(bundle2.bundlepriority))
-
-    return proto.addcapabilities(repo, caps)
-
-# If you are writing an extension and consider wrapping this function. Wrap
-# `_capabilities` instead.
-@wireprotocommand('capabilities', permission='pull',
-                  transportpolicy=POLICY_V1_ONLY)
-def capabilities(repo, proto):
-    caps = _capabilities(repo, proto)
-    return wireprototypes.bytesresponse(' '.join(sorted(caps)))
-
-@wireprotocommand('changegroup', 'roots', transportpolicy=POLICY_V1_ONLY,
-                  permission='pull')
-def changegroup(repo, proto, roots):
-    nodes = wireprototypes.decodelist(roots)
-    outgoing = discovery.outgoing(repo, missingroots=nodes,
-                                  missingheads=repo.heads())
-    cg = changegroupmod.makechangegroup(repo, outgoing, '01', 'serve')
-    gen = iter(lambda: cg.read(32768), '')
-    return wireprototypes.streamres(gen=gen)
-
-@wireprotocommand('changegroupsubset', 'bases heads',
-                  transportpolicy=POLICY_V1_ONLY,
-                  permission='pull')
-def changegroupsubset(repo, proto, bases, heads):
-    bases = wireprototypes.decodelist(bases)
-    heads = wireprototypes.decodelist(heads)
-    outgoing = discovery.outgoing(repo, missingroots=bases,
-                                  missingheads=heads)
-    cg = changegroupmod.makechangegroup(repo, outgoing, '01', 'serve')
-    gen = iter(lambda: cg.read(32768), '')
-    return wireprototypes.streamres(gen=gen)
-
-@wireprotocommand('debugwireargs', 'one two *',
-                  permission='pull', transportpolicy=POLICY_V1_ONLY)
-def debugwireargs(repo, proto, one, two, others):
-    # only accept optional args from the known set
-    opts = options('debugwireargs', ['three', 'four'], others)
-    return wireprototypes.bytesresponse(repo.debugwireargs(
-        one, two, **pycompat.strkwargs(opts)))
-
-def find_pullbundle(repo, proto, opts, clheads, heads, common):
-    """Return a file object for the first matching pullbundle.
-
-    Pullbundles are specified in .hg/pullbundles.manifest similar to
-    clonebundles.
-    For each entry, the bundle specification is checked for compatibility:
-    - Client features vs the BUNDLESPEC.
-    - Revisions shared with the clients vs base revisions of the bundle.
-      A bundle can be applied only if all its base revisions are known by
-      the client.
-    - At least one leaf of the bundle's DAG is missing on the client.
-    - Every leaf of the bundle's DAG is part of node set the client wants.
-      E.g. do not send a bundle of all changes if the client wants only
-      one specific branch of many.
-    """
-    def decodehexstring(s):
-        return set([h.decode('hex') for h in s.split(';')])
-
-    manifest = repo.vfs.tryread('pullbundles.manifest')
-    if not manifest:
-        return None
-    res = exchange.parseclonebundlesmanifest(repo, manifest)
-    res = exchange.filterclonebundleentries(repo, res)
-    if not res:
-        return None
-    cl = repo.changelog
-    heads_anc = cl.ancestors([cl.rev(rev) for rev in heads], inclusive=True)
-    common_anc = cl.ancestors([cl.rev(rev) for rev in common], inclusive=True)
-    compformats = clientcompressionsupport(proto)
-    for entry in res:
-        if 'COMPRESSION' in entry and entry['COMPRESSION'] not in compformats:
-            continue
-        # No test yet for VERSION, since V2 is supported by any client
-        # that advertises partial pulls
-        if 'heads' in entry:
-            try:
-                bundle_heads = decodehexstring(entry['heads'])
-            except TypeError:
-                # Bad heads entry
-                continue
-            if bundle_heads.issubset(common):
-                continue # Nothing new
-            if all(cl.rev(rev) in common_anc for rev in bundle_heads):
-                continue # Still nothing new
-            if any(cl.rev(rev) not in heads_anc and
-                   cl.rev(rev) not in common_anc for rev in bundle_heads):
-                continue
-        if 'bases' in entry:
-            try:
-                bundle_bases = decodehexstring(entry['bases'])
-            except TypeError:
-                # Bad bases entry
-                continue
-            if not all(cl.rev(rev) in common_anc for rev in bundle_bases):
-                continue
-        path = entry['URL']
-        repo.ui.debug('sending pullbundle "%s"\n' % path)
-        try:
-            return repo.vfs.open(path)
-        except IOError:
-            repo.ui.debug('pullbundle "%s" not accessible\n' % path)
-            continue
-    return None
-
-@wireprotocommand('getbundle', '*', permission='pull',
-                  transportpolicy=POLICY_V1_ONLY)
-def getbundle(repo, proto, others):
-    opts = options('getbundle', wireprototypes.GETBUNDLE_ARGUMENTS.keys(),
-                   others)
-    for k, v in opts.iteritems():
-        keytype = wireprototypes.GETBUNDLE_ARGUMENTS[k]
-        if keytype == 'nodes':
-            opts[k] = wireprototypes.decodelist(v)
-        elif keytype == 'csv':
-            opts[k] = list(v.split(','))
-        elif keytype == 'scsv':
-            opts[k] = set(v.split(','))
-        elif keytype == 'boolean':
-            # Client should serialize False as '0', which is a non-empty string
-            # so it evaluates as a True bool.
-            if v == '0':
-                opts[k] = False
-            else:
-                opts[k] = bool(v)
-        elif keytype != 'plain':
-            raise KeyError('unknown getbundle option type %s'
-                           % keytype)
-
-    if not bundle1allowed(repo, 'pull'):
-        if not exchange.bundle2requested(opts.get('bundlecaps')):
-            if proto.name == 'http-v1':
-                return wireprototypes.ooberror(bundle2required)
-            raise error.Abort(bundle2requiredmain,
-                              hint=bundle2requiredhint)
-
-    prefercompressed = True
-
-    try:
-        clheads = set(repo.changelog.heads())
-        heads = set(opts.get('heads', set()))
-        common = set(opts.get('common', set()))
-        common.discard(nullid)
-        if (repo.ui.configbool('server', 'pullbundle') and
-            'partial-pull' in proto.getprotocaps()):
-            # Check if a pre-built bundle covers this request.
-            bundle = find_pullbundle(repo, proto, opts, clheads, heads, common)
-            if bundle:
-                return wireprototypes.streamres(gen=util.filechunkiter(bundle),
-                                                prefer_uncompressed=True)
-
-        if repo.ui.configbool('server', 'disablefullbundle'):
-            # Check to see if this is a full clone.
-            changegroup = opts.get('cg', True)
-            if changegroup and not common and clheads == heads:
-                raise error.Abort(
-                    _('server has pull-based clones disabled'),
-                    hint=_('remove --pull if specified or upgrade Mercurial'))
-
-        info, chunks = exchange.getbundlechunks(repo, 'serve',
-                                                **pycompat.strkwargs(opts))
-        prefercompressed = info.get('prefercompressed', True)
-    except error.Abort as exc:
-        # cleanly forward Abort error to the client
-        if not exchange.bundle2requested(opts.get('bundlecaps')):
-            if proto.name == 'http-v1':
-                return wireprototypes.ooberror(pycompat.bytestr(exc) + '\n')
-            raise # cannot do better for bundle1 + ssh
-        # bundle2 request expect a bundle2 reply
-        bundler = bundle2.bundle20(repo.ui)
-        manargs = [('message', pycompat.bytestr(exc))]
-        advargs = []
-        if exc.hint is not None:
-            advargs.append(('hint', exc.hint))
-        bundler.addpart(bundle2.bundlepart('error:abort',
-                                           manargs, advargs))
-        chunks = bundler.getchunks()
-        prefercompressed = False
-
-    return wireprototypes.streamres(
-        gen=chunks, prefer_uncompressed=not prefercompressed)
-
-@wireprotocommand('heads', permission='pull', transportpolicy=POLICY_V1_ONLY)
-def heads(repo, proto):
-    h = repo.heads()
-    return wireprototypes.bytesresponse(wireprototypes.encodelist(h) + '\n')
-
-@wireprotocommand('hello', permission='pull', transportpolicy=POLICY_V1_ONLY)
-def hello(repo, proto):
-    """Called as part of SSH handshake to obtain server info.
-
-    Returns a list of lines describing interesting things about the
-    server, in an RFC822-like format.
-
-    Currently, the only one defined is ``capabilities``, which consists of a
-    line of space separated tokens describing server abilities:
-
-        capabilities: <token0> <token1> <token2>
-    """
-    caps = capabilities(repo, proto).data
-    return wireprototypes.bytesresponse('capabilities: %s\n' % caps)
-
-@wireprotocommand('listkeys', 'namespace', permission='pull',
-                  transportpolicy=POLICY_V1_ONLY)
-def listkeys(repo, proto, namespace):
-    d = sorted(repo.listkeys(encoding.tolocal(namespace)).items())
-    return wireprototypes.bytesresponse(pushkeymod.encodekeys(d))
-
-@wireprotocommand('lookup', 'key', permission='pull',
-                  transportpolicy=POLICY_V1_ONLY)
-def lookup(repo, proto, key):
-    try:
-        k = encoding.tolocal(key)
-        n = repo.lookup(k)
-        r = hex(n)
-        success = 1
-    except Exception as inst:
-        r = stringutil.forcebytestr(inst)
-        success = 0
-    return wireprototypes.bytesresponse('%d %s\n' % (success, r))
-
-@wireprotocommand('known', 'nodes *', permission='pull',
-                  transportpolicy=POLICY_V1_ONLY)
-def known(repo, proto, nodes, others):
-    v = ''.join(b and '1' or '0'
-                for b in repo.known(wireprototypes.decodelist(nodes)))
-    return wireprototypes.bytesresponse(v)
-
-@wireprotocommand('protocaps', 'caps', permission='pull',
-                  transportpolicy=POLICY_V1_ONLY)
-def protocaps(repo, proto, caps):
-    if proto.name == wireprototypes.SSHV1:
-        proto._protocaps = set(caps.split(' '))
-    return wireprototypes.bytesresponse('OK')
-
-@wireprotocommand('pushkey', 'namespace key old new', permission='push',
-                  transportpolicy=POLICY_V1_ONLY)
-def pushkey(repo, proto, namespace, key, old, new):
-    # compatibility with pre-1.8 clients which were accidentally
-    # sending raw binary nodes rather than utf-8-encoded hex
-    if len(new) == 20 and stringutil.escapestr(new) != new:
-        # looks like it could be a binary node
-        try:
-            new.decode('utf-8')
-            new = encoding.tolocal(new) # but cleanly decodes as UTF-8
-        except UnicodeDecodeError:
-            pass # binary, leave unmodified
-    else:
-        new = encoding.tolocal(new) # normal path
-
-    with proto.mayberedirectstdio() as output:
-        r = repo.pushkey(encoding.tolocal(namespace), encoding.tolocal(key),
-                         encoding.tolocal(old), new) or False
-
-    output = output.getvalue() if output else ''
-    return wireprototypes.bytesresponse('%d\n%s' % (int(r), output))
-
-@wireprotocommand('stream_out', permission='pull',
-                  transportpolicy=POLICY_V1_ONLY)
-def stream(repo, proto):
-    '''If the server supports streaming clone, it advertises the "stream"
-    capability with a value representing the version and flags of the repo
-    it is serving. Client checks to see if it understands the format.
-    '''
-    return wireprototypes.streamreslegacy(
-        streamclone.generatev1wireproto(repo))
-
-@wireprotocommand('unbundle', 'heads', permission='push',
-                  transportpolicy=POLICY_V1_ONLY)
-def unbundle(repo, proto, heads):
-    their_heads = wireprototypes.decodelist(heads)
-
-    with proto.mayberedirectstdio() as output:
-        try:
-            exchange.check_heads(repo, their_heads, 'preparing changes')
-            cleanup = lambda: None
-            try:
-                payload = proto.getpayload()
-                if repo.ui.configbool('server', 'streamunbundle'):
-                    def cleanup():
-                        # Ensure that the full payload is consumed, so
-                        # that the connection doesn't contain trailing garbage.
-                        for p in payload:
-                            pass
-                    fp = util.chunkbuffer(payload)
-                else:
-                    # write bundle data to temporary file as it can be big
-                    fp, tempname = None, None
-                    def cleanup():
-                        if fp:
-                            fp.close()
-                        if tempname:
-                            os.unlink(tempname)
-                    fd, tempname = tempfile.mkstemp(prefix='hg-unbundle-')
-                    repo.ui.debug('redirecting incoming bundle to %s\n' %
-                        tempname)
-                    fp = os.fdopen(fd, pycompat.sysstr('wb+'))
-                    r = 0
-                    for p in payload:
-                        fp.write(p)
-                    fp.seek(0)
-
-                gen = exchange.readbundle(repo.ui, fp, None)
-                if (isinstance(gen, changegroupmod.cg1unpacker)
-                    and not bundle1allowed(repo, 'push')):
-                    if proto.name == 'http-v1':
-                        # need to special case http because stderr do not get to
-                        # the http client on failed push so we need to abuse
-                        # some other error type to make sure the message get to
-                        # the user.
-                        return wireprototypes.ooberror(bundle2required)
-                    raise error.Abort(bundle2requiredmain,
-                                      hint=bundle2requiredhint)
-
-                r = exchange.unbundle(repo, gen, their_heads, 'serve',
-                                      proto.client())
-                if util.safehasattr(r, 'addpart'):
-                    # The return looks streamable, we are in the bundle2 case
-                    # and should return a stream.
-                    return wireprototypes.streamreslegacy(gen=r.getchunks())
-                return wireprototypes.pushres(
-                    r, output.getvalue() if output else '')
-
-            finally:
-                cleanup()
-
-        except (error.BundleValueError, error.Abort, error.PushRaced) as exc:
-            # handle non-bundle2 case first
-            if not getattr(exc, 'duringunbundle2', False):
-                try:
-                    raise
-                except error.Abort:
-                    # The old code we moved used procutil.stderr directly.
-                    # We did not change it to minimise code change.
-                    # This need to be moved to something proper.
-                    # Feel free to do it.
-                    procutil.stderr.write("abort: %s\n" % exc)
-                    if exc.hint is not None:
-                        procutil.stderr.write("(%s)\n" % exc.hint)
-                    procutil.stderr.flush()
-                    return wireprototypes.pushres(
-                        0, output.getvalue() if output else '')
-                except error.PushRaced:
-                    return wireprototypes.pusherr(
-                        pycompat.bytestr(exc),
-                        output.getvalue() if output else '')
-
-            bundler = bundle2.bundle20(repo.ui)
-            for out in getattr(exc, '_bundle2salvagedoutput', ()):
-                bundler.addpart(out)
-            try:
-                try:
-                    raise
-                except error.PushkeyFailed as exc:
-                    # check client caps
-                    remotecaps = getattr(exc, '_replycaps', None)
-                    if (remotecaps is not None
-                            and 'pushkey' not in remotecaps.get('error', ())):
-                        # no support remote side, fallback to Abort handler.
-                        raise
-                    part = bundler.newpart('error:pushkey')
-                    part.addparam('in-reply-to', exc.partid)
-                    if exc.namespace is not None:
-                        part.addparam('namespace', exc.namespace,
-                                      mandatory=False)
-                    if exc.key is not None:
-                        part.addparam('key', exc.key, mandatory=False)
-                    if exc.new is not None:
-                        part.addparam('new', exc.new, mandatory=False)
-                    if exc.old is not None:
-                        part.addparam('old', exc.old, mandatory=False)
-                    if exc.ret is not None:
-                        part.addparam('ret', exc.ret, mandatory=False)
-            except error.BundleValueError as exc:
-                errpart = bundler.newpart('error:unsupportedcontent')
-                if exc.parttype is not None:
-                    errpart.addparam('parttype', exc.parttype)
-                if exc.params:
-                    errpart.addparam('params', '\0'.join(exc.params))
-            except error.Abort as exc:
-                manargs = [('message', stringutil.forcebytestr(exc))]
-                advargs = []
-                if exc.hint is not None:
-                    advargs.append(('hint', exc.hint))
-                bundler.addpart(bundle2.bundlepart('error:abort',
-                                                   manargs, advargs))
-            except error.PushRaced as exc:
-                bundler.newpart('error:pushraced',
-                                [('message', stringutil.forcebytestr(exc))])
-            return wireprototypes.streamreslegacy(gen=bundler.getchunks())

tests/test-batching.py

@@ -1,206 +1,206 b''
 # test-batching.py - tests for transparent command batching
 #
 # Copyright 2011 Peter Arrenbrecht <[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 __future__ import absolute_import, print_function
 
 from mercurial import (
     error,
     peer,
     util,
-    wireproto,
+    wireprotov1peer,
 )
 
 # equivalent of repo.repository
 class thing(object):
     def hello(self):
         return "Ready."
 
 # equivalent of localrepo.localrepository
 class localthing(thing):
     def foo(self, one, two=None):
         if one:
             return "%s and %s" % (one, two,)
         return "Nope"
     def bar(self, b, a):
         return "%s und %s" % (b, a,)
     def greet(self, name=None):
         return "Hello, %s" % name
     def batchiter(self):
         '''Support for local batching.'''
         return peer.localiterbatcher(self)
 
 # usage of "thing" interface
 def use(it):
 
     # Direct call to base method shared between client and server.
     print(it.hello())
 
     # Direct calls to proxied methods. They cause individual roundtrips.
     print(it.foo("Un", two="Deux"))
     print(it.bar("Eins", "Zwei"))
 
     # Batched call to a couple of proxied methods.
     batch = it.batchiter()
     # The calls return futures to eventually hold results.
     foo = batch.foo(one="One", two="Two")
     bar = batch.bar("Eins", "Zwei")
     bar2 = batch.bar(b="Uno", a="Due")
 
     # Future shouldn't be set until we submit().
     assert isinstance(foo, peer.future)
     assert not util.safehasattr(foo, 'value')
     assert not util.safehasattr(bar, 'value')
     batch.submit()
     # Call results() to obtain results as a generator.
     results = batch.results()
 
     # Future results shouldn't be set until we consume a value.
     assert not util.safehasattr(foo, 'value')
     foovalue = next(results)
     assert util.safehasattr(foo, 'value')
     assert foovalue == foo.value
     print(foo.value)
     next(results)
     print(bar.value)
     next(results)
     print(bar2.value)
 
     # We should be at the end of the results generator.
     try:
         next(results)
     except StopIteration:
         print('proper end of results generator')
     else:
         print('extra emitted element!')
 
     # Attempting to call a non-batchable method inside a batch fails.
     batch = it.batchiter()
     try:
         batch.greet(name='John Smith')
     except error.ProgrammingError as e:
         print(e)
 
     # Attempting to call a local method inside a batch fails.
     batch = it.batchiter()
     try:
         batch.hello()
     except error.ProgrammingError as e:
         print(e)
 
 # local usage
 mylocal = localthing()
 print()
 print("== Local")
 use(mylocal)
 
 # demo remoting; mimicks what wireproto and HTTP/SSH do
 
 # shared
 
 def escapearg(plain):
     return (plain
             .replace(':', '::')
             .replace(',', ':,')
             .replace(';', ':;')
             .replace('=', ':='))
 def unescapearg(escaped):
     return (escaped
             .replace(':=', '=')
             .replace(':;', ';')
             .replace(':,', ',')
             .replace('::', ':'))
 
 # server side
 
 # equivalent of wireproto's global functions
 class server(object):
     def __init__(self, local):
         self.local = local
     def _call(self, name, args):
         args = dict(arg.split('=', 1) for arg in args)
         return getattr(self, name)(**args)
     def perform(self, req):
         print("REQ:", req)
         name, args = req.split('?', 1)
         args = args.split('&')
         vals = dict(arg.split('=', 1) for arg in args)
         res = getattr(self, name)(**vals)
         print("  ->", res)
         return res
     def batch(self, cmds):
         res = []
         for pair in cmds.split(';'):
             name, args = pair.split(':', 1)
             vals = {}
             for a in args.split(','):
                 if a:
                     n, v = a.split('=')
                     vals[n] = unescapearg(v)
             res.append(escapearg(getattr(self, name)(**vals)))
         return ';'.join(res)
     def foo(self, one, two):
         return mangle(self.local.foo(unmangle(one), unmangle(two)))
     def bar(self, b, a):
         return mangle(self.local.bar(unmangle(b), unmangle(a)))
     def greet(self, name):
         return mangle(self.local.greet(unmangle(name)))
 myserver = server(mylocal)
 
 # local side
 
 # equivalent of wireproto.encode/decodelist, that is, type-specific marshalling
 # here we just transform the strings a bit to check we're properly en-/decoding
 def mangle(s):
     return ''.join(chr(ord(c) + 1) for c in s)
 def unmangle(s):
     return ''.join(chr(ord(c) - 1) for c in s)
 
 # equivalent of wireproto.wirerepository and something like http's wire format
 class remotething(thing):
     def __init__(self, server):
         self.server = server
     def _submitone(self, name, args):
         req = name + '?' + '&'.join(['%s=%s' % (n, v) for n, v in args])
         return self.server.perform(req)
     def _submitbatch(self, cmds):
         req = []
         for name, args in cmds:
             args = ','.join(n + '=' + escapearg(v) for n, v in args)
             req.append(name + ':' + args)
         req = ';'.join(req)
         res = self._submitone('batch', [('cmds', req,)])
         for r in res.split(';'):
             yield r
 
     def batchiter(self):
-        return wireproto.remoteiterbatcher(self)
+        return wireprotov1peer.remoteiterbatcher(self)
 
     @peer.batchable
     def foo(self, one, two=None):
         encargs = [('one', mangle(one),), ('two', mangle(two),)]
         encresref = peer.future()
         yield encargs, encresref
         yield unmangle(encresref.value)
 
     @peer.batchable
     def bar(self, b, a):
         encresref = peer.future()
         yield [('b', mangle(b),), ('a', mangle(a),)], encresref
         yield unmangle(encresref.value)
 
     # 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
     # greet is done in its own roundtrip.
     def greet(self, name=None):
         return unmangle(self._submitone('greet', [('name', mangle(name),)]))
 
 # demo remote usage
 
 myproxy = remotething(myserver)
 print()
 print("== Remote")
 use(myproxy)

tests/test-wireproto.py

@@ -1,98 +1,99 b''
 from __future__ import absolute_import, print_function
 
 from mercurial import (
     error,
     pycompat,
     ui as uimod,
     util,
     wireproto,
     wireprototypes,
+    wireprotov1peer,
 )
 stringio = util.stringio
 
 class proto(object):
     def __init__(self, args):
         self.args = args
         self.name = 'dummyproto'
 
     def getargs(self, spec):
         args = self.args
         args.setdefault(b'*', {})
         names = spec.split()
         return [args[n] for n in names]
 
     def checkperm(self, perm):
         pass
 
 wireprototypes.TRANSPORTS['dummyproto'] = {
     'transport': 'dummy',
     'version': 1,
 }
 
-class clientpeer(wireproto.wirepeer):
+class clientpeer(wireprotov1peer.wirepeer):
     def __init__(self, serverrepo, ui):
         self.serverrepo = serverrepo
         self.ui = ui
 
     def url(self):
         return b'test'
 
     def local(self):
         return None
 
     def peer(self):
         return self
 
     def canpush(self):
         return True
 
     def close(self):
         pass
 
     def capabilities(self):
         return [b'batch']
 
     def _call(self, cmd, **args):
         args = pycompat.byteskwargs(args)
         res = wireproto.dispatch(self.serverrepo, proto(args), cmd)
         if isinstance(res, wireprototypes.bytesresponse):
             return res.data
         elif isinstance(res, bytes):
             return res
         else:
             raise error.Abort('dummy client does not support response type')
 
     def _callstream(self, cmd, **args):
         return stringio(self._call(cmd, **args))
 
-    @wireproto.batchable
+    @wireprotov1peer.batchable
     def greet(self, name):
-        f = wireproto.future()
+        f = wireprotov1peer.future()
         yield {b'name': mangle(name)}, f
         yield unmangle(f.value)
 
 class serverrepo(object):
     def greet(self, name):
         return b"Hello, " + name
 
     def filtered(self, name):
         return self
 
 def mangle(s):
     return b''.join(pycompat.bytechr(ord(c) + 1) for c in pycompat.bytestr(s))
 def unmangle(s):
     return b''.join(pycompat.bytechr(ord(c) - 1) for c in pycompat.bytestr(s))
 
 def greet(repo, proto, name):
     return mangle(repo.greet(unmangle(name)))
 
 wireproto.commands[b'greet'] = (greet, b'name',)
 
 srv = serverrepo()
 clt = clientpeer(srv, uimod.ui())
 
 print(clt.greet(b"Foobar"))
 b = clt.iterbatch()
 list(map(b.greet, (b'Fo, =;:<o', b'Bar')))
 b.submit()
 print([r for r in b.results()])