code: def _log_main(self) -> None:
      reconnecting = False
      while not self.client_worker._in_shutdown:
          if reconnecting:
              # Refresh queue and retry last request
              self.request_queue = queue.Queue()
              if self.last_req:
                  self.request_queue.put(self.last_req)
          stub = ray_client_pb2_grpc.RayletLogStreamerStub(
              self.client_worker.channel
          )
          try:
              log_stream = stub.Logstream(
                  iter(self.request_queue.get, None),
                  metadata=self._metadata,
              )
          except ValueError:
              # Trying to use the stub on a cancelled channel will raise
              # ValueError. This should only happen when the data client
              # is attempting to reset the connection -- sleep and try
              # again.
              time.sleep(0.5)
              continue
          try:
              for record in log_stream:
                  if record.level < 0:
                      self.stdstream(level=record.level, msg=record.msg)
                  self.log(level=record.level, msg=record.msg)
              return
          except grpc.RpcError as e:
              reconnecting = self._process_rpc_error(e)
              if not reconnecting:
                  return
  
file_path: ray/python/ray/util/client/logsclient.py
imports: import sys
  import logging
  import queue
  import threading
  import time
  import grpc
  from typing import TYPE_CHECKING
  import ray.core.generated.ray_client_pb2
  import ray.core.generated.ray_client_pb2_grpc
  from ray.util.debug import log_once
label_desc: Use image node_id 3 for calling the LogstreamClient obj's underlying member method code with example usage: obj._log_main() without return types