class_docstr: This object collects chunks from async get requests via __call__, and
  calls the underlying callback when the object is fully received, or if an
  exception while retrieving the object occurs.
  
  This is not used in synchronous gets (synchronous gets interact with the
  raylet servicer directly, not through the datapath).
  
  __call__ returns true once the underlying call back has been called.
class_docstr_tok: ["This","object","collects","chunks","from","async","get","requests","via","__call__",",","and","calls","the","underlying","callback","when","the","object","is","fully","received",",","or","if","an","exception","while","retrieving","the","object","occurs",".","This","is","not","used","in","synchronous","gets","(","synchronous","gets","interact","with","the","raylet","servicer","directly",",","not","through","the","datapath",")",".","__call__","returns","true","once","the","underlying","call","back","has","been","called","."]
code: def __call__(
      self, response: Union[ray_client_pb2.DataResponse, Exception]
  ) -> bool:
      if isinstance(response, Exception):
          self.callback(response)
          return True
      get_resp = response.get
      if not get_resp.valid:
          self.callback(response)
          return True
      if (
          get_resp.total_size > OBJECT_TRANSFER_WARNING_SIZE
          and log_once("client_object_transfer_size_warning")
      ):
          size_gb = get_resp.total_size / 2**30
          warnings.warn(
              "Ray Client is attempting to retrieve a "
              f"{size_gb:.2f} GiB object over the network, which may "
              "be slow. Consider serializing the object to a file and "
              "using rsync or S3 instead.",
              UserWarning,
          )
      chunk_data = get_resp.data
      chunk_id = get_resp.chunk_id
      if chunk_id == self.last_seen_chunk + 1:
          self.data.extend(chunk_data)
          self.last_seen_chunk = chunk_id
          # If we disconnect partway through, restart the get request
          # at the first chunk we haven't seen
          self.request.get.start_chunk_id = self.last_seen_chunk + 1
      elif chunk_id > self.last_seen_chunk + 1:
          # A chunk was skipped. This shouldn't happen in practice since
          # grpc guarantees that chunks will arrive in order.
          msg = (
              f"Received chunk {chunk_id} when we expected "
              f"{self.last_seen_chunk + 1} for request {response.req_id}"
          )
          logger.warning(msg)
          self.callback(RuntimeError(msg))
          return True
      else:
          # We received a chunk that've already seen before. Ignore, since
          # it should already be appended to self.data.
          logger.debug(
              f"Received a repeated chunk {chunk_id} "
              f"from request {response.req_id}."
          )
  
      if get_resp.chunk_id == get_resp.total_chunks - 1:
          self.callback(self.data)
          return True
      else:
          # Not done yet
          return False
  
imports: import math
  import logging
  import queue
  import threading
  import warnings
  import grpc
  from collections import OrderedDict
  from typing import Any, Callable, Dict, TYPE_CHECKING, Optional, Union
  import ray.core.generated.ray_client_pb2
  import ray.core.generated.ray_client_pb2_grpc
  from ray.util.client.common import INT32_MAX, OBJECT_TRANSFER_CHUNK_SIZE, OBJECT_TRANSFER_WARNING_SIZE
  from ray.util.debug import log_once
label_desc: Use image node_id 2 for calling the ChunkCollector obj's underlying member method code with example usage: obj.__call__(response) and returns: True, True, True, False, True