最近Pythonの非同期処理、asyncioを使ったプログラムを書いています。
今までは非同期だとNode.jsを使っていたんですが、aiohttpや周辺ライブラリーが揃ってきたようなので使っています。
Node.jsの場合、Streamは stream.pipe(other_stream) のようにPipeできるのですが、それをaiohttpの WebSocketResponse でやる方法を書きます。
Server Reference — aiohttp 3.4.4 documentation
asyncioの Streams でも要領は同じなので参考になると思います。
（注意: Pythonのasyncioやaiohttpはまだ新しいものなので、情報が古くなってないか気をつけてください）
Pipeを1回やる（一方通行の）例
1つのストリームから読み込んで、もう片方のストリームに書き込むのは簡単です。
async for msg in ws:
await to_ws.send_str(msg.data)
WebSocketでアクセスを受け付けて、バックエンドの別のWebSocketにつなぎ込むaiohttpのView関数を考えると、以下のようになります。
from aiohttp import web
routes = web.RouteTableDef()
@routes.post('...')
async def websocket_gateway(request):
ws = web.WebSocketResponse()
await ws.prepare(request)
session = aiohttp.ClientSession()
to_ws = await session.ws_connect('...')
async for msg in ws:
await to_ws.send_str(msg.data)
await ws.close()
await to_ws.close()
この場合、サーバー側のWebSocket ws がクローズすると処理が終了します。
他にも書き込み先のWebSocket to_ws が先にクローズした場合の処理や、 msg.data がバイト列の場合に to_ws.send_bytes にする処理なども必要そうです。
相互にPipeする例
上記の例ではWebSocketは一方通行にしか仲介されていません。ここで、 to_ws からの入力も、 ws に返すようにしましょう。
その場合、 2つの非同期処理を同時に実行する必要があります 。
1つの while ループの中で2つのストリームの読み書きをしても、片方が書き込まないと、もう片方の入力が受け取れないようになるので難しいです。
以下のように別々のループをして、それを asyncio.gather するとうまくいきます。
import asyncio
from aiohttp import web
from aiohttp.http import WSMsgType
WS_CLOSE_TYPES = (WSMsgType.CLOSE, WSMsgType.CLOSING, WSMsgType.CLOSED)
routes = web.RouteTableDef()
async def pipe(f, t):
while not f.closed and not f.closed:
msg = await f.receive()
if msg.type in WS_CLOSE_TYPES:
break
await t.send_str(msg.data)
return
@routes.post('...')
async def websocket_gateway(request):
ws = web.WebSocketResponse()
await ws.prepare(request)
session = aiohttp.ClientSession()
to_ws = await session.ws_connect('...')
await asyncio.gather(
pipe(ws, cs),
pipe(cs, ws)
)
await ws.close()
await to_ws.close()
asyncio.gather を使うことで、2つの処理を同時に実行して、両方が終了するのを待ってくれます。
優秀なやつですが、今回は片方のWebSocketが終わったら両方終了してほしいので、 pipe 関数内に工夫をしています。
pipe 関数の中で async for msg in ws せずに while ループを使っているのは、両方のWebSocketがクローズしていない限り処理を続けるようにするためです。
そうしないと、片方のWebSocketがクローズしてるのに、反対のWebSocketから書き込まれようとしてエラーになるからです。
こんな感じで、 asyncio や aiohttp はかなり刺激的で面白いのでぜひ試してみてください。
誰得情報かもしれませんが、 aiohttp や asyncio についての情報を出していけたらなと思います。
