async def read_frame(self) -> DataFrame:
        """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.
        """
        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