ts-intersrc

Thread-sharing source for inter-pipelines communication.

ts-intersrc is an element that proxies events travelling downstream, non-serialized queries and buffers (including metas) from another pipeline that contains a matching ts-intersink element. The purpose is to allow one to many decoupled pipelines to function as though they were one without having to manually shuttle buffers, events, queries, etc.

The ts-intersink & ts-intersrc elements take advantage of the threadshare runtime, reducing the number of threads & context switches which would be necessary with other forms of inter-pipelines elements.

Usage

 use futures::prelude::*;
 use gst::prelude::*;

 let g_ctx = gst::glib::MainContext::default();
 gst::init().unwrap();

 // An upstream pipeline producing a 1 second Opus encoded audio stream
 let pipe_up = gst::parse::launch(
     "
         audiotestsrc is-live=true num-buffers=50 volume=0.02
         ! opusenc
         ! ts-intersink inter-context=my-inter-ctx
     ",
 )
 .unwrap()
 .downcast::<gst::Pipeline>()
 .unwrap();

 // A downstream pipeline which will receive the Opus encoded audio stream
 // and render it locally.
 let pipe_down = gst::parse::launch(
     "
         ts-intersrc inter-context=my-inter-ctx context=ts-group-01 context-wait=20
         ! opusdec
         ! audioconvert
         ! audioresample
         ! ts-queue context=ts-group-01 context-wait=20 max-size-buffers=1 max-size-bytes=0 max-size-time=0
         ! autoaudiosink
     ",
 )
 .unwrap()
 .downcast::<gst::Pipeline>()
 .unwrap();

 // Both pipelines must agree on the timing information or we'll get glitches
 // or overruns/underruns. Ideally, we should tell pipe_up to use the same clock
 // as pipe_down, but since that will be set asynchronously to the audio clock, it
 // is simpler and likely accurate enough to use the system clock for both
 // pipelines. If no element in either pipeline will provide a clock, this
 // is not needed.
 let clock = gst::SystemClock::obtain();
 pipe_up.set_clock(Some(&clock)).unwrap();
 pipe_down.set_clock(Some(&clock)).unwrap();

 // This is not really needed in this case since the pipelines are created and
 // started at the same time. However, an application that dynamically
 // generates pipelines must ensure that all the pipelines that will be
 // connected together share the same base time.
 pipe_up.set_base_time(gst::ClockTime::ZERO);
 pipe_up.set_start_time(gst::ClockTime::NONE);
 pipe_down.set_base_time(gst::ClockTime::ZERO);
 pipe_down.set_start_time(gst::ClockTime::NONE);

 pipe_up.set_state(gst::State::Playing).unwrap();
 pipe_down.set_state(gst::State::Playing).unwrap();

 g_ctx.block_on(async {
     use gst::MessageView::*;

     let mut bus_up_stream = pipe_up.bus().unwrap().stream();
     let mut bus_down_stream = pipe_down.bus().unwrap().stream();

     loop {
         futures::select! {
             msg = bus_up_stream.next() => {
                 let Some(msg) = msg else { continue };
                 match msg.view() {
                     Latency(_) => {
                         let _ = pipe_down.recalculate_latency();
                     }
                     Error(err) => {
                         eprintln!("Error with downstream pipeline {err:?}");
                         break;
                     }
                     _ => (),
                 }
             }
             msg = bus_down_stream.next() => {
                 let Some(msg) = msg else { continue };
                 match msg.view() {
                     Latency(_) => {
                         let _ = pipe_down.recalculate_latency();
                     }
                     Eos(_) => {
                         println!("Got EoS");
                         break;
                     }
                     Error(err) => {
                         eprintln!("Error with downstream pipeline {err:?}");
                         break;
                     }
                     _ => (),
                 }
             }
         };
     }
 });

 pipe_up.set_state(gst::State::Null).unwrap();
 pipe_down.set_state(gst::State::Null).unwrap();