Реализация redis pub/sub в tornado?

Question

[Xaip]

Пытаюсь реализовать websocket чат на tornado. В качестве хандлера новых сообщений решил использовать redis, но вешать listen() на основной поток нельзя( А все реализации ввиде bruwka, toredis и другие просто ужасны. Видел где-то на гитхабе изящное решение, с помощью доп event IOLoop потока. Буду признателен, в решении данного вопроса

Comments

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

Объясните подробнее зачем вам Redis.

[Xaip]

Сергей Горностаев, DRF при сохранение в бд делает так же запись в redis по типу thread_id: message
На фронтенде React который открывает websocket с Tornado и подписывается на обновления сообщений к Redis.
И вот именно этап с subscribe я не могу выполнить.

class EchoWebSocket(tornado.websocket.WebSocketHandler):

    def open(self):
        print("WebSocket opened")
        redis_client = redis.StrictRedis()
        sub = redis_client.pubsub()
        sub.subscribe('channel')
        for message in sub.listen():         # поместить в основной цикл не могу
            print(message.body)


    def send_new_message(self, evt):
        print(evt.body)

    def check_origin(self, origin):
        return True

    def on_message(self, data):
        json_data = json.loads(data)
        self.token = json_data['token']
        self.message = json_data['message']
        self.thread = json_data['thread']
        self.post()

Answer 1

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

Библиотеки brukva и toredis уже не обновлялись по нескольку лет. Зато Tornado в последнем обновлении окончательно перешёл на использование очереди событий asyncio, что даёт возможность использовать свежие и хорошие библиотеки. Например aioredis:

import asyncio
import aioredis
from tornado import web, websocket
from tornado.ioloop import IOLoop

connections = []

class WSHandler(websocket.WebSocketHandler):
    def open(self):
        connections.append(self)

    def on_message(self, message):
        ...

    def on_close(self):
        connections.remove(self)


class GetHandler(web.RequestHandler):
    def get(self):
        self.render("chat.html")


async def consumer(channel):
    while await channel.wait_message():
        msg = await channel.get(encoding='utf-8')
        for connection in connections:
            await connection.write_message(msg)


async def setup():
    connection = await aioredis.create_redis('redis://localhost')
    channel = await connection.subscribe('notifications')
    asyncio.ensure_future(consumer(channel))


application = web.Application([
    (r'/', GetHandler),
    (r'/chat/', WSHandler),
])    


if __name__ == '__main__':
    application.listen(8000)
    loop = IOLoop.current()
    loop.add_callback(setup)
    loop.start()

Естественно, это максимально упрощённый пример, в реальном коде соединения не стоит держать в глобальной переменной, а при завершении работы сервера стоит отписаться от канала и закрыть соединение с redis.