Comment créer un transport personnalisé pour asyncio?

Question

Je lis la documentation pour les classes Protocol et Transport dans le paquet asyncio. Plus précisément:

Lors du sous-classement d'une classe de protocole, il est recommandé de remplacer certaines méthodes. Ces méthodes sont des rappels: elles seront appelées par le transport sur certains événements (par exemple lors de la réception de certaines données); vous ne devez pas les appeler vous-même, sauf si vous implémentez un transport.

Non souligné dans

Ainsi, en principe, il devrait être possible de mettre en place un transport, mais ...

Transports sont des classes fournies par asyncio afin de faire abstraction divers types de canaux de communication. En général, vous n'instanciez pas vous-même un transport. à la place, appelez une méthode AbstractEventLoop(laquelle?) qui créera le transport et tentera de lancer le canal de communication sous-jacent, vous rappelant quand il réussit.

Encore une fois, l'accent ajouté

La lecture de la section sur AbstractEventLoop le long et à travers, je ne vois aucun moyen de créer un transport personnalisé. Le plus proche, il est dans AbstractEventLoop.create_connection(protocol_factory, host=...) qui implique seulement qu'il va créer une sorte de socket ...

Eh bien, mon but ultime est d'utiliser un transport sur mesure qui n'est pas n'importe quel type de socket (peut-être un StringIO un curseur de connexion à la base de données, peut-être un autre protocole).

---

Alors, est-ce juste un bien intentionné mais jamais mis en œuvre erreurs dans les documents, ou est-il réellement un moyen de mettre en œuvre le transport personnalisé sans singe ragréage asyncio et sacrifier firstborns?

Answer 1

Un peu de code plaque de la chaudière est nécessaire, même si elle est pas trop:

from asyncio import streams, transports, get_event_loop 


class CustomTransport(transports.Transport): 

    def __init__(self, loop, protocol, *args, **kwargs): 
     self._loop = loop 
     self._protocol = protocol 

    def get_protocol(self): 
     return self._protocol 

    def set_protocol(self, protocol): 
     return self._protocol 

    # Implement read/write methods 

    # [...] 


async def custom_create_connection(protocol_factory, *args, **kwargs): 
    loop = get_event_loop() 
    protocol = protocol_factory() 
    transport = CustomTransport(loop, protocol, *args, **kwargs) 
    return transport, protocol 


async def custom_open_connection(*args, **kwargs): 
    reader = streams.StreamReader() 
    protocol = streams.StreamReaderProtocol(reader) 
    factory = lambda: protocol 
    transport, _ = await custom_create_connection(factory, *args, **kwargs) 
    writer = streams.StreamWriter(transport, protocol, reader) 
    return reader, writer