GStreamer emits not-linked error even when elements are already linked

Viewed 89

I'm trying to create a capture-to-streaming pipeline with GStreamer but after creating the pipeline and linking all of the elements, I still get warnings that the source elements are not linked, even though the dot file for the pipeline shows that they are.

Code:

//TODO: Send all logs to tracing and filter from there instead of using gstreamer filter
//gst::debug_set_default_threshold(gst::DebugLevel::Count);
tracing_gstreamer::integrate_events();
gst::debug_remove_default_log_function();
gst::init()?;
tracing_gstreamer::integrate_spans();
*HANDLES_COUNT.lock().unwrap() += 1;

//Create a new GStreamer pipeline
let pipeline = gst::Pipeline::new(None);

//--VIDEO--

//Create a new ximagesrc to get video from the X server
let ximagesrc = gst::ElementFactory::make("ximagesrc", None)?;

let videoscale = gst::ElementFactory::make("videoscale", None)?;

//Creating a capsfilter to set the resolution and the fps
let capsfilter = gst::ElementFactory::make("capsfilter", None)?;

let fps_frac = gst::Fraction::new(fps, 1);

//Create a vector containing the option of the gst caps
let mut caps_options: Vec<(&str, &(dyn ToSendValue + Sync))> = vec![("framerate", &fps_frac)];

//If the resolution is specified, add it to the caps
let width = resolution.width as i32;
let height = resolution.height as i32;
if resolution.is_fixed {
    caps_options.push(("width", &width));
    caps_options.push(("height", &height));
};


capsfilter.set_property("caps", &gst::Caps::new_simple(
    "video/x-raw",
    caps_options.as_ref()
));

ximagesrc.set_property_from_str("show-pointer", "1");
//Set xid based on constructor parameter to get video only from the specified X window
ximagesrc.set_property("xid", xid as u64);

//Create a new videoconvert to allow encoding of the raw video
let videoconvert = gst::ElementFactory::make("videoconvert", None)?;

//Chose encoder and rtp encapsulator based on constructor params
let (encoder, encoder_pay) = match encoder_to_use {
    VideoEncoderType::H264(settings) => {
        (
            //Use nvidia encoder based on settings
            if settings.nvidia_encoder {
                gst::ElementFactory::make("nvh264enc", None)?
            } else {
                gst::ElementFactory::make("x264enc", None)?
            },
            gst::ElementFactory::make("rtph264pay", None)?
        )
    }
    VideoEncoderType::VP8 => {
        (
            gst::ElementFactory::make("vp8enc", None)?,
            gst::ElementFactory::make("rtpvp8pay", None)?
        )
    }
    VideoEncoderType::VP9 => {
        (
            gst::ElementFactory::make("vp9enc", None)?,
            gst::ElementFactory::make("rtpvp9pay", None)?
        )
    }
};

encoder_pay.set_property("ssrc", video_ssrc);


//--AUDIO--

// Caps filter for audio from conversion to encoding
let audio_capsfilter = gst::ElementFactory::make("capsfilter", None)?;

//Create a vector containing the option of the gst caps
let caps_options: Vec<(&str, &(dyn ToSendValue + Sync))> = vec![("channels", &2), ("rate", &48000)];

audio_capsfilter.set_property("caps", &gst::Caps::new_simple(
    "audio/x-raw",
    caps_options.as_ref()
));

//Create a new pulsesrc to get audio from the PulseAudio server
let pulsesrc = gst::ElementFactory::make("pulsesrc", None)?;
//Set the audio device based on constructor parameter (should be the sink of the audio application)
pulsesrc.set_property_from_str("device", "tuxphones.monitor");

//Create a new audioconvert to allow encoding of the raw audio
let audioconvert = gst::ElementFactory::make("audioconvert", None)?;
//Encoder for the raw audio to opus
let opusenc = gst::ElementFactory::make("opusenc", None)?;
//Opus encapsulator for rtp
let rtpopuspay = gst::ElementFactory::make("rtpopuspay", None)?;
rtpopuspay.set_property("ssrc", audio_ssrc);


//--DESTINATION--

//mux
let rtpmux = gst::ElementFactory::make("rtpmux", None)?;
rtpmux.set_property("ssrc", rtx_ssrc);
rtpmux.add_pad(&gst::GhostPad::new(Some("vsink"), gst::PadDirection::Sink))?;
rtpmux.add_pad(&gst::GhostPad::new(Some("asink"), gst::PadDirection::Sink))?;
let video_sink = rtpmux.static_pad("vsink").unwrap();
let audio_sink = rtpmux.static_pad("asink").unwrap();

