gst-plugin-webrtc

https://docs.rs/gst-plugin-webrtc/0.15.0/
Admin
A GStreamer plugin that provides WebRTC capabilities, including WebRTC sink and source elements for...
Tokens:477,583
Snippets:2,526
Trust Score:9.1
Update:5 months ago
Show doc for...
Tokens:
Raw
### LiveKit WebRTC Source Example

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/src/gstrswebrtc/webrtcsrc/mod.rs.html

Example of using `livekitwebrtcsrc` to receive a video stream. Configure signaller details to connect to a room.

```shell
gst-launch-1.0 \
    livekitwebrtcsrc \
      signaller::ws-url=ws://127.0.0.1:7880 \
      signaller::api-key=devkey \
      signaller::secret-key=secret \
      signaller::room-name=testroom \
      signaller::identity=gst-consumer \
      signaller::participant-name=gst-consumer \
    ! queue ! videoconvert ! autovideosink
```

--------------------------------

### Get Start Bitrate Property

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/src/gstrswebrtc/webrtcsink/imp.rs.html

Retrieves the 'start-bitrate' property from the settings.

```rust
            "start-bitrate" => {
                let settings = self.settings.lock().unwrap();

```

--------------------------------

### LiveKit WebRTC Sink Example

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/src/gstrswebrtc/webrtcsrc/mod.rs.html

Example of using `livekitwebrtcsink` to send a video stream. Configure signaller details and video caps.

```shell
gst-launch-1.0 \
    videotestsrc is-live=1 \
    ! video/x-raw,width=640,height=360,framerate=15/1 \
    ! timeoverlay ! videoconvert ! queue \
    ! livekitwebrtcsink name=sink \
        signaller::ws-url=ws://127.0.0.1:7880 \
        signaller::api-key=devkey \
        signaller::secret-key=secret \
        signaller::room-name=testroom \
        signaller::identity=gst-producer \
        signaller::participant-name=gst-producer \
        video-caps='video/x-vp8'
```

--------------------------------

### Setup and Connect Input Stream

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/src/gstrswebrtc/webrtcsink/imp.rs.html

Sets up the payloading chain for a given stream and connects its producer to the WebRTC pad. Handles errors during setup and connection, marking the session for removal if failures occur.

```rust
match state
                .streams
                .get(stream_name)
                .and_then(|stream| stream.producer.clone()) {
                Some(producer) => {
                    drop(state_guard);

                    let peer_id = session.peer_id.clone();
                    let session_codecs = session.codecs.clone().unwrap_or_else(|| codecs.clone());
                    let pipeline = session.pipeline.clone();

                    drop(session);

                    let res = match self.setup_session_payloading_chain(
                        &peer_id,
                        webrtc_pad,
                        &session_codecs,
                        &sdp,
                        &pipeline,
                    ) {
                        Err(err) => {
                            gst::error!(
                                CAT,
                                imp = self,
                                "Failed to setup elements {} for session {}: {}",
                                stream_name,
                                session_id,
                                err
                            );
                            remove = true;
                            state_guard = self.state.lock().unwrap();
                            session = session_clone.lock().unwrap();
                            state = state_guard.deref_mut();
                            break;
                        }
                        Ok(Some(res)) => res,
                        Ok(None) => {
                            state_guard = self.state.lock().unwrap();
                            session = session_clone.lock().unwrap();
                            state = state_guard.deref_mut();
                            continue;
                        }
                    };

                    state_guard = self.state.lock().unwrap();
                    session = session_clone.lock().unwrap();
                    state = state_guard.deref_mut();

                    if let Err(err) = 
                        session.connect_input_stream(&self.obj(), &producer, webrtc_pad, res) 
                    {
                        gst::error!(
                            CAT,
                            imp = self,
                            "Failed to connect input stream {} for session {}: {}",
                            stream_name,
                            session.id,
                            err
                        );
                        remove = true;
                        break;
                    }
                }
                _ => {
                    if !removed_streams.contains(stream_name) {
                        gst::error!(
                            CAT,
                            imp = self,
                            "No producer to connect session {} to",
                            session.id,
                        );
                        remove = true;
                        break;
                    }
                }
            }
```

--------------------------------

### Slice Length Example

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/gstrswebrtc/utils/struct.Codecs.html

Demonstrates how to get the number of elements in a slice.

```rust
let a = [1, 2, 3];
assert_eq!(a.len(), 3);
```

--------------------------------

### Start WhepServer

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/src/gstrswebrtc/whep_signaller/server.rs.html

Initiates the WhepServer by starting its asynchronous serving task and storing the handle for potential future management.

```rust
fn start(&self) {
        gst::info!(CAT, imp = self, "starting the WHEP server");
        let jh = self.serve();
        let mut settings = self.settings.lock().unwrap();
        settings.server_handle = jh;
    }
```

--------------------------------

### Janus VR WebRTC Source Example

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/src/gstrswebrtc/webrtcsrc/mod.rs.html

Example of using `janusvrwebrtcsrc` to receive a video stream from a Janus Gateway room. Requires room ID, producer peer ID, and Janus endpoint.

```bash
$ gst-launch-1.0 janusvrwebrtcsrc signaller::room-id=1234 signaller::producer-peer-id=777 signaller::janus-endpoint=wss://janus.conf.meetecho.com/ws ! videoconvert ! autovideosink
```

--------------------------------

### Janus VR WebRTC Sink Example

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/src/gstrswebrtc/webrtcsrc/mod.rs.html

Example of using `janusvrwebrtcsink` to send a video stream to a Janus Gateway room. Requires room ID, feed ID, and Janus endpoint.

