Method _process_client_data

stream/proxy_server.py:292–352

()

        request_context = {"request_ts": 0.0}

        async def _process_client_data():
            nonlocal client_buffer, should_sniff
            try:
                while True:
                    data = await client_reader.read(8192)
                    if not data:
                        break
                    client_buffer.extend(data)

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

                        lines = headers_data.split(b"\r\n")
                        request_line = lines[0].decode("utf-8")

                        try:
                            _method, path, _ = request_line.split(" ")
                        except ValueError:
                            server_writer.write(client_buffer)
                            await server_writer.drain()
                            client_buffer.clear()
                            continue

                        if "GenerateContent" in path or "generateContent" in path:
                            should_sniff = True
                            request_context["request_ts"] = time.time()
                            # Reset interceptor state for new request to prevent
                            # state leakage from previous requests
                            self.interceptor.reset_for_new_request()
                            self.logger.debug(
                                f"[Proxy] Detected GenerateContent request: {path[:60]}..."
                            )
                            processed_body = await self.interceptor.process_request(
                                bytes(body_data), host, path
                            )
                            server_writer.write(headers_data)
                            if isinstance(processed_body, bytes):
                                server_writer.write(processed_body)
                        else:
                            should_sniff = False
                            server_writer.write(client_buffer)

                        await server_writer.drain()
                        client_buffer.clear()
                    else:
                        server_writer.write(data)
                        await server_writer.drain()
                        client_buffer.clear()
            except ConnectionResetError:
                self.logger.debug("Connection reset by peer processing client data.")
            except Exception as e:
                if "Broken pipe" in str(e) or "Connection reset" in str(e):
                    self.logger.debug(f"[Proxy] Client disconnected: {e}")
                else:
                    self.logger.error(
                        f"Error processing client data: {e}", exc_info=True