如何 websocket.send() 来自异步函数的数据

Question

我正在学习 asyncio 并试图弄清楚如何将数据从一个异步函数传递到另一个异步函数中的 websocket 循环。

在我的场景中，数据由第三方POST编辑到网络API。我想将 POST 数据回显给连接的 websocket 客户端。

Sanic 和 Rx 不是必需的，但这是我开始的路径。以下是我到目前为止的想法：

#!/usr/bin/env python

from sanic import Sanic
from sanic import response
from sanic.response import file
from rx import Observable

app = Sanic(__name__)

@app.route('/')
async def index(request):
    return await file('ws.html')

async def observable_message(message):
    return Observable.from_(message)

@app.route('/post', methods=["POST"])
async def message_inbound(request):
    payload = request.json
    await observable_message(payload)
    return response.json({"status": "OK"})

@app.websocket('/feed')
async def feed(request, ws):
    while True:
        message = await app.stream
        print('Sending: ' + message)
        await ws.send(message)

@app.listener('before_server_start')
async def setup_observable_stream(app, loop):
    app.stream = Observable.start_async(observable_message)

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=9000, workers=4, debug=True)

这显然行不通，因为 observable_message() 需要消息作为参数，而我正试图将它用于 start_async()，所以我很困惑。我怎样才能连接这些东西？

客户端可以很简单：

<!DOCTYPE html>
<html>
<head><title>POST data</title></head>
<body>
<script>
    var ws = new WebSocket('ws://' + document.domain + ':' + location.port + '/feed'),
        messages = document.createElement('ul');
    ws.onmessage = function (event) {
        var messages = document.getElementsByTagName('ul')[0],
            message = document.createElement('li'),
            content = document.createTextNode('Received: ' + event.data);
        message.appendChild(content);
        messages.appendChild(message);
    };
    document.body.appendChild(messages);
</script>
</body>
</html>

Answer 1

可能有更好的方法让它与 Rx 一起工作，但我发现简化有助于通过它推理。这有效：

#!/usr/bin/env python

import asyncio
import uvloop

from sanic import Sanic
from sanic import response
from sanic.response import file
from sanic.websocket import ConnectionClosed

app = Sanic(__name__)

app.ws_clients = set()
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

@app.route('/')
async def index(request):
    return await file('ws.html')

@app.route('/post', methods=["POST"])
async def message_inbound(request):
    payload = request.body.decode("utf-8")
    await broadcast(payload)
    return response.json({"status": "OK"})

async def broadcast(message):
    broadcasts = [ws.send(message) for ws in app.ws_clients]
    for result in asyncio.as_completed(broadcasts):
        try:
            await result
        except ConnectionClosed:
            print("ConnectionClosed")
        except Exception as ex:
            template = "An exception of type {0} occurred. Arguments:\n{1!r}"
            message = template.format(type(ex).__name__, ex.args)
            print(message)

@app.websocket("/ws")
async def websocket(request, ws):
    app.ws_clients.add(ws)
    await ws.send("Connected.")
    print(f'{len(app.ws_clients)} clients')
    while True:
        data = await ws.recv()
        print('Received: ' + data)

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=9005, workers=1, debug=False)

如何 websocket.send() 来自异步函数的数据

How to websocket.send() data from an async function

python

websocket

rx-py

sanic