```bash
$ gst-launch-1.0 videotestsrc ! janusvrwebrtcsink signaller::room-id=1234 signaller::feed-id=777 signaller::janus-endpoint=wss://janus.conf.meetecho.com/ws
```

--------------------------------

### Vector Length Example

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/gstrswebrtc/utils/struct.Codecs.html

Demonstrates how to get the number of elements in a vector.

```rust
let a = vec![1, 2, 3];
assert_eq!(a.len(), 3);
```

--------------------------------

### start Signal

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/src/gstrswebrtc/signaller/iface.rs.html

Starts the signaller, connecting it to the signalling server.

```APIDOC
## start Signal

### Description
Starts the signaller, connecting it to the signalling server.

### Response
No specific response defined for this signal.
```

--------------------------------

### Start WHIP Server Task

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/src/gstrswebrtc/whip_signaller/imp.rs.html

Starts the WHIP server task and stores its handle. This is typically called when the server is initialized.

```rust
fn start(&self) {
        gst::info!(CAT, imp = self, "starting the WHIP server");
        let jh = self.serve();
        let mut settings = self.settings.lock().unwrap();
        settings.server_handle = jh;
    }
```

--------------------------------

### Start Bitrate Property

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/src/gstrswebrtc/webrtcsink/imp.rs.html

Sets the initial bitrate for the connection.

```APIDOC
## Property: start-bitrate

### Description
Sets the initial bitrate for the connection.

### Type
`UINT`

### Settable
Yes
```

--------------------------------

### SignallableExt Implementation: Start and Stop

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/src/gstrswebrtc/signaller/iface.rs.html

Implements the `start` and `stop` methods for the `SignallableExt` trait by emitting corresponding signals.

```rust
impl<Obj: IsA<super::Signallable>> SignallableExt for Obj {
    fn start(&self) {
        self.emit_by_name::<bool>("start", &[]);
    }

    fn stop(&self) {
        self.emit_by_name::<bool>("stop", &[]);
    }

```

--------------------------------

### Initiate WHEP Session Start

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/src/gstrswebrtc/whep_signaller/client.rs.html

Starts the WHEP signaller by setting the initial state to 'Post' and emitting 'session-started' and 'session-requested' signals. It ensures the WHEP endpoint URL is configured before proceeding.

```rust
        if self.settings.lock().unwrap().whep_endpoint.is_none() {
            self.raise_error("WHEP endpoint URL must be set".to_string());
            return;
        }

        let mut state = self.state.lock().unwrap();
        *state = State::Post { redirects: 0 };
        drop(state);

        let this_weak = self.downgrade();

        RUNTIME.spawn(async move {
            if let Some(this) = this_weak.upgrade() {
                this.obj().emit_by_name::<()>( 
                    "session-started", 
                    &[&WHEP_CLIENT_OFFER, &WHEP_CLIENT_OFFER],
                );

                this.obj().emit_by_name::<()>( 
                    "session-requested", 
                    &[

```

--------------------------------

### Start Stream Discovery if Needed

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/src/gstrswebrtc/webrtcsink/imp.rs.html

Initiates stream discovery if it hasn't started yet. It determines codecs and creates discovery information, then spawns an asynchronous task to look up caps.

```rust
fn start_stream_discovery_if_needed(&self, stream_name: &str) {
        let (codecs, discovery_info) = {
            let mut state = self.state.lock().unwrap();

            let discovery_info = {
                let stream = state.streams.get_mut(stream_name).unwrap();

                // Initial discovery already happened... nothing to do here.
                if stream.initial_discovery_started {
                    return;
                }

                stream.initial_discovery_started = true;

                stream.create_discovery()
            };

            let codecs = if !state.codecs.is_empty() {
                Codecs::from_map(&state.codecs)
            } else {
                drop(state);
                let settings = self.settings.lock().unwrap();
                let codecs = Codecs::list_encoders_and_payloaders(
                    settings.video_caps.iter().chain(settings.audio_caps.iter()),
                );

                state = self.state.lock().unwrap();
                state.codecs = codecs.to_map();
                codecs
            };

            (codecs, discovery_info)
        };

        let stream_name_clone = stream_name.to_owned();
        RUNTIME.spawn(glib::clone!(
            #[to_owned(rename_to = this)]
            self,
            #[strong]
            discovery_info,
            async move {
                let (fut, handle) = futures::future::abortable(this.lookup_caps(
                    discovery_info.clone(),
                    stream_name_clone.clone(),
                    gst::Caps::new_any(),
                    &codecs,
                ));

                let (codecs_done_sender, codecs_done_receiver) =
                    futures::channel::oneshot::channel();

                // Compiler isn't budged by dropping state before await, 
                // so let's make a new scope instead.
                {
                    let mut state = this.state.lock().unwrap();
                    state.codecs_abort_handles.push(handle);
                    state.codecs_done_receivers.push(codecs_done_receiver);
                }

                match fut.await {
                    Ok(Err(err)) => {
                        gst::error!(CAT, imp = this, "Error running discovery: {err:?}");
                        gst::element_error!(
                            this.obj(),
                            gst::StreamError::CodecNotFound,
                            ["Failed to look up output caps: {err:?}"]
                        );
                    }
                    Ok(Ok(_)) => {
                        let settings_clone = this.settings.lock().unwrap().clone();

                        let (sessions, stream) = {
                            let mut state = this.state.lock().unwrap();

                            state.codec_discovery_done = state
                                .streams
                                .values()
                                .all(|stream| stream.out_caps.is_some());

                            (
                                state.sessions.clone(),
                                state.streams.get(&stream_name_clone).cloned(),
                            )
                        };

                        if let Some(stream) = stream {
                            for session in sessions.values() {
                                gst::debug!(
                                    CAT,
                                    imp = this,
                                    "renegotiating session {} from discovery done",
                                    session.0.lock().unwrap().id
                                );
                                this.renegotiate(&settings_clone, session, vec![stream.clone()])
                                    .await;
                            }
                        }

                        let mut state = this.state.lock().unwrap();

                        if state.should_start_signaller(&this.obj()) {
                            state.signaller_state = SignallerState::Started;
                        }
                    }
                }
            }
        ));
    }
```

