code: def setup_receive_proxy(
      shared_ray_instance,
  ) -> Generator[Tuple[ASGIReceiveProxy, ActorHandle], None, None]:
      @ray.remote
      class ASGIReceive:
          def __init__(self):
              self._message_queue = MessageQueue()
  
          def ready(self):
              pass
  
          async def put(self, message: Union[Exception, Message]):
              await self._message_queue(message)
  
          async def receive_asgi_messages(
              self, request_id: str
          ) -> bytes:
              await self._message_queue.wait_for_message()
              messages = self._message_queue.get_messages_nowait()
              for message in messages:
                  if isinstance(message, Exception):
                      raise message
  
              return pickle.dumps(messages)
  
      actor = ASGIReceive.remote()
      ray.get(actor.ready.remote())
      loop = get_or_create_event_loop()
      asgi_receive_proxy = ASGIReceiveProxy("", actor)
      receiver_task = loop.create_task(
          asgi_receive_proxy.fetch_until_disconnect()
      )
      try:
          yield asgi_receive_proxy, actor
      except Exception:
          receiver_task.cancel()
  
file_path: ray/python/ray/serve/tests/test_http_util.py
imports: import asyncio
  import pickle
  from typing import Generator, Tuple, Union
  import pytest
  from starlette.types import Message
  import ray
  from ray._private.utils import get_or_create_event_loop
  from ray.actor import ActorHandle
  from ray.serve._private.http_util import ASGIReceiveProxy, MessageQueue
label_desc: Use image node_id 2 for calling a global function with example usage: setup_receive_proxy(shared_ray_instance) and returns: pickle