comparison mercurial/bundle2.py @ 24686:e0e28e910fa3

bundle2: rename format, parts and config to final names

It is finally time to freeze the bundle2 format! To do so we:
- rename HG2Y to HG20,
- drop "b2x:" prefix from all part names,
- rename capability to "bundle2-exp" to "bundle2"
- rename the hook flag from 'bundle2-exp' to 'bundle2'

24685:b3d78d82d84c

            handler(op, part)
        finally:
            if output is not None:
                output = op.ui.popbuffer()
        if output:
            outpart = op.reply.newpart('b2x:output', data=output,
                                       mandatory=False)
            outpart.addparam('in-reply-to', str(part.id), mandatory=False)
    finally:
        # consume the part content to not corrupt the stream.
        part.seek(0, 2)

    Use the `addparam` method to add stream level parameter. and `newpart` to
    populate it. Then call `getchunks` to retrieve all the binary chunks of
    data that compose the bundle2 container."""

    _magicstring = 'HG2Y'

    def __init__(self, ui, capabilities=()):
        self.ui = ui
        self._params = []
        self._parts = []

        return None

    def compressed(self):
        return False

formatmap = {'2Y': unbundle20}

class bundlepart(object):
    """A bundle2 part contains application level payload

    The part `type` is used to route the part to the application level

                yield chunk
        except Exception, exc:
            # backup exception data for later
            exc_info = sys.exc_info()
            msg = 'unexpected error: %s' % exc
            interpart = bundlepart('b2x:error:abort', [('message', msg)],
                                   mandatory=False)
            interpart.id = 0
            yield _pack(_fpayloadsize, -1)
            for chunk in interpart.getchunks():
                yield chunk

            adjust = self.read(internaloffset)
            if len(adjust) != internaloffset:
                raise util.Abort(_('Seek failed\n'))
            self._pos = newpos

capabilities = {'HG2Y': (),
                'b2x:listkeys': (),
                'b2x:pushkey': (),
                'digests': tuple(sorted(util.DIGESTS.keys())),
                'b2x:remote-changegroup': ('http', 'https'),
               }

def getrepocaps(repo, allowpushback=False):
    """return the bundle2 capabilities for a given repo

    Exists to allow extensions (like evolution) to mutate the capabilities.
    """
    caps = capabilities.copy()
    caps['b2x:changegroup'] = tuple(sorted(changegroup.packermap.keys()))
    if obsolete.isenabled(repo, obsolete.exchangeopt):
        supportedformat = tuple('V%i' % v for v in obsolete.formats)
        caps['b2x:obsmarkers'] = supportedformat
    if allowpushback:
        caps['b2x:pushback'] = ()
    return caps

def bundle2caps(remote):
    """return the bundle capabilities of a peer as dict"""
    raw = remote.capable('bundle2-exp')
    if not raw and raw != '':
        return {}
    capsblob = urllib.unquote(remote.capable('bundle2-exp'))
    return decodecaps(capsblob)

def obsmarkersversion(caps):
    """extract the list of supported obsmarkers versions from a bundle2caps dict
    """
    obscaps = caps.get('b2x:obsmarkers', ())
    return [int(c[1:]) for c in obscaps if c.startswith('V')]

@parthandler('b2x:changegroup', ('version',))
def handlechangegroup(op, inpart):
    """apply a changegroup part on the repo

    This is a very early implementation that will massive rework before being
    inflicted to any end-user.

    ret = changegroup.addchangegroup(op.repo, cg, 'bundle2', 'bundle2')
    op.records.add('changegroup', {'return': ret})
    if op.reply is not None:
        # This is definitely not the final form of this
        # return. But one need to start somewhere.
        part = op.reply.newpart('b2x:reply:changegroup', mandatory=False)
        part.addparam('in-reply-to', str(inpart.id), mandatory=False)
        part.addparam('return', '%i' % ret, mandatory=False)
    assert not inpart.read()

_remotechangegroupparams = tuple(['url', 'size', 'digests'] +
    ['digest:%s' % k for k in util.DIGESTS.keys()])
@parthandler('b2x:remote-changegroup', _remotechangegroupparams)
def handleremotechangegroup(op, inpart):
    """apply a bundle10 on the repo, given an url and validation information

    All the information about the remote bundle to import are given as
    parameters. The parameters include:

    try:
        raw_url = inpart.params['url']
    except KeyError:
        raise util.Abort(_('remote-changegroup: missing "%s" param') % 'url')
    parsed_url = util.url(raw_url)
    if parsed_url.scheme not in capabilities['b2x:remote-changegroup']:
        raise util.Abort(_('remote-changegroup does not support %s urls') %
            parsed_url.scheme)

    try:
        size = int(inpart.params['size'])

    ret = changegroup.addchangegroup(op.repo, cg, 'bundle2', 'bundle2')
    op.records.add('changegroup', {'return': ret})
    if op.reply is not None:
        # This is definitely not the final form of this
        # return. But one need to start somewhere.
        part = op.reply.newpart('b2x:reply:changegroup')
        part.addparam('in-reply-to', str(inpart.id), mandatory=False)
        part.addparam('return', '%i' % ret, mandatory=False)
    try:
        real_part.validate()
    except util.Abort, e:
        raise util.Abort(_('bundle at %s is corrupted:\n%s') %
            (util.hidepassword(raw_url), str(e)))
    assert not inpart.read()

