comparison mercurial/bundle2.py @ 30351:f81002f736d7

bundle2: use new compression engine API for compression

Now that we have a new API to define compression engines, let's put it
to use!

The new code stores a reference to the compression engine instead of
a low-level compressor object. This will allow us to more easily
transition to different APIs on the compression engine interface
once we implement them.

As part of this, we change the registration in bundletypes to use 'UN'
instead of None. Previously, util.compressors had the no-op compressor
registered under both the 'UN' and None keys. Since we're switching to
a new API, I don't see the point in carrying this dual registration
forward.

30350:358cda0af6ee

            ca = "%s=%s" % (ca, ','.join(vals))
        chunks.append(ca)
    return '\n'.join(chunks)

bundletypes = {
    "": ("", None),       # only when using unbundle on ssh and old http servers
                          # since the unification ssh accepts a header but there
                          # is no capability signaling it.
    "HG20": (), # special-cased below
    "HG10UN": ("HG10UN", None),
    "HG10BZ": ("HG10", 'BZ'),
    "HG10GZ": ("HG10GZ", 'GZ'),
}

# hgweb uses this list to communicate its preferred type

    def __init__(self, ui, capabilities=()):
        self.ui = ui
        self._params = []
        self._parts = []
        self.capabilities = dict(capabilities)
        self._compressor = util.compressors[None]()

    def setcompression(self, alg):
        """setup core part compression to <alg>"""
        if alg is None:
            return
        assert not any(n.lower() == 'Compression' for n, v in self._params)
        self.addparam('Compression', alg)
        self._compressor = util.compressors[alg]()

    @property
    def nbparts(self):
        """total number of parts added to the bundler"""
        return len(self._parts)

        outdebug(self.ui, 'bundle parameter: %s' % param)
        yield _pack(_fstreamparamsize, len(param))
        if param:
            yield param
        # starting compression
        for chunk in self._getcorechunk():
            data = self._compressor.compress(chunk)
            if data:
                yield data
        yield self._compressor.flush()

    def _paramchunk(self):
        """return a encoded version of all stream parameters"""
        blocks = []
        for par, value in self._params:

        assert compression is None
        if cg.version != '01':
            raise error.Abort(_('old bundle types only supports v1 '
                                'changegroups'))
        header, comp = bundletypes[bundletype]
        if comp not in util.compressors:
            raise error.Abort(_('unknown stream compression type: %s')
                              % comp)
        z = util.compressors[comp]()
        subchunkiter = cg.getchunks()
        def chunkiter():
            yield header
            for chunk in subchunkiter:
                data = z.compress(chunk)
                if data:
                    yield data
            yield z.flush()
        chunkiter = chunkiter()

    # parse the changegroup data, otherwise we will block
    # in case of sshrepo because we don't know the end of the stream
    return changegroup.writechunks(ui, chunkiter, filename, vfs=vfs)

30351:f81002f736d7

            ca = "%s=%s" % (ca, ','.join(vals))
        chunks.append(ca)
    return '\n'.join(chunks)

bundletypes = {
    "": ("", 'UN'),       # only when using unbundle on ssh and old http servers
                          # since the unification ssh accepts a header but there
                          # is no capability signaling it.
    "HG20": (), # special-cased below
    "HG10UN": ("HG10UN", 'UN'),
    "HG10BZ": ("HG10", 'BZ'),
    "HG10GZ": ("HG10GZ", 'GZ'),
}

# hgweb uses this list to communicate its preferred type

    def __init__(self, ui, capabilities=()):
        self.ui = ui
        self._params = []
        self._parts = []
        self.capabilities = dict(capabilities)
        self._compengine = util.compengines.forbundletype('UN')

    def setcompression(self, alg):
        """setup core part compression to <alg>"""
        if alg is None:
            return
        assert not any(n.lower() == 'Compression' for n, v in self._params)
        self.addparam('Compression', alg)
        self._compengine = util.compengines.forbundletype(alg)

    @property
    def nbparts(self):
        """total number of parts added to the bundler"""
        return len(self._parts)

        outdebug(self.ui, 'bundle parameter: %s' % param)
        yield _pack(_fstreamparamsize, len(param))
        if param:
            yield param
        # starting compression
        compressor = self._compengine.compressorobj()
        for chunk in self._getcorechunk():
            data = compressor.compress(chunk)
            if data:
                yield data
        yield compressor.flush()

    def _paramchunk(self):
        """return a encoded version of all stream parameters"""
        blocks = []
        for par, value in self._params:

        assert compression is None
        if cg.version != '01':
            raise error.Abort(_('old bundle types only supports v1 '
                                'changegroups'))
        header, comp = bundletypes[bundletype]
        if comp not in util.compengines.supportedbundletypes:
            raise error.Abort(_('unknown stream compression type: %s')
                              % comp)
        compengine = util.compengines.forbundletype(comp)
        compressor = compengine.compressorobj()
        subchunkiter = cg.getchunks()
        def chunkiter():
            yield header
            for chunk in subchunkiter:
                data = compressor.compress(chunk)
                if data:
                    yield data
            yield compressor.flush()
        chunkiter = chunkiter()

    # parse the changegroup data, otherwise we will block
    # in case of sshrepo because we don't know the end of the stream
    return changegroup.writechunks(ui, chunkiter, filename, vfs=vfs)