portier/source/config/models.py

import json
import uuid

from django.conf import settings
from django.db import models
from django.urls import reverse
from django.utils.translation import gettext as _
from django.db.models.signals import pre_delete
from portier.common import handlers
from config import signals_shared
from config.util import validate_stream_url


class Stream(models.Model):
    stream = models.UUIDField(unique=True, default=uuid.uuid4, help_text=_('stream_stream_help'))
    name = models.CharField(max_length=100, help_text=_('stream_name_help'))

    # the same stream uuid can be published multiple times to different origin
    # servers. this is a valid scheme to achieve a failover on the origin layer.
    # thus we need to keep track if a stream is published at least once,
    # and only send signals when we are going to / coming from 0 published streams.
    publish_counter = models.PositiveIntegerField(default=0)

    def save(self, *args, **kwargs):
        super().save(*args, **kwargs)
        if self.publish_counter > 0:
            signals_shared.stream_active.send(sender=self.__class__,
                                              stream=str(self.stream),
                                              param=None
                                              )
        else:
            signals_shared.stream_inactive.send(sender=self.__class__,
                                                stream=str(self.stream),
                                                param=None
                                                )


    def get_absolute_url(self):
        return reverse('config:stream_detail', kwargs={'pk': self.pk})

    def class_name(self):
        return _('stream_class_name')

    def __str__(self):
        return str(self.name)


pre_delete.connect(handlers.remove_obj_perms_connected_with_user, sender=Stream)


class SRSNode(models.Model):
    name = models.CharField(max_length=100, help_text=_('srsnode_name_help'))
    api_base = models.CharField(max_length=256, help_text=_('srsnode_api_base_help'))
    rtmp_base = models.CharField(max_length=256, help_text=_('srsnode_rtmp_base_help'))
    active = models.BooleanField(help_text=_('srsnode_active_help'))

    class Meta:
        verbose_name = _('srsnode_verbose_name')
        verbose_name_plural = _('srsnode_verbose_name_plural')

    def __str__(self):
        return self.name


class SRSStreamInstance(models.Model):
    node = models.ForeignKey(SRSNode, on_delete=models.CASCADE, help_text=_('srsstreaminstance_node_help'))
    stream = models.ForeignKey(Stream, on_delete=models.CASCADE, help_text=_('srsstreaminstance_stream_help'))
    last_update = models.DateTimeField(auto_now=True, help_text=_('srsstreaminstance_last_update_help'))
    statusdata = models.TextField(default="{}", help_text=_('srsstreaminstance_statusdata_help'))

    class Meta:
        verbose_name = _('srsstreaminstance_verbose_name')
        verbose_name_plural = _('srsstreaminstance_verbose_name_plural')

    def __str__(self):
        return f"{self.stream} on {self.node}"


class Pull(models.Model):
    stream = models.ForeignKey(Stream, on_delete=models.CASCADE, help_text=_('pull_stream_help'))
    source = models.CharField(max_length=500, validators=[validate_stream_url], help_text=_('pull_source_help'))
    active = models.BooleanField(help_text=_('pull_activate_help'))
    name = models.CharField(max_length=100, help_text=_('pull_name_help'))

    class Meta:
        verbose_name = _('pull_verbose_name')
        verbose_name_plural = _('pull_verbose_name_plural')

    def class_name(self):
        return _('pull_class_name')

    def get_absolute_url(self):
        return reverse('config:pull_detail', kwargs={'pk': self.pk})

    def __str__(self):
        return str(self.name)

    def save(self, *args, **kwargs):
        super().save(*args, **kwargs)
        if self.active:
            signals_shared.pull_active.send(
                sender=self.__class__,
                pull_id=self.id,
            )
        else:
            signals_shared.pull_inactive.send(
                sender=self.__class__,
                pull_id=self.id,
            )

    def get_concierge_configuration(self):
        if not self.active:
            return None

        # Select a random node that is active to ingest the pulled stream.
        node = SRSNode.objects.filter(active=True).order_by('?').first()

        return {
            "type": "pull",
            "active": self.active,
            "name": self.name,
            "source": self.source,
            "target": f"{node.rtmp_base}/{settings.GLOBAL_STREAM_NAMESPACE}/{self.stream.stream}",
        }


pre_delete.connect(handlers.remove_obj_perms_connected_with_user, sender=Pull)


class Restream(models.Model):
    FORMATS = (
        ('flv', 'flv (RTMP)'),
        ('mpegts', 'mpegts (SRT)'),
    )
    stream = models.ForeignKey(Stream, on_delete=models.CASCADE, help_text=_('restream_stream_help'))
    target = models.CharField(max_length=500, validators=[validate_stream_url], help_text=_('restream_target_help'))
    name = models.CharField(max_length=100, help_text=_('restream_name_help'))
    active = models.BooleanField(help_text=_('restream_activate_help'))
    format = models.CharField(max_length=6, choices=FORMATS, default='flv', help_text=_('restream_format_help'))

    class Meta:
        verbose_name = _('restream_verbose_name')
        verbose_name_plural = _('restream_verbose_name_plural')

    def class_name(self):
        return _('restream_class_name')

    def get_absolute_url(self):
        return reverse('config:restream_detail', kwargs={'pk': self.pk})

    def __str__(self):
        return '{} to {}'.format(self.stream, self.name)

    def get_concierge_configuration(self):
        stream_instances = SRSStreamInstance.objects.filter(stream=self.stream).all()

        if not stream_instances:
            return None

        # We use the actual SRS node of the first stream instance to make sure
        # that we use the same node for the restreaming. This avoids unnecessary
        # traffic between the SRS nodes. In theory a clustered SRS setup could
        # provide the same stream on all edges, but that would mean east-west
        # traffic between the nodes.

        # This assumption breaks when a second stream instance appears on another
        # SRS node, and only afterwards the first stream instance is removed.
        # In that case, the restreaming would not be retriggered, and the edge
        # would automatically start pulling the stream from the other SRS node.

        # It's a tradeoff between stability and efficiency. We choose stability.
        rtmp_base = stream_instances[0].node.rtmp_base

        return {
            'config_version': 1,
            'stream_source_url': f"{rtmp_base}/{settings.GLOBAL_STREAM_NAMESPACE}/{self.stream.stream}",
            'stream_target_url': self.target,
            'stream_target_transport': self.format,
        }


pre_delete.connect(handlers.remove_obj_perms_connected_with_user, sender=Restream)