HowTo: Как подружить Django с WebSocket (socket.io, sockjs)

Version: 0.2

Возникла у меня потребность атомарного обновления в реальном времени страницы у некоторого количества пользователей в зависимости от действий других пользователей (не гербалайф чат). Понятное дело, можно всё выкинуть в помойку и, по-молодецки, запилить с нуля на tornado/twisted.web, но явно не самый продуктивный путь (да и я не мо́лодец ни разу) когда всё что надо — уже работает на Django и нужно всего-то чуть-чуть… Естественным образом, по сути своей, сюда просится WebSocket. И всё бы ничего но Django WSGI приложение, а этот стандарт не предполагает таких выкрутасов даже близко (пока). Гугления интернетов навели, в очередной раз, на труд известного python-гуру kmike (это без сарказма, т.к. его работы выручали меня лично уже не однократно, за что нижайший ему поклон!).

Итак если вы хотите скрестить ваш Django проект с websocket посредством js библиотеки socket.io или sockjs — вилькоммен!

Об обновлении

Первая версия статьи была посвящена только варианту с Socket.io. В процессе работы возникло странное ощущение, что библиотека не всегда определяет отвалившееся подключение. Т.е. при отвале сокета на долгое время она не пыталась зацепиться обратно. Вполне допускаю, что я сам сделал что-то не так. В комментариях к первой версии статьи люди рекомендовали sockjs (по своей инициативе, т.к. в итоге перешли на него) и я решил проверить этот вариант. В итоге либа держится за коннект всеми зубами, но со стороны сервера возникла странная ситуация, когда он перестаёт отвечать на запросы (также скорее всего по нашей вине, но причина пока ещё в процессе выяснения). В любом случае я решил дополнить статью для полноты картины (не без настоятельной просьбы товарища из гуглоплюса). В общем оставляю выбор реализации за вами.

Вступление

Давно хотелось попробовать чего-нибудь асинхронного, да всё не было хорошего повода, Тут появилась необходимость, а откуда стартовать было совершенно не ясно. Собственно здесь я попытаюсь создать эту самую актуальную (мной самим за отправную точку был взят упомянутый выше доку́мент но он довольно стар и уже появились некоторые усовершенствования) отправную точку для старта. Будет знакомый островок Django к которому я покажу как подпустить свежего ветерку…

Кстати из труда kmike пара функций использована без изменений, надеюсь автор не против.

Что получим

В результате мы получим асинхронный сервис, который крутится рядом с основным django сайтом, знает какой django пользователь посылает/получает запросы, и [сервис] может получать команды от django, выполняя на их основе какие-то действия в браузере юзера.

Пример

Возьмём для примера гипотетическую биржу. У неё есть модераторы и клиенты. Всё работало у вас нормально и тут понадобилось дать модераторам возможность в реальном времени видеть изменения позиций на бирже. При этом модераторы могут как-то оперировать с позициями на бирже и нельзя просто перезагружать страницу.

До этого у вас все дружно колбасили F5… И, в общем, highload, как таковой, нас интересует не особо.

Инструменты

Для работы нам понадобятся:

pip install redis tornado-redis
и 
pip install tornadio2 
или
pip install sockjs-tornado
в зависимости от вашего выбора.

А также библиотека socket.io или sockjs

Теория

Для работы с socket.io будем использовать библиотеку tornadio2, а для sockjs — sockjs-tornado которые, естественным образом основаны на асинхронном фреймворке tornado. Запускаться это дело будет как manage команда django (привет supervisor). Особых проблем с исполнением джанговского когда в tornadio нет, а вот в обратку у нас небольшой затык, который решается PubSub возможностями Redis (вкратце это такие каналы или очереди сообщений в которые publisher'ы пихают сообщения, а subscriber'ы их получают).

Отмазка

По ходу пьесы внимательный читатель может заметить нестыковочки, вроде использования django функций, которые, по сути своей, синхронны, но это небольшая жертва в угоду быстрой разработке. Кроме того речи о хайлоад изначально не идёт ну и это не всестороннее решение, а отправная точка. Так что развлекаться с нюансами вашей реализации и узкими местами вашего кода я оставляю вам, за что прошу меня великодушно простить…

