Function stream_anext

couchbase/logic/streaming.py:45–75

**INTERNAL** Drive a single async streaming iteration for ``req`` and apply uniform teardown/error handling.

(req)

async def stream_anext(req):
    """
    **INTERNAL**

    Drive a single async streaming iteration for ``req`` and apply uniform teardown/error handling.
    """
    try:
        # this is a blocking operation, so it is offloaded to the request's executor
        row = await req._loop.run_in_executor(req._tp_executor, req._get_next_row)
        # We want to end the streaming op span once we have a response from the C++ core.
        # Unfortunately right now, that means we need to wait until we have the first row (or we
        # have an error).  As this is idempotent, it is safe to call for each row (it will only do
        # work for the first call).
        req._process_core_span()
        return row
    except StopAsyncIteration:
        req._done_streaming = True
        req._finalize()
        req._get_metadata()
        raise
    except CouchbaseException as ex:
        req._finalize(exc_val=ex)
        raise
    except Exception as ex:
        excptn = _internal_exception(str(ex))
        req._finalize(exc_val=excptn)
        raise excptn
    except BaseException as ex:
        # asyncio.CancelledError / KeyboardInterrupt / SystemExit -- see module note above.
        req._finalize(exc_val=ex)
        raise


def stream_next(req):