--------------------------------

### SignallableExt::start Method

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/gstrswebrtc/signaller/trait.SignallableExt.html

Initiates the WebRTC signaling process for a given session.

```rust
fn start(&self);
```

--------------------------------

### Setting Up Decodebin3 for Decoding

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/src/gstrswebrtc/webrtcsrc/imp.rs.html

Initializes and configures a 'decodebin3' element for media decoding. It sets the downstream capabilities and connects a pad-added signal handler to manage the ghost pad.

```rust
let decodebin = gst::ElementFactory::make("decodebin3")
                    .build()
                    .expect("decodebin3 needs to be present!");

                let downstream_caps = srcpad.peer_query_caps(None);
                gst::debug!(
                    CAT,
                    obj = element,
                    "Stopping decoding at caps {downstream_caps:#?}"
                );

                decodebin.set_property("caps", &downstream_caps);

                bin.add(&decodebin).unwrap();
                decodebin.sync_state_with_parent().unwrap();
                decodebin.connect_pad_added(glib::clone!(
                    #[weak]
                    ghostpad,
                    move |_webrtcbin, pad| {
                        if pad.direction() == gst::PadDirection::Sink {
                            return;
                        }

                        ghostpad.set_target(Some(pad)).unwrap();
                    }
                ));

                gst::debug!(
                    CAT,
                    obj = element,
                    "Decoding for {}",
                    srcpad.imp().stream_id()
                );
```

--------------------------------

### Iterate over mutable slice chunks from the end (rchunks_mut)

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/gstrswebrtc/utils/struct.Codecs.html

Use `rchunks_mut` to get an iterator over mutable slices of a specified size, starting from the end. This is useful for in-place modifications.

```rust
let v = &mut [0, 0, 0, 0, 0];
let mut count = 1;

for chunk in v.rchunks_mut(2) {
    for elem in chunk.iter_mut() {
        *elem += count;
    }
    count += 1;
}
assert_eq!(v, &[3, 2, 2, 1, 1]);
```

--------------------------------

### Iterate over exact mutable slice chunks from the end (rchunks_exact_mut)

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/gstrswebrtc/utils/struct.Codecs.html

Use `rchunks_exact_mut` to get an iterator over mutable slices of a specified size, starting from the end. The remainder can be accessed via `into_remainder()`.

```rust
let v = &mut [0, 0, 0, 0, 0];
let mut count = 1;

for chunk in v.rchunks_exact_mut(2) {
    for elem in chunk.iter_mut() {
        *elem += count;
    }
    count += 1;
}
assert_eq!(v, &[0, 2, 2, 1, 1]);
```

--------------------------------

### SignallableImpl Start Method

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/src/gstrswebrtc/janusvr_signaller/imp.rs.html

Initiates the signaling process by connecting to the Janus server and creating a session. Includes a check for producer peer ID in Consumer role.

```rust
fn start(&self) {
        {
            let settings = self.settings.lock().unwrap();

            if let (WebRTCSignallerRole::Consumer, None) =
                (&settings.role, &settings.producer_peer_id)
            {
                panic!("producer-peer-id should be set in Consumer role");
            }
        }

        let this = self.obj().clone();
        let imp = self.downgrade();
        RUNTIME.spawn(async move {
            if let Some(imp) = imp.upgrade() {
                match imp.connect().await {
                    Err(err) => {
                        this.emit_by_name::<()>( "error", &[&format!("{:?}", anyhow!(err))]);
                    }
                    _ => {
                        imp.create_session();
                    }
                }
            }
        });
    }
```

--------------------------------

### Iterate over exact slice chunks from the end (rchunks_exact)

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/gstrswebrtc/utils/struct.Codecs.html

Use `rchunks_exact` to get an iterator over mutable slices of a specified size, starting from the end. Elements that do not form a full chunk are available via `remainder()`.

```rust
let slice = ['l', 'o', 'r', 'e', 'm'];
let mut iter = slice.rchunks_exact(2);
assert_eq!(iter.next().unwrap(), &['e', 'm']);
assert_eq!(iter.next().unwrap(), &['o', 'r']);
assert!(iter.next().is_none());
assert_eq!(iter.remainder(), &['l']);
```

--------------------------------

### Iterate over slice chunks from the end (rchunks)

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/gstrswebrtc/utils/struct.Codecs.html

Use `rchunks` to get an iterator over mutable slices of a specified size, starting from the end. The last chunk may be smaller if the slice length is not divisible by `chunk_size`.

```rust
let slice = ['l', 'o', 'r', 'e', 'm'];
let mut iter = slice.rchunks(2);
assert_eq!(iter.next().unwrap(), &['e', 'm']);
assert_eq!(iter.next().unwrap(), &['o', 'r']);
assert_eq!(iter.next().unwrap(), &['l']);
assert!(iter.next().is_none());
```

--------------------------------

### Get Element Offset in Slice

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/gstrswebrtc/utils/struct.Codecs.html

Use `element_offset` to find the index of a given element reference within a slice. This method relies on pointer arithmetic and does not compare element values. It returns `None` if the element reference is not aligned with the start of an element.

