tractor/examples/asynchronous_generators.py

from itertools import repeat
from typing import AsyncIterator

import trio
import tractor


async def stream_forever() -> AsyncIterator[str]:
    '''
    Stream the same message indefinitely.

    '''
    message: str
    for message in repeat(
        'I can see these little future bubble things',
    ):
        # each yielded value is sent over the ``Channel`` to the
        # parent actor
        yield message
        await trio.sleep(0.01)


async def main() -> None:
    '''
    Print messages streamed from a subactor.

    '''
    an: tractor.ActorNursery
    async with tractor.open_nursery() as an:

        portal: tractor.Portal = await an.start_actor(
            'donny',
            enable_modules=[__name__],
        )

        # this async for loop streams values from the above
        # async generator running in a separate process
        stream: tractor.MsgStream
        async with portal.open_stream_from(stream_forever) as stream:
            count: int = 0
            message: str
            async for message in stream:
                print(message)
                count += 1

                if count > 50:
                    break

        print('stream terminated')

        await portal.cancel_actor()


if __name__ == '__main__':
    trio.run(main)