diff --git a/examples/src/bin/rtpfecclient.rs b/examples/src/bin/rtpfecclient.rs new file mode 100644 index 000000000..e49c4141d --- /dev/null +++ b/examples/src/bin/rtpfecclient.rs @@ -0,0 +1,293 @@ +#[macro_use] +extern crate gstreamer as gst; +use gst::prelude::*; + +extern crate glib; + +use std::env; +use std::error::Error as StdError; + +#[path = "../examples-common.rs"] +mod examples_common; + +extern crate failure; +use failure::Error; + +#[macro_use] +extern crate failure_derive; + +#[derive(Debug, Fail)] +#[fail(display = "Missing element {}", _0)] +struct MissingElement(&'static str); + +#[derive(Debug, Fail)] +#[fail(display = "No such pad {} in {}", _0, _1)] +struct NoSuchPad(&'static str, String); + +#[derive(Debug, Fail)] +#[fail(display = "Unknown payload type {}", _0)] +struct UnknownPT(u32); + +#[derive(Debug, Fail)] +#[fail(display = "Usage: {} (play | record) DROP_PROBABILITY", _0)] +struct UsageError(String); + +#[derive(Debug, Fail)] +#[fail(display = "Received error from {}: {} (debug: {:?})", src, error, debug)] +struct ErrorMessage { + src: String, + error: String, + debug: Option, + #[cause] cause: glib::Error, +} + +fn make_element<'a, P: Into>>( + factory_name: &'static str, + element_name: P, +) -> Result { + match gst::ElementFactory::make(factory_name, element_name.into()) { + Some(elem) => Ok(elem), + None => Err(Error::from(MissingElement(factory_name))), + } +} + +fn get_static_pad(element: &gst::Element, pad_name: &'static str) -> Result { + match element.get_static_pad(pad_name) { + Some(pad) => Ok(pad), + None => { + let element_name = element.get_name(); + Err(Error::from(NoSuchPad(pad_name, element_name))) + } + } +} + +fn get_request_pad(element: &gst::Element, pad_name: &'static str) -> Result { + match element.get_request_pad(pad_name) { + Some(pad) => Ok(pad), + None => { + let element_name = element.get_name(); + Err(Error::from(NoSuchPad(pad_name, element_name))) + } + } +} + +fn connect_rtpbin_srcpad(src_pad: &gst::Pad, sink: &gst::Element) -> Result<(), Error> { + let name = src_pad.get_name(); + let split_name = name.split("_"); + let split_name = split_name.collect::>(); + let pt = split_name[5].parse::()?; + + match pt { + 96 => { + let sinkpad = get_static_pad(sink, "sink")?; + src_pad.link(&sinkpad).into_result()?; + Ok(()) + } + _ => Err(Error::from(UnknownPT(pt))), + } +} + +fn make_fec_decoder(rtpbin: &gst::Element, sess_id: u32) -> Result { + let fecdec = make_element("rtpulpfecdec", None)?; + let internal_storage = rtpbin + .emit("get-internal-storage", &[&sess_id.to_value()]) + .unwrap() + .unwrap(); + + fecdec.set_property("storage", &internal_storage.to_value())?; + fecdec.set_property("pt", &100u32.to_value())?; + + Ok(fecdec) +} + +fn example_main() -> Result<(), Error> { + gst::init()?; + + let args: Vec<_> = env::args().collect(); + + if args.len() != 3 { + return Err(Error::from(UsageError(args[0].clone()))); + } + + let drop_probability = args[2].parse::()?; + + let pipeline = gst::Pipeline::new(None); + let src = make_element("udpsrc", None)?; + let netsim = make_element("netsim", None)?; + let rtpbin = make_element("rtpbin", None)?; + let depay = make_element("rtpvp8depay", None)?; + let dec = make_element("vp8dec", None)?; + let conv = make_element("videoconvert", None)?; + let scale = make_element("videoscale", None)?; + let filter = make_element("capsfilter", None)?; + + pipeline.add_many(&[ + &src, + &netsim, + &rtpbin, + &depay, + &dec, + &conv, + &scale, + &filter, + ])?; + gst::Element::link_many(&[&depay, &dec, &conv, &scale, &filter])?; + + match args[1].as_str() { + "play" => { + let sink = make_element("autovideosink", None)?; + pipeline.add(&sink)?; + filter.link(&sink)?; + } + "record" => { + let enc = make_element("x264enc", None)?; + let mux = make_element("matroskamux", None)?; + let sink = make_element("filesink", None)?; + + pipeline.add_many(&[&enc, &mux, &sink])?; + gst::Element::link_many(&[&filter, &enc, &mux, &sink])?; + sink.set_property("location", &"out.mkv".to_value())?; + enc.set_property_from_str("tune", "zerolatency"); + eprintln!("Recording to out.mkv"); + } + _ => return Err(Error::from(UsageError(args[0].clone()))), + } + + src.link(&netsim)?; + + rtpbin.connect("new-storage", false, |values| { + let storage = values[1].get::().expect("Invalid argument"); + storage + .set_property("size-time", &250_000_000u64.to_value()) + .unwrap(); + + None + })?; + + rtpbin.connect("request-pt-map", false, |values| { + let pt = values[2].get::().expect("Invalid argument"); + match pt { + 100 => Some( + gst::Caps::new_simple( + "application/x-rtp", + &[ + ("media", &"video"), + ("clock-rate", &90000i32), + ("is-fec", &true), + ], + ).to_value(), + ), + 96 => Some( + gst::Caps::new_simple( + "application/x-rtp", + &[ + ("media", &"video"), + ("clock-rate", &90000i32), + ("encoding-name", &"VP8"), + ], + ).to_value(), + ), + _ => None, + } + })?; + + rtpbin.connect("request-fec-decoder", false, |values| { + let rtpbin = values[0].get::().expect("Invalid argument"); + let sess_id = values[1].get::().expect("Invalid argument"); + + match make_fec_decoder(&rtpbin, sess_id) { + Ok(elem) => Some(elem.to_value()), + Err(err) => { + gst_element_error!( + rtpbin, + gst::LibraryError::Failed, + ("Failed to make FEC decoder"), + ["{}", err] + ); + None + } + } + })?; + + let srcpad = get_static_pad(&netsim, "src")?; + let sinkpad = get_request_pad(&rtpbin, "recv_rtp_sink_0")?; + srcpad.link(&sinkpad).into_result()?; + + let depay_clone = depay.clone(); + rtpbin.connect_pad_added(move |rtpbin, src_pad| { + match connect_rtpbin_srcpad(&src_pad, &depay_clone) { + Ok(_) => (), + Err(err) => { + gst_element_error!( + rtpbin, + gst::LibraryError::Failed, + ("Failed to link srcpad"), + ["{}", err] + ); + () + } + } + }); + + let rtp_caps = gst::Caps::new_simple("application/x-rtp", &[("clock-rate", &90000i32)]); + + let video_caps = + gst::Caps::new_simple("video/x-raw", &[("width", &1920i32), ("height", &1080i32)]); + + src.set_property("address", &"127.0.0.1".to_value())?; + src.set_property("caps", &rtp_caps.to_value())?; + netsim.set_property("drop-probability", &drop_probability.to_value())?; + rtpbin.set_property("do-lost", &true.to_value())?; + filter.set_property("caps", &video_caps.to_value())?; + + let bus = pipeline + .get_bus() + .expect("Pipeline without bus. Shouldn't happen!"); + + let ret = pipeline.set_state(gst::State::Playing); + assert_ne!(ret, gst::StateChangeReturn::Failure); + + while let Some(msg) = bus.timed_pop(gst::CLOCK_TIME_NONE) { + use gst::MessageView; + + match msg.view() { + MessageView::Eos(..) => break, + MessageView::Error(err) => { + return Err( + ErrorMessage { + src: msg.get_src() + .map(|s| s.get_path_string()) + .unwrap_or(String::from("None")), + error: err.get_error().description().into(), + debug: err.get_debug(), + cause: err.get_error(), + }.into(), + ); + } + MessageView::StateChanged(s) => match msg.get_src() { + Some(element) => if element == pipeline && s.get_current() == gst::State::Playing { + eprintln!("PLAYING"); + gst::debug_bin_to_dot_file( + &pipeline, + gst::DebugGraphDetails::all(), + "client-playing", + ); + }, + None => (), + }, + _ => (), + } + } + + let ret = pipeline.set_state(gst::State::Null); + assert_ne!(ret, gst::StateChangeReturn::Failure); + + Ok((())) +} + +fn main() { + match examples_common::run(example_main) { + Ok(r) => r, + Err(e) => eprintln!("Error! {}", e), + } +} diff --git a/examples/src/bin/rtpfecserver.rs b/examples/src/bin/rtpfecserver.rs new file mode 100644 index 000000000..ed46de7ed --- /dev/null +++ b/examples/src/bin/rtpfecserver.rs @@ -0,0 +1,223 @@ +#[macro_use] +extern crate gstreamer as gst; +use gst::prelude::*; + +extern crate glib; + +use std::error::Error as StdError; + +#[path = "../examples-common.rs"] +mod examples_common; + +use std::env; + +extern crate failure; +use failure::Error; + +#[macro_use] +extern crate failure_derive; + +#[derive(Debug, Fail)] +#[fail(display = "Missing element {}", _0)] +struct MissingElement(&'static str); + +#[derive(Debug, Fail)] +#[fail(display = "No such pad {} in {}", _0, _1)] +struct NoSuchPad(&'static str, String); + +#[derive(Debug, Fail)] +#[fail(display = "Usage: {} URI FEC_PERCENTAGE", _0)] +struct UsageError(String); + +#[derive(Debug, Fail)] +#[fail(display = "Received error from {}: {} (debug: {:?})", src, error, debug)] +struct ErrorMessage { + src: String, + error: String, + debug: Option, + #[cause] cause: glib::Error, +} + +fn make_element<'a, P: Into>>( + factory_name: &'static str, + element_name: P, +) -> Result { + match gst::ElementFactory::make(factory_name, element_name.into()) { + Some(elem) => Ok(elem), + None => Err(Error::from(MissingElement(factory_name))), + } +} + +fn get_static_pad(element: &gst::Element, pad_name: &'static str) -> Result { + match element.get_static_pad(pad_name) { + Some(pad) => Ok(pad), + None => { + let element_name = element.get_name(); + Err(Error::from(NoSuchPad(pad_name, element_name))) + } + } +} + +fn get_request_pad(element: &gst::Element, pad_name: &'static str) -> Result { + match element.get_request_pad(pad_name) { + Some(pad) => Ok(pad), + None => { + let element_name = element.get_name(); + Err(Error::from(NoSuchPad(pad_name, element_name))) + } + } +} + +fn connect_decodebin_pad(src_pad: &gst::Pad, sink: &gst::Element) -> Result<(), Error> { + let sinkpad = get_static_pad(&sink, "sink")?; + src_pad.link(&sinkpad).into_result()?; + + Ok(()) +} + +fn make_fec_encoder(fec_percentage: u32) -> Result { + let fecenc = make_element("rtpulpfecenc", None)?; + + fecenc.set_property("pt", &100u32.to_value())?; + fecenc.set_property("mux-seq", &true.to_value())?; + fecenc.set_property("multipacket", &true.to_value())?; + fecenc.set_property("percentage", &fec_percentage.to_value())?; + + Ok(fecenc) +} + +fn example_main() -> Result<(), Error> { + gst::init()?; + + let args: Vec<_> = env::args().collect(); + + if args.len() != 3 { + return Err(Error::from(UsageError(args[0].clone()))); + } + + let uri = &args[1]; + let fec_percentage = args[2].parse::()?; + + let pipeline = gst::Pipeline::new(None); + let src = make_element("uridecodebin", None)?; + let conv = make_element("videoconvert", None)?; + let q1 = make_element("queue", None)?; + let enc = make_element("vp8enc", None)?; + let q2 = make_element("queue", None)?; + let pay = make_element("rtpvp8pay", None)?; + let rtpbin = make_element("rtpbin", None)?; + let sink = make_element("udpsink", None)?; + + pipeline.add_many(&[&src, &conv, &q1, &enc, &q2, &pay, &rtpbin, &sink])?; + + conv.link(&q1)?; + q1.link(&enc)?; + enc.link(&pay)?; + pay.link(&q2)?; + + rtpbin.connect("request-fec-encoder", false, move |values| { + let rtpbin = values[0].get::().expect("Invalid argument"); + + match make_fec_encoder(fec_percentage) { + Ok(elem) => Some(elem.to_value()), + Err(err) => { + gst_element_error!( + rtpbin, + gst::LibraryError::Failed, + ("Failed to make FEC encoder"), + ["{}", err] + ); + None + } + } + })?; + + let srcpad = get_static_pad(&q2, "src")?; + let sinkpad = get_request_pad(&rtpbin, "send_rtp_sink_0")?; + srcpad.link(&sinkpad).into_result()?; + + let srcpad = get_static_pad(&rtpbin, "send_rtp_src_0")?; + let sinkpad = get_static_pad(&sink, "sink")?; + srcpad.link(&sinkpad).into_result()?; + + let convclone = conv.clone(); + src.connect_pad_added(move |decodebin, src_pad| { + match connect_decodebin_pad(&src_pad, &convclone) { + Ok(_) => (), + Err(err) => { + gst_element_error!( + decodebin, + gst::LibraryError::Failed, + ("Failed to link decodebin srcpad"), + ["{}", err] + ); + () + } + } + }); + + let video_caps = gst::Caps::new_simple("video/x-raw", &[]); + + src.set_property_from_str("pattern", "ball"); + sink.set_property("host", &"127.0.0.1".to_value())?; + sink.set_property("sync", &true.to_value())?; + enc.set_property("keyframe-max-dist", &30i32.to_value())?; + enc.set_property("threads", &12i32.to_value())?; + enc.set_property("cpu-used", &(-16i32).to_value())?; + enc.set_property("deadline", &1i64.to_value())?; + enc.set_property_from_str("error-resilient", "default"); + src.set_property("expose-all-streams", &false.to_value())?; + src.set_property("caps", &video_caps.to_value())?; + src.set_property("uri", &uri.to_value())?; + + let bus = pipeline + .get_bus() + .expect("Pipeline without bus. Shouldn't happen!"); + + let ret = pipeline.set_state(gst::State::Playing); + assert_ne!(ret, gst::StateChangeReturn::Failure); + + while let Some(msg) = bus.timed_pop(gst::CLOCK_TIME_NONE) { + use gst::MessageView; + + match msg.view() { + MessageView::Eos(..) => break, + MessageView::Error(err) => { + return Err( + ErrorMessage { + src: msg.get_src() + .map(|s| s.get_path_string()) + .unwrap_or(String::from("None")), + error: err.get_error().description().into(), + debug: err.get_debug(), + cause: err.get_error(), + }.into(), + ); + } + MessageView::StateChanged(s) => match msg.get_src() { + Some(element) => if element == pipeline && s.get_current() == gst::State::Playing { + eprintln!("PLAYING"); + gst::debug_bin_to_dot_file( + &pipeline, + gst::DebugGraphDetails::all(), + "server-playing", + ); + }, + None => (), + }, + _ => (), + } + } + + let ret = pipeline.set_state(gst::State::Null); + assert_ne!(ret, gst::StateChangeReturn::Failure); + + Ok((())) +} + +fn main() { + match examples_common::run(example_main) { + Ok(r) => r, + Err(e) => eprintln!("Error! {}", e), + } +}