Skip to content

BinaryReader

faststream.redis.parser.binary.BinaryReader #

BinaryReader(data: bytes)
Source code in faststream/redis/parser/binary.py 
145
146
147
def __init__(self, data: bytes) -> None:
    self.data = data
    self.offset = 0

data instance-attribute #

data = data

offset instance-attribute #

offset = 0

read_until #

read_until(offset: int) -> bytes
Source code in faststream/redis/parser/binary.py 
149
150
151
152
def read_until(self, offset: int) -> bytes:
    data = self.data[self.offset : self.offset + offset]
    self.offset += offset
    return data

shift_offset_to #

shift_offset_to(offset: int) -> None
Source code in faststream/redis/parser/binary.py 
154
155
def shift_offset_to(self, offset: int) -> None:
    self.offset = offset

read_short #

read_short() -> int
Source code in faststream/redis/parser/binary.py 
157
158
159
160
def read_short(self) -> int:
    data = unpack(">H", self.data[self.offset : self.offset + 2])[0]
    self.offset += 2
    return int(data)

read_int #

read_int() -> int
Source code in faststream/redis/parser/binary.py 
162
163
164
165
def read_int(self) -> int:
    data = unpack(">I", self.data[self.offset : self.offset + 4])[0]
    self.offset += 4
    return int(data)

read_string #

read_string() -> str
Source code in faststream/redis/parser/binary.py 
167
168
169
170
171
def read_string(self) -> str:
    str_len = self.read_short()
    data = self.data[self.offset : self.offset + str_len]
    self.offset += str_len
    return data.decode()

read_bytes #

read_bytes() -> bytes
Source code in faststream/redis/parser/binary.py 
173
174
def read_bytes(self) -> bytes:
    return self.data[self.offset :]