backgrounddating.com tornado chat. This source code is licensed under the terms of the GNU GPL 3.0.

tornadoapp.py

import os
import datetime
import json
import time
import urllib

import brukva
import tornado.httpserver
import tornado.web
import tornado.websocket
import tornado.ioloop
import tornado.httpclient

from django.conf import settings
from django.utils.importlib import import_module

session_engine = import_module(settings.SESSION_ENGINE)

from dating.models import UserProfile
from privatemessages.models import Thread

c = brukva.Client()
c.connect()

class MainHandler(tornado.web.RequestHandler):
    def get(self):
        self.set_header('Content-Type', 'text/plain')
        self.write('Hello. :)')

class MessagesHandler(tornado.websocket.WebSocketHandler):
    def __init__(self, *args, **kwargs):
        super(MessagesHandler, self).__init__(*args, **kwargs)
        self.client = brukva.Client()
        self.client.connect()

    def open(self, thread_id):
        session_key = self.get_cookie(settings.SESSION_COOKIE_NAME)
        session = session_engine.SessionStore(session_key)
        try:
            user_id = session["_auth_user_id"]
        except KeyError:
            self.close()
            return
        if not Thread.objects.filter(id=thread_id, participants__id=user_id):
            self.close()
            return
        self.channel = "".join(['thread_', thread_id,'_messages'])
        self.client.subscribe(self.channel)
        self.sender_name = UserProfile.objects.get(user_id=user_id).name
        self.user_id = user_id
        self.thread_id = thread_id
        self.client.listen(self.show_new_message)
        c.hincrby("".join(["user_", str(user_id), "_open_chats"]), thread_id, 1)

    def handle_request(self, response):
        pass

    def on_message(self, message):
        if not message:
            return
        if len(message) > 10000:
            return
        c.publish(self.channel, json.dumps({
            "timestamp": int(time.time()),
            "sender": self.sender_name,
            "text": message,
        }))
        http_client = tornado.httpclient.AsyncHTTPClient()
        request = tornado.httpclient.HTTPRequest(
            "".join([
                        settings.SEND_MESSAGE_API_URL,
                        "/",
                        self.thread_id,
                        "/"
                    ]),
            method="POST",
            body=urllib.urlencode({
                "message": message.encode("utf-8"),
                "api_key": settings.API_KEY,
                "sender_id": self.user_id,
            })
        )
        http_client.fetch(request, self.handle_request)

    def show_new_message(self, result):
        self.write_message(str(result.body))

    def on_close(self):
        try:
            self.client.unsubscribe(self.channel)
        except AttributeError:
            pass
        else:
            c.hincrby(
                "".join(["user_", str(self.user_id), "_open_chats"]),
                self.thread_id, -1
            )
        def check():
            if self.client.connection.in_progress:
                tornado.ioloop.IOLoop.instance().add_timeout(datetime.timedelta(0.00001), check)
            else:
                self.client.disconnect()
        tornado.ioloop.IOLoop.instance().add_timeout(datetime.timedelta(0.00001), check)

class NewMessagesCountHandler(tornado.websocket.WebSocketHandler):
    def __init__(self, *args, **kwargs):
        super(NewMessagesCountHandler, self).__init__(*args, **kwargs)
        self.client = brukva.Client()
        self.client.connect()

    def on_new_messages_count_retrieval(self, result):
        if result:
            self.write_message(result)

        self.channel = "".join(
            ['user_', self.user_id, '_new_messages_count_updates']
        )
        self.client.subscribe(self.channel)
        self.client.listen(self.send_updated_message_count)

    def open(self):
        session_key = self.get_cookie(settings.SESSION_COOKIE_NAME)
        session = session_engine.SessionStore(session_key)
        try:
            self.user_id = str(session["_auth_user_id"])
        except KeyError:
            self.close()
            return
        c.get(
            "".join(['user_', self.user_id, '_new_messages_total']),
            self.on_new_messages_count_retrieval
        )

    def send_updated_message_count(self, result):
        self.write_message(str(result.body))

    def on_close(self):
        try:
            self.client.unsubscribe(self.channel)
        except AttributeError:
            pass
        def check():
            if self.client.connection.in_progress:
                tornado.ioloop.IOLoop.instance().add_timeout(datetime.timedelta(0.00001), check)
            else:
                self.client.disconnect()
        tornado.ioloop.IOLoop.instance().add_timeout(datetime.timedelta(0.00001), check)

application = tornado.web.Application([
    (r"/", MainHandler),
    (r"/new_messages_count/", NewMessagesCountHandler),
    (r'/(?P<thread_id>\d+)/', MessagesHandler),
])