0

I want to create messenger via websockets. My logic is: User_1 send a message (json) to User_2 via tornado handler, a message is checked (def send_message_to_RDB_parallel) on the tornado server (some requests to RDB, PostgreSQL) and then User_1 recieve the response and User_2 recieve a message.

Checking with requests to RDB (def send_message_to_RDB_parallel) - might block my tornado server. Because of it I want to do it via Celery (with RabbitMQ) or just yielded it. As I understand it can help me unblock tornado server. But I need to get the response back when it will be done. I can launch it with Celery or without, but I cant get response.. And when I break my tornado server (push Ctrl-C) then I see an error like "... object is not callable"

How can I get the response and send it (self.write_message())?

In this example I try to do it just with yield

class MessagesHandler(tornado.websocket.WebSocketHandler):
    ...

    def on_message(self, mess):
    ...
        self.send_message_to_RDB(thread_id=thread_id,
                                     sender_id=self.user_id,
                                     recipient_id=recipient_id,
                                     message=message['msg'],
                                     time=datetime.datetime.now(datetime.timezone.utc),
                                     check=True)
    ...                              

    @tornado.gen.coroutine
    def send_message_to_RDB(self, thread_id, sender_id, recipient_id, message, time, check):
        response = yield tornado.gen.Task(send_message_to_RDB_parallel(thread_id=thread_id,
                                             sender_id=sender_id,
                                             recipient_id=recipient_id,
                                             message=message,
                                             time=time,
                                             check=check))
        if response.result[0] is False:
            self.write_message(response.result[1])



def send_message_to_RDB_parallel(thread_id, sender_id, recipient_id, message, time, check=False):
    """
    Send message to rdb. Check thread. One recipient_id !
    """
  #  tf__ = False
    if check is True:
        if recipient_id == sender_id:
            return False, to_json_error(MessengerRecipientEqualSenderServerMessage[1])

        if User.objects.filter(id=recipient_id,
                               is_deleted=False,
                               is_active=True,
                               is_blocked=False).exists() is False:
            return False, to_json_error("Wrong User")
    ...    

    else:
        me = Message()
        me.text = message
        me.thread_id = thread_id
        me.sender_id = sender_id
        me.datetime = time
        me.save()

    return True, None
S-Man
  • 22,521
  • 7
  • 40
  • 63
Anton
  • 420
  • 5
  • 15

1 Answers1

1

There are couple general errors:

  1. send_message_to_RDB_parallel is not async even doesn't have callback arg, but you trying to use it with gen.Task - no result will be set
  2. on_message is a coroutine, it's called in send_message_to_RDB, nut it isn't yielded (awaited)
  3. gen.Task takes a function (and optional additional arguments) and runs it, but it the code actually you are calling it not passing

Because of 2) any further error that occurs are not raised, and that way you see them after ^C. Must take a read of http://www.tornadoweb.org/en/stable/guide/async.html

Solution

Of course you could use celery and asynchronously wait for results (Tornado celery integration hacks)..

But if you using Postgres I would recommend to use existing async library (Saving API output async using SQLAlchemy and Tornado):

  • momoko - postgres Tornado-based client, it is not an ORM,
  • aiopg - postgres asyncio-based client (Tornado 4.3 and above), support for sqlalchemy query builders
kwarunek
  • 12,141
  • 4
  • 43
  • 48