```rust
let nums: &[u32] = &[1, 7, 1, 1];
let num = &nums[2];

assert_eq!(num, &1);
assert_eq!(nums.element_offset(num), Some(2));
```

```rust
let arr: &[[u32; 2]] = &[[0, 1], [2, 3]];
let flat_arr: &[u32] = arr.as_flattened();

let ok_elm: &[u32; 2] = flat_arr[0..2].try_into().unwrap();
let weird_elm: &[u32; 2] = flat_arr[1..3].try_into().unwrap();

assert_eq!(ok_elm, &[0, 1]);
assert_eq!(weird_elm, &[1, 2]);

assert_eq!(arr.element_offset(ok_elm), Some(0)); // Points to element 0
assert_eq!(arr.element_offset(weird_elm), None); // Points between element 0 and 1
```

--------------------------------

### From

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/gstrswebrtc/webrtcsink/struct.WebRTCSinkPad.html

Provides a way to create a WebRTCSinkPad from itself.

```APIDOC
## fn from(t: T) -> T

Returns the argument unchanged.

### Parameters
- **t**: The value to convert.

### Returns
- `T`: The unchanged value.
```

--------------------------------

### Initialize Logging

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/src/gstrswebrtc/webrtcsink/imp.rs.html

Sets up the global tracing subscriber with environment filtering and a formatted layer. This should be called once at the start of the application.

```rust
fn initialize_logging(env_var: &str) -> Result<(), Error> {
    let env_filter = tracing_subscriber::EnvFilter::try_from_default_env()
        .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info"));
    let fmt_layer = tracing_subscriber::fmt::layer()
        .with_thread_ids(true)
        .with_target(true)
        .with_span_events(
            tracing_subscriber::fmt::format::FmtSpan::NEW
                | tracing_subscriber::fmt::format::FmtSpan::CLOSE,
        );
    let subscriber = tracing_subscriber::Registry::default()
        .with(env_filter)
        .with(fmt_layer);
    tracing::subscriber::set_global_default(subscriber)?;

    Ok(())
}
```

--------------------------------

### Property Get Operations

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/gstrswebrtc/signaller/struct.Signallable.html

Operations for getting property values.

```APIDOC
## get<R, F>

### Description
Get the value of a property using a closure.

### Method
`fn get<R, F>(&self, f: F) -> R`
where F: Fn(&<T as PropertyGet>::Value) -> R
```

--------------------------------

### Prepare WebRTC Sink

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/src/gstrswebrtc/webrtcsink/imp.rs.html

Initializes the WebRTC sink, optionally spawning a web server if the 'web_server' feature is enabled and configured.

```rust
fn prepare(&self) -> Result<(), Error> {
        gst::debug!(CAT, imp = self, "preparing");

        #[cfg(feature = "web_server")]
        {
            let settings = self.settings.lock().unwrap();

            if settings.run_web_server {
                let mut state = self.state.lock().unwrap();
                let (web_shutdown_tx, web_join_handle) = 
                    BaseWebRTCSink::spawn_web_server(&settings)?;
                state.web_shutdown_tx = Some(web_shutdown_tx);
                state.web_join_handle = Some(web_join_handle);
            }
        }

        Ok(())
    }
```

--------------------------------

### Start Signaller if Conditions Met

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/src/gstrswebrtc/webrtcsrc/imp.rs.html

Starts the WebRTC signaller if it is currently stopped and the GStreamer element is in the Paused state or higher. Logs a message indicating the signaller has started.

```rust
fn maybe_start_signaller(&self) {
        let obj = self.obj();
        let mut state = self.state.lock().unwrap();
        if state.signaller_state == SignallerState::Stopped
            && obj.current_state() >= gst::State::Paused
        {
            self.signaller().start();

            gst::info!(CAT, imp = self, "Started signaller");
            state.signaller_state = SignallerState::Started;
        }
    }
```

--------------------------------

### Check if Signaller Should Start

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/src/gstrswebrtc/webrtcsink/imp.rs.html

Determines if the signaller should be started based on its current state, the element's state, and whether codec discovery is complete. This ensures the signaller starts only when conditions are met.

```rust
fn should_start_signaller(&mut self, element: &super::BaseWebRTCSink) -> bool {
        self.signaller_state == SignallerState::Stopped
            && element.current_state() >= gst::State::Paused
            && self.codec_discovery_done
    }
}
```

--------------------------------

### GstRSWebRTCSignallableIface: start Signal

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/src/gstrswebrtc/signaller/iface.rs.html

Signal to start the signaller and connect it to the signalling server. It uses run_last semantics.

```rust
Signal::builder("start")
                    .run_last()

```

--------------------------------

### Prepare Input Stream

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/src/gstrswebrtc/webrtcsink/imp.rs.html

Prepares a new input stream by calling its `prepare` method. If preparation fails, the sink pad is removed and deactivated, and an error is logged.

```rust
if let Err(err) = input_stream.prepare(&element) {
            element.remove_pad(&sink_pad).unwrap();
            sink_pad.set_active(false).unwrap();
            gst::error!(CAT, "failed to prepare input stream: {err:?}");
            return None;
        }
```

--------------------------------

### Generate Offer and Set Local Description

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/src/gstrswebrtc/webrtcsrc/imp.rs.html

Handles the creation of an SDP offer, setting it as the local description, and sending it via the signaller. Errors during offer generation are logged and handled by ending the session.

