Skip to content

BinaryWriter

faststream.redis.parser.binary.BinaryWriter #

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

data instance-attribute #

data = bytearray()

write #

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

write_short #

write_short(number: int) -> None
Source code in faststream/redis/parser/binary.py 
119
120
121
def write_short(self, number: int) -> None:
    int_bytes = pack(">H", number)
    self.write(int_bytes)

write_int #

write_int(number: int) -> None
Source code in faststream/redis/parser/binary.py 
123
124
125
def write_int(self, number: int) -> None:
    int_bytes = pack(">I", number)
    self.write(int_bytes)

write_string #

write_string(data: str | bytes) -> None
Source code in faststream/redis/parser/binary.py 
127
128
129
130
131
132
133
def write_string(self, data: 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() -> bytes
Source code in faststream/redis/parser/binary.py 
135
136
def get_bytes(self) -> bytes:
    return bytes(self.data)