How to send data periodically with asyncio.Protoco

问题:

I have asyncio.Protocol subclass

class MyProtocol(Protocol):
    def __init__(self, exit_future):
        self.exit_future = exit_future

    def connection_made(self, transport):
        self.transport = transport

    def data_received(self, data):
        pass

    def eof_received(self):
        self.exit_future.set_result(True)

    def connection_lost(self, exc):
        self.exit_future.set_result(True)

and network connection created with

while True:
    try:
        exit_future = Future(loop=loop)
        transport, protocol = await loop.create_connection(lambda: MyProtocol(exit_future), host, port)

        await exit_future
        transport.close()
    except:
        pass

Now the question is: how can I send some data on some external event occurs? For instance when asyncio.Queue is not empty (queue.get will not block), what fills that queue is not related to asyncio? What is the most correct way to call transport.write when something happens?

回答1:

how can I send some data on some external event occurs?

The easiest way is to spawn the coroutine in connection_made and leave it to handle the event in the "background":

def connection_made(self, transport):
    self.transport = transport
    loop = asyncio.get_event_loop()
    self._interesting_events = asyncio.Queue()
    self.monitor = loop.create_task(self._monitor_impl())

def connection_lost(self, exc):
    self.exit_future.set_result(True)
    self.monitor.cancel()

async def _monitor_impl(self):
    while True:
        # this can also await asyncio.sleep() or whatever is needed
        event = await self._interesting_events.get()
        self.transport.write(...)

Note that in the long run it might be worth it to replace create_connection with open_connection and use the streams API from the ground up. That way you can use coroutines all the way without worrying about the callback/coroutine impedance mismatch.

On an unrelated note, try followed by except: pass is an anti-pattern - consider catching a specific exception instead, or at least logging the exception.