```rust
match reply
                    .value("offer")
                    .map(|offer| offer.get::<gst_webrtc::WebRTCSessionDescription>().unwrap())
                {
                    Ok(offer_sdp) => {
                        webrtcbin.emit_by_name::<()>(
                            "set-local-description",
                            &[&offer_sdp, &None::<gst::Promise>],
                        );

                        gst::log!(
                            CAT,
                            obj = ele,
                            "Sending SDP, {}",
                            offer_sdp.sdp().to_string()
                        );

                        let signaller = ele.imp().signaller();
                        signaller.send_sdp(sess_id.as_str(), &offer_sdp);
                    }
                    _ => {
                        let error = reply
                            .value("error")
                            .expect("structure must have an error value")
                            .get::<glib::Error>()
                            .expect("value must be a GLib error");

                        gst::error!(
                            CAT,
                            obj = ele,
                            "generate offer::Promise returned with error: {}",
                            error
                        );
                        ele.imp().signaller().end_session(sess_id.as_str());
                    }
                }
            }
        }));

        webrtcbin
            .clone()
            .emit_by_name::<()]("create-offer", &[&None::<gst::Structure>, &promise]);
```

--------------------------------

### Flush Source Pads

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/src/gstrswebrtc/webrtcsrc/imp.rs.html

Iterates over all source pads and pushes a `FlushStart` event to each one. This is crucial to prevent deadlocks when removing sessions.

```rust
let obj = self.obj();

        for pad in obj.src_pads().iter() {
            if !pad.push_event(gst::event::FlushStart::new()) {
                return Err(anyhow::anyhow!(
                    "Failed to send flush-start for pad: "
                    ,
                    pad.name()
                ));
            }
        }
```

--------------------------------

### Load Root Certificates from File

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/src/gstrswebrtc/utils.rs.html

Asynchronously reads and parses certificates from a file to create a RootCertStore. Supports PEM format.

```rust
async fn get_root_certstore<P: AsRef<Path>>(
    cafile_path: P,
) -> Result<rustls::RootCertStore, std::io::Error> {
    let mut root_cert_store = rustls::RootCertStore::empty();

    let certs = {
        use rustls_pki_types::pem::PemObject;

        let cert_file = tokio::fs::read(&cafile_path).await?;
        let cert_iter = rustls_pki_types::CertificateDer::pem_slice_iter(&cert_file);
        cert_iter
            .into_iter()
            .map(|c| c.unwrap())
            .collect::<Vec<_>>()
    };

    let _ = root_cert_store.add_parsable_certificates(certs);

    Ok(root_cert_store)
}
```

--------------------------------

### Vector Capacity Example

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/gstrswebrtc/utils/struct.Codecs.html

Demonstrates how to check the capacity of a vector. Note that vectors with zero-sized elements have a capacity of usize::MAX.

```rust
let mut vec: Vec<i32> = Vec::with_capacity(10);
vec.push(42);
assert!(vec.capacity() >= 10);
```

```rust
#[derive(Clone)]
struct ZeroSized;

fn main() {
    assert_eq!(std::mem::size_of::<ZeroSized>(), 0);
    let v = vec![ZeroSized; 0];
    assert_eq!(v.capacity(), usize::MAX);
}
```

--------------------------------

### Get Mutable Element or Subslice

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/gstrswebrtc/utils/struct.Codecs.html

Safely gets a mutable reference to an element or subslice. Returns `None` if the index is out of bounds.

```rust
let x = &mut [0, 1, 2];

if let Some(elem) = x.get_mut(1) {
    *elem = 42;
}
assert_eq!(x, &[0, 42, 2]);
```

--------------------------------

### Start WebRTC Session

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/src/gstrswebrtc/webrtcsrc/imp.rs.html

Initiates a new WebRTC session. It prevents multiple sessions and configures the WebRTC bin with STUN/TURN servers from the plugin's settings.

```rust
fn start_session(&self, session_id: &str) -> Result<(), Error> {
        let state = self.state.lock().unwrap();

        if state.session.is_some() {
            return Err(anyhow::anyhow!("webrtcsrc only supports a single session"));
        };
        drop(state);

        let session = SessionInner::new(session_id)?;

        let webrtcbin = session.webrtcbin();

        {
            let settings = self.settings.lock().unwrap();

            if let Some(stun_server) = settings.stun_server.as_ref() {
                webrtcbin.set_property("stun-server", stun_server);
            }

            for turn_server in settings.turn_servers.iter() {
                webrtcbin.emit_by_name::<bool>("add-turn-server", &[&turn_server]);
            }
        }

        let bin = gst::Bin::new();

        bin.connect_pad_removed(glib::clone!(
            #[weak(rename_to = this)]
            self,
            move |_, pad| {
                let mut state = this.state.lock().unwrap();
                if let Some(ref mut session) = state.session {
                    let session = session.0.lock().unwrap();
                    session.flow_combiner.lock().unwrap().remove_pad(pad);
                };
            }
        ));
        bin.connect_pad_added(glib::clone!(
            #[weak(rename_to = this)]
            self,
            move |_, pad| {
                let mut state = this.state.lock().unwrap();
                if let Some(ref mut session) = state.session {
                    let session = session.0.lock().unwrap();
                    session.flow_combiner.lock().unwrap().add_pad(pad);
                };
            }
        ));

        webrtcbin.connect_pad_added(glib::clone!(
            #[weak(rename_to = this)]
            self,
            #[weak]
            bin,
            move |_webrtcbin, pad| {
                if pad.direction() == gst::PadDirection::Sink {
                    return;
                }
                let mut state = this.state.lock().unwrap();
                let Some(ref mut session) = state.session else {
                    gst::error!(CAT, imp = this, "no current session");
                    return;
                };
                let bin_ghostpad =
                    session
                        .0
                        .lock()
                        .unwrap()
                        .handle_webrtc_src_pad(&bin, pad, &this.obj());
                drop(state);
                bin.add_pad(&bin_ghostpad)
                    .expect("Adding ghostpad to the bin should always work");
            }
        ));

        webrtcbin.connect_pad_removed(glib::clone!(
            #[weak(rename_to = this)]
            self,
            move |_webrtcbin, pad| {
                let mut state = this.state.lock().unwrap();
                if let Some(ref mut session) = state.session {
                    let session = session.0.lock().unwrap();
                    session.flow_combiner.lock().unwrap().remove_pad(pad);
                };
            }
        ));

        webrtcbin.connect_closure(
            "on-ice-candidate",
            false,
            glib::closure!(
                #[weak(rename_to = this)]
                self,
                move |_webrtcbin: gst::Bin, sdp_m_line_index: u32, candidate: String| {
                    let mut state = this.state.lock().unwrap();
                    let Some(ref mut session) = state.session else {
                        gst::error!(CAT, imp = this, "no current session");
                        return;
                    };
                    let session = session.0.lock().unwrap();
                    session.on_ice_candidate(sdp_m_line_index, candidate, &this.obj());
                }
            ),
        );

        webrtcbin.connect_closure(
            "on-data-channel",
            
```

