如何对连接到websocket的每个客户端使用唯一查询

Question

我试图在我的websocket中只有一个数据库连接，并将此信息返回给每个连接的客户端。有可能做到吗？

这是我当前的代码：

import asyncio
import aiopg
import websockets
import logging
import sys
import configparser
config = configparser.ConfigParser()
config.read('config/config.ini')
logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
logger = logging.getLogger('websockets.server')
logger2 = logging.getLogger('asyncio')
logger2.setLevel(logging.DEBUG)
logger.addHandler(logging.StreamHandler())
logger2.addHandler(logging.StreamHandler())
async def listen(websocket, path):
async with aiopg.create_pool(config.get('default', 'connexion_bd'), maxsize=1, pool_recycle=0) as pool:
async with pool.acquire() as conn1:
async with conn1.cursor() as cur:
await cur.execute(config.get('default', 'pg_listen'))
while True:
msg = await conn1.notifies.get()
if msg.payload == 'finish':
return
else:
await websocket.send(msg.payload)

start_server = websockets.serve(listen, 'localhost', config.getint('default', 'port_websocket_server'))
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()

目前，每次客户端监听我的websocket时，我的数据库中都会有一个新的连接。每个客户端都会得到相同的内容，所以我不需要他们连接到数据库，只需要我的websocket。

我试图分割我的代码(连接到另一个正在连接这个代码的网络套接字上(，但我遇到了同样的问题。

如有任何提示，我们将不胜感激。

感谢

Answer 1

得到

需要将websocket.serve放在查询之后。

import asyncio
import functools
import aiopg
import websockets
import logging
import sys
import configparser
config = configparser.ConfigParser()
config.read('config/config.ini')
logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
logger = logging.getLogger('websockets.server')
logger2 = logging.getLogger('asyncio')
logger2.setLevel(logging.DEBUG)
logger.addHandler(logging.StreamHandler())
logger2.addHandler(logging.StreamHandler())
# Pour starter
# env/bin/python3 websocket_server_aiopg.py
USERS = set()

async def listen(websocket, path, conn1):
USERS.add(websocket)
while True:
msg = await conn1.notifies.get()
if msg.payload == 'finish':
return
else:
await asyncio.wait(([user.send(msg.payload) for user in USERS]))

async def run_server():
async with aiopg.create_pool(config.get('default', 'connexion_bd')) as pool:
async with pool.acquire() as conn1:
async with conn1.cursor() as cur:
await cur.execute(config.get('default', 'pg_listen'))
async with websockets.serve(functools.partial(listen, conn1=conn1), 'localhost', config.getint('default', 'port_websocket_server')) as ws:
await asyncio.sleep(1_000_000_000)  # consider waiting on an exit condition instead
asyncio.get_event_loop().run_until_complete(run_server())