2

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

在我的场景中，某些第三方将数据发布到 Web 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() 需要消息作为 arg，而我正试图用它来 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>
4

1 回答 1

2

可能有更好的方法让它与 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)
于 2018-02-05T02:03:48.880 回答