파이썬 - asyncio 프로토콜/서버 간 통신

Question

텔넷으로 연결할 수있는 서버 사이드 이벤트 서버를 작성하고 텔넷 컨텐츠를 브라우저에 푸시하려고합니다. 파이썬과 asyncio를 사용하는 배후의 아이디어는 가능한 한 작은 CPU를 사용하여 이것이 Raspberry Pi에서 실행될 수 있도록하는 것입니다.

지금까지 나는 여기에 발견 된 라이브러리를 사용하고 있습니다 : https://pypi.python.org/pypi/asyncio-sse/0.1 asyncio를 사용합니다.

그리고 asyncio도 사용하는 텔넷 서버를 복사했습니다.

둘 다 별도로 작동하지만 두 가지를 서로 묶는 방법을 모릅니다. 내가 이해할 때 SSEHandler 클래스의 send()을 Telnet.data_received 내부에서 호출해야하지만 액세스 방법을 모르겠습니다. 이 두 '서버'는 새로운 연결을 수락하거나 데이터를 밀어 넣기 위해 루프로 실행되어야합니다.

누구든지 도와 주거나 다른 방향으로 나를 가리킬 수 있습니까? 나는 더 이상 무엇을 해야할지 모르겠다. 무덤에있다. 무엇이 더 이상 google인지 모른다.

import asyncio 
import sse 

# Get an instance of the asyncio event loop 
loop = asyncio.get_event_loop() 

# Setup SSE address and port 
sse_host, sse_port = '192.168.2.25', 8888 

class Telnet(asyncio.Protocol): 
    def connection_made(self, transport): 
     print("Connection received!"); 
     self.transport = transport 

    def data_received(self, data): 
     print(data) 
     self.transport.write(b'echo:') 
     self.transport.write(data) 

     # This is where I want to send data via SSE 
     # SSEHandler.send(data) 

     # Things I've tried :(
     #loop.call_soon_threadsafe(SSEHandler.handle_request()); 
     #loop.call_soon_threadsafe(sse_server.send("PAH!")); 

    def connection_lost(self, esc): 
     print("Connection lost!") 
     telnet_server.close() 

class SSEHandler(sse.Handler): 
    @asyncio.coroutine 
    def handle_request(self): 
     self.send('Working') 

# SSE server 
sse_server = sse.serve(SSEHandler, sse_host, sse_port) 

# Telnet server 
telnet_server = loop.run_until_complete(loop.create_server(Telnet, '192.168.2.25', 7777)) 

#telnet_server.something = sse_server; 

loop.run_until_complete(sse_server) 
loop.run_until_complete(telnet_server.wait_closed()) 

Answer 1

서버 쪽 이벤트는 일종의 http 프로토콜이다. 주어진 순간에 여러 개의 동시 HTTP 요청이있을 수 있습니다. 아무도 연결되어 있지 않으면 0이 될 수 있습니다. 이 뉘앙스는 모두 두 개의 sse.serve 및 sse.Handler 구조로 묶여 있습니다. 전자는 하나의 청취 포트를 나타내며 각각의 개별 클라이언트 요청을 후자 포트로 발송합니다.

또한 sse.Handler.handle_request()은 각 클라이언트마다 한 번 호출되며 해당 공동 루틴이 종료되면 클라이언트 연결이 끊어집니다. 코드에서 해당 코 루틴은 즉시 종료되므로 클라이언트는 단일 "작업 중"이벤트를 보게됩니다. 그래서, 우리는 영원히 더 많이 또는 더 기다려야합니다. yield from을 입력하면 asyncio.Future()이됩니다.

두 번째 문제는 어떻게 든 SSEHandler()의 별도 인스턴스를 모두 보유해야하며 어떻게 든 각각 send() 메서드를 사용해야한다는 것입니다. 자, 우리는 각각 handle_request() 방법으로 각자 등록 할 수 있습니다; 개별 처리기 인스턴스를 대기중인 미래에 맵핑하는 dict에 각각을 추가합니다. 이제

class SSEHandler(sse.Handler): 
    _instances = {} 

    @asyncio.coroutine 
    def handle_request(self): 
     self.send('Working') 
     my_future = asyncio.Future() 
     SSEHandler._instances[self] = my_future 
     yield from my_future 

, 우리는 단지 우리가 만든 DICT에 등록 된 SSEHandler의 모든 인스턴스를 방문 듣고 하나 하나에 send()를 사용하여 모든에 이벤트를 보낼 수 있습니다.

class SSEHandler(sse.Handler): 

    #... 

    @classmethod 
    def broadcast(cls, message): 
     for instance, future in cls._instances.items(): 
      instance.send(message) 

class Telnet(asyncio.Protocol): 

    #... 

    def data_received(self, data): 
     #... 
     SSEHandler.broadcast(data.decode('ascii')) 

마지막으로 텔넷 연결이 닫히면 코드가 종료됩니다. 괜찮아요.하지만 그때도 정리해야합니다. 다행히도, 그 여기에 내 그림이 명확하지 않다 경우 전체, 작업 덤프의 핸들러

class SSEHandler(sse.Handler): 

    #... 

    @classmethod 
    def abort(cls): 
     for instance, future in cls._instances.items(): 
      future.set_result(None) 
     cls._instances = {} 

class Telnet(asyncio.Protocol): 

    #... 

    def connection_lost(self, esc): 
     print("Connection lost!") 
     SSEHandler.abort() 
     telnet_server.close() 

---

의 모든 선물의 모든에 결과를 설정의 문제입니다.

import asyncio 
import sse 

loop = asyncio.get_event_loop() 
sse_host, sse_port = '0.0.0.0', 8888 

class Telnet(asyncio.Protocol): 
    def connection_made(self, transport): 
     print("Connection received!"); 
     self.transport = transport 

    def data_received(self, data): 
     SSEHandler.broadcast(data.decode('ascii')) 

    def connection_lost(self, esc): 
     print("Connection lost!") 
     SSEHandler.abort() 
     telnet_server.close() 

class SSEHandler(sse.Handler): 
    _instances = {} 
    @classmethod 
    def broadcast(cls, message): 
     for instance, future in cls._instances.items(): 
      instance.send(message) 

    @classmethod 
    def abort(cls): 
     for instance, future in cls._instances.items(): 
      future.set_result(None) 
     cls._instances = {} 

    @asyncio.coroutine 
    def handle_request(self): 
     self.send('Working') 
     my_future = asyncio.Future() 
     SSEHandler._instances[self] = my_future 
     yield from my_future 

sse_server = sse.serve(SSEHandler, sse_host, sse_port) 
telnet_server = loop.run_until_complete(loop.create_server(Telnet, '0.0.0.0', 7777)) 
loop.run_until_complete(sse_server) 
loop.run_until_complete(telnet_server.wait_closed())