@parthandler('b2x:reply:changegroup', ('return', 'in-reply-to'))
def handlereplychangegroup(op, inpart):
    ret = int(inpart.params['return'])
    replyto = int(inpart.params['in-reply-to'])
    op.records.add('changegroup', {'return': ret}, replyto)

@parthandler('b2x:check:heads')
def handlecheckheads(op, inpart):
    """check that head of the repo did not change

    This is used to detect a push race when using unbundle.
    This replaces the "heads" argument of unbundle."""

    assert not h
    if heads != op.repo.heads():
        raise error.PushRaced('repository changed while pushing - '
                              'please try again')

@parthandler('b2x:output')
def handleoutput(op, inpart):
    """forward output captured on the server to the client"""
    for line in inpart.read().splitlines():
        op.ui.write(('remote: %s\n' % line))

@parthandler('b2x:replycaps')
def handlereplycaps(op, inpart):
    """Notify that a reply bundle should be created

    The payload contains the capabilities information for the reply"""
    caps = decodecaps(inpart.read())
    if op.reply is None:
        op.reply = bundle20(op.ui, caps)

@parthandler('b2x:error:abort', ('message', 'hint'))
def handlereplycaps(op, inpart):
    """Used to transmit abort error over the wire"""
    raise util.Abort(inpart.params['message'], hint=inpart.params.get('hint'))

@parthandler('b2x:error:unsupportedcontent', ('parttype', 'params'))
def handlereplycaps(op, inpart):
    """Used to transmit unknown content error over the wire"""
    kwargs = {}
    parttype = inpart.params.get('parttype')
    if parttype is not None:

    if params is not None:
        kwargs['params'] = params.split('\0')

    raise error.UnsupportedPartError(**kwargs)

@parthandler('b2x:error:pushraced', ('message',))
def handlereplycaps(op, inpart):
    """Used to transmit push race error over the wire"""
    raise error.ResponseError(_('push failed:'), inpart.params['message'])

@parthandler('b2x:listkeys', ('namespace',))
def handlelistkeys(op, inpart):
    """retrieve pushkey namespace content stored in a bundle2"""
    namespace = inpart.params['namespace']
    r = pushkey.decodekeys(inpart.read())
    op.records.add('listkeys', (namespace, r))

@parthandler('b2x:pushkey', ('namespace', 'key', 'old', 'new'))
def handlepushkey(op, inpart):
    """process a pushkey request"""
    dec = pushkey.decode
    namespace = dec(inpart.params['namespace'])
    key = dec(inpart.params['key'])

              'key': key,
              'old': old,
              'new': new}
    op.records.add('pushkey', record)
    if op.reply is not None:
        rpart = op.reply.newpart('b2x:reply:pushkey')
        rpart.addparam('in-reply-to', str(inpart.id), mandatory=False)
        rpart.addparam('return', '%i' % ret, mandatory=False)

@parthandler('b2x:reply:pushkey', ('return', 'in-reply-to'))
def handlepushkeyreply(op, inpart):
    """retrieve the result of a pushkey request"""
    ret = int(inpart.params['return'])
    partid = int(inpart.params['in-reply-to'])
    op.records.add('pushkey', {'return': ret}, partid)

@parthandler('b2x:obsmarkers')
def handleobsmarker(op, inpart):
    """add a stream of obsmarkers to the repo"""
    tr = op.gettransaction()
    new = op.repo.obsstore.mergemarkers(tr, inpart.read())
    if new:
        op.repo.ui.status(_('%i new obsolescence markers\n') % new)
    op.records.add('obsmarkers', {'new': new})
    if op.reply is not None:
        rpart = op.reply.newpart('b2x:reply:obsmarkers')
        rpart.addparam('in-reply-to', str(inpart.id), mandatory=False)
        rpart.addparam('new', '%i' % new, mandatory=False)


