我正在学习 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>
请您参考如下方法:
可能有更好的方法让它与 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)
