Handling errors with gst-rtsp-server Python bindings

Question

I have a simple Python program creates an RTSP stream using gst-rtsp-server. It works, but as-is there's no error handling. If the pipeline has a typo or there's some issue connecting to the video source, I don't see a stack trace or any logging. Where would I hook in code to handle problems like this?

I should mention that I'm a complete beginner to the GObject world. I suspect there is a standard way for these libraries to report errors but I haven't been able to find anything in the documentation I've read about how that's done.

In case it's helpful, here is my code as I have it right now:

from threading import Thread
from time import sleep
import signal

import gi
gi.require_version("Gst", "1.0")
gi.require_version("GstRtsp", "1.0")
gi.require_version("GstRtspServer", "1.0")
from gi.repository import GLib, GObject, Gst, GstRtsp, GstRtspServer

PIPELINE = (
    "( videotestsrc ! vp8enc ! rtpvp8pay name=pay0 pt=96 )")


def main():
    GObject.threads_init()
    Gst.init(None)

    server = GstRtspServer.RTSPServer.new()
    server.props.service = "3000"

    server.attach(None)

    loop = GLib.MainLoop.new(None, False)

    def on_sigint(_sig, _frame):
        print("Got a SIGINT, closing...")
        loop.quit()
    signal.signal(signal.SIGINT, on_sigint)

    def run_main_loop():
        loop.run()

    main_loop_thread = Thread(target=run_main_loop)

    main_loop_thread.start()

    media_factory = GstRtspServer.RTSPMediaFactory.new()
    media_factory.set_launch(PIPELINE)
    media_factory.set_shared(True)
    server.get_mount_points().add_factory("/test", media_factory)
    print("Stream ready at rtsp://127.0.0.1:3000/test")


    while loop.is_running():
        sleep(0.1)


if __name__ == "__main__":
    main()

Answer 1

So you can override do_handle_message in Gst.Bin in the following way:

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

class SubclassBin(Gst.Bin):
    def do_handle_message(self, message):
        if message.type == Gst.MessageType.ERROR:
            error, message = message.parse_error()
            # TODO: Do something with the error
        # Call the base Gst.Bin do_handle_message
        super().do_handle_message(self, message)

subclass_bin = SubclassBin("mybin")

That said, I'm not sure how to tell GstRtspServer.RTSPMediaFactory to use SubclassBin instead of Gst.Bin because as far as I can tell, the only way to connect any Bin to RTSPMediaFactory is through the set_launch method which wants a pipeline string instead of a prebuilt bin. If there was a way of adding a prebuilt bin to RTSPMediaFactory, this would be a complete answer...but unfortunately this is as far as I can get.

Answer 2

Edit: I would recommend using bkanuka's solution (Subclassing Gst.Bin) instead of the below one, which has many drawbacks.

After some more experimentation with the GStreamer and the RTSP server library, the error handling situation is complicated.

The Normal Way

The canonical way to see errors on a GStreamer pipeline is to add a watcher to the pipeline's bus and listen for error messages.

def watcher(bus, message, *user_data);
    if message.type == Gst.MessageType.ERROR:
        error, message = message.parse_error()
        # TODO: Do something with the error

my_pipeline.get_bus().add_watch(
    GLib.PRIORITY_DEFAULT,
    watcher,
    None)

However, you cannot do this with pipelines you supply to a GstRtspServer. This is because GstRtspServer expects to be able to install its own watcher on the pipeline's bus and only one watcher can be attached to a bus at once. This is especially unfortunate because this prevents us from listening to any events on the pipeline, not just errors.

My Workaround

We can split the pipeline into two pieces: one that's in charge of the error-prone process of connecting to the source and decoding frames, and another that is in charge of encoding the resulting frames and payloading them for the GstRtspServer. We can then use the intervideo plugin to communicate between the two.

For example, let's say you're trying to stream from a file in the VP8 format. Our first pipeline that's in charge of reading and decoding frames would look like this:

filesrc location="{filepath}" ! decodebin ! intervideosink channel="file-channel"

... and our second pipeline that's in charge of encoding and payloading the frame would look like this:

intervideosrc channel="file-channel" ! videoconvert ! vp8enc deadline=1 ! rtpvp8pay name=pay0 pt=96

The key here is that only the second pipeline has to be managed by GstRtspServer, since it's the pipeline that provides the payloaded data. The first one is managed by us and we can attach our own watcher to it that responds intelligently to errors and does whatever else we need. This of course is not a perfect solution because we aren't able to respond to errors related to encoding and payloading, but we've gained the ability to receive errors related to reading the file and decoding it. We're now also able to do other message-related tasks, like intercepting an end-of-stream message to loop the video file.