Repository: gradio-app/gradio
File: gradio/routes.py (Python)

Code before:
    head: bytearray
    subs: list[asyncio.Queue[bytes | None]]
    resp: asyncio.Future[httpx.Response | None]


class NodeProxyCache:
    def __init__(self, client: httpx.AsyncClient):
        self.client = client
        self.cache: dict[str, NodeProxyCacheEntry | None] = {}

    async def get(
        self,
        method: str,
        url: str,
        headers: dict[str, str],
    ):
        key = f"{method} {url}"
        key += "::".join(map(":".join, headers.items()))
        res: asyncio.Queue[bytes | None] = asyncio.Queue()
        if (entry := self.cache.get(key, None)) is None:
            loop = asyncio.get_running_loop()
            entry = NodeProxyCacheEntry(bytearray(), [], loop.create_future())
            asyncio.create_task(self.fetch(key, entry, method, url, headers))
            self.cache[key] = entry
        entry.subs.append(res)
        head = bytes(entry.head)
        if (resp := await entry.resp) is None:
            raise Error("Error while proxying request to Node server")
        return resp.status_code, resp.headers, NodeProxyCache.iter_body(head, res)

    async def fetch(
        self,
        key: str,
        entry: NodeProxyCacheEntry,
        method: str,
        url: str,
        headers: dict[str, str],
    ):
        request = self.client.build_request(method, httpx.URL(url), headers=headers)
        try:
            response = await self.client.send(request, stream=True)
        except Exception:
            entry.resp.set_result(None)
            raise
        entry.resp.set_result(response)
        try:
            async for bytes_chunk in response.aiter_raw():
                entry.head.extend(bytes_chunk)
                for sub in entry.subs:
                    sub.put_nowait(bytes_chunk)
        finally:

Review comment (suggestion):
The use of `head` is really clever. But is there a risk of duplication here? Between `entry.subs.append(res)` and `head = bytes(entry.head)` the same piece of data could be added to both the head and the subs queue.

Code after:
    FastAPI App Wrapper
    """

    app_port = None

    def __init__(
        self,
        auth_dependency: Callable[[fastapi.Request], str | None] | None = None,
        **kwargs,
    ):
        self.tokens = {}
        self.auth = None
        self.analytics_key = secrets.token_urlsafe(16)
        self.monitoring_enabled = False
        self.blocks: gradio.Blocks | None = None
        self.state_holder = StateHolder()
        self.iterators: dict[str, AsyncIterator] = {}
        self.iterators_to_reset: set[str] = set()
        self.lock = utils.safe_get_lock()
        self.stop_event = utils.safe_get_stop_event()
        self.cookie_id = secrets.token_urlsafe(32)
        self.queue_token = secrets.token_urlsafe(32)
        self.startup_events_triggered = False
        self.uploaded_file_dir = get_upload_folder()
        self.change_count: int = 0
        self.change_type: Literal["reload", "error"] | None = None
        self.reload_error_message: str | None = None
        self._asyncio_tasks: list[asyncio.Task] = []
        self.auth_dependency = auth_dependency
        self.api_info = None
        self.all_app_info = None

        # Allow user to manually set `docs_url` and `redoc_url`
        # when instantiating an App; when they're not set, disable docs and redoc.
        kwargs.setdefault("docs_url", None)
        kwargs.setdefault("redoc_url", None)
        self.custom_component_hashes: dict[str, str] = {}
        super().__init__(**kwargs)

    # Create a single client to be reused across requests
    # We're not overriding any defaults here

    client = httpx.AsyncClient()
    proxy_cache = NodeProxyCache(client)

    @staticmethod
    async def proxy_to_node(
        request: fastapi.Request,
        server_name: str,
        node_port: int,
        python_port: int,