#!/usr/bin/env python3
from threading import Thread
import os
import sys
import time

import gi

gi.require_version("Gst", "1.0")
from gi.repository import Gst


Gst.init(sys.argv)
Gst.debug_set_default_threshold(Gst.DebugLevel.FIXME)
Gst.debug_set_threshold_for_name("ebmlread", Gst.DebugLevel.ERROR)

producer = Gst.parse_launch("videotestsrc ! x264enc key-int-max=30 tune=zerolatency "
                          "! matroskamux name=mux streamable=1 "
                          "! multifdsink name=fdsink units-max=100 sync-method=next-keyframe")
fdsink = producer.get_by_name("fdsink")

fdsink.connect("client-fd-removed", lambda _, fd: os.close(fd))

producer.set_name("producer")
producer.set_state(Gst.State.READY)


def mk_consumer(index):
    consumer = Gst.parse_launch("fdsrc name=fdsrc ! matroskademux ! fakesink")
    fdsrc = consumer.get_by_name("fdsrc")

    consumer.set_name("consumer" + str(index))

    pipe_read, pipe_write = os.pipe()
    fdsink.emit("add", pipe_write)
    fdsrc.set_property("fd", pipe_read)

    return consumer


consumers = [mk_consumer(i) for i in range(5)]


def setup_bus(pipeline):
    bus = pipeline.get_bus()

    def handler():
        while True:
            message = bus.timed_pop(Gst.CLOCK_TIME_NONE)
            if message.type == Gst.MessageType.EOS:
                print("Pipeline {} posted EOS".format(pipeline.name))
                break
            if message.type == Gst.MessageType.ERROR:
                print("Pipeline {} posted ERROR".format(pipeline.name))
                break
        pipeline.set_state(Gst.State.NULL)

    Thread(target=handler).start()


for p in (*consumers, producer):
    setup_bus(p)
    p.set_state(Gst.State.PLAYING)


def provoke(consumer):
    time.sleep(1)
    consumer.set_state(Gst.State.PAUSED)
    time.sleep(5)
    consumer.set_state(Gst.State.PLAYING)


provoke(consumers[2])

producer.send_event(Gst.Event.new_eos())