--------------------------------

### Stop Signaller if Started

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/src/gstrswebrtc/webrtcsrc/imp.rs.html

Conditionally stops the signaller if its current state is 'Started'. Ensures the signaller is properly shut down before any state changes.

```rust
fn maybe_stop_signaller(&self) {
        let mut state = self.state.lock().unwrap();
        if state.signaller_state == SignallerState::Started {
            state.signaller_state = SignallerState::Stopped;
            drop(state);
            self.signaller().stop();
            gst::info!(CAT, imp = self, "Stopped signaller");
        }
    }
```

--------------------------------

### Get Element or Subslice

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/gstrswebrtc/utils/struct.Codecs.html

Safely gets a reference to an element or subslice using a position or range. Returns `None` if the index is out of bounds.

```rust
let v = [10, 40, 30];
assert_eq!(Some(&40), v.get(1));
assert_eq!(Some(&[10, 40][..]), v.get(0..2));
assert_eq!(None, v.get(3));
assert_eq!(None, v.get(0..4));
```

--------------------------------

### WhipServer on_webrtcbin_ready Method

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/src/gstrswebrtc/whip_signaller/imp.rs.html

Callback function to be invoked when the webrtcbin is ready, used for initializing the WHIP server signaller.

```rust
impl WhipServer {
    pub fn on_webrtcbin_ready(&self) -> RustClosure {
        glib::closure!(|signaller: &super::WhipServerSignaller,
                        session_id: &str,

```

--------------------------------

### SignallableExt::start

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/src/gstrswebrtc/signaller/iface.rs.html

Starts a WebRTC session. This method is used to initiate the signaling process for a new session.

```APIDOC
## start

### Description
Starts a WebRTC session.

### Method
`start()`

### Parameters
None

### Request Example
```rust
// Assuming `signaller` is an object implementing SignallableExt
signaller.start();
```

### Response
None directly, but initiates a session.

### Response Example
None
```

--------------------------------

### Handle WebRTC Source Pad Creation

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/src/gstrswebrtc/webrtcsrc/imp.rs.html

This function creates and configures a ghost pad for a WebRTC source pad. It handles stream start events, sets up decoding requirements based on negotiated caps, and configures the ghost pad's chain and event functions.

```rust
fn handle_webrtc_src_pad(
        &mut self,
        bin: &gst::Bin,
        webrtcbin_pad: &gst::Pad,
        element: &super::BaseWebRTCSrc,
    ) -> gst::GhostPad {
        let srcpad_and_caps = self.take_pending_src_pad(webrtcbin_pad);
        if let Some((ref srcpad, ref caps)) = srcpad_and_caps {
            let stream_id = srcpad.imp().stream_id();
            let mut builder = gst::event::StreamStart::builder(&stream_id);
            if let Some(stream_start) = webrtcbin_pad.sticky_event::<gst::event::StreamStart>(0) {
                builder = builder
                    .seqnum(stream_start.seqnum())
                    .group_id(stream_start.group_id().unwrap_or_else(gst::GroupId::next));
            }

            gst::debug!(
                CAT,
                obj = element,
                "Storing id {stream_id} on {webrtcbin_pad:?}",
            );
            webrtcbin_pad.store_sticky_event(&builder.build()).ok();

            srcpad.imp().set_webrtc_pad(webrtcbin_pad.downgrade());

            // if srcpad is request pad, it might have already been added to the element
            if srcpad.parent().is_none() {
                element
                    .add_pad(srcpad)
                    .expect("Adding ghost pad should never fail");
            }
            let media_type = caps
                .structure(0)
                .expect("Passing empty caps is invalid")
                .get::<&str>("media")
                .expect("Only caps with a `media` field are expected when creating the pad");

            let settings = element.imp().settings.lock().unwrap();

            let (raw_caps, encoded_caps) = if media_type == "video" {
                let encoded_video_caps = settings
                    .video_codecs
                    .iter()
                    .map(|c| c.caps.clone())
                    .collect::<gst::Caps>();
                (VIDEO_CAPS.clone(), encoded_video_caps)
            } else if media_type == "audio" {
                let encoded_audio_caps = settings
                    .audio_codecs
                    .iter()
                    .map(|c| c.caps.clone())
                    .collect::<gst::Caps>();
                (AUDIO_CAPS.clone(), encoded_audio_caps)
            } else {
                unreachable!()
            };
            drop(settings);

            let caps_with_raw = [caps.clone(), raw_caps.clone(), encoded_caps.clone()]
                .into_iter()
                .collect::<gst::Caps>();

            let downstream_caps = srcpad.peer_query_caps(Some(&caps_with_raw));
            if let Some(first_struct) = downstream_caps.structure(0)
                && (first_struct.has_name(raw_caps.structure(0).unwrap().name())
                    || !first_struct.has_name("application/x-rtp"))
            {
                srcpad.imp().set_needs_decoding(true)
            }
        }

        let ghostpad = gst::GhostPad::builder(gst::PadDirection::Src)
            .proxy_pad_chain_function(glib::clone!(
                #[weak]
                element,
                #[upgrade_or_panic]
                move |pad, parent, buffer| {
                    let padret = gst::ProxyPad::chain_default(pad, parent, buffer);
                    let state = element.imp().state.lock().unwrap();
                    if let Some(ref session) = state.session {
                        let session = session.0.lock().unwrap();

                        session.flow_combiner.lock().unwrap().update_flow(padret)
                    } else {
                        padret
                    }
                }
            ))
            .proxy_pad_event_function(glib::clone!(
                #[weak]
                element,
                #[weak(rename_to = webrtcpad)]
                webrtcbin_pad,
                #[upgrade_or_panic]
                move |pad, parent, event| {

```

