python - 当客户端不干净地退出时，如何检测服务器上 Python aiohttp web 套接字的关闭

Question

我有一个简单的命令和控制服务器 server.py(完全不安全 - 不要使用)，一个被动客户端 update_client.py 和另一个可以发送命令 update_commander.py。在 http://0.0.0.0:8080/ 处有一个 http 端点其中列出了连接的客户端。当 update_commander.py 脚本退出时，它的客户端会被正确清理。当 update_client.py 断开连接时，服务器没有注意到断开连接，并且在 update_commander.py 发送的进一步消息上，我收到错误 socket.send()引发异常。 对于每个幽灵客户端连接。清理代码标记为### CLEANUP CODE ###

我觉得我应该做的是在尝试发送到套接字但没有引发异常时捕获错误，只是向标准输出发送一条消息。

server.py

import uuid
import asyncio
import aiohttp
from aiohttp import web


class Client(object):

    def __init__(self):
        self.websocket = None
        self.name = None


class ClientList(web.View):
    async def get(self):

        clients = self.request.app['clients']

        client_list = [client.name for name, client in clients.items()]
        txt = ", ".join(client_list)
        return web.Response(text=txt)


class WebSocket(web.View):
    async def get(self):
        ws = web.WebSocketResponse()
        await ws.prepare(self.request)

        # session = await get_session(self.request)
        # user = User(self.request.db, {'id': session.get('user')})
        # login = await user.get_login()
        login = str(uuid.uuid4())
        client = Client()
        client.name = login
        client.websocket = ws
        self.request.app['clients'][client.name] = client
        print('%s connected.' % login)

        for _ws in [c.websocket for name, c in self.request.app['clients'].items()]:
            _ws.send_str('%s joined' % login)

        async for msg in ws:
            if msg.tp == aiohttp.WSMsgType.text:
                if msg.data == 'close':
                    await ws.close()
                else:
                    # do something here like save it
                    print('%s sent: %s' % (login, msg.data))
                    # Send message to all clients other clients
                    for _ws in [c.websocket for name, c in self.request.app['clients'].items()]:
                        try:
                            _ws.send_str('(%s) %s' % (login, msg.data))
                            asyncio.sleep(0)
                        except:
                            print(ws.exception())
            elif msg.tp == aiohttp.WSMsgType.error:
                print('ws connection closed with exception %s' % ws.exception())

        ### CLEANUP CODE ###
        await client.websocket.close()
        del self.request.app['clients'][client.name]
        for _ws in [c.websocket for name, c in self.request.app['clients'].items()]:
            _ws.send_str('%s disconected' % login)
        print('%s disconnected' % login)

        return ws


routes = [
    ('GET', '/',        ClientList,  'main'),
    ('GET', '/ws',      WebSocket,   'chat'),
]

app = web.Application()

for route in routes:
    app.router.add_route(route[0], route[1], route[2], name=route[3])
app['clients'] = {}
web.run_app(app)

指挥官.py

import asyncio
import aiohttp
import os

HOST = os.getenv('HOST', '0.0.0.0')
PORT = int(os.getenv('PORT', 8080))

URL = f'http://{HOST}:{PORT}/ws'


async def main():
    session = aiohttp.ClientSession()
    async with session.ws_connect(URL) as ws:

        await prompt_and_send(ws)
        async for msg in ws:
            print('Message received from server:', msg.data)
            await prompt_and_send(ws)

            if msg.type in (aiohttp.WSMsgType.CLOSED,
                            aiohttp.WSMsgType.ERROR):
                break


async def prompt_and_send(ws):
    new_msg_to_send = input('Type a message to send to the server: ')
    if new_msg_to_send == 'exit':
        print('Exiting!')
        raise SystemExit(0)
    await ws.send_str(new_msg_to_send)


if __name__ == '__main__':
    print('Type "exit" to quit')
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

客户端.py

import aiohttp
import asyncio
import os
import time
import logging
import sys

logging.basicConfig(
        level=logging.DEBUG,

        format='%(name)s: %(message)s',
        stream=sys.stderr,
        )
log = logging.getLogger('main')

HOST = os.getenv('HOST', '0.0.0.0')
PORT = int(os.getenv('PORT', 8080))

URL = f'http://{HOST}:{PORT}/ws'


async def callback(msg):

    if msg == 'time':
        #ws.send_str(time.time())
        print(msg)
    else:
        print(msg)

async def main():
    session = aiohttp.ClientSession()
    async with session.ws_connect(URL) as ws:
        async for msg in ws:
            if msg.type == aiohttp.WSMsgType.TEXT:
                await callback(msg.data)
            elif msg.type == aiohttp.WSMsgType.CLOSED:
                print("CLOSED")
                break
            elif msg.type == aiohttp.WSMsgType.ERROR:
                print("error")
                break

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    tasks = loop.run_until_complete(main())
    try:
        loop.run_forever()
    except KeyboardInterrupt as e:
        print("Caught keyboard interrupt. Canceling tasks...")
        tasks.cancel()
        loop.run_forever()
    finally:
        log.debug('closing server')
        loop.run_until_complete(loop.shutdown_asyncgens())  # python 3.6 only
        log.debug('closing event loop')
        loop.close()

Answer 1

因此，有时仅向其他人阐明问题会有所帮助。我通过像这样在 try/finally 中包装整个 async for msg in ws: 来修复它。

    try:
        async for msg in ws:
            if msg.tp == aiohttp.WSMsgType.text:
                if msg.data == 'close':
                    await ws.close()
                else:
                    # do something here like save it
                    print('%s sent: %s' % (login, msg.data))
                    # Send message to all clients other clients
                    for _ws in [c.websocket for name, c in self.request.app['clients'].items()]:

                            _ws.send_str('(%s) %s' % (login, msg.data))
            elif msg.tp == aiohttp.WSMsgType.error:
                print('ws connection closed with exception %s' % ws.exception())
    finally:
        await client.websocket.close()
        del self.request.app['clients'][client.name]
        for _ws in [c.websocket for name, c in self.request.app['clients'].items()]:
            _ws.send_str('%s disconected' % login)
        print('%s disconnected' % login)