How to use Gstreamer to ingress live stream on Azure Media Services using RTMP in a long run?

Question

Cross-posting on Microsoft Q&A

With some inspiration from another post, I can successfully have a live stream on Azure Media Services Basic Pass-through Live Event using Gstreamer and RTMP.

However, it won't stay for long. I want to have the stream working 24/7, but after some hours, the preview and the egress output get more and more buffering/loading with many browser console 404 errors.

I suspect that it is because I restart the stream once every 6 hours? How can I fix that?

Here is my Gstreamer command:

gst-launch-1.0 v4l2src device=/dev/video0 ! video/x-raw, width=640, height=480, format=BGR \                                                                                                 
    ! vspmfilter dmabuf-use=true ! video/x-raw, format=NV12 \                                                                                                                                
    ! omxh264enc control-rate=2 target-bitrate=10485760 \                                                                                                                                    
    ! video/x-h264,profile=\(string\)high,level=\(string\)4.2 \                                                                                                                              
    ! h264parse config-interval=-1 \                                                                                                                                                         
    ! mux. audiotestsrc is-live=true wave=silence ! audio/x-raw,rate=48000 ! faac bitrate=96000 ! aacparse ! audio/mpeg, mpegversion=4! mux. flvmux streamable=true name=mux \               
    ! rtmpsink location="${dest} live=1 flashver=FMLE/3.0(compatibble;FMSc/1.0)" 

Answer 1

To maintain a 24/7 live stream on Azure Media Services using GStreamer and RTMP without experiencing increased buffering/loading times or browser console 404 errors, you might need to avoid restarting the stream every 6 hours. That method can introduce interruptions and potential resource allocation issues over time.

You can try and set the librtmp timeout to 0, as suggested in your original thread. But GStreamer's RTMP sink (rtmpsink) does not natively support automatic reconnection or error recovery. You will need to manage these aspects at a higher level, possibly with a custom script or application logic.

For long-term stability, consider adding a mechanism to monitor the pipeline's state and automatically attempt to reconnect if the pipeline fails. You would typically manage this through an external script or a GStreamer application written in Python or C, where you can catch errors and restart the pipeline as needed.

#!/bin/bash

while true; do
    gst-launch-1.0 your_pipeline_here
    echo "Stream stopped, restarting in 5 seconds..."
    sleep 5
done

You are contradicting yourself by saying "you might need to avoid restarting the stream" and then providing a suggestion script that uses GStreamer in a loop (which is restarting)

The intention behind suggesting a loop script was to make sure the stream could recover from failures automatically, not to contradict the advice against manual or scheduled restarts every 6 hours.
The key difference lies in the context: automated recovery from errors versus preemptive, time-based restarts.

---

Instead of a script that restarts the GStreamer pipeline for any failure, consider more sophisticated logic that attempts to reconnect or resolve issues without a full pipeline restart.
That could involve checking the status of the RTMP connection and only restarting components of the pipeline that are experiencing issues.

The following Python script incorporates dynamic adjustments and error handling, using GStreamer directly through gi.repository.Gst (as used here):

import gi
gi.require_version('Gst', '1.0')
from gi.repository import Gst, GObject
import time

# Initialize GStreamer
Gst.init(None)

class Streamer:
    def __init__(self, rtmp_server):
        self.pipeline = Gst.parse_launch(f"""
            appsrc name=source ! videoconvert ! video/x-raw,format=I420 !
            x264enc tune=zerolatency byte-stream=true threads=4 !
            flvmux name=mux streamable=true !
            rtmpsink location='{rtmp_server} live=true flashver=FMLE/3.0(compatibble;FMSc/1.0)' 
            audiotestsrc is-live=true ! audioconvert ! audioresample !
            audio/x-raw,rate=48000 ! voaacenc bitrate=96000 ! audio/mpeg !
            aacparse ! audio/mpeg, mpegversion=4 ! mux.
        """)
        self.bus = self.pipeline.get_bus()
        self.bus.add_signal_watch()
        self.bus.connect("message::eos", self.on_eos)
        self.bus.connect("message::error", self.on_error)

    def start(self):
        print("Starting stream")
        self.pipeline.set_state(Gst.State.PLAYING)

    def stop(self):
        print("Stopping stream")
        self.pipeline.set_state(Gst.State.NULL)

    def on_eos(self, bus, msg):
        print("End of stream")
        self.restart_stream()

    def on_error(self, bus, msg):
        err, debug = msg.parse_error()
        print(f"Error: {err}, {debug}")
        self.restart_stream()

    def restart_stream(self):
        print("Restarting stream...")
        self.stop()
        time.sleep(5)  # Wait a bit before restarting to handle transient issues
        self.start()

if __name__ == "__main__":
    # Replace 'rtmp_server' with your Azure RTMP server URL
    rtmp_server = 'rtmp://xxxx.media.azure.net:1935/live/xxxx/mystream'
    streamer = Streamer(rtmp_server)
    try:
        streamer.start()
        GObject.MainLoop().run()
    except KeyboardInterrupt:
        streamer.stop()
        print("Stream stopped")

By using GStreamer's Python bindings (gi.repository.Gst), you would benefit from direct access to GStreamer's capabilities, allowing for more complex logic and interaction with the streaming pipeline than what might be achievable with a simple command-line invocation or OpenCV's VideoWriter.

You can extend it with additional logic, such as dynamic source selection, quality adjustments based on network conditions, or integration with monitoring systems for advanced error detection and handling.