brokers.py

from logging import getLogger
from json import dumps

from django.db import models
from django.core.cache import cache

import pika
from pika.credentials import PlainCredentials

LOGGER = getLogger(__name__)


def publish_to_user(user, obj):
    from . import serializers
    RabbitMQInstance.load().publish_to_user(
        user,
        dumps(serializers.UpdateSerializer(obj).data),
    )


# singleton for simple configs
# https://steelkiwi.com/blog/practical-application-singleton-design-pattern/
class SingletonModel(models.Model):
    class Meta:
        abstract = True

    def set_cache(self):
        cache.set(self.__class__.__name__, self)

    # pylint: disable=invalid-name, arguments-differ
    def save(self, *args, **kwargs):
        self.pk = 1
        super(SingletonModel, self).save(*args, **kwargs)
        self.set_cache()

    @classmethod
    def load(cls):
        if cache.get(cls.__name__) is None:
            obj, created = cls.objects.get_or_create(pk=1)
            if not created:
                obj.set_cache()
        return cache.get(cls.__name__)


# The rabbitmq instance located on this host (localhost)
class RabbitMQInstance(SingletonModel):

    @property
    def host(self):
        return 'localhost'

    @property
    def vhost(self):
        return '/'

    @property
    def exchanges(self):
        return [
            'entitlements',
            'groups',
            'users',
            'services',
        ]

    def _init_exchanges(self, channel):
        # this is no error
        for exchange in self.exchanges:
            channel.exchange_declare(
                exchange=exchange,
                durable=True,  # so exchange survive a broker restart
                auto_delete=False,
                exchange_type='topic',
            )

    @property
    def _params(self):
        # we set NO port here, we use the default (probably 5672)
        # this requires the
        return pika.ConnectionParameters(
            host=self.host,
            virtual_host=self.vhost,
            credentials=PlainCredentials(
                'guest',
                'guest',
            ),
        )

    def __str__(self):
        return self.host

    def msg(self, msg):
        return '[RabbitMQ:{}] {}'.format(self.host, msg)

    def _open_connection(self):
        try:
            # start a new connection
            return pika.BlockingConnection(self._params)
        except Exception as e:
            LOGGER.exception('RabbitMQ connection error: %s', e)
            raise e

    def _publish(self, exchange, routing_key, body):
        connection = self._open_connection()

        channel = connection.channel()
        # put in confirm mode
        channel.confirm_delivery()

        channel.basic_publish(
            exchange=exchange,
            routing_key=routing_key,
            body=body,
            properties=pika.BasicProperties(
                delivery_mode=1,
            ),
        )

        # close channel and connection
        connection.close()

    # PUBLIC API

    # called on client registration to make sure the  exchanges exists
    def initialize(self):
        connection = self._open_connection()
        channel = connection.channel()
        self._init_exchanges(channel)
        connection.close()

    def publish_deployment_state(self, deployment_state):
        from .serializers.clients import DeploymentStateSerializer
        self._publish(
            'services',
            deployment_state.service.name,
            dumps(DeploymentStateSerializer(deployment_state).data),
        )

    def publish_to_user(self, user, msg):
        self._publish(
            'users',
            str(user.id),
            msg,
        )


# we keep this, as removing this breaks migrations
def exchanges_default():
    return []