dtlssrt example

#!/usr/bin/env python3

# create two endpoints, each with tx and rx pipelines using the DTLS
# elements and let audio flowing for a while to give time for a packet capture

import time
#gi.require_version('Gst','1.0')
from gi.repository import Gst, GObject, GLib
GObject.threads_init()
Gst.init(None)


def _start_pipeline(pipeline):
    pipeline.set_state(Gst.State.PLAYING)
    pipeline.get_state(Gst.CLOCK_TIME_NONE)


def _sleep_while_iterating_gloop(secs):
    """ block for secs seconds but iterate the gloop while you do """
    for _ in range(10 * secs):
        gloop = GLib.MainLoop()
        gloop_context = gloop.get_context()
        gloop_context.iteration(may_block=False)
        time.sleep(0.1)

def dtls_tx_pipeline_description(name, is_client, port):
    return ' ! '.join([
        'audiotestsrc is-live=true',
        'audio/x-raw, rate=8000, format=S16LE, channels=1',
        'opusenc frame-size=10',
        'rtpopuspay pt=103',
        '.rtp_sink_0 dtlssrtpenc connection-id={name} is-client={client} .src',
        'udpsink port={port}'
    ]).format(name=name, client=is_client, port=port)


def dtls_rx_pipeline_description(name, port):
    return ' ! '.join([
        'udpsrc port={port}',
        '.sink dtlssrtpdec connection-id={name} .rtp_src',
        'queue',
        'fakesink async=false'
    ]).format(name=name, port=port)


class Endpoint:
    def __init__(self, name, is_client, tx_port, rx_port):
        self.name = name
        tx_pipeline_description = dtls_tx_pipeline_description(
            name, is_client, tx_port
        )
        rx_pipeline_description = dtls_rx_pipeline_description(name, rx_port)
        print(rx_pipeline_description)
        print(tx_pipeline_description)

        self.rx_pipeline = Gst.parse_launch(rx_pipeline_description)
        self.tx_pipeline = Gst.parse_launch(tx_pipeline_description)

    def start(self):
				# Start RX first, otherwise it fails due to the current implementation
        self.start_rx_pipeline()
        self.start_tx_pipeline()

    def start_rx_pipeline(self):
        _start_pipeline(self.rx_pipeline)

    def start_tx_pipeline(self):
        _start_pipeline(self.tx_pipeline)

    def stop(self):
        def stop_pipeline(p):
            p.set_state(Gst.State.NULL)
            p.get_state(Gst.CLOCK_TIME_NONE)
        stop_pipeline(self.tx_pipeline)
        stop_pipeline(self.rx_pipeline)

blue = Endpoint("blue", is_client=True, tx_port=23000, rx_port=23002)
red = Endpoint("red", is_client=False, tx_port=23002, rx_port=23000)
print('Start')

red.start()
blue.start()

_sleep_while_iterating_gloop(3)

red.stop()
blue.stop()
print('End')