### 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> SignallableExt for Obj { fn start(&self) { self.emit_by_name::("start", &[]); } fn stop(&self) { self.emit_by_name::("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", &[ ``` -------------------------------- ### 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 ### Description Get the value of a property using a closure. ### Method `fn get(&self, f: F) -> R` where F: Fn(&::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::().unwrap()) { Ok(offer_sdp) => { webrtcbin.emit_by_name::<()>( "set-local-description", &[&offer_sdp, &None::], ); 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::() .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::, &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>( cafile_path: P, ) -> Result { 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::>() }; 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 = Vec::with_capacity(10); vec.push(42); assert!(vec.capacity() >= 10); ``` ```rust #[derive(Clone)] struct ZeroSized; fn main() { assert_eq!(std::mem::size_of::(), 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::("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::(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::(); (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::(); (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::(); 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::() .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::() .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 { if index == 0 { Some(self.signaller().upcast()) } else { None } } fn children_count(&self) -> u32 { 1 } fn child_by_name(&self, name: &str) -> Option { 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, control_data_channel: Option, n_video_pads: AtomicU16, n_audio_pads: AtomicU16, flow_combiner: Mutex, request_counter: u64, pending_srcpads: HashMap, } #[derive(Clone)] struct Session(Arc>); struct State { session: Option, signaller_state: SignallerState, signaller_signals: Option, finalizing_session: Arc<(Mutex>, 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::() .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::() .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::().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()); } ``` -------------------------------- ### 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; } } } } )); } ```