code: def test_responses_actually_streamed(
      serve_instance,
      use_fastapi: bool,
      use_async: bool,
      use_multiple_replicas: bool,
  ):
      """Checks that responses are streamed as they are yielded.
  
      Also checks that responses can be streamed concurrently from a single replica
      or from multiple replicas.
      """
      signal_actor = SignalActor.remote()
  
      async def wait_on_signal_async():
          yield f"{os.getpid()}: before signal"
          await signal_actor.wait.remote()
          yield f"{os.getpid()}: after signal"
  
      def wait_on_signal_sync():
          yield f"{os.getpid()}: before signal"
          ray.get(signal_actor.wait.remote())
          yield f"{os.getpid()}: after signal"
  
      if use_fastapi:
          app = FastAPI()
  
          @serve.deployment
          @serve.ingress(app)
          class SimpleGenerator:
              @app.get("/")
              def stream(self, request: Request) -> StreamingResponse:
                  gen = (
                      wait_on_signal_async()
                      if use_async
                      else wait_on_signal_sync()
                  )
                  return StreamingResponse(gen, media_type="text/plain")
  
      else:
  
          @serve.deployment
          class SimpleGenerator:
              def __call__(self, request: Request) -> StreamingResponse:
                  gen = (
                      wait_on_signal_async()
                      if use_async
                      else wait_on_signal_sync()
                  )
                  return StreamingResponse(gen, media_type="text/plain")
  
      serve.run(
          SimpleGenerator.options(
              ray_actor_options={"num_cpus": 0},
              num_replicas=2 if use_multiple_replicas else 1,
          ).bind()
      )
  
      requester = StreamingRequester.remote()
      gen1 = requester.make_request.options(
          num_returns="streaming"
      ).remote()
      gen2 = requester.make_request.options(
          num_returns="streaming"
      ).remote()
  
      # Check that we get the first responses before the signal is sent
      # (so the generator is still hanging after the first yield).
      gen1_result = ray.get(next(gen1))
      gen2_result = ray.get(next(gen2))
      assert gen1_result.endswith("before signal")
      assert gen2_result.endswith("before signal")
      gen1_pid = gen1_result.split(":")[0]
      gen2_pid = gen2_result.split(":")[0]
      if use_multiple_replicas:
          assert gen1_pid != gen2_pid
      else:
          assert gen1_pid == gen2_pid
  
      # Check that the next obj_ref is not ready yet for both generators.
      assert gen1._next_sync(timeout_s=0.01).is_nil()
      assert gen2._next_sync(timeout_s=0.01).is_nil()
  
      # Now send signal to actor, second yield happens and we should get responses.
      ray.get(signal_actor.send.remote())
      gen1_result = ray.get(next(gen1))
      gen2_result = ray.get(next(gen2))
      assert gen1_result.startswith(gen1_pid)
      assert gen2_result.startswith(gen2_pid)
      assert gen1_result.endswith("after signal")
      assert gen2_result.endswith("after signal")
  
      # Client should be done getting messages.
      with pytest.raises(StopIteration):
          next(gen1)
  
      with pytest.raises(StopIteration):
          next(gen2)
  
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 2 for calling a global function with example usage: test_responses_actually_streamed(serve_instance, use_fastapi, use_async, use_multiple_replicas) and returns: StreamingResponse, StreamingResponse