class_docstr: Pushes chunks of files to our streaming endpoint.
  
  This class is used as a singleton. It has a thread that serializes access to
  the streaming endpoint and performs rate-limiting and batching.
  
  TODO: Differentiate between binary/text encoding.
code: def _thread_body(self) -> None:
      posted_data_time = time.time()
      posted_anything_time = time.time()
      ready_chunks = []
      uploaded: Set[str] = set()
      finished: Optional[FileStreamApi.Finish] = None
      while finished is None:
          items = self._read_queue()
          for item in items:
              if isinstance(item, self.Finish):
                  finished = item
              elif isinstance(item, self.Preempting):
                  request_with_retry(
                      self._client.post,
                      self._endpoint,
                      json={
                          "complete": False,
                          "preempting": True,
                          "dropped": self._dropped_chunks,
                          "uploaded": list(uploaded),
                      },
                  )
                  uploaded = set()
              elif isinstance(item, self.PushSuccess):
                  uploaded.add(item.save_name)
              else:
                  # item is Chunk
                  ready_chunks.append(item)
  
          cur_time = time.time()
  
          if ready_chunks and (
              finished
              or cur_time - posted_data_time > self.rate_limit_seconds()
          ):
              posted_data_time = cur_time
              posted_anything_time = cur_time
              success = self._send(ready_chunks, uploaded=uploaded)
              ready_chunks = []
              if success:
                  uploaded = set()
  
          # If there aren't ready chunks or uploaded files, we still want to
          # send regular heartbeats so the backend doesn't erroneously mark this
          # run as crashed.
          if cur_time - posted_anything_time > self.heartbeat_seconds:
              posted_anything_time = cur_time
  
              # If we encountered an error trying to publish the
              # list of uploaded files, don't reset the `uploaded`
              # list. Retry publishing the list on the next attempt.
              if not isinstance(
                  request_with_retry(
                      self._client.post,
                      self._endpoint,
                      json={
                          "complete": False,
                          "failed": False,
                          "dropped": self._dropped_chunks,
                          "uploaded": list(uploaded),
                      },
                  ),
                  Exception,
              ):
                  uploaded = set()
  
      # post the final close message. (item is self.Finish instance now)
      request_with_retry(
          self._client.post,
          self._endpoint,
          json={
              "complete": True,
              "exitcode": int(finished.exitcode),
              "dropped": self._dropped_chunks,
              "uploaded": list(uploaded),
          },
      )
  
imports: import base64
  import functools
  import itertools
  import logging
  import os
  import queue
  import random
  import sys
  import threading
  import time
  from types import TracebackType
  from typing import TYPE_CHECKING, Any, Callable, Dict, List, NamedTuple, Optional, Set, Tuple, Type, Union
  import requests
  import wandb
  from wandb import util
  from wandb.sdk.internal import internal_api
  from ..lib import file_stream_utils
label_desc: Use image node_id 9 for calling the FileStreamApi obj's underlying member method code with example usage: obj._thread_body() without return types