Простой чат на tornado и websocket?

Question

[Xaip]

И снова я, и снова тут.
Данный вопрос я уже создавал и мне на него ответили(Спасибо). https://toster.ru/q/540718
Но данное решение не подходит для чата потому что создает новое соединения для каждого класса, и в итоге получается целая лавина сообщений с экспоненциальным ростом. Я думаю проблему можно решить при помощи отдельного хандлера для redis который будет работать отдельно и итерироваться по списку. Но проблема в том что GIL не позволит мне не даст одновременно обращаться к этому списку. Как мне синхронизировать потоки так что бы при обращении к переменой из вебсокета приостонавливался. Хотя тут тоже вытекают свои недостатки, я же могу потерять сообщения за время приостановки хандлера редиса. Прошу помощи.

# Тут держу список соединений каждое соединение тоже представляет
# список из строки с названием треда и экземпляр класса 
connections = []
# Только единожды инициализирую редис
r = redis.StrictRedis()
p = r.pubsub()
# и подписываю на паттерн сообщений
p.psubscribe("thread:"*)



class EchoWebSocket(tornado.websocket.WebSocketHandler):

    async def open(self):
        print("WebSocket opened")

    def check_origin(self, origin):
        return True

    async def on_message(self, data):
        json_data = json.loads(data)
        if json_data['type'] == "SEND_MESSAGE":
            token = json_data['token']
            message = json_data['message']
            thread = json_data['thread']
            await self.post(message, thread, token)
        if json_data['type'] == "SUBSCRIBE_THREAD":
            thread = "thread:" + str(json_data['id'])
            connections.append([thread, self])


    async def post(self, message, thread, token):
        http_client = httpclient.AsyncHTTPClient()
        url = "http://127.0.0.1:8000/api/v0/thread/add/"
        headers = {'Authorization': 'JWT ' + token, "Content-Type":"application/json"}
        context = {'thread':thread, 'message':message}
        body = json.dumps(context)
        request = await http_client.fetch(request=url, method="POST", headers=headers, body=body)
        http_client.close()

    def on_close(self):
        # Пока не проработал отписку от треда
        self.connection.close()
        print("WebSocket closed")

async def reader():
    while True:
        for evt in p.listen():
            for channel, connection in connections:
                if channel == evt['channel'].decode("UTF-8"):
                    await connection.write_message(evt["data"])



application = tornado.web.Application([
    (r"/", EchoWebSocket),
])
application.autoreload = True
application.listen(8888, '127.0.0.1')
tornado.autoreload.start()
try:
    loop = tornado.ioloop.IOLoop.current()
    loop.start()
except KeyboardInterrupt:
    tornado.ioloop.IOLoop.current().stop()

Comments

[Пасечник Кузьмич]

Ничего не понятно, но очень похоже на попытки применять Tornado и Redis не по назначению.

[Stanislav Pugachev]

Ничего не понятно

да уж

[Stanislav Pugachev]

# Тут держу список соединений каждое соединение тоже представляет
# список из строки с названием треда и экземпляр класса

это зачем так сделано то?

[Xaip]

Pavel Denisov, Redis - PUB/SUB. Django - Pub. При сохранении нового сообщения в Django:

payload = {"id":last_msg.id,
                       "text":last_msg.text,
                       "datetime":str(last_msg.datetime),
                       "sender":last_msg.sender.username,
                       "thread":last_msg.thread.id}
            r.publish("".join("thread:" + str(thread.id)), payload)

В торнадо пытаюсь сделать единый обработчик который принимает новые сообщения и итерируется по списку на наличие подписчиков.

[Пасечник Кузьмич]

Xaip, если это обычный чат, то Django и Redis - лишние звенья.

[Xaip]

Pavel Denisov, Два потока. функция reader() должна выполняться бесконечно, и внутри итерироваться по коннектам.
Пасечник Кузьмич, Нет, это приложение с Django раздающим api, реактом на фронтенде и tornado с Redis для чата)

[Пасечник Кузьмич]

Pavel Denisov, connections - обычный список, так что async for был бы бесполезен.

[Пасечник Кузьмич]

Xaip, похоже, многие особенности реализации, важные для понимания вопроса, остались за кадром. Вам стоит подробно описать логику работы всех элементов системы.

[Пасечник Кузьмич]

Pavel Denisov, async for вызывает anext асинхронного итератора, то есть асинхронно получает значение в переменную цикла (тыц). Для асинхронных операций в теле цикла он не подходит, надо обычный wait вызывать.

[Xaip]

Pavel Denisov, изначально подразумевался aioredis. Потом решил что смогу сделать с помощью стандартного клиента redis-py и конструкцией for msg in p.listen(): потом узнал что aioredis не просто так и он построен на очередях

[Пасечник Кузьмич]

Pavel Denisov, теперь понял! Вы про эту строчку for evt in p.listen():. Простите за невнимательность.

[maxfox]

Что-то я смотрю на ваш код и не очень понимаю, что вы делаете. Я делал похожую штуку, был falcon для rest api, react на клиенте и потребность посылать уведомления по websocket. Решал с помощью aiohttp и redis, falcon делал pub в redis, aiohttp слушал redis и в зависимости от события рассылал уведомления всем подписчикам по websocket. Просто в цикле проходил по списку подписчиков и смотрел, если есть подписка на данное событие, то отправляем его подписчику. Проблем никаких не было, правда потом websocket воркер на ноду переписал, но это другая история. Вот я и не очень понимаю, что у вас за сложности, в чем конкретно проблема? О каком экспотенциальном росте речь идет?

[Xaip]

Пасечник Кузьмич, Pavel Denisov,
Я понимаю вот это код правильно работает, но мне почему то кажется не лучшим решением. Что будет при 1000 пользователях? Придется держать 1000 конектов и циклов? Мне почему то кажется что редису нужен отдельный обработчик который будет ждать сообщений по определенному паттерну.

class WSHandler(tornado.websocket.WebSocketHandler):

    def open(self):
        print("websocket opened")

    def check_origin(self, origin):
        return True

    async def on_message(self, evt):
        data = json.loads(evt)
        print(data)
        await self.setup(data['id'])

    async def setup(self, id):
        self.conn = await aioredis.create_redis("redis://127.0.0.1")
        self.thread = "thread:{!s}".format(id)
        channel = await self.conn.psubscribe(self.thread)
        await asyncio.ensure_future(self.send_message(channel, self))

    async def send_message(self, channel, obj):
        print("this")
        # redis = await aioredis.create_redis("redis://127.0.0.1")
        # channel_pattern = await redis.psubscribe("thread:*",)
        while await channel[0].wait_message():
            msg = await channel[0].get(encoding="utf-8")
            print(msg)
            # await obj.write_message(msg)





    async def on_close(self):
        print("WS close")
        await self.conn.unsubscribe("thread:{!s}".format(id))
        await self.conn.wait_closed()

[Xaip]

maxfox, Хочу свой велосипед. Я уже отписал что впринципе решил проблему но считаю решение не корректным. Может у кого есть интересные примеры.

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

Xaip, ну так вы подписку делайте не в WebSocketHandler, а в отдельной сопрограмме, как я вам показывал. Будет одно асинхронное соединенте с Redis пересылающее сообщения в тысячи соединений websocket.

[Xaip]

Сергей Горностаев, Пойду тестить бенчмарками