mercurial/revlog.py

51676:031d66801d5f

import io
import os
import struct
import weakref
import zlib

# import stuff from node for others to import from revlog
from .node import (
    bin,
    hex,

            return None
        t = self._default_compression_header
        c = self._get_decompressor(t)
        return c.decompress

    def _get_decompressor(self, t):
        try:
            compressor = self._decompressors[t]
        except KeyError:
            try:
                engine = util.compengines.forrevlogheader(t)

                raise error.RevlogError(
                    _(b'unknown compression type %s') % binascii.hexlify(t)
                )
        return compressor

    def compress(self, data):
        """Generate a possibly-compressed representation of data."""
        if not data:
            return b'', data

        compressed = self._compressor.compress(data)

        if data[0:1] == b'\0':
            return b'', data
        return b'u', data

    def decompress(self, data):
        """Decompress a revlog chunk.

        The chunk is expected to begin with a header identifying the
        format type so it can be routed to an appropriate decompressor.
        """

        _format_flags = header & ~0xFFFF
        _format_version = header & 0xFFFF

        features = FEATURES_BY_VERSION[_format_version]
        return features[b'inline'](_format_flags)

    def __init__(
        self,
        opener,
        target,

                cachedelta,
                deltacomputer=deltacomputer,
                sidedata=sidedata,
            )

    def compress(self, data):
        return self._inner.compress(data)

    def decompress(self, data):
        return self._inner.decompress(data)

51677:df6ce326936f

import io
import os
import struct
import weakref
import zlib

from typing import (
    Optional,
    Tuple,
)

# import stuff from node for others to import from revlog
from .node import (
    bin,
    hex,

            return None
        t = self._default_compression_header
        c = self._get_decompressor(t)
        return c.decompress

    def _get_decompressor(self, t: bytes):
        try:
            compressor = self._decompressors[t]
        except KeyError:
            try:
                engine = util.compengines.forrevlogheader(t)

                raise error.RevlogError(
                    _(b'unknown compression type %s') % binascii.hexlify(t)
                )
        return compressor

    def compress(self, data: bytes) -> Tuple[bytes, bytes]:
        """Generate a possibly-compressed representation of data."""
        if not data:
            return b'', data

        compressed = self._compressor.compress(data)

        if data[0:1] == b'\0':
            return b'', data
        return b'u', data

    def decompress(self, data: bytes):
        """Decompress a revlog chunk.

        The chunk is expected to begin with a header identifying the
        format type so it can be routed to an appropriate decompressor.
        """

        _format_flags = header & ~0xFFFF
        _format_version = header & 0xFFFF

        features = FEATURES_BY_VERSION[_format_version]
        return features[b'inline'](_format_flags)

    _docket_file: Optional[bytes]

    def __init__(
        self,
        opener,
        target,

                cachedelta,
                deltacomputer=deltacomputer,
                sidedata=sidedata,
            )

    def compress(self, data: bytes) -> Tuple[bytes, bytes]:
        return self._inner.compress(data)

    def decompress(self, data):
        return self._inner.decompress(data)