code: async def read_frame(self) -> DataFrame:
          
          if self._data_frames.qsize() == 0 and self.closed:
              raise StreamConsumedError(self.id)
          frame = await self._data_frames.get()
          self._data_frames.task_done()
          if frame is None:
              raise StreamConsumedError(self.id)
          return frame
docs: Read a single frame from the local buffer.
  
          If no frames are available but the stream is still open, waits until
          more frames arrive. Otherwise, raises StreamConsumedError.
  
          When a stream is closed, a single `None` is added to the data frame
          Queue to wake up any waiting `read_frame` coroutines.