Redis pub/sub implementation in tornado?

Question

Redis

I'm trying to implement websocket chat on tornado. I decided to use redis as a handler for new messages, but you can’t hang listen() on the main thread (And all implementations in the form of bruwka, toredis and others are simply terrible. I saw an elegant solution somewhere on the github, using the additional event IOLoop thread. I would be grateful, in resolving this issue

Answer 1

The brukva and toredis libraries have not been updated for several years. But in the last update, Tornado finally switched to using the asyncio event queue, which makes it possible to use fresh and good libraries. For example 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()

Naturally, this is the most simplified example, in the real code, the connection should not be kept in a global variable, and when the server shuts down, it is worth unsubscribing from the channel and closing the connection to redis.