--------------------------------

### Get Last Element of Slice

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/gstrswebrtc/utils/struct.Codecs.html

Use `last` to get an immutable reference to the last element of a slice. Returns `None` if the slice is empty.

```rust
let v = [10, 40, 30];
assert_eq!(Some(&30), v.last());

let w: &[i32] = &[];
assert_eq!(None, w.last());
```

--------------------------------

### Create and Probe Source Pad

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/src/gstrswebrtc/webrtcsrc/imp.rs.html

Attempts to create and probe a new source pad for a given media stream. This is part of the process of establishing a media connection.

```rust
element.imp().create_and_probe_src_pad(&caps, &stream_id, self)
```

--------------------------------

### Get Mutable Reference to Last Element

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/gstrswebrtc/utils/struct.Codecs.html

Use `last_mut` to get a mutable reference to the last element of a slice. Returns `None` if the slice is empty.

```rust
let x = &mut [0, 1, 2];

if let Some(last) = x.last_mut() {
    *last = 10;
}
assert_eq!(x, &[0, 1, 10]);

let y: &mut [i32] = &mut [];
assert_eq!(None, y.last_mut());
```

--------------------------------

### Get Mutable Reference to First Element

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/gstrswebrtc/utils/struct.Codecs.html

Use `first_mut` to get a mutable reference to the first element of a slice. Returns `None` if the slice is empty.

```rust
let x = &mut [0, 1, 2];

if let Some(first) = x.first_mut() {
    *first = 5;
}
assert_eq!(x, &[5, 1, 2]);

let y: &mut [i32] = &mut [];
assert_eq!(None, y.first_mut());
```

--------------------------------

### Get Unchecked Element or Subslice

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/gstrswebrtc/utils/struct.Codecs.html

Unsafely gets a reference to an element or subslice without bounds checking. Use with caution, as out-of-bounds access is undefined behavior.

```rust
let x = &[1, 2, 4];

unsafe {
    assert_eq!(x.get_unchecked(1), &2);
}
```

--------------------------------

### Create and Add Source Pad

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/src/gstrswebrtc/webrtcsrc/imp.rs.html

Creates a new source pad based on a template, sets its properties, and adds it to the element. Logs information about pad creation and usage.

