Skip to content

BinaryWriter

faststream.redis.parser.BinaryWriter #

BinaryWriter()
Source code in faststream/redis/parser.py 
48
49
def __init__(self) -> None:
    self.data = bytearray()

data instance-attribute #

data = bytearray()

write #

write(data)
Source code in faststream/redis/parser.py 
51
52
def write(self, data: bytes) -> None:
    self.data.extend(data)

write_short #

write_short(number)
Source code in faststream/redis/parser.py 
54
55
56
def write_short(self, number: int) -> None:
    int_bytes = pack(">H", number)
    self.write(int_bytes)

write_int #

write_int(number)
Source code in faststream/redis/parser.py 
58
59
60
def write_int(self, number: int) -> None:
    int_bytes = pack(">I", number)
    self.write(int_bytes)

write_string #

write_string(data)
Source code in faststream/redis/parser.py 
62
63
64
65
66
67
68
def write_string(self, data: Union[str, bytes]) -> None:
    str_len = len(data)
    self.write_short(str_len)
    if isinstance(data, bytes):
        self.write(data)
    else:
        self.write(data.encode())

get_bytes #

get_bytes()
Source code in faststream/redis/parser.py 
70
71
def get_bytes(self) -> bytes:
    return bytes(self.data)