kitty/kittens/transfer/rsync.pyi

from typing import Callable, Union

from kitty.typing import ReadableBuffer, WriteableBuffer

class RsyncError(Exception):
    pass

class Hasher:
    def __init__(self, which: str, data: ReadableBuffer = b''): ...
    def update(self, data: ReadableBuffer) -> None: ...
    def reset(self) -> None: ...
    def digest(self) -> bytes: ...
    def hexdigest(self) -> str: ...

    @property
    def digest_size(self) -> int: ...
    @property
    def block_size(self) -> int: ...
    @property
    def name(self) -> str: ...

def xxh128_hash(data: ReadableBuffer) -> bytes: ...
def xxh128_hash_with_seed(data: ReadableBuffer, seed: int) -> bytes: ...


class Patcher:

    def __init__(self, expected_input_size: int = 0): ...
    def signature_header(self, output: WriteableBuffer) -> int: ...
    def sign_block(self, block: ReadableBuffer, output: WriteableBuffer) -> int: ...
    def apply_delta_data(self, data: ReadableBuffer, read: Callable[[int, WriteableBuffer], int], write: Callable[[ReadableBuffer], None]) -> None: ...
    def finish_delta_data(self) -> None: ...

    @property
    def block_size(self) -> int: ...
    @property
    def total_data_in_delta(self) -> int: ...


class Differ:

    def add_signature_data(self, data: ReadableBuffer) -> None: ...
    def finish_signature_data(self) -> None: ...
    def next_op(self, read: Callable[[WriteableBuffer], int], write: Callable[[ReadableBuffer], None]) -> bool: ...


def parse_ftc(x: Union[str, ReadableBuffer], callback: Callable[[memoryview, memoryview], None]) -> None: ...