code: def test_http_disconnect(serve_instance):
      """Test that response generators are cancelled when the client disconnects."""
      signal_actor = SignalActor.remote()
  
      @serve.deployment
      class SimpleGenerator:
          def __call__(self, request: Request) -> StreamingResponse:
              async def wait_for_disconnect():
                  try:
                      yield "hi"
                      await asyncio.sleep(100)
                  except asyncio.CancelledError:
                      print("Cancelled!")
                      signal_actor.send.remote()
  
              return StreamingResponse(wait_for_disconnect())
  
      serve.run(SimpleGenerator.bind())
  
      with requests.get("http://localhost:8000", stream=True):
          with pytest.raises(TimeoutError):
              ray.get(signal_actor.wait.remote(), timeout=1)
  
      ray.get(signal_actor.wait.remote(), timeout=5)
  
file_path: ray/python/ray/serve/tests/test_streaming_response.py
imports: import asyncio
  import os
  from typing import AsyncGenerator
  import pytest
  import requests
  from fastapi import FastAPI
  from starlette.requests import Request
  from starlette.responses import StreamingResponse
  import ray
  from ray import serve
  from ray._private.test_utils import SignalActor
  from ray.serve.handle import RayServeHandle
label_desc: Use image node_id 6 for calling a global function with example usage: test_http_disconnect(serve_instance) and returns: StreamingResponse