async def METHOD_NAME(self) -> AsyncGenerator[_Rowbatch, None]:
    if not self._subscribed:
        raise ValueError("Table '{}' not subscribed.".format(self.name))
    # Get the batch from somewhere, continue yielding until done.
    while True:
        rb: _Rowbatch = await self._rowbatch_q.get()
        # rb.close_table is how we communicate whether a connection has been closed
        # for a particular cluster. If this happens before we receive eos from each stream,
        # we throw an error.
        if rb.close_table:
            raise ValueError("Closed before receiving end-of-stream.")
        yield rb
        self._rowbatch_q.task_done()
        if rb.batch.eos:
            break