class_docstr: Task log reader.
class_docstr_tok: ["Task","log","reader","."]
docstr: Continuously read log to the end.
  
  :param ti: The Task Instance
  :param try_number: the task try number
  :param metadata: A dictionary containing information about how to read the task log
docstr_tok: ["Continuously","read","log","to","the","end",".",":","param","ti",":","The","Task","Instance",":","param","try_number",":","the","task","try","number",":","param","metadata",":","A","dictionary","containing","information","about","how","to","read","the","task","log"]
code: def read_log_stream(
      self, ti: TaskInstance, try_number: int | None, metadata: dict
  ) -> Iterator[str]:
      """
      Continuously read log to the end.
  
      :param ti: The Task Instance
      :param try_number: the task try number
      :param metadata: A dictionary containing information about how to read the task log
      """
      if try_number is None:
          next_try = ti.next_try_number
          try_numbers = list(range(1, next_try))
      else:
          try_numbers = [try_number]
      for current_try_number in try_numbers:
          metadata.pop("end_of_log", None)
          metadata.pop("max_offset", None)
          metadata.pop("offset", None)
          metadata.pop("log_pos", None)
          while True:
              logs, metadata = self.read_log_chunks(
                  ti, current_try_number, metadata
              )
              for host, log in logs[0]:
                  yield "\n".join([host or "", log]) + "\n"
              if "end_of_log" not in metadata or (
                  not metadata["end_of_log"]
                  and ti.state
                  not in (
                      TaskInstanceState.RUNNING,
                      TaskInstanceState.DEFERRED,
                  )
              ):
                  if not logs[0]:
                      # we did not receive any logs in this loop
                      # sleeping to conserve resources / limit requests on external services
                      time.sleep(self.STREAM_LOOP_SLEEP_SECONDS)
              else:
                  break
  
file_path: airflow/airflow/utils/log/log_reader.py
imports: from __future__ import annotations
  import logging
  import time
  from functools import cached_property
  from typing import TYPE_CHECKING, Iterator
  from airflow.configuration import conf
  from airflow.utils.helpers import render_log_filename
  from airflow.utils.log.logging_mixin import ExternalLoggingMixin
  from airflow.utils.session import NEW_SESSION, provide_session
  from airflow.utils.state import TaskInstanceState
label_desc: Use image node_id 2 for calling the TaskLogReader obj's underlying member method code with example usage: obj.read_log_stream(ti, try_number, metadata) without return types