comparison mercurial/pure/parsers.py @ 46407:095fa99ae5f5

revlog: prepare pure parser for being overloaded

The current class uses module-level variables which don't allow for reusing the
current logic for a slightly different revlog version.

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

46406:95054317e172

def dirstatetuple(*x):
    # x is a tuple
    return x


indexformatng = b">Qiiiiii20s12x"
indexfirst = struct.calcsize(b'Q')
sizeint = struct.calcsize(b'i')
indexsize = struct.calcsize(indexformatng)
nullitem = (0, 0, 0, -1, -1, -1, -1, nullid)


def gettype(q):
    return int(q & 0xFFFF)


def offset_type(offset, type):
    return int(int(offset) << 16 | type)


class BaseIndexObject(object):
    @property
    def nodemap(self):
        msg = b"index.nodemap is deprecated, use index.[has_node|rev|get_rev]"
        util.nouideprecwarn(msg, b'5.3', stacklevel=2)
        return self._nodemap

        return self._lgt + len(self._extra)

    def append(self, tup):
        if '_nodemap' in vars(self):
            self._nodemap[tup[7]] = len(self)
        data = _pack(indexformatng, *tup)
        self._extra.append(data)

    def _check_index(self, i):
        if not isinstance(i, int):
            raise TypeError(b"expecting int indexes")
        if i < 0 or i >= len(self):
            raise IndexError

    def __getitem__(self, i):
        if i == -1:
            return nullitem
        self._check_index(i)
        if i >= self._lgt:
            data = self._extra[i - self._lgt]
        else:
            index = self._calculate_index(i)
            data = self._data[index : index + indexsize]
        r = _unpack(indexformatng, data)
        if self._lgt and i == 0:
            r = (offset_type(0, gettype(r[0])),) + r[1:]
        return r


class IndexObject(BaseIndexObject):
    def __init__(self, data):
        assert len(data) % indexsize == 0
        self._data = data
        self._lgt = len(data) // indexsize
        self._extra = []

    def _calculate_index(self, i):
        return i * indexsize

    def __delitem__(self, i):
        if not isinstance(i, slice) or not i.stop == -1 or i.step is not None:
            raise ValueError(b"deleting slices only supports a:-1 with step 1")
        i = i.start
        self._check_index(i)
        self._stripnodes(i)
        if i < self._lgt:
            self._data = self._data[: i * indexsize]
            self._lgt = i
            self._extra = []
        else:
            self._extra = self._extra[: i - self._lgt]

    def _inline_scan(self, lgt):
        off = 0
        if lgt is not None:
            self._offsets = [0] * lgt
        count = 0
        while off <= len(self._data) - indexsize:
            (s,) = struct.unpack(
                b'>i', self._data[off + indexfirst : off + sizeint + indexfirst]
            )
            if lgt is not None:
                self._offsets[count] = off
            count += 1
            off += indexsize + s
        if off != len(self._data):
            raise ValueError(b"corrupted data")
        return count

    def __delitem__(self, i):

46407:095fa99ae5f5

def dirstatetuple(*x):
    # x is a tuple
    return x


def gettype(q):
    return int(q & 0xFFFF)


def offset_type(offset, type):
    return int(int(offset) << 16 | type)


class BaseIndexObject(object):
    index_format = b">Qiiiiii20s12x"
    big_int_size = struct.calcsize(b'Q')
    int_size = struct.calcsize(b'i')
    index_size = struct.calcsize(index_format)
    null_item = (0, 0, 0, -1, -1, -1, -1, nullid)

    @property
    def nodemap(self):
        msg = b"index.nodemap is deprecated, use index.[has_node|rev|get_rev]"
        util.nouideprecwarn(msg, b'5.3', stacklevel=2)
        return self._nodemap

        return self._lgt + len(self._extra)

    def append(self, tup):
        if '_nodemap' in vars(self):
            self._nodemap[tup[7]] = len(self)
        data = _pack(self.index_format, *tup)
        self._extra.append(data)

    def _check_index(self, i):
        if not isinstance(i, int):
            raise TypeError(b"expecting int indexes")
        if i < 0 or i >= len(self):
            raise IndexError

    def __getitem__(self, i):
        if i == -1:
            return self.null_item
        self._check_index(i)
        if i >= self._lgt:
            data = self._extra[i - self._lgt]
        else:
            index = self._calculate_index(i)
            data = self._data[index : index + self.index_size]
        r = _unpack(self.index_format, data)
        if self._lgt and i == 0:
            r = (offset_type(0, gettype(r[0])),) + r[1:]
        return r


class IndexObject(BaseIndexObject):
    def __init__(self, data):
        assert len(data) % self.index_size == 0
        self._data = data
        self._lgt = len(data) // self.index_size
        self._extra = []

    def _calculate_index(self, i):
        return i * self.index_size

    def __delitem__(self, i):
        if not isinstance(i, slice) or not i.stop == -1 or i.step is not None:
            raise ValueError(b"deleting slices only supports a:-1 with step 1")
        i = i.start
        self._check_index(i)
        self._stripnodes(i)
        if i < self._lgt:
            self._data = self._data[: i * self.index_size]
            self._lgt = i
            self._extra = []
        else:
            self._extra = self._extra[: i - self._lgt]

    def _inline_scan(self, lgt):
        off = 0
        if lgt is not None:
            self._offsets = [0] * lgt
        count = 0
        while off <= len(self._data) - self.index_size:
            start = off + self.big_int_size
            (s,) = struct.unpack(
                b'>i',
                self._data[start : start + self.int_size],
            )
            if lgt is not None:
                self._offsets[count] = off
            count += 1
            off += self.index_size + s
        if off != len(self._data):
            raise ValueError(b"corrupted data")
        return count

    def __delitem__(self, i):