Method agent

tests/unit/test_streams.py:179–190

(stream)

        @app.agent(channel)
        async def agent(stream):
            with ExitStack() as stack:
                if raises:
                    stack.enter_context(pytest.raises(raises))
                async for value in stream:
                    if value is sentinel:
                        got_sentinel.set()
                        break
                    assert value is event.value
                    if side_effect:
                        got_sentinel.set()
                        raise side_effect

        s: faust.StreamT = None
        async with agent as _agent: