How To Add A Filetransport Cleanly To Asyncio?

I'm writing an application which reads text data and acts on it. The text data could come from a TCP port, or from a text file (which contains data earlier read from the TCP port a

Solution 1:

According to the documentation of APIcreate_connection(), it takes a protocol and creates a streaming transport, which is a TCP connection. So it is not supposed to be an API for custom transports.

However, the idea to reuse the same protocol for either TCP transports or custom file transports is valid. It won't be a "completely different implementation", but at least not using create_connection(). Let's assume it is read_file():

def my_protocol_factory():
    return your_protocol

if using_tcp_port:
    transport, protocol = await loop.create_connection(my_protocol_factory, host, port)
else:
    transport, protocol = await read_file(loop, my_protocol_factory, path_to_file)

Then you would have something like this:

from asyncio import transports

import aiofiles  # https://github.com/Tinche/aiofilesdefread_file(loop, protocol_factory, path):
    protocol = protocol_factory()
    transport = FileTransport(path, loop)
    transport.set_protocol(protocol)
    return transport, protocol


classFileTransport(transports.ReadTransport):
    def__init__(self, path, loop):
        super().__init__()
        self._path = path
        self._loop = loop
        self._closing = Falsedefis_closing(self):
        return self._closing

    defclose(self):
        self._closing = Truedefset_protocol(self, protocol):
        self._protocol = protocol
        self._loop.create_task(self._do_read())

    defget_protocol(self):
        return self._protocol

    asyncdef_do_read(self):
        try:
            asyncwith aiofiles.open(self._path) as f:
                self._loop.call_soon(self._protocol.connection_made, self)
                asyncfor line in f:
                    self._loop.call_soon(self._protocol.data_received, line)
                    if self._closing:
                        break
                self._loop.call_soon(self._protocol.eof_received)
        except Exception as ex:
            self._loop.call_soon(self._protocol.connection_lost, ex)
        else:
            self._loop.call_soon(self._protocol.connection_lost, None)