Method _decode_chunked

stream/interceptors.py:529–558

(response_body: bytes)

    @staticmethod
    def _decode_chunked(response_body: bytes) -> Tuple[bytes, bool]:
        chunked_data = bytearray()
        while True:
            length_crlf_idx = response_body.find(b"\r\n")
            if length_crlf_idx == -1:
                break

            hex_length = response_body[:length_crlf_idx]
            try:
                length = int(hex_length, 16)
            except ValueError as e:
                logging.error(f"Parsing chunked length failed: {e}")
                break

            if length == 0:
                length_crlf_idx = response_body.find(b"0\r\n\r\n")
                if length_crlf_idx != -1:
                    return bytes(chunked_data), True

            if length + 2 > len(response_body):
                break

            chunked_data.extend(
                response_body[length_crlf_idx + 2 : length_crlf_idx + 2 + length]
            )
            if length_crlf_idx + 2 + length + 2 > len(response_body):
                break

            response_body = response_body[length_crlf_idx + 2 + length + 2 :]
        return bytes(chunked_data), False