Method _process_server_data

stream/proxy_server.py:354–435

()

                self._safe_close(server_writer)

        async def _process_server_data():
            nonlocal server_buffer, should_sniff
            try:
                while True:
                    data = await server_reader.read(8192)
                    if not data:
                        break

                    server_buffer.extend(data)
                    if b"\r\n\r\n" in server_buffer:
                        headers_end = server_buffer.find(b"\r\n\r\n") + 4
                        headers_data = server_buffer[:headers_end]
                        body_data = server_buffer[headers_end:]

                        lines = headers_data.split(b"\r\n")
                        status_code = 200
                        status_message = "OK"
                        if lines and lines[0]:
                            try:
                                status_line = lines[0].decode("utf-8")
                                parts = status_line.split(" ", 2)
                                if len(parts) >= 2:
                                    status_code = int(parts[1])
                                    status_message = parts[2] if len(parts) > 2 else ""
                            except (ValueError, UnicodeDecodeError):
                                pass

                        headers: dict[str, str] = {}
                        for i in range(1, len(lines)):
                            if not lines[i]:
                                continue
                            try:
                                key, value = lines[i].decode("utf-8").split(":", 1)
                                headers[key.strip()] = value.strip()
                            except ValueError:
                                continue

                        if should_sniff:
                            try:
                                if status_code >= 400:
                                    self.logger.error(
                                        f"[UPSTREAM ERROR] {status_code} {status_message}"
                                    )
                                    if self.queue is not None:
                                        error_payload = {
                                            "error": True,
                                            "status": status_code,
                                            "message": f"{status_code} {status_message}",
                                            "done": True,
                                        }
                                        self.queue.put(json.dumps(error_payload))
                                else:
                                    resp = await self.interceptor.process_response(
                                        bytes(body_data), host, "", headers
                                    )
                                    if self.queue is not None:
                                        payload = {
                                            "ts": request_context.get("request_ts", 0),