media_pp/elements/source/compositor/
sw_video_compositor.rs

1use std::{
2    collections::{HashMap, HashSet},
3    sync::{
4        Arc, Mutex, Weak,
5        atomic::{AtomicU64, Ordering},
6    },
7    thread,
8    time::{Duration, Instant},
9};
10
11use crate::pp_log::{PpLog, pp_info};
12use arc_swap::ArcSwapOption;
13use ffmpeg_next as ffmpeg;
14use thiserror::Error as ThisError;
15
16use super::video_layer::{
17    self, LayerGeometry, MAX_DIMENSION, VideoFit, VideoInputId, VideoLayer, VideoLayerError,
18    VideoRect,
19};
20use crate::{
21    buffer::MediaBuffer,
22    bus::{Bus, BusEvent},
23    color::Color,
24    control::{ControlMsg, ControlReceiver, drain_control},
25    element::{Element, ElementType, Sink, Source, SourceElement, element_pp_log},
26    error::Result,
27    pad::SrcPad,
28    pool::{UnboundObjectPool, UnboundObjectPoolRef},
29    schedule::PeriodicSchedule,
30};
31
32const OUTPUT_POOL_SIZE: usize = 4;
33const CONTROL_POLL_INTERVAL: Duration = Duration::from_millis(5);
34
35/// The compositor's fixed output definition. Every emitted frame is an
36/// opaque [`ffmpeg::format::Pixel::BGRA`] frame at `width` x `height`.
37#[derive(Debug, Clone, Copy)]
38pub struct VideoCompositorOptions {
39    pub width: u32,
40    pub height: u32,
41    pub frame_rate: ffmpeg::Rational,
42    pub background: Color,
43}
44
45impl Default for VideoCompositorOptions {
46    fn default() -> Self {
47        Self {
48            width: 1920,
49            height: 1080,
50            frame_rate: ffmpeg::Rational::new(30, 1),
51            background: Color::BLACK,
52        }
53    }
54}
55
56/// Errors specific to [`SwVideoCompositor`].
57#[derive(Debug, ThisError)]
58pub enum SwVideoCompositorError {
59    #[error("ffmpeg error: {0}")]
60    Ffmpeg(#[from] ffmpeg::Error),
61
62    #[error(
63        "invalid output dimensions {width}x{height}; each dimension must be 1..={MAX_DIMENSION}"
64    )]
65    InvalidOutputDimensions { width: u32, height: u32 },
66
67    #[error("invalid frame rate {0}; numerator and denominator must both be positive")]
68    InvalidFrameRate(ffmpeg::Rational),
69
70    #[error(
71        "invalid layer dimensions {width}x{height}; each dimension must be 1..={MAX_DIMENSION}"
72    )]
73    InvalidLayerDimensions { width: u32, height: u32 },
74
75    #[error("layer opacity must be finite and between 0.0 and 1.0, got {0}")]
76    InvalidOpacity(f32),
77
78    #[error("input frame has invalid dimensions {width}x{height}")]
79    InvalidInputDimensions { width: u32, height: u32 },
80
81    #[error("scaled layer would exceed {MAX_DIMENSION}px: {width}x{height}")]
82    ScaledLayerTooLarge { width: u32, height: u32 },
83
84    #[error("the compositor input has been removed")]
85    SourceRemoved,
86
87    #[error(
88        "SwVideoCompositorInputSink only accepts decoded Video frames, got a {0}; link it after a decoder or video source"
89    )]
90    UnsupportedBuffer(&'static str),
91
92    #[error("SwVideoCompositor doesn't support seeking a live composition")]
93    SeekUnsupported,
94}
95
96struct VideoInput {
97    id: VideoInputId,
98    /// The hot producer/consumer path is an atomic latest-value slot:
99    /// input pipelines replace the pointer without taking the layer lock,
100    /// and the compositor acquires a stable Arc snapshot independently.
101    latest_frame: ArcSwapOption<UnboundObjectPoolRef<ffmpeg::frame::Video>>,
102    /// Layer changes are infrequent and update several related fields as
103    /// one coherent value, so a small dedicated lock remains appropriate.
104    layer: Mutex<VideoLayer>,
105}
106
107struct CompositorShared {
108    inputs: Mutex<HashMap<Arc<str>, Arc<VideoInput>>>,
109    next_input_id: AtomicU64,
110}
111
112/// A cheaply cloneable handle for adding and removing compositor inputs.
113/// It mirrors [`crate::elements::MixerHandle`], but each registration also
114/// returns a [`SwVideoLayerHandle`] for changing that input's placement.
115#[derive(Clone)]
116pub struct SwVideoCompositorHandle {
117    shared: Weak<CompositorShared>,
118}
119
120/// The two endpoints created for one compositor input registration.
121/// Move `sink` into the upstream pipeline and retain `layer` in application
122/// code for runtime placement changes.
123pub struct SwVideoCompositorInput {
124    pub sink: Box<dyn Sink>,
125    pub layer: SwVideoLayerHandle,
126}
127
128impl SwVideoCompositorHandle {
129    /// Registers an input and returns its terminal Sink plus independent
130    /// runtime layer control. Reusing `name` replaces the old registration;
131    /// old sinks and layer handles become harmlessly stale.
132    pub fn add_source(
133        &self,
134        name: impl Into<String>,
135        layer: VideoLayer,
136    ) -> std::result::Result<Option<SwVideoCompositorInput>, SwVideoCompositorError> {
137        validate_layer(layer)?;
138        let Some(shared) = self.shared.upgrade() else {
139            return Ok(None);
140        };
141        let name: Arc<str> = name.into().into();
142        let id = VideoInputId(shared.next_input_id.fetch_add(1, Ordering::Relaxed));
143        let input = Arc::new(VideoInput {
144            id,
145            latest_frame: ArcSwapOption::empty(),
146            layer: Mutex::new(layer),
147        });
148        shared
149            .inputs
150            .lock()
151            .unwrap()
152            .insert(name.clone(), input.clone());
153
154        Ok(Some(SwVideoCompositorInput {
155            sink: Box::new(SwVideoCompositorInputSink {
156                name: name.clone(),
157                pp_log: element_pp_log(ElementType::SwVideoCompositor, &name, None),
158                shared: self.shared.clone(),
159                input: Arc::downgrade(&input),
160            }),
161            layer: SwVideoLayerHandle {
162                id,
163                name,
164                input: Arc::downgrade(&input),
165            },
166        }))
167    }
168
169    /// Removes `name` immediately. Existing input sinks and layer handles
170    /// become disconnected and cannot affect a later same-name source.
171    pub fn remove_source(&self, name: &str) {
172        if let Some(shared) = self.shared.upgrade() {
173            shared.inputs.lock().unwrap().remove(name);
174        }
175    }
176
177    pub fn source_count(&self) -> usize {
178        self.shared
179            .upgrade()
180            .map(|shared| shared.inputs.lock().unwrap().len())
181            .unwrap_or(0)
182    }
183}
184
185/// Thread-safe runtime placement control for one compositor input.
186/// Retaining it does not keep the input or compositor alive.
187#[derive(Clone)]
188pub struct SwVideoLayerHandle {
189    id: VideoInputId,
190    name: Arc<str>,
191    input: Weak<VideoInput>,
192}
193
194impl SwVideoLayerHandle {
195    pub fn id(&self) -> VideoInputId {
196        self.id
197    }
198
199    pub fn name(&self) -> Arc<str> {
200        self.name.clone()
201    }
202
203    pub fn layer(&self) -> Option<VideoLayer> {
204        self.input
205            .upgrade()
206            .map(|input| *input.layer.lock().unwrap())
207    }
208
209    pub fn set_layer(&self, layer: VideoLayer) -> std::result::Result<(), SwVideoCompositorError> {
210        validate_layer(layer)?;
211        self.update(|current| *current = layer)
212    }
213
214    pub fn set_rect(&self, rect: VideoRect) -> std::result::Result<(), SwVideoCompositorError> {
215        validate_rect(rect)?;
216        self.update(|layer| layer.rect = rect)
217    }
218
219    pub fn set_opacity(&self, opacity: f32) -> std::result::Result<(), SwVideoCompositorError> {
220        validate_opacity(opacity)?;
221        self.update(|layer| layer.opacity = opacity)
222    }
223
224    pub fn set_z_index(&self, z_index: i32) -> std::result::Result<(), SwVideoCompositorError> {
225        self.update(|layer| layer.z_index = z_index)
226    }
227
228    pub fn set_visible(&self, visible: bool) -> std::result::Result<(), SwVideoCompositorError> {
229        self.update(|layer| layer.visible = visible)
230    }
231
232    pub fn set_fit(&self, fit: VideoFit) -> std::result::Result<(), SwVideoCompositorError> {
233        self.update(|layer| layer.fit = fit)
234    }
235
236    fn update(
237        &self,
238        update: impl FnOnce(&mut VideoLayer),
239    ) -> std::result::Result<(), SwVideoCompositorError> {
240        let input = self
241            .input
242            .upgrade()
243            .ok_or(SwVideoCompositorError::SourceRemoved)?;
244        update(&mut input.layer.lock().unwrap());
245        Ok(())
246    }
247}
248
249/// One terminal video input returned by
250/// [`SwVideoCompositorHandle::add_source`]. It stores only the latest frame,
251/// so a fast producer cannot build an unbounded queue behind a slower
252/// compositor output rate.
253pub struct SwVideoCompositorInputSink {
254    pp_log: PpLog,
255    name: Arc<str>,
256    shared: Weak<CompositorShared>,
257    input: Weak<VideoInput>,
258}
259
260impl SwVideoCompositorInputSink {
261    fn detach(&self) {
262        let (Some(shared), Some(input)) = (self.shared.upgrade(), self.input.upgrade()) else {
263            return;
264        };
265        let mut inputs = shared.inputs.lock().unwrap();
266        let is_current = inputs
267            .get(&self.name)
268            .is_some_and(|current| Arc::ptr_eq(current, &input));
269        if is_current {
270            inputs.remove(&self.name);
271        }
272    }
273}
274
275impl Element for SwVideoCompositorInputSink {
276    fn name(&self) -> Arc<str> {
277        self.name.clone()
278    }
279
280    fn element_type(&self) -> ElementType {
281        ElementType::SwVideoCompositor
282    }
283
284    fn pp_log(&self) -> &PpLog {
285        &self.pp_log
286    }
287
288    fn pp_log_mut(&mut self) -> &mut PpLog {
289        &mut self.pp_log
290    }
291}
292
293impl Sink for SwVideoCompositorInputSink {
294    fn consume(&mut self, buf: MediaBuffer) -> Result<()> {
295        let Some(input) = self.input.upgrade() else {
296            return Ok(());
297        };
298        match buf {
299            MediaBuffer::Video(frame) => {
300                input.latest_frame.store(Some(frame));
301                Ok(())
302            }
303            MediaBuffer::Eos => {
304                self.detach();
305                Ok(())
306            }
307            MediaBuffer::Packet(_) => {
308                Err(SwVideoCompositorError::UnsupportedBuffer("Packet").into())
309            }
310            MediaBuffer::Audio(_) => Err(SwVideoCompositorError::UnsupportedBuffer("Audio").into()),
311        }
312    }
313
314    fn control(&mut self, msg: ControlMsg) -> Result<()> {
315        match msg {
316            ControlMsg::Stop => self.detach(),
317            ControlMsg::Seek(_) => {
318                if let Some(input) = self.input.upgrade() {
319                    input.latest_frame.store(None);
320                }
321            }
322            ControlMsg::Pause | ControlMsg::Resume => {}
323        }
324        Ok(())
325    }
326}
327
328#[derive(Clone)]
329struct InputSnapshot {
330    id: VideoInputId,
331    layer: VideoLayer,
332    frame: Option<Arc<UnboundObjectPoolRef<ffmpeg::frame::Video>>>,
333}
334
335#[derive(Debug, Clone, Copy, PartialEq, Eq)]
336struct ScaleDefinition {
337    source_format: ffmpeg::format::Pixel,
338    source_width: u32,
339    source_height: u32,
340    target_width: u32,
341    target_height: u32,
342}
343
344struct InputScaler {
345    definition: Option<ScaleDefinition>,
346    context: Option<ffmpeg::software::scaling::Context>,
347    output: ffmpeg::frame::Video,
348}
349
350impl InputScaler {
351    fn new() -> Self {
352        Self {
353            definition: None,
354            context: None,
355            output: ffmpeg::frame::Video::empty(),
356        }
357    }
358
359    fn scale(
360        &mut self,
361        source: &ffmpeg::frame::Video,
362        width: u32,
363        height: u32,
364    ) -> std::result::Result<&ffmpeg::frame::Video, ffmpeg::Error> {
365        let definition = ScaleDefinition {
366            source_format: source.format(),
367            source_width: source.width(),
368            source_height: source.height(),
369            target_width: width,
370            target_height: height,
371        };
372        if self.definition != Some(definition) {
373            self.context = Some(ffmpeg::software::scaling::Context::get(
374                definition.source_format,
375                definition.source_width,
376                definition.source_height,
377                ffmpeg::format::Pixel::BGRA,
378                width,
379                height,
380                ffmpeg::software::scaling::Flags::BILINEAR,
381            )?);
382            self.output = ffmpeg::frame::Video::new(ffmpeg::format::Pixel::BGRA, width, height);
383            self.definition = Some(definition);
384        }
385        self.context
386            .as_mut()
387            .expect("created with a changed definition or retained from a matching one")
388            .run(source, &mut self.output)?;
389        Ok(&self.output)
390    }
391}
392
393/// Composites the latest frames from any number of independent input
394/// pipelines into one fixed-rate opaque BGRA video stream.
395///
396/// Like [`crate::elements::AudioMixer`], this is a [`SourceElement`], not
397/// a conventional one-input filter: upstream pipelines terminate at the
398/// sinks returned by [`SwVideoCompositorHandle::add_source`], while this
399/// element's own pipeline drives output on its independent clock. Input
400/// frame PTS values therefore do not become output PTS; output advances by
401/// one tick in [`SwVideoCompositor::time_base`] for every composed frame.
402pub struct SwVideoCompositor {
403    pp_log: PpLog,
404    name: Arc<str>,
405    shared: Arc<CompositorShared>,
406    options: VideoCompositorOptions,
407    frame_interval: Duration,
408    frame_index: i64,
409    scalers: HashMap<VideoInputId, InputScaler>,
410    output_pool: UnboundObjectPool<ffmpeg::frame::Video>,
411    pad: SrcPad,
412}
413
414// SAFETY: `SwsContext` has no thread affinity and every scaling context is
415// exclusively accessed through `&mut self` on the compositor's one source
416// thread. ffmpeg-next simply omits Send for this wrapper, as with SwScaler.
417unsafe impl Send for SwVideoCompositor {}
418
419impl SwVideoCompositor {
420    pub fn new(
421        name: impl Into<String>,
422        options: VideoCompositorOptions,
423    ) -> std::result::Result<(Self, SwVideoCompositorHandle), SwVideoCompositorError> {
424        validate_output_options(options)?;
425        let name: Arc<str> = name.into().into();
426        let pp_log = element_pp_log(ElementType::SwVideoCompositor, &name, None);
427        let shared = Arc::new(CompositorShared {
428            inputs: Mutex::new(HashMap::new()),
429            next_input_id: AtomicU64::new(1),
430        });
431        let frame_interval = Duration::from_secs_f64(
432            options.frame_rate.denominator() as f64 / options.frame_rate.numerator() as f64,
433        );
434        let (width, height) = (options.width, options.height);
435        let output_pool = UnboundObjectPool::new(
436            OUTPUT_POOL_SIZE,
437            move || ffmpeg::frame::Video::new(ffmpeg::format::Pixel::BGRA, width, height),
438            |_| {},
439        );
440        pp_info!(
441            pp_log: &pp_log,
442            "created: {}x{}, frame_rate={}, format=BGRA",
443            width,
444            height,
445            options.frame_rate
446        );
447        Ok((
448            Self {
449                name: name.clone(),
450                pp_log,
451                shared: shared.clone(),
452                options,
453                frame_interval,
454                frame_index: 0,
455                scalers: HashMap::new(),
456                output_pool,
457                pad: SrcPad::new(format!("{name}_src")),
458            },
459            SwVideoCompositorHandle {
460                shared: Arc::downgrade(&shared),
461            },
462        ))
463    }
464
465    pub fn format(&self) -> ffmpeg::format::Pixel {
466        ffmpeg::format::Pixel::BGRA
467    }
468
469    pub fn width(&self) -> u32 {
470        self.options.width
471    }
472
473    pub fn height(&self) -> u32 {
474        self.options.height
475    }
476
477    pub fn frame_rate(&self) -> ffmpeg::Rational {
478        self.options.frame_rate
479    }
480
481    pub fn time_base(&self) -> ffmpeg::Rational {
482        ffmpeg::Rational::new(
483            self.options.frame_rate.denominator(),
484            self.options.frame_rate.numerator(),
485        )
486    }
487
488    fn snapshots(&self) -> Vec<InputSnapshot> {
489        let inputs: Vec<_> = self
490            .shared
491            .inputs
492            .lock()
493            .unwrap()
494            .values()
495            .cloned()
496            .collect();
497        inputs
498            .into_iter()
499            .map(|input| InputSnapshot {
500                id: input.id,
501                layer: *input.layer.lock().unwrap(),
502                frame: input.latest_frame.load_full(),
503            })
504            .collect()
505    }
506
507    fn compose_frame(
508        &mut self,
509    ) -> std::result::Result<UnboundObjectPoolRef<ffmpeg::frame::Video>, SwVideoCompositorError>
510    {
511        let mut snapshots = self.snapshots();
512        let active: HashSet<_> = snapshots.iter().map(|snapshot| snapshot.id).collect();
513        self.scalers.retain(|id, _| active.contains(id));
514        snapshots.sort_by(|left, right| {
515            left.layer
516                .z_index
517                .cmp(&right.layer.z_index)
518                .then_with(|| left.id.cmp(&right.id))
519        });
520
521        let mut output = self.output_pool.get();
522        fill_background(&mut output, self.options.background);
523        for snapshot in snapshots {
524            if !snapshot.layer.visible || snapshot.layer.opacity == 0.0 {
525                continue;
526            }
527            let Some(frame) = snapshot.frame else {
528                continue;
529            };
530            let geometry = layer_geometry(
531                frame.width(),
532                frame.height(),
533                snapshot.layer.rect,
534                snapshot.layer.fit,
535            )?;
536            let scaled = self
537                .scalers
538                .entry(snapshot.id)
539                .or_insert_with(InputScaler::new)
540                .scale(&frame, geometry.image_width, geometry.image_height)
541                .map_err(SwVideoCompositorError::from)?;
542            blend_bgra(&mut output, scaled, geometry, snapshot.layer.opacity);
543        }
544        output.set_pts(Some(self.frame_index));
545        self.frame_index += 1;
546        Ok(output)
547    }
548
549    fn push_frame(&mut self, bus: &Bus) -> std::result::Result<(), SwVideoCompositorError> {
550        let output = self.compose_frame()?;
551        if let Err(error) = self.pad.push(MediaBuffer::Video(Arc::new(output))) {
552            bus.post(
553                &self.pp_log,
554                BusEvent::Error {
555                    element_type: ElementType::SwVideoCompositor,
556                    name: self.name.clone(),
557                    error,
558                },
559            );
560        }
561        Ok(())
562    }
563}
564
565impl Element for SwVideoCompositor {
566    fn name(&self) -> Arc<str> {
567        self.name.clone()
568    }
569
570    fn element_type(&self) -> ElementType {
571        ElementType::SwVideoCompositor
572    }
573
574    fn pp_log(&self) -> &PpLog {
575        &self.pp_log
576    }
577
578    fn pp_log_mut(&mut self) -> &mut PpLog {
579        &mut self.pp_log
580    }
581}
582
583impl Source for SwVideoCompositor {
584    fn src_pads(&mut self) -> &mut [SrcPad] {
585        std::slice::from_mut(&mut self.pad)
586    }
587}
588
589impl SourceElement for SwVideoCompositor {
590    fn run(&mut self, control: &ControlReceiver, bus: &Bus) -> Result<()> {
591        pp_info!(self, "started");
592        let mut schedule = PeriodicSchedule::new(self.frame_interval, Instant::now());
593        loop {
594            let outcome = drain_control(control, self, bus)?;
595            if outcome.stopped {
596                pp_info!(self, "stopped");
597                return Ok(());
598            }
599            if outcome.paused_for > Duration::ZERO {
600                schedule.resume_after_pause(outcome.paused_for, Instant::now());
601            }
602
603            let now = Instant::now();
604            if !schedule.is_due(now) {
605                thread::sleep(schedule.remaining(now).min(CONTROL_POLL_INTERVAL));
606                continue;
607            }
608
609            self.push_frame(bus)?;
610            schedule.advance_after_tick(Instant::now());
611        }
612    }
613
614    fn seek(&mut self, _target: Duration) -> Result<Duration> {
615        Err(SwVideoCompositorError::SeekUnsupported.into())
616    }
617}
618
619fn validate_output_options(
620    options: VideoCompositorOptions,
621) -> std::result::Result<(), SwVideoCompositorError> {
622    if options.width == 0
623        || options.height == 0
624        || options.width > MAX_DIMENSION
625        || options.height > MAX_DIMENSION
626    {
627        return Err(SwVideoCompositorError::InvalidOutputDimensions {
628            width: options.width,
629            height: options.height,
630        });
631    }
632    if options.frame_rate.numerator() <= 0 || options.frame_rate.denominator() <= 0 {
633        return Err(SwVideoCompositorError::InvalidFrameRate(options.frame_rate));
634    }
635    Ok(())
636}
637
638/// Thin adapters over the shared, backend-agnostic logic in
639/// [`super::video_layer`] — translate its [`VideoLayerError`] into this
640/// backend's own [`SwVideoCompositorError`] variants so every existing call
641/// site/error consumer here keeps seeing the same error shape it always
642/// has.
643fn map_layer_error(error: VideoLayerError) -> SwVideoCompositorError {
644    match error {
645        VideoLayerError::InvalidDimensions { width, height } => {
646            SwVideoCompositorError::InvalidLayerDimensions { width, height }
647        }
648        VideoLayerError::InvalidOpacity(opacity) => SwVideoCompositorError::InvalidOpacity(opacity),
649        VideoLayerError::InvalidInputDimensions { width, height } => {
650            SwVideoCompositorError::InvalidInputDimensions { width, height }
651        }
652        VideoLayerError::ScaledLayerTooLarge { width, height } => {
653            SwVideoCompositorError::ScaledLayerTooLarge { width, height }
654        }
655    }
656}
657
658fn validate_layer(layer: VideoLayer) -> std::result::Result<(), SwVideoCompositorError> {
659    video_layer::validate_layer(layer).map_err(map_layer_error)
660}
661
662fn validate_rect(rect: VideoRect) -> std::result::Result<(), SwVideoCompositorError> {
663    video_layer::validate_rect(rect).map_err(map_layer_error)
664}
665
666fn validate_opacity(opacity: f32) -> std::result::Result<(), SwVideoCompositorError> {
667    video_layer::validate_opacity(opacity).map_err(map_layer_error)
668}
669
670fn layer_geometry(
671    source_width: u32,
672    source_height: u32,
673    rect: VideoRect,
674    fit: VideoFit,
675) -> std::result::Result<LayerGeometry, SwVideoCompositorError> {
676    video_layer::layer_geometry(source_width, source_height, rect, fit).map_err(map_layer_error)
677}
678
679fn fill_background(frame: &mut ffmpeg::frame::Video, color: Color) {
680    let width = frame.width() as usize;
681    let height = frame.height() as usize;
682    let stride = frame.stride(0);
683    let data = frame.data_mut(0);
684    for row in 0..height {
685        for pixel in data[row * stride..row * stride + width * 4].chunks_exact_mut(4) {
686            pixel.copy_from_slice(&[color.blue, color.green, color.red, 255]);
687        }
688    }
689}
690
691fn blend_bgra(
692    destination: &mut ffmpeg::frame::Video,
693    source: &ffmpeg::frame::Video,
694    geometry: LayerGeometry,
695    opacity: f32,
696) {
697    let output_width = i64::from(destination.width());
698    let output_height = i64::from(destination.height());
699    let clip_left = i64::from(geometry.clip.x).max(0);
700    let clip_top = i64::from(geometry.clip.y).max(0);
701    let clip_right =
702        (i64::from(geometry.clip.x) + i64::from(geometry.clip.width)).min(output_width);
703    let clip_bottom =
704        (i64::from(geometry.clip.y) + i64::from(geometry.clip.height)).min(output_height);
705    let left = geometry.image_x.max(clip_left);
706    let top = geometry.image_y.max(clip_top);
707    let right = (geometry.image_x + i64::from(geometry.image_width)).min(clip_right);
708    let bottom = (geometry.image_y + i64::from(geometry.image_height)).min(clip_bottom);
709    if left >= right || top >= bottom {
710        return;
711    }
712
713    let source_stride = source.stride(0);
714    let destination_stride = destination.stride(0);
715    let source_data = source.data(0);
716    let destination_data = destination.data_mut(0);
717    for output_y in top..bottom {
718        let source_y = (output_y - geometry.image_y) as usize;
719        let destination_y = output_y as usize;
720        for output_x in left..right {
721            let source_x = (output_x - geometry.image_x) as usize;
722            let destination_x = output_x as usize;
723            let source_offset = source_y * source_stride + source_x * 4;
724            let destination_offset = destination_y * destination_stride + destination_x * 4;
725            let source_pixel = &source_data[source_offset..source_offset + 4];
726            let destination_pixel =
727                &mut destination_data[destination_offset..destination_offset + 4];
728            let alpha = (f32::from(source_pixel[3]) / 255.0) * opacity;
729            let inverse = 1.0 - alpha;
730            for channel in 0..3 {
731                destination_pixel[channel] = (f32::from(source_pixel[channel]) * alpha
732                    + f32::from(destination_pixel[channel]) * inverse)
733                    .round()
734                    .clamp(0.0, 255.0) as u8;
735            }
736            destination_pixel[3] = 255;
737        }
738    }
739}
740
741#[cfg(test)]
742mod tests {
743    use std::{
744        sync::{
745            Mutex as StdMutex,
746            atomic::{AtomicBool, Ordering as AtomicOrdering},
747        },
748        thread,
749    };
750
751    use super::*;
752
753    struct CapturingSink {
754        pp_log: PpLog,
755        received: Arc<StdMutex<Vec<MediaBuffer>>>,
756    }
757
758    impl Element for CapturingSink {
759        fn name(&self) -> Arc<str> {
760            "capture".into()
761        }
762
763        fn element_type(&self) -> ElementType {
764            ElementType::Other
765        }
766
767        fn pp_log(&self) -> &PpLog {
768            &self.pp_log
769        }
770
771        fn pp_log_mut(&mut self) -> &mut PpLog {
772            &mut self.pp_log
773        }
774    }
775
776    impl Sink for CapturingSink {
777        fn consume(&mut self, buf: MediaBuffer) -> Result<()> {
778            self.received.lock().unwrap().push(buf);
779            Ok(())
780        }
781
782        fn control(&mut self, _msg: ControlMsg) -> Result<()> {
783            Ok(())
784        }
785    }
786
787    fn options(width: u32, height: u32) -> VideoCompositorOptions {
788        VideoCompositorOptions {
789            width,
790            height,
791            frame_rate: ffmpeg::Rational::new(30, 1),
792            background: Color::BLACK,
793        }
794    }
795
796    fn solid_frame(
797        width: u32,
798        height: u32,
799        color: Color,
800    ) -> Arc<UnboundObjectPoolRef<ffmpeg::frame::Video>> {
801        let pool = UnboundObjectPool::new(
802            0,
803            move || ffmpeg::frame::Video::new(ffmpeg::format::Pixel::BGRA, width, height),
804            |_| {},
805        );
806        let mut frame = pool.get();
807        fill_background(&mut frame, color);
808        Arc::new(frame)
809    }
810
811    fn pixel(frame: &ffmpeg::frame::Video, x: usize, y: usize) -> [u8; 4] {
812        let offset = y * frame.stride(0) + x * 4;
813        frame.data(0)[offset..offset + 4].try_into().unwrap()
814    }
815
816    fn input(
817        handle: &SwVideoCompositorHandle,
818        name: &str,
819        layer: VideoLayer,
820    ) -> (Box<dyn Sink>, SwVideoLayerHandle) {
821        let input = handle.add_source(name, layer).unwrap().unwrap();
822        (input.sink, input.layer)
823    }
824
825    #[test]
826    fn composes_inputs_in_z_order_and_preserves_output_contract() {
827        let (mut compositor, handle) = SwVideoCompositor::new("compositor", options(4, 4)).unwrap();
828        let mut background = VideoLayer::new(VideoRect::new(0, 0, 4, 4));
829        background.fit = VideoFit::Stretch;
830        let (mut red_sink, _) = input(&handle, "red", background);
831        let mut overlay = VideoLayer::new(VideoRect::new(1, 1, 2, 2));
832        overlay.z_index = 1;
833        overlay.fit = VideoFit::Stretch;
834        let (mut blue_sink, _) = input(&handle, "blue", overlay);
835        red_sink
836            .consume(MediaBuffer::Video(solid_frame(4, 4, Color::new(255, 0, 0))))
837            .unwrap();
838        blue_sink
839            .consume(MediaBuffer::Video(solid_frame(2, 2, Color::new(0, 0, 255))))
840            .unwrap();
841
842        let frame = compositor.compose_frame().unwrap();
843        assert_eq!(frame.format(), ffmpeg::format::Pixel::BGRA);
844        assert_eq!((frame.width(), frame.height()), (4, 4));
845        assert_eq!(frame.pts(), Some(0));
846        assert_eq!(pixel(&frame, 0, 0), [0, 0, 255, 255]);
847        assert_eq!(pixel(&frame, 1, 1), [255, 0, 0, 255]);
848    }
849
850    #[test]
851    fn layer_handle_moves_blends_and_hides_a_live_source() {
852        let (mut compositor, handle) = SwVideoCompositor::new("compositor", options(3, 1)).unwrap();
853        let layer = VideoLayer::new(VideoRect::new(0, 0, 1, 1));
854        let (mut sink, layer_handle) = input(&handle, "white", layer);
855        sink.consume(MediaBuffer::Video(solid_frame(1, 1, Color::WHITE)))
856            .unwrap();
857
858        layer_handle.set_rect(VideoRect::new(1, 0, 1, 1)).unwrap();
859        layer_handle.set_opacity(0.5).unwrap();
860        let blended = compositor.compose_frame().unwrap();
861        assert_eq!(pixel(&blended, 0, 0), [0, 0, 0, 255]);
862        assert_eq!(pixel(&blended, 1, 0), [128, 128, 128, 255]);
863
864        layer_handle.set_visible(false).unwrap();
865        let hidden = compositor.compose_frame().unwrap();
866        assert_eq!(pixel(&hidden, 1, 0), [0, 0, 0, 255]);
867        assert_eq!(hidden.pts(), Some(1));
868    }
869
870    #[test]
871    fn input_keeps_only_the_latest_frame() {
872        let (mut compositor, handle) = SwVideoCompositor::new("compositor", options(1, 1)).unwrap();
873        let (mut sink, _) = input(
874            &handle,
875            "latest",
876            VideoLayer::new(VideoRect::new(0, 0, 1, 1)),
877        );
878        sink.consume(MediaBuffer::Video(solid_frame(1, 1, Color::new(255, 0, 0))))
879            .unwrap();
880        sink.consume(MediaBuffer::Video(solid_frame(1, 1, Color::new(0, 255, 0))))
881            .unwrap();
882
883        let frame = compositor.compose_frame().unwrap();
884        assert_eq!(pixel(&frame, 0, 0), [0, 255, 0, 255]);
885    }
886
887    #[test]
888    fn frame_replacement_and_composition_run_concurrently() {
889        let (mut compositor, handle) = SwVideoCompositor::new("compositor", options(1, 1)).unwrap();
890        let (mut sink, _) = input(&handle, "live", VideoLayer::new(VideoRect::new(0, 0, 1, 1)));
891        let red = solid_frame(1, 1, Color::new(255, 0, 0));
892        let green = solid_frame(1, 1, Color::new(0, 255, 0));
893        let done = Arc::new(AtomicBool::new(false));
894        let producer_done = done.clone();
895        let producer = thread::spawn(move || {
896            for index in 0..2_000 {
897                let frame = if index % 2 == 0 {
898                    red.clone()
899                } else {
900                    green.clone()
901                };
902                sink.consume(MediaBuffer::Video(frame)).unwrap();
903            }
904            // Make the final observable value deterministic after the
905            // concurrent replacement phase ends.
906            sink.consume(MediaBuffer::Video(green)).unwrap();
907            producer_done.store(true, AtomicOrdering::Release);
908        });
909
910        while !done.load(AtomicOrdering::Acquire) {
911            let frame = compositor.compose_frame().unwrap();
912            assert!(matches!(
913                pixel(&frame, 0, 0),
914                [0, 0, 0, 255] | [0, 0, 255, 255] | [0, 255, 0, 255]
915            ));
916        }
917        producer.join().unwrap();
918        let final_frame = compositor.compose_frame().unwrap();
919        assert_eq!(pixel(&final_frame, 0, 0), [0, 255, 0, 255]);
920    }
921
922    #[test]
923    fn replacing_a_name_invalidates_old_sink_and_layer_handle() {
924        let (mut compositor, handle) = SwVideoCompositor::new("compositor", options(1, 1)).unwrap();
925        let layer = VideoLayer::new(VideoRect::new(0, 0, 1, 1));
926        let (mut old_sink, old_layer) = input(&handle, "camera", layer);
927        let (mut new_sink, _) = input(&handle, "camera", layer);
928        assert!(matches!(
929            old_layer.set_visible(false),
930            Err(SwVideoCompositorError::SourceRemoved)
931        ));
932        old_sink
933            .consume(MediaBuffer::Video(solid_frame(1, 1, Color::new(255, 0, 0))))
934            .unwrap();
935        new_sink
936            .consume(MediaBuffer::Video(solid_frame(1, 1, Color::new(0, 0, 255))))
937            .unwrap();
938
939        let frame = compositor.compose_frame().unwrap();
940        assert_eq!(pixel(&frame, 0, 0), [255, 0, 0, 255]);
941        assert_eq!(handle.source_count(), 1);
942    }
943
944    #[test]
945    fn stop_removes_only_the_current_registration() {
946        let (_compositor, handle) = SwVideoCompositor::new("compositor", options(1, 1)).unwrap();
947        let layer = VideoLayer::new(VideoRect::new(0, 0, 1, 1));
948        let (mut old_sink, _) = input(&handle, "camera", layer);
949        let (_new_sink, _) = input(&handle, "camera", layer);
950        old_sink.control(ControlMsg::Stop).unwrap();
951        assert_eq!(handle.source_count(), 1);
952
953        handle.remove_source("camera");
954        assert_eq!(handle.source_count(), 0);
955    }
956
957    #[test]
958    fn contain_and_cover_preserve_aspect_ratio() {
959        let rect = VideoRect::new(10, 20, 100, 100);
960        let contain = layer_geometry(160, 90, rect, VideoFit::Contain).unwrap();
961        assert_eq!((contain.image_width, contain.image_height), (100, 56));
962        assert_eq!((contain.image_x, contain.image_y), (10, 42));
963
964        let cover = layer_geometry(160, 90, rect, VideoFit::Cover).unwrap();
965        assert_eq!((cover.image_width, cover.image_height), (178, 100));
966        assert_eq!((cover.image_x, cover.image_y), (-29, 20));
967    }
968
969    #[test]
970    fn rejects_invalid_layers_and_non_video_buffers() {
971        let (_compositor, handle) = SwVideoCompositor::new("compositor", options(1, 1)).unwrap();
972        let invalid = VideoLayer {
973            opacity: 1.5,
974            ..VideoLayer::new(VideoRect::new(0, 0, 1, 1))
975        };
976        assert!(matches!(
977            handle.add_source("invalid", invalid),
978            Err(SwVideoCompositorError::InvalidOpacity(1.5))
979        ));
980
981        let (mut sink, _) = input(
982            &handle,
983            "valid",
984            VideoLayer::new(VideoRect::new(0, 0, 1, 1)),
985        );
986        let error = sink
987            .consume(MediaBuffer::Packet(Arc::new(ffmpeg::Packet::empty())))
988            .unwrap_err();
989        assert!(matches!(
990            error,
991            crate::Error::SwVideoCompositorError(SwVideoCompositorError::UnsupportedBuffer(
992                "Packet"
993            ))
994        ));
995    }
996
997    #[test]
998    fn pushes_fixed_format_frames_with_contiguous_pts() {
999        let (mut compositor, _) = SwVideoCompositor::new("compositor", options(2, 2)).unwrap();
1000        let received = Arc::new(StdMutex::new(Vec::new()));
1001        compositor.src_pads()[0].link(Box::new(CapturingSink {
1002            received: received.clone(),
1003            pp_log: element_pp_log(ElementType::Other, "capture", None),
1004        }));
1005        let (bus, _) = Bus::new();
1006        compositor.push_frame(&bus).unwrap();
1007        compositor.push_frame(&bus).unwrap();
1008
1009        let received = received.lock().unwrap();
1010        let pts: Vec<_> = received
1011            .iter()
1012            .filter_map(|buffer| match buffer {
1013                MediaBuffer::Video(frame) => Some(frame.pts()),
1014                _ => None,
1015            })
1016            .collect();
1017        assert_eq!(pts, vec![Some(0), Some(1)]);
1018    }
1019
1020    struct TimestampSink {
1021        pp_log: PpLog,
1022        tx: crossbeam_channel::Sender<Instant>,
1023    }
1024
1025    impl Element for TimestampSink {
1026        fn name(&self) -> Arc<str> {
1027            "timestamp-recorder".into()
1028        }
1029        fn element_type(&self) -> ElementType {
1030            ElementType::Other
1031        }
1032        fn pp_log(&self) -> &PpLog {
1033            &self.pp_log
1034        }
1035        fn pp_log_mut(&mut self) -> &mut PpLog {
1036            &mut self.pp_log
1037        }
1038    }
1039
1040    impl Sink for TimestampSink {
1041        fn consume(&mut self, buf: MediaBuffer) -> Result<()> {
1042            if matches!(buf, MediaBuffer::Video(_)) {
1043                let _ = self.tx.send(Instant::now());
1044            }
1045            Ok(())
1046        }
1047        fn control(&mut self, _msg: ControlMsg) -> Result<()> {
1048            Ok(())
1049        }
1050    }
1051
1052    /// Regression test: `SwVideoCompositor::run` never folded
1053    /// `ControlOutcome::paused_for` back into `next_due` — a `Pause` let
1054    /// real time blow straight past the stale deadline, so the loop
1055    /// iteration right after `Resume` always found `next_due` already in
1056    /// the past and pushed immediately, resetting the output cadence's
1057    /// phase to the resume instant instead of preserving wherever it was
1058    /// before the freeze. A slow 10fps (100ms/frame) rate keeps the
1059    /// expected gap (near-zero vs. near-one-interval) well clear of
1060    /// scheduling jitter. Pausing is triggered synchronously right after a
1061    /// frame is observed, so `next_due` is a known ~100ms away the instant
1062    /// `Pause` is drained (`run`'s own control check only happens at the
1063    /// top of the loop, after that deadline has already been advanced).
1064    #[test]
1065    fn resuming_after_a_pause_preserves_output_phase() {
1066        use crate::pipeline::Pipeline;
1067
1068        let (tx, rx) = crossbeam_channel::unbounded();
1069        let sink = TimestampSink {
1070            tx,
1071            pp_log: element_pp_log(ElementType::Other, "timestamp-recorder", None),
1072        };
1073        let (compositor, _handle) = SwVideoCompositor::new(
1074            "compositor",
1075            VideoCompositorOptions {
1076                frame_rate: ffmpeg::Rational::new(10, 1),
1077                ..options(2, 2)
1078            },
1079        )
1080        .unwrap();
1081
1082        let pipeline = Pipeline::new("phase-test", compositor, |source, ctx| {
1083            let branch = ctx.branch().to(Box::new(sink))?;
1084            ctx.attach(source, 0, branch)?;
1085            Ok(())
1086        })
1087        .expect("test pipeline wiring must succeed");
1088
1089        pipeline.run();
1090        // Warm up, then pause the instant a frame is observed — `next_due`
1091        // is then a known one interval away.
1092        for _ in 0..2 {
1093            rx.recv_timeout(Duration::from_millis(500))
1094                .expect("expected steady frames before pausing");
1095        }
1096        pipeline.pause();
1097        thread::sleep(Duration::from_millis(500));
1098
1099        let resumed_at = Instant::now();
1100        pipeline.resume();
1101        let first_after_resume = rx
1102            .recv_timeout(Duration::from_millis(500))
1103            .expect("expected a frame after resume");
1104        pipeline.stop();
1105        pipeline.bus().log_events();
1106
1107        let gap = first_after_resume.saturating_duration_since(resumed_at);
1108        assert!(
1109            gap >= Duration::from_millis(50),
1110            "expected the post-pause frame to land close to a full 100ms \
1111             interval after resume (phase preserved from before the \
1112             pause), not almost immediately (phase reset to the resume \
1113             instant): got {gap:?}"
1114        );
1115    }
1116}
media_pp/elements/source/compositor/sw_video_compositor.rs

media_pp/elements/source/compositor/
sw_video_compositor.rs