```rust
            format!("{}{}", VIDEO_REQUEST_PAD_NAME_PREFIX, self.n_request_video_pads.fetch_add(1, Ordering::SeqCst))
        ) {
            }
        } else if templ.name().starts_with(AUDIO_REQUEST_PAD_NAME_PREFIX) {
            format!("{}{}", AUDIO_REQUEST_PAD_NAME_PREFIX, self.n_request_audio_pads.fetch_add(1, Ordering::SeqCst))
        } else {
            return None;
        };

        if let Some(n) = name {
            gst::info!(
                CAT,
                imp = self,
                "not using the pad name {n} requested, pad serial is autoenumerated internally"
            );
        }

        gst::info!(CAT, imp = self, "Creating new pad {pad_name}");

        let src_pad = gst::GhostPad::builder_from_template(templ)
            .name(pad_name)
            .build()
            .downcast::<WebRTCSrcPad>()
            .unwrap();

        src_pad.set_active(true).unwrap();
        src_pad.use_fixed_caps();
        self.obj().add_pad(&src_pad).unwrap();

        Some(src_pad.into())
    }

    fn release_pad(&self, pad: &gst::Pad) {
        let element = self.obj();
        let ws = element
            .upcast_ref::<crate::webrtcsrc::BaseWebRTCSrc>()
            .imp();

        gst::info!(CAT, obj = element, "releasing {pad:?}");

        let state = ws.state.lock().unwrap();
        if let Some(s) = &state.session {
            // remove the pad from the pending src pads hashmap in case it is not used yet
            let mut session = s.0.lock().unwrap();
            let _ = session.take_pending_src_pad(pad);
        }
        drop(state);

        if let Err(e) = self.obj().remove_pad(pad) {
            gst::error!(CAT, obj = element, "could remove {pad:?}: {e:?}");
        };
    }
}

impl GstObjectImpl for BaseWebRTCSrc {}

impl BinImpl for BaseWebRTCSrc {}

impl ChildProxyImpl for BaseWebRTCSrc {
    fn child_by_index(&self, index: u32) -> Option<glib::Object> {
        if index == 0 {
            Some(self.signaller().upcast())
        } else {
            None
        }
    }

    fn children_count(&self) -> u32 {
        1
    }

    fn child_by_name(&self, name: &str) -> Option<glib::Object> {
        match name {
            "signaller" => {
                gst::info!(CAT, imp = self, "Getting signaller");
                Some(self.signaller().upcast())
            }
            _ => None,
        }
    }
}

#[derive(PartialEq)]
enum SignallerState {
    Started,
    Stopped,
}

struct SessionInner {
    id: String,
    webrtcbin: gst::Element,
    navigation_data_channel: Option<WebRTCDataChannel>,
    control_data_channel: Option<WebRTCDataChannel>,
    n_video_pads: AtomicU16,
    n_audio_pads: AtomicU16,
    flow_combiner: Mutex<gst_base::UniqueFlowCombiner>,
    request_counter: u64,
    pending_srcpads: HashMap<String, (WebRTCSrcPad, gst::Caps)>,
}

#[derive(Clone)]
struct Session(Arc<Mutex<SessionInner>>);

struct State {
    session: Option<Session>,
    signaller_state: SignallerState,
    signaller_signals: Option<SignallerSignals>,
    finalizing_session: Arc<(Mutex<Option<Session>>, Condvar)>,
}

impl Default for State {
    fn default() -> Self {
        Self {
            signaller_state: SignallerState::Stopped,
            session: None,
            signaller_signals: Default::default(),
            finalizing_session: Arc::new((Mutex::new(None), Condvar::new()))
        }
    }
}

impl State {
    fn finalize_session(&mut self, element: &super::BaseWebRTCSrc, session: Session) {
        let inner = session.0.lock().unwrap();
        drop(inner);

        gst::info!(CAT, "Ending session");

        let finalizing_session_opt = self.finalizing_session.clone();
        let (finalizing_session, _cvar) = &*finalizing_session_opt;
        *finalizing_session.lock().unwrap() = Some(session);

        let element = element.clone();
        RUNTIME.spawn_blocking(move || {
            let (finalizing_session, cvar) = &*finalizing_session_opt;
            let mut finalizing_session = finalizing_session.lock().unwrap();
            let session = finalizing_session.take().unwrap();
            let session = session.0.lock().unwrap();

            let bin = session
                .webrtcbin()
                .parent()
                .and_downcast::<gst::Bin>()
                .unwrap();

```

--------------------------------

### GstRSWebRTCSignallableIface::start Signal Handler

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/src/gstrswebrtc/signaller/iface.rs.html

Handles the 'start' signal for the GstRSWebRTCSignallableIface. It retrieves the signaller object, accesses its virtual table, and calls the start function. Returns false.

```rust
.return_type::<bool>()
                    .class_handler(|args| {
                        let this = args[0usize]
                            .get::<&super::Signallable>()
                            .unwrap_or_else(|e| {
                                panic!("Wrong type for argument {}: {:?}", 0usize, e)
                            });
                        let vtable = this.interface::<super::Signallable>().unwrap();
                        let vtable = vtable.as_ref();
                        (vtable.start)(this);

                        Some(false.into())
                    })
                    .accumulator(move |_hint, _acc, value| {
                        // First signal handler wins
                        std::ops::ControlFlow::Break(value.clone())
                    })
                    .build()
```

--------------------------------

### Comparable

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/gstrswebrtc/webrtcsink/struct.WebRTCSinkPad.html

Enables comparison of WebRTCSinkPad instances with other types.

```APIDOC
## fn compare(&self, key: &K) -> Ordering

Compare self to `key` and return their ordering.

### Parameters
- **&self**: A reference to the WebRTCSinkPad instance.
- **key**: A reference to the key to compare against.

### Returns
- `Ordering`: The ordering of self relative to key.
```

--------------------------------

### Get First Chunk of Slice

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/gstrswebrtc/utils/struct.Codecs.html

Use `first_chunk` to get an immutable reference to the first `N` elements of a slice as an array. Returns `None` if the slice is shorter than `N`.

```rust
let u = [10, 40, 30];
assert_eq!(Some(&[10, 40]), u.first_chunk::<2>());

let v: &[i32] = &[10];
assert_eq!(None, v.first_chunk::<2>());

let w: &[i32] = &[];
assert_eq!(Some(&[]), w.first_chunk::<0>());
```

--------------------------------

### Start Session with Producer

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/src/gstrswebrtc/signaller/imp.rs.html

Initiates a session with a producer if the signaller role is `Consumer`. It retrieves the producer's peer ID and sends a `StartSession` message.

```rust
pub fn start_session(&self) {
        let role = self.settings.lock().unwrap().role;
        if matches!(role, super::WebRTCSignallerRole::Consumer) {
            let target_producer = self.producer_peer_id().unwrap();

            self.send(p::IncomingMessage::StartSession(p::StartSessionMessage {
                peer_id: target_producer.clone(),
                offer: None,
            }));

            gst::info!(
                CAT,
                imp = self,
                "Started session with producer peer id {target_producer}",
            );
        }
    }
```

--------------------------------

### Start Discovery Pipeline and Queue Stream

Source: https://docs.rs/gst-plugin-webrtc/0.15.0/src/gstrswebrtc/webrtcsink/imp.rs.html

Sets the GStreamer pipeline to the Playing state and queues the discovery information for the specified stream. This initiates the discovery process.

```rust
pipe.0
            .set_state(gst::State::Playing)
            .with_context(|| format!("Running discovery pipeline for caps {input_caps}"))?;

        {
            let mut state = self.state.lock().unwrap();
            state.queue_discovery(stream_name, discovery_info.clone());
        }
```
Tokens:
Raw