media_pp\core\pipeline/
builder.rs

1use std::sync::{
2    Arc, Mutex,
3    atomic::{AtomicBool, AtomicUsize},
4};
5
6use crate::{
7    bus::{Bus, BusReceiver},
8    clock::Clock,
9    control::{self, ControlReceiver, ControlSender},
10    element::{Context, SourceElement, element_pp_log, pipeline_pp_log},
11    error::Result,
12    graph::{ElementId, PipelineGraph},
13    playback_clock::PlaybackClock,
14};
15
16use super::Pipeline;
17
18/// Accumulates one or more sources into a single [`Pipeline`] — the
19/// multi-source generalization of what [`Pipeline::new`] does for exactly
20/// one. Each [`PipelineBuilder::add_source`] call gets its own background
21/// thread once [`PipelineBuilder::build`]'s [`Pipeline::run`] starts, but
22/// they all share one [`Bus`] (so [`Pipeline::bus`] sees every source's
23/// events on one channel), one [`Clock`] (so every [`crate::elements::Pacer`]
24/// anywhere in the pipeline — regardless of which source's chain it's
25/// under — agrees on the same t=0/pause timeline), and one
26/// [`PipelineGraph`] (so [`Pipeline::topology`] renders every source's
27/// own branches together).
28///
29/// [`Pipeline::new`] is exactly `PipelineBuilder::new(id).add_source(source,
30/// wire).build()` — the ergonomic single-source special case, kept as its
31/// own entry point so existing single-source callers don't need to change.
32/// Reach for `PipelineBuilder` directly once there's more than one live
33/// source to combine into one file/output — e.g. a video capture and an
34/// audio capture both feeding the same [`crate::elements::Mp4Muxer`]: two
35/// independent sources under today's [`crate::element::SourceElement`]
36/// model, but one [`Pipeline`] so `run()`/`pause()`/`resume()`/`stop()`
37/// only need to be called once, not once per source.
38pub(super) type SourceEntry = (ElementId, Box<dyn SourceElement>);
39
40pub struct PipelineBuilder {
41    id: Arc<str>,
42    bus: Bus,
43    bus_rx: BusReceiver,
44    clock: Arc<Clock>,
45    playback_clock: Arc<PlaybackClock>,
46    graph: PipelineGraph,
47    sources: Vec<SourceEntry>,
48    control_pairs: Vec<(ControlSender, ControlReceiver)>,
49}
50
51impl PipelineBuilder {
52    pub fn new(id: impl Into<String>) -> Self {
53        let id: Arc<str> = id.into().into();
54        let (bus, bus_rx) = Bus::new();
55        let clock = Arc::new(Clock::new());
56        Self {
57            id,
58            bus,
59            bus_rx,
60            playback_clock: Arc::new(PlaybackClock::new(clock.clone())),
61            clock,
62            graph: PipelineGraph::new(),
63            sources: Vec::new(),
64            control_pairs: Vec::new(),
65        }
66    }
67
68    /// Registers one more source. `wire` receives a source-scoped
69    /// [`Context`]; build detached branches with [`Context::branch`] and
70    /// commit them with [`Context::attach`]. A wiring error aborts the
71    /// builder without publishing a partially built pipeline.
72    pub fn add_source<S: SourceElement + 'static>(
73        mut self,
74        mut source: S,
75        wire: impl FnOnce(&mut S, &Arc<Context>) -> Result<()>,
76    ) -> Result<Self> {
77        *source.pp_log_mut() =
78            element_pp_log(source.element_type(), &source.name(), Some(&self.id));
79        let source_id = self.graph.add_source(source.element_type(), source.name());
80        let context = Arc::new(Context {
81            bus: self.bus.clone(),
82            pipeline_id: self.id.clone(),
83            graph: self.graph.clone(),
84            clock: self.clock.clone(),
85            playback_clock: self.playback_clock.clone(),
86            source_id,
87        });
88        wire(&mut source, &context)?;
89        self.sources.push((source_id, Box::new(source)));
90        self.control_pairs.push(control::channel());
91        Ok(self)
92    }
93
94    /// Finishes construction. At least one [`PipelineBuilder::add_source`]
95    /// call must have happened — an empty [`Pipeline`] has nothing for
96    /// [`Pipeline::run`] to ever drive, and [`Pipeline::bus`] would block
97    /// forever waiting for a source thread that will never start (nothing
98    /// left holding a [`Bus`] sender to eventually drop).
99    pub fn build(self) -> Arc<Pipeline> {
100        assert!(
101            !self.sources.is_empty(),
102            "PipelineBuilder::build called with no sources added"
103        );
104        let (control_txs, control_rxs): (Vec<_>, Vec<_>) = self.control_pairs.into_iter().unzip();
105        let pp_log = pipeline_pp_log(&self.id);
106        Arc::new(Pipeline {
107            id: self.id,
108            pp_log,
109            sources: Mutex::new(Some(self.sources)),
110            bus: Mutex::new(Some(self.bus)),
111            control_txs,
112            control_rxs: Mutex::new(Some(control_rxs)),
113            clock: self.clock,
114            playback_clock: self.playback_clock,
115            bus_rx: self.bus_rx,
116            running: Arc::new(AtomicUsize::new(0)),
117            paused: AtomicBool::new(false),
118            workers: Mutex::new(Vec::new()),
119            graph: self.graph,
120        })
121    }
122}
media_pp\core\pipeline/builder.rs

media_pp\core\pipeline/
builder.rs
