How to convert callback-based async function to async generator

Question

It's currently possible to start asyncio servers by providing a callback that would fire on each incoming connection:

async def on_connection(reader, writer):
    # this is invoked each time a new connection is made
    pass

server = await asyncio.start_server(on_connection, host, port)

I would like to get rid of callback and use async for instead, so it would look like this:

async for reader, writer in my_start_server(host, port):
    # this should be invoked each time a new connection is made
    pass

Unfortunately, it does not seem to be very easy:

async def my_start_server(host, port):
    async def on_connection(reader, writer):
        # here I have to somehow yield (reader, writer) tuple
        # but if I do just `yield (reader, writer)`, it would 
        # make `on_connection` itself a generator, but would 
        # not affect `my_start_server`
        pass

    server = await asyncio.start_server(on_connection, host, port)

I have thought about having a class with __aiter__ implementation there, but result seems to be overcomplicated a lot. So, is this the only way to go, or am I missing any easy approaches on converting async callbacks to async generators?

Usually, you want to be able to handle multiple clients at the same time. The async for approach doesn't allow that. — Vincent
– Vincent, Commented Apr 30, 2017 at 11:59
@Vincent, hm, you seem to be right - async for won't fetch next connection from the server until you have completed processing current one. I wish there was async for with "consume as soon as possible" semantics. — toriningen
– toriningen, Commented Apr 30, 2017 at 15:15

Mikhail Gerasimov · Accepted Answer · 2017-04-30 18:15:06Z

As Vincent noticed, you shouldn't use async for to fetch connections since code inside it would block other connections from handling. Consider each connection as separate task that should be run regardless of others.

Here's example of how you can yield inside my_start_server using queue which also shows that we still will return to some sort of on_connection:

async def my_start_server(host, port):
    queue = asyncio.Queue()

    async def put(reader, writer):
        await queue.put((reader, writer,))

    await asyncio.start_server(put, host, port)

    while True:
        yield (await queue.get())

.

async for (reader, writer) in my_start_server(host, port):

    # If we will just handle reader/writer here, others will be suspended from handling.
    # We can avoid it starting some task,
    # but in this case 'handle' would be nothing different from 'on_connection'

    asyncio.Task(handle(reader, writer))

Collectives™ on Stack Overflow

How to convert callback-based async function to async generator

1 Answer 1

Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related