如何使用autobahn asyncio实现交互式websocket客户端？

Question

我试图实现一个使用高速公路| python 和asyncio的websocket/wamp客户端，虽然它有点工作，但也有一些零件不知道 。

我真正想要做的是在qt5/QML中实现WAMP，但这个 看起来像是一个更容易的路径。

这个简化的客户端大部分都是从网上复制过来的。它在onJoin发生时读取 时间服务。

我想要做的就是触发这个从外部来源读取。

我采取的复杂方法是​​在 线程中运行asyncio事件循环，然后通过套接字发送命令来触发读取。 I 到目前为止还无法确定在哪里放置例程/协程，以便可以从读者例行程序中找到它。

我怀疑有一个更简单的方法去做这件事，但我还没有发现它 呢。欢迎提出建议。

#!/usr/bin/python3 
try: 
    import asyncio 
except ImportError: 
    ## Trollius >= 0.3 was renamed 
    import trollius as asyncio 

from autobahn.asyncio import wamp, websocket 
import threading 
import time 
from socket import socketpair 

rsock, wsock = socketpair() 

def reader() : 
    data = rsock.recv(100) 
    print("Received:", data.decode()) 

class MyFrontendComponent(wamp.ApplicationSession): 
    def onConnect(self): 
     self.join(u"realm1") 



    @asyncio.coroutine 
    def onJoin(self, details): 
     print('joined') 
     ## call a remote procedure 
     ## 
     try: 
      now = yield from self.call(u'com.timeservice.now') 
     except Exception as e: 
      print("Error: {}".format(e)) 
     else: 
      print("Current time from time service: {}".format(now)) 



    def onLeave(self, details): 
     self.disconnect() 

    def onDisconnect(self): 
     asyncio.get_event_loop().stop() 



def start_aloop() : 
    loop = asyncio.new_event_loop() 
    asyncio.set_event_loop(loop) 
    transport_factory = websocket.WampWebSocketClientFactory(session_factory, 
        debug = False, 
        debug_wamp = False) 
    coro = loop.create_connection(transport_factory, '127.0.0.1', 8080) 
    loop.add_reader(rsock,reader) 
    loop.run_until_complete(coro) 
    loop.run_forever() 
    loop.close() 

if __name__ == '__main__': 
    session_factory = wamp.ApplicationSessionFactory() 
    session_factory.session = MyFrontendComponent 

    ## 4) now enter the asyncio event loop 
    print('starting thread') 
    thread = threading.Thread(target=start_aloop) 
    thread.start() 
    time.sleep(5) 
    print("IN MAIN") 
    # emulate an outside call 
    wsock.send(b'a byte string') 

Answer 1

您可以监听事件循环内的插座异步，使用loop.sock_accept。你可以只调用一个协同程序设置的onConnect或onJoin插座内：

try: 
    import asyncio 
except ImportError: 
    ## Trollius >= 0.3 was renamed 
    import trollius as asyncio 

from autobahn.asyncio import wamp, websocket 
import socket 

class MyFrontendComponent(wamp.ApplicationSession): 
    def onConnect(self): 
     self.join(u"realm1") 

    @asyncio.coroutine 
    def setup_socket(self): 
     # Create a non-blocking socket 
     self.sock = socket.socket() 
     self.sock.setblocking(0) 
     self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) 
     self.sock.bind(('localhost', 8889)) 
     self.sock.listen(5) 
     loop = asyncio.get_event_loop() 
     # Wait for connections to come in. When one arrives, 
     # call the time service and disconnect immediately. 
     while True: 
      conn, address = yield from loop.sock_accept(self.sock) 
      yield from self.call_timeservice() 
      conn.close() 

    @asyncio.coroutine 
    def onJoin(self, details): 
     print('joined') 
     # Setup our socket server 
     asyncio.async(self.setup_socket()) 

     ## call a remote procedure 
     ## 
     yield from self.call_timeservice() 

    @asyncio.coroutine 
    def call_timeservice(self): 
     try: 
      now = yield from self.call(u'com.timeservice.now') 
     except Exception as e: 
      print("Error: {}".format(e)) 
     else: 
      print("Current time from time service: {}".format(now)) 

    ... # The rest is the same 

Answer 2

感谢您的答复端午。不完全是我需要的解决方案，但它指出了我正确的方向。是的，我希望客户端可以通过外部触发器远程调用RPC。

我来到了，让我通过为特定调用的字符串（虽然只有一个实现现在）

这就是我想出了以下，虽然我不知道如何优雅它是。

import asyncio 
from autobahn.asyncio import wamp, websocket 
import threading 
import time 
import socket 


rsock, wsock = socket.socketpair() 

class MyFrontendComponent(wamp.ApplicationSession): 
    def onConnect(self): 
     self.join(u"realm1") 

    @asyncio.coroutine 
    def setup_socket(self): 
     # Create a non-blocking socket 
     self.sock = rsock 
     self.sock.setblocking(0) 
     loop = asyncio.get_event_loop() 
     # Wait for connections to come in. When one arrives, 
     # call the time service and disconnect immediately. 
     while True: 
      rcmd = yield from loop.sock_recv(rsock,80) 
      yield from self.call_service(rcmd.decode()) 

    @asyncio.coroutine 
    def onJoin(self, details): 
     # Setup our socket server 
     asyncio.async(self.setup_socket()) 


    @asyncio.coroutine 
    def call_service(self,rcmd): 
     print(rcmd) 
     try: 
      now = yield from self.call(rcmd) 
     except Exception as e: 
      print("Error: {}".format(e)) 
     else: 
      print("Current time from time service: {}".format(now)) 



    def onLeave(self, details): 
     self.disconnect() 

    def onDisconnect(self): 
     asyncio.get_event_loop().stop() 



def start_aloop() : 
    loop = asyncio.new_event_loop() 
    asyncio.set_event_loop(loop) 
    transport_factory = websocket.WampWebSocketClientFactory(session_factory, 
        debug = False, 
        debug_wamp = False) 
    coro = loop.create_connection(transport_factory, '127.0.0.1', 8080) 
    loop.run_until_complete(coro) 
    loop.run_forever() 
    loop.close() 

if __name__ == '__main__': 
    session_factory = wamp.ApplicationSessionFactory() 
    session_factory.session = MyFrontendComponent 

    ## 4) now enter the asyncio event loop 
    print('starting thread') 
    thread = threading.Thread(target=start_aloop) 
    thread.start() 
    time.sleep(5) 
    wsock.send(b'com.timeservice.now') 
    time.sleep(5) 
    wsock.send(b'com.timeservice.now') 
    time.sleep(5) 
    wsock.send(b'com.timeservice.now')