我正在使用以下代码来测试本地 websocket 服务器是否正在运行:
import asyncio
import websockets
async def hello():
async with websockets.connect('ws://127.0.0.1:8000/ws/registration/123') as websocket:
await websocket.send(json_data)
asyncio.get_event_loop().run_until_complete(hello())
有没有更简单的方法来做到这一点而不使用 asyncio?例如:
import asyncio
import websockets
conn = websockets.connect('ws://127.0.0.1:8000/ws/registration/123')
conn.send('hello')
基本上,我只是想找到最简单的方法来测试我的 websocket 服务器是否正在侦听和接收特定 url 的消息。
不会async_to_sync
让这更复杂吗?为什么不创建一个普通test_url
函数:
def test_url(url, data=""):
async def inner():
async with websockets.connect(url) as websocket:
await websocket.send(data)
return asyncio.get_event_loop().run_until_complete(inner())
test_url("ws://127.0.0.1:8000/ws/registration/123")
本文收集自互联网,转载请注明来源。
如有侵权,请联系 [email protected] 删除。
我来说两句