asyncio에 대한 사용자 지정 전송을 만드는 방법은 무엇입니까?

Question

나는 Protocol 및 Transport 클래스에 대한 설명서를 asyncio 패키지로 읽고 있습니다. 구체적으로 :

프로토콜 클래스를 서브 클래 싱하는 경우 특정 방법을 재정의하는 것이 좋습니다. 이러한 메소드는 콜백입니다. 특정 이벤트에서 전송에 의해 호출됩니다 (예 : 일부 데이터가 수신되는 경우). 전송을 구현하지 않는 한 직접 전화하지 마십시오 ().

는 강조 원칙적으로는 ... 전송을 구현하는 것이 가능해야하지만,

그래서을 추가

수송선 추상 다양한하기 위해 asyncio에서 제공하는 클래스입니다 통신 채널의 종류. 일반적으로 운송 수단을 직접 인스턴스화하지 않습니다. 대신 AbstractEventLoop 메서드(어느 하나?)을 호출하여 전송을 만들고 기본 통신 채널을 시작하고 성공할 때 다시 전화를 겁니다.

다시 한번 강조 함께 나는 사용자 정의 전송을 만드는 어떤 방법이 표시되지 않습니다에 걸쳐 AbstractEventLoop 섹션을 읽기

을 추가했다. 가장 가까운 주소는 AbstractEventLoop.create_connection(protocol_factory, host=...)입니다. 소켓의 종류가 무엇인지 알 수 있습니다.

글쎄, 내 궁극적 인 목표는 어떤 종류의 소켓도 아닌 사용자 정의 전송을 사용하는 것입니다 (아마도 StringIO, 아마도 데이터베이스 연결 커서, 아마도 다른 프로토콜).

---

그래서 문서에서이 단지 잘 것이 아니라 구현 결코 실수, 또는 asyncio을 원숭이 - 패치와 장자를 희생하지 않고 사용자 정의 전송을 구현하는 방법은 실제로 존재? 너무 많이하지 비록

Answer 1

보일러 플레이트 코드의 비트가 필요하다 :

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