Также см. отмазки kmike в документе, на который я всё время ссылаюсь.

Практика

Практика будет практичной, потому много объяснений в комментариях в исходниках.

service.py

Собственно сам сервис, который будет поддерживать соединения с браузером, получать команды из django, отправляя их клиентам (и аналогично в обратную сторону).

Метод on_message обязателен к реализации, но в приведённом примере он не нужен, т.к. всё реализуется на новомодной событийной модели (для socket.io).

Реализация для socket.io

# -*- coding: utf-8 -*-

import tornado
import tornadoredis
from tornadio2 import SocketConnection
from tornadio2.conn import event
import django
from django.utils.importlib import import_module
from django.conf import settings
from django.utils import simplejson

# start of kmike's sources
_engine = import_module(settings.SESSION_ENGINE)


def get_session(session_key):
    return _engine.SessionStore(session_key)


def get_user(session):
    class Dummy(object):
        pass

    django_request = Dummy()
    django_request.session = session

    return django.contrib.auth.get_user(django_request)
# end of kmike's sources


# конфиг для подключения к redis можно хранить в настройках django
ORDERS_REDIS_HOST = getattr(settings, 'ORDERS_REDIS_HOST', 'localhost')
ORDERS_REDIS_PORT = getattr(settings, 'ORDERS_REDIS_PORT', 6379)
ORDERS_REDIS_PASSWORD = getattr(settings, 'ORDERS_REDIS_PASSWORD', None)
ORDERS_REDIS_DB = getattr(settings, 'ORDERS_REDIS_DB', None)

# немного удобства
unjson = simplejson.loads
json = simplejson.dumps


class Connection(SocketConnection):
    def __init__(self, *args, **kwargs):
        super(Connection, self).__init__(*args, **kwargs)
        self.listen_redis()

    @tornado.gen.engine
    def listen_redis(self):
        """
        Вешаем подписчиков на каналы сообщений.
        """
        self.redis_client = tornadoredis.Client(
                host=ORDERS_REDIS_HOST,
                port=ORDERS_REDIS_PORT,
                password=ORDERS_REDIS_PASSWORD,
                selected_db=ORDERS_REDIS_DB
            )
        self.redis_client.connect()

        yield tornado.gen.Task(self.redis_client.subscribe, [
            'order_lock',
            'order_done'
        ])
        self.redis_client.listen(self.on_redis_queue)  # при получении сообщения
                           #  вызываем self.on_redis_queue

    def on_open(self, info):
        """
        Определяем сессию django.
        """
        self.django_session = get_session(info.get_cookie('sessionid').value)

    @event  # событие, произошедшее в браузере
    def login(self):
        """
        Определение пользователя и его возможностей
        """
        # это просто для примера входящей команды, определять юзера можно и в on_open
        self.user = get_user(self.django_session)
        self.is_client = self.user.has_perm('order.lock')
        self.is_moder = self.user.has_perm('order.delete')

    def on_message(self):
        """
        Обязательный метод.
        """
        pass

    def on_redis_queue(self, message):
        """
        Обновление в списке заказов
        """
        if message.kind == 'message':  # сообщения у редиса бывают разного типа, 
                           # много сервисных, нам нужны только эти
            message_body = unjson(message.body)  # разворачиваем сабж, как вы  
                                   #  поняли я передаю данные в JSON

            # в зависимости от канала получения распределяем сообщения
            if message.channel == 'order_lock':
                self.on_lock(message_body)

            if message.channel == 'order_done:
                self.on_done(message_body)

    def on_lock(self, message):
        """
        Заказ закреплён
        """
        if message['user'] != self.user.pk:  # юзеру-источнику действия сообщать о нём не надо
            self.emit('lock', message)

    def on_done(self, message):
        """
        Заказ выполнен
        """
        if message['user'] != self.user.pk:
            if self.is_client:
                message['action'] = 'hide'
            else:
                message['action'] = 'highlight'

            self.emit('done', message)

    def on_close(self):
        """
        При закрытии соединения отписываемся от сообщений
        """
        self.redis_client.unsubscribe([
            'order_lock',
            'order_done'
        ])
        self.redis_client.disconnect()

Реализация для sockjs

# -*- coding: utf-8 -*-

