code: def _app_logs_reader(
      logs_api_client: _LightningLogsSocketAPI,
      project_id: str,
      app_id: str,
      component_names: List[str],
      follow: bool,
      on_error_callback: Optional[Callable] = None,
  ) -> Iterator[_LogEvent]:
      read_queue = queue.PriorityQueue()
  
      # We will use a socket per component
      log_sockets = [
          logs_api_client.create_lightning_logs_socket(
              project_id=project_id,
              app_id=app_id,
              component=component_name,
              on_message_callback=_push_log_events_to_read_queue_callback(
                  component_name, read_queue
              ),
              on_error_callback=on_error_callback or _error_callback,
          )
          for component_name in component_names
      ]
  
      # And each socket on separate thread pushing log event to print queue
      #   run_forever() will run until we close() the connection from outside
      log_threads = [
          Thread(target=work.run_forever, daemon=True)
          for work in log_sockets
      ]
  
      # Establish connection and begin pushing logs to the print queue
      for th in log_threads:
          th.start()
  
      # Print logs from queue when log event is available
      flow = "Your app has started."
      work = "USER_RUN_WORK"
      start_timestamps = {}
  
      # Print logs from queue when log event is available
      try:
          while True:
              log_event: _LogEvent = read_queue.get(
                  timeout=None if follow else 1.0
              )
  
              token = (
                  flow if log_event.component_name == "flow" else work
              )
              if token in log_event.message:
                  start_timestamps[
                      log_event.component_name
                  ] = log_event.timestamp
  
              timestamp = start_timestamps.get(
                  log_event.component_name, None
              )
              if (
                  timestamp
                  and log_event.timestamp >= timestamp
                  and "launcher" not in log_event.message
              ):
                  yield log_event
  
      except queue.Empty:
          # Empty is raised by queue.get if timeout is reached. Follow = False case.
          pass
  
      except KeyboardInterrupt:
          # User pressed CTRL+C to exit, we should respect that
          pass
  
      finally:
          # Close connections - it will cause run_forever() to finish -> thread as finishes aswell
          for socket in log_sockets:
              socket.close()
  
          # Because all socket were closed, we can just wait for threads to finish.
          for th in log_threads:
              th.join()
  
file_path: pytorch-lightning/src/lightning/app/utilities/app_logs.py
imports: import json
  import queue
  from dataclasses import dataclass
  from threading import Thread
  from typing import Callable, Iterator, List, Optional
  import dateutil.parser
  from websocket import WebSocketApp
  from lightning.app.utilities.log_helpers import _error_callback, _OrderedLogEntry
  from lightning.app.utilities.logs_socket_api import _LightningLogsSocketAPI
label_desc: Use image node_id 2 for calling a global function with example usage: _app_logs_reader(logs_api_client, project_id, app_id, component_names, follow, on_error_callback) without return types