Как преиодически отправлять данные от сервера клиенту через websocket?

Question

[strannik7j]

Добый день!
Клиент подключается к серверу, устанавливает соединение по websoket и запрашивает данные. Сервер ему их отправляет.
Но через 2 секунды данные меняются и нужна повторная отправка. Как реализовать её в рамках доной сессии?
Вот рабочий код сервера:

import asyncio
import websockets
from json import dumps

async def hello(websocket, path):
    Client = await websocket.recv()
    print("< {}".format(Client))
    
    Data = BSession.GetData()
    await websocket.send(dumps(Data))

start_server = websockets.serve(hello, '', 55333)

asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()

Но в этом случае сервер отпраляет данные единожды за сессию.

Я пробовал сделать так:

import asyncio
import websockets
from json import dumps

async def hello(websocket, path):
    Client = await websocket.recv()
    print("< {}".format(Client))

    while True:
        Data = BSession.GetData()
        await websocket.send(dumps(Data))
        sleep(2)

start_server = websockets.serve(hello, '', 55333)

asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()

но в этом случае сервер перестаёт принимать новые запросы и циклится на одном клиенте.
Как сделать правильно?

Answer 1

[Сергей Горностаев]

Насколько я понял, вы пытаетесь опрашивать базу данных и выводить изменения, а BSession.GetData() как раз данные из базы получает с помощью SQLAlchemy.

Во-первых, обращение к базе синхронное и в моменты выполнения запросов будет ставить колом все установленные websocket-соединения. Использовать для решения это проблемы можно sqlalchemy_aio или SQLAlchemy+aiopg. А ещё лучше asyncpg без SQLAlchemy.

Во-вторых, чтобы пользователи не получали каждый раз всё содержимое базы, стоит хранить где-нибудь последний извлечённый id и запрашивать из базы только записи, у которых id больше.

Наконец, что касается организации двусторонней асинхронности, должен помочь немного модифицированный вариант из документации:

async def consumer_handler(websocket, path):
    while True:
        message = await websocket.recv()
        # делаем что-нибудь с сообщением от клиента


async def producer_handler(websocket, path):
    while True:
        data = await BSession.GetData()
        if data:
            await websocket.send(data)


async def handler(websocket, path):
    consumer_task = asyncio.ensure_future(consumer_handler(websocket))
    producer_task = asyncio.ensure_future(producer_handler(websocket))
    done, pending = await asyncio.wait(
        [consumer_task, producer_task],
        return_when=asyncio.FIRST_COMPLETED,
    )

    for task in pending:
        task.cancel()


loop = asyncio.get_event_loop()
loop.run_until_complete(websockets.serve(handler, 'localhost', 8765))
loop.run_forever()

Answer 2

[Влад]

await asyncio.sleep(2)