import tornado
import tornadoredis
from sockjs.tornado import SockJSConnection
import django
from django.utils.importlib import import_module
from django.conf import settings
from django.utils import simplejson

# start of kmike's sources
_engine = import_module(settings.SESSION_ENGINE)


def get_session(session_key):
    return _engine.SessionStore(session_key)


def get_user(session):
    class Dummy(object):
        pass

    django_request = Dummy()
    django_request.session = session

    return django.contrib.auth.get_user(django_request)
# end of kmike's sources


# конфиг для подключения к redis можно хранить в настройках django
ORDERS_REDIS_HOST = getattr(settings, 'ORDERS_REDIS_HOST', 'localhost')
ORDERS_REDIS_PORT = getattr(settings, 'ORDERS_REDIS_PORT', 6379)
ORDERS_REDIS_PASSWORD = getattr(settings, 'ORDERS_REDIS_PASSWORD', None)
ORDERS_REDIS_DB = getattr(settings, 'ORDERS_REDIS_DB', None)

# немного удобства
unjson = simplejson.loads
json = simplejson.dumps


class Connection(SocketConnection):
    def __init__(self, *args, **kwargs):
        super(Connection, self).__init__(*args, **kwargs)
        self.listen_redis()

    @tornado.gen.engine
    def listen_redis(self):
        """
        Вешаем подписчиков на каналы сообщений.
        """
        self.redis_client = tornadoredis.Client(
                host=ORDERS_REDIS_HOST,
                port=ORDERS_REDIS_PORT,
                password=ORDERS_REDIS_PASSWORD,
                selected_db=ORDERS_REDIS_DB
            )
        self.redis_client.connect()

        yield tornado.gen.Task(self.redis_client.subscribe, [
            'order_lock',
            'order_done'
        ])
        self.redis_client.listen(self.on_redis_queue)  # при получении сообщения
                           #  вызываем self.on_redis_queue

    def send(self, msg_type, message):
        """
        Оправка сообщений.
        """
        return super(Connection, self).send({
                'type': msg_type,
                'data': message,
            })

    def on_open(self, info):
        """
        Определяем сессию django.
        """
        self.django_session = get_session(info.get_cookie('sessionid').value)
        self.user = get_user(self.django_session)
        self.is_client = self.user.has_perm('order.lock')
        self.is_moder = self.user.has_perm('order.delete')

    def on_message(self):
        """
        Обязательный метод.
        """
        pass

    def on_redis_queue(self, message):
        """
        Обновление в списке заказов
        """
        if message.kind == 'message':  # сообщения у редиса бывают разного типа, 
                           # много сервисных, нам нужны только эти
            message_body = unjson(message.body)  # разворачиваем сабж, как вы  
                                   #  поняли я передаю данные в JSON

            # в зависимости от канала получения распределяем сообщения
            if message.channel == 'order_lock':
                self.on_lock(message_body)

            if message.channel == 'order_done:
                self.on_done(message_body)

    def on_lock(self, message):
        """
        Заказ закреплён
        """
        if message['user'] != self.user.pk:  # юзеру-источнику действия сообщать о нём не надо
            self.send('lock', message)

    def on_done(self, message):
        """
        Заказ выполнен
        """
        if message['user'] != self.user.pk:
            if self.is_client:
                message['action'] = 'hide'
            else:
                message['action'] = 'highlight'

            self.send('done', message)

    def on_close(self):
        """
        При закрытии соединения отписываемся от сообщений
        """
        self.redis_client.unsubscribe([
            'order_lock',
            'order_done'
        ])
        self.redis_client.disconnect()

models.py

Источник изменений. Пускай это будет модель.

# -*- coding: utf-8 -*-

import redis
from django.conf import settings
from django.db import models


ORDERS_FREE_LOCK_TIME = getattr(settings, 'ORDERS_FREE_LOCK_TIME', 0)
ORDERS_REDIS_HOST = getattr(settings, 'ORDERS_REDIS_HOST', 'localhost')
ORDERS_REDIS_PORT = getattr(settings, 'ORDERS_REDIS_PORT', 6379)
ORDERS_REDIS_PASSWORD = getattr(settings, 'ORDERS_REDIS_PASSWORD', None)
ORDERS_REDIS_DB = getattr(settings, 'ORDERS_REDIS_DB', 0)