//encryption
let srtpenc = gst::ElementFactory::make("srtpenc", None)?;
srtpenc.set_property_from_str("rtcp-cipher", encryption_algorithm.to_gst_str());
srtpenc.set_property("key", gst::Buffer::from_slice(key));
srtpenc.add_pad(&gst::GhostPad::new(Some("src"), gst::PadDirection::Src))?;

//Create a new webrtcbin to connect the pipeline to the WebRTC peer
let webrtcbin = gst::ElementFactory::make("webrtcbin", None)?;
webrtcbin.add_pad(&gst::GhostPad::new(Some("sink"), gst::PadDirection::Sink))?;

let mut sdp = SDPMessage::new();
sdp.set_connection("IN", "IP4", discord_address, 1, 1);

let webrtc_desc = WebRTCSessionDescription::new(
    WebRTCSDPType::Offer,
    sdp
);
webrtcbin.emit_by_name::<()>("set-remote-description", &[&webrtc_desc, &None::<gst::Promise>]);

//Add elements to the pipeline
pipeline.add_many(&[
    &ximagesrc, &videoscale, &capsfilter, &videoconvert, &encoder, &encoder_pay,
    &pulsesrc, &audioconvert, &audio_capsfilter, &opusenc, &rtpopuspay,
    &rtpmux, &srtpenc,
    &webrtcbin])?;

//Link video elements
Element::link_many(&[&ximagesrc, &videoscale, &capsfilter, &videoconvert, &encoder, &encoder_pay])?;

//Link audio elements
Element::link_many(&[&pulsesrc, &audioconvert, &audio_capsfilter, &opusenc, &rtpopuspay])?;

rtpmux.link(&srtpenc)?;
srtpenc.link(&webrtcbin)?;
//gst::Pad::link(&srtpenc.static_pad("rtp_src_0").unwrap(), &webrtcbin.static_pad("sink").unwrap())?;

//Link encoderpay to rtpmux video sink
gst::Pad::link(&encoder_pay.static_pad("src").unwrap(), &video_sink)?;

//Link rtpopuspay to rtpmux audio sink
gst::Pad::link(&rtpopuspay.static_pad("src").unwrap(), &audio_sink)?;

// Generate dot file (not shown)

After this is run, the pipeline is started immediately after:

self.pipeline.set_state(gst::State::Playing)

Log:

2022-07-01T16:23:02.640204Z  WARN gstreamer::basesrc: error: Internal data stream error. gobject.address=140187265069712 gobject.type="GstPulseSrc" gstobject.name="pulsesrc0" gstelement.state="playing" gstelement.pending_state="void-pending"
2022-07-01T16:23:02.640245Z  WARN gstreamer::basesrc: error: streaming stopped, reason not-linked (-1) gobject.address=140187265069712 gobject.type="GstPulseSrc" gstobject.name="pulsesrc0" gstelement.state="playing" gstelement.pending_state="void-pending"
2022-07-01T16:23:04.806453Z  WARN gstreamer::basesrc: error: Internal data stream error. gobject.address=140187264845024 gobject.type="GstXImageSrc" gstobject.name="ximagesrc0" gstelement.state="playing" gstelement.pending_state="void-pending"
2022-07-01T16:23:04.806520Z  WARN gstreamer::basesrc: error: streaming stopped, reason not-linked (-1) gobject.address=140187264845024 gobject.type="GstXImageSrc" gstobject.name="ximagesrc0" gstelement.state="playing" gstelement.pending_state="void-pending"

Rendered Dot File: GStreamer Pipeline

Rendered dot file after negotiation: GStreamer Pipeline Negotiated

What am I not doing correctly here?

1 Answers

Your code will probably work if you get rid of the rtpmux and srtpenc, and instead link the RTP payloaders directly to (two!) webrtcbin sink pads.

webrtcbin is doing everything after RTP payloading internally, including bundling (what you seem to try with rtpmux here) and SRTP encryption.

If you look at the debug logs you'll see that webrtcbin simply doesn't know what to do with the incoming SRTP stream and because of that doesn't link it further internally but instead just leaves it unlinked.

Related