class_docstr: The asynchronous long polling client.
  
  Args:
      host_actor: handle to actor embedding LongPollHost.
      key_listeners: a dictionary mapping keys to
        callbacks to be called on state update for the corresponding keys.
      call_in_event_loop: an asyncio event loop
        to post the callback into.
code: def _process_update(self, updates: Dict[str, UpdatedObject]):
      if isinstance(updates, (ray.exceptions.RayActorError)):
          # This can happen during shutdown where the controller is
          # intentionally killed, the client should just gracefully
          # exit.
          logger.debug(
              "LongPollClient failed to connect to host. Shutting down."
          )
          self.is_running = False
          return
  
      if isinstance(updates, ConnectionError):
          logger.warning(
              "LongPollClient connection failed, shutting down."
          )
          self.is_running = False
          return
  
      if isinstance(updates, (ray.exceptions.RayTaskError)):
          # Some error happened in the controller. It could be a bug or
          # some undesired state.
          logger.error("LongPollHost errored\n" + updates.traceback_str)
          # We must call this in event loop so it works in Ray Client.
          # See https://github.com/ray-project/ray/issues/20971
          self._schedule_to_event_loop(self._poll_next)
          return
  
      if updates == LongPollState.TIME_OUT:
          logger.debug("LongPollClient polling timed out. Retrying.")
          self._schedule_to_event_loop(self._poll_next)
          return
  
      logger.debug(
          f"LongPollClient {self} received updates for keys: "
          f"{list(updates.keys())}.",
          extra={"log_to_stderr": False},
      )
      for key, update in updates.items():
          self.snapshot_ids[key] = update.snapshot_id
          callback = self.key_listeners[key]
  
          # Bind the parameters because closures are late-binding.
          # https://docs.python-guide.org/writing/gotchas/#late-binding-closures # noqa: E501
          def chained(callback=callback, arg=update.object_snapshot):
              callback(arg)
              self._on_callback_completed(trigger_at=len(updates))
  
          self._schedule_to_event_loop(chained)
  
imports: import asyncio
  import logging
  import os
  import random
  from asyncio.events import AbstractEventLoop
  from collections import defaultdict
  from dataclasses import dataclass
  from enum import Enum, auto
  from typing import Any, Callable, DefaultDict, Dict, Optional, Set, Tuple, Union
  import ray
  from ray._private.utils import get_or_create_event_loop
  from ray.serve._private.common import ReplicaName
  from ray.serve._private.constants import SERVE_LOGGER_NAME
  from ray.serve._private.utils import format_actor_name
  from ray.serve.generated.serve_pb2 import ActorNameList
  from ray.serve.generated.serve_pb2 import EndpointInfo
  from ray.serve.generated.serve_pb2 import EndpointSet, LongPollRequest, LongPollResult
  from ray.serve.generated.serve_pb2 import UpdatedObject
  from ray.util import metrics
label_desc: Use image node_id 5 for calling the LongPollClient obj's underlying member method code with example usage: obj._process_update(updates) without return types