# опять удобства
service_queue = redis.StrictRedis(
    host=ORDERS_REDIS_HOST,
    port=ORDERS_REDIS_PORT,
    db=ORDERS_REDIS_DB,
    password=ORDERS_REDIS_PASSWORD
).publish
json = simplejson.dumps

class Order(models.Model)
    …

    def lock(self):
        """
        Закрепление заказа
        """
        …

        service_queue('order_lock', json({
                'user': self.client.pk,
                'order': self.pk,
            }))

    def done(self):
        """
        Завершение заказа
        """
        …

        service_queue('order_done', json({
                'user': self.client.pk,
                'order': self.pk,
            }))

Собственно здесь методы lock и done после выполнения какой-то бизнес-логики отправляют сообщения с необходимой информацией. Эта информация будет получена вышеописанным сервисом, обработана и разослана клиентским браузерам.

Т.е. действие выполнено пользователем по стандартной схеме: он кликнул ссылку/нажал кнопку, django отработал необходимые действия, послал уведомление в канал для рассылки через websocket и вернул юзеру классический ответ.

client.js

Не забывайте загрузить в html socket.io.js или sockjs.js в зависимости от вашего выбора (ссылки в начале статьи)!

Собственно апофигей всего этого действа — работа на клиентской стороне.

Реализация для socket.io

    var socket = io.connect('http://' + window.location.host + ':8989');  // ваш порт для асинхронного сервиса
    // при соединении вызываем событие login, которое будет выполнено на серверной стороне
    socket.on('connect', function(){
        socket.emit('login');
    });

    // при дисконнекте - пытаемся вернуть соединение
    socket.on('disconnect', function() {
        setTimeout(socket.socket.reconnect, 5000);
    });

    // при возникновении события "lock" вызываем "ws_order_lock" с полученным сообщением в качестве параметра
    socket.on('lock', function(msg){
        ws_order_lock(msg);
    });

    socket.on('done', function(msg){
        ws_order_done(msg);
    });

function ws_order_lock(msg){
    if (msg.action == 'highlight'){
        $('.id_order_row__' + msg.order).addClass('order-row_is_locked');
    }else{
        $('.id_info_renew_orders').addClass('hidden');
    }
}

…

Реализация для sockjs

socket_connect();

function socket_connect() {
    socket = new SockJS('http://' + window.location.host + ':8989/orders');  // ваш порт для асинхронного сервиса
    // при соединении вызываем событие login, которое будет выполнено на серверной стороне

    socket.onmessage = function(msg){
        window['ws_order_' + msg.data.type](msg.data.data);  // роутер, выполняет функцию согласно типу сообщения
    }

    socket.onclose = function(e){
        setTimeout(socket_connect, 5000);
    };
}

function ws_order_lock(msg){
    if (msg.action == 'highlight'){
        $('.id_order_row__' + msg.order).addClass('order-row_is_locked');
    }else{
        $('.id_info_renew_orders').addClass('hidden');
    }
}

…

async_server.py

Это manage команда, файл надо класть в папку myProject/orderApp/management/commands не забываем также, в каждой из подпапок файлик __init__.py.

Реализация для socket.io

# -*- coding: utf-8 -*-

import tornado
import tornadio2 as tornadio
from django.core.management.base import NoArgsCommand

from myProject.order.tornado.service import Connection


class Command(NoArgsCommand):
    def handle_noargs(self, **options):
        router = tornadio.TornadioRouter(Connection)
        app = tornado.web.Application(router.urls, socket_io_port=8989)  # ваш порт для асинхронного сервиса
        tornadio.SocketServer(app)

Реализация для sockjs

# -*- coding: utf-8 -*-

import tornado
import tornadio2 as tornadio
from django.core.management.base import NoArgsCommand

from myProject.order.tornado.service import Connection


class Command(NoArgsCommand):
    def handle_noargs(self, **options):
        router = SockJSRouter(Connection, '/orders')  # sockjs не захотел работать с корнем :(
        app = tornado.web.Application(router.urls)
        app.listen(8989)
        tornado.ioloop.IOLoop.instance().start()

Теперь можно стартовать сервис python manage.py async_server.