@parthandler('b2x:reply:obsmarkers', ('new', 'in-reply-to'))
def handlepushkeyreply(op, inpart):
    """retrieve the result of a pushkey request"""
    ret = int(inpart.params['new'])
    partid = int(inpart.params['in-reply-to'])
    op.records.add('obsmarkers', {'new': ret}, partid)

24686:e0e28e910fa3

            handler(op, part)
        finally:
            if output is not None:
                output = op.ui.popbuffer()
        if output:
            outpart = op.reply.newpart('output', data=output,
                                       mandatory=False)
            outpart.addparam('in-reply-to', str(part.id), mandatory=False)
    finally:
        # consume the part content to not corrupt the stream.
        part.seek(0, 2)

    Use the `addparam` method to add stream level parameter. and `newpart` to
    populate it. Then call `getchunks` to retrieve all the binary chunks of
    data that compose the bundle2 container."""

    _magicstring = 'HG20'

    def __init__(self, ui, capabilities=()):
        self.ui = ui
        self._params = []
        self._parts = []

        return None

    def compressed(self):
        return False

formatmap = {'20': unbundle20}

class bundlepart(object):
    """A bundle2 part contains application level payload

    The part `type` is used to route the part to the application level

                yield chunk
        except Exception, exc:
            # backup exception data for later
            exc_info = sys.exc_info()
            msg = 'unexpected error: %s' % exc
            interpart = bundlepart('error:abort', [('message', msg)],
                                   mandatory=False)
            interpart.id = 0
            yield _pack(_fpayloadsize, -1)
            for chunk in interpart.getchunks():
                yield chunk

            adjust = self.read(internaloffset)
            if len(adjust) != internaloffset:
                raise util.Abort(_('Seek failed\n'))
            self._pos = newpos

capabilities = {'HG20': (),
                'listkeys': (),
                'pushkey': (),
                'digests': tuple(sorted(util.DIGESTS.keys())),
                'remote-changegroup': ('http', 'https'),
               }

def getrepocaps(repo, allowpushback=False):
    """return the bundle2 capabilities for a given repo

    Exists to allow extensions (like evolution) to mutate the capabilities.
    """
    caps = capabilities.copy()
    caps['changegroup'] = tuple(sorted(changegroup.packermap.keys()))
    if obsolete.isenabled(repo, obsolete.exchangeopt):
        supportedformat = tuple('V%i' % v for v in obsolete.formats)
        caps['obsmarkers'] = supportedformat
    if allowpushback:
        caps['pushback'] = ()
    return caps

def bundle2caps(remote):
    """return the bundle capabilities of a peer as dict"""
    raw = remote.capable('bundle2')
    if not raw and raw != '':
        return {}
    capsblob = urllib.unquote(remote.capable('bundle2'))
    return decodecaps(capsblob)

def obsmarkersversion(caps):
    """extract the list of supported obsmarkers versions from a bundle2caps dict
    """
    obscaps = caps.get('obsmarkers', ())
    return [int(c[1:]) for c in obscaps if c.startswith('V')]

@parthandler('changegroup', ('version',))
def handlechangegroup(op, inpart):
    """apply a changegroup part on the repo

    This is a very early implementation that will massive rework before being
    inflicted to any end-user.

    ret = changegroup.addchangegroup(op.repo, cg, 'bundle2', 'bundle2')
    op.records.add('changegroup', {'return': ret})
    if op.reply is not None:
        # This is definitely not the final form of this
        # return. But one need to start somewhere.
        part = op.reply.newpart('reply:changegroup', mandatory=False)
        part.addparam('in-reply-to', str(inpart.id), mandatory=False)
        part.addparam('return', '%i' % ret, mandatory=False)
    assert not inpart.read()

_remotechangegroupparams = tuple(['url', 'size', 'digests'] +
    ['digest:%s' % k for k in util.DIGESTS.keys()])
@parthandler('remote-changegroup', _remotechangegroupparams)
def handleremotechangegroup(op, inpart):
    """apply a bundle10 on the repo, given an url and validation information

    All the information about the remote bundle to import are given as
    parameters. The parameters include:

    try:
        raw_url = inpart.params['url']
    except KeyError:
        raise util.Abort(_('remote-changegroup: missing "%s" param') % 'url')
    parsed_url = util.url(raw_url)
    if parsed_url.scheme not in capabilities['remote-changegroup']:
        raise util.Abort(_('remote-changegroup does not support %s urls') %
            parsed_url.scheme)

    try:
        size = int(inpart.params['size'])

    ret = changegroup.addchangegroup(op.repo, cg, 'bundle2', 'bundle2')
    op.records.add('changegroup', {'return': ret})
    if op.reply is not None:
        # This is definitely not the final form of this
        # return. But one need to start somewhere.
        part = op.reply.newpart('reply:changegroup')
        part.addparam('in-reply-to', str(inpart.id), mandatory=False)
        part.addparam('return', '%i' % ret, mandatory=False)
    try:
        real_part.validate()
    except util.Abort, e:
        raise util.Abort(_('bundle at %s is corrupted:\n%s') %
            (util.hidepassword(raw_url), str(e)))
    assert not inpart.read()

@parthandler('reply:changegroup', ('return', 'in-reply-to'))
def handlereplychangegroup(op, inpart):
    ret = int(inpart.params['return'])
    replyto = int(inpart.params['in-reply-to'])
    op.records.add('changegroup', {'return': ret}, replyto)

@parthandler('check:heads')
def handlecheckheads(op, inpart):
    """check that head of the repo did not change

    This is used to detect a push race when using unbundle.
    This replaces the "heads" argument of unbundle."""

    assert not h
    if heads != op.repo.heads():
        raise error.PushRaced('repository changed while pushing - '
                              'please try again')

@parthandler('output')
def handleoutput(op, inpart):
    """forward output captured on the server to the client"""
    for line in inpart.read().splitlines():
        op.ui.write(('remote: %s\n' % line))

@parthandler('replycaps')
def handlereplycaps(op, inpart):
    """Notify that a reply bundle should be created

    The payload contains the capabilities information for the reply"""
    caps = decodecaps(inpart.read())
    if op.reply is None:
        op.reply = bundle20(op.ui, caps)

@parthandler('error:abort', ('message', 'hint'))
def handlereplycaps(op, inpart):
    """Used to transmit abort error over the wire"""
    raise util.Abort(inpart.params['message'], hint=inpart.params.get('hint'))

@parthandler('error:unsupportedcontent', ('parttype', 'params'))
def handlereplycaps(op, inpart):
    """Used to transmit unknown content error over the wire"""
    kwargs = {}
    parttype = inpart.params.get('parttype')
    if parttype is not None:

    if params is not None:
        kwargs['params'] = params.split('\0')

    raise error.UnsupportedPartError(**kwargs)

@parthandler('error:pushraced', ('message',))
def handlereplycaps(op, inpart):
    """Used to transmit push race error over the wire"""
    raise error.ResponseError(_('push failed:'), inpart.params['message'])

@parthandler('listkeys', ('namespace',))
def handlelistkeys(op, inpart):
    """retrieve pushkey namespace content stored in a bundle2"""
    namespace = inpart.params['namespace']
    r = pushkey.decodekeys(inpart.read())
    op.records.add('listkeys', (namespace, r))

@parthandler('pushkey', ('namespace', 'key', 'old', 'new'))
def handlepushkey(op, inpart):
    """process a pushkey request"""
    dec = pushkey.decode
    namespace = dec(inpart.params['namespace'])
    key = dec(inpart.params['key'])

              'key': key,
              'old': old,
              'new': new}
    op.records.add('pushkey', record)
    if op.reply is not None:
        rpart = op.reply.newpart('reply:pushkey')
        rpart.addparam('in-reply-to', str(inpart.id), mandatory=False)
        rpart.addparam('return', '%i' % ret, mandatory=False)

@parthandler('reply:pushkey', ('return', 'in-reply-to'))
def handlepushkeyreply(op, inpart):
    """retrieve the result of a pushkey request"""
    ret = int(inpart.params['return'])
    partid = int(inpart.params['in-reply-to'])
    op.records.add('pushkey', {'return': ret}, partid)

@parthandler('obsmarkers')
def handleobsmarker(op, inpart):
    """add a stream of obsmarkers to the repo"""
    tr = op.gettransaction()
    new = op.repo.obsstore.mergemarkers(tr, inpart.read())
    if new:
        op.repo.ui.status(_('%i new obsolescence markers\n') % new)
    op.records.add('obsmarkers', {'new': new})
    if op.reply is not None:
        rpart = op.reply.newpart('reply:obsmarkers')
        rpart.addparam('in-reply-to', str(inpart.id), mandatory=False)
        rpart.addparam('new', '%i' % new, mandatory=False)


@parthandler('reply:obsmarkers', ('new', 'in-reply-to'))
def handlepushkeyreply(op, inpart):
    """retrieve the result of a pushkey request"""
    ret = int(inpart.params['new'])
    partid = int(inpart.params['in-reply-to'])
    op.records.add('obsmarkers', {'new': ret}, partid)