media_pp/elements/sink/
frame_counter.rs

1use std::sync::{
2    Arc,
3    atomic::{AtomicUsize, Ordering},
4};
5
6use crate::pp_log::{PpLog, pp_info};
7
8use crate::{
9    buffer::MediaBuffer,
10    contract::{InputContract, MediaKindSet, MemoryDomainSet, PortContract},
11    control::ControlMsg,
12    element::{Element, ElementType, Sink, element_pp_log},
13    error::Result,
14};
15
16/// Terminal sink that counts decoded frames (video or audio). Backed by
17/// an `Arc<AtomicUsize>` so the count can be read from outside the
18/// pipeline even when this sink ends up running on a `Queue` worker
19/// thread.
20pub struct FrameCounter {
21    pp_log: PpLog,
22    name: Arc<str>,
23    count: Arc<AtomicUsize>,
24}
25
26impl FrameCounter {
27    /// Creates a sink and a shared counter that increments for each video frame.
28    pub fn new(name: impl Into<String>) -> (Self, Arc<AtomicUsize>) {
29        let count = Arc::new(AtomicUsize::new(0));
30        let name: Arc<str> = name.into().into();
31        let pp_log = element_pp_log(ElementType::FrameCounter, &name, None);
32        pp_info!(pp_log: &pp_log, "created");
33        (
34            Self {
35                name,
36                pp_log,
37                count: count.clone(),
38            },
39            count,
40        )
41    }
42}
43
44impl Element for FrameCounter {
45    fn name(&self) -> Arc<str> {
46        self.name.clone()
47    }
48
49    fn element_type(&self) -> ElementType {
50        ElementType::FrameCounter
51    }
52
53    fn pp_log(&self) -> &PpLog {
54        &self.pp_log
55    }
56
57    fn pp_log_mut(&mut self) -> &mut PpLog {
58        &mut self.pp_log
59    }
60}
61
62impl Sink for FrameCounter {
63    /// Counts decoded buffers of either medium — it only tallies them,
64    /// so it neither reads the samples nor cares which memory they live
65    /// in. PacketCounter is the encoded-side counterpart.
66    fn input_contract(&self) -> InputContract {
67        InputContract::Fixed(PortContract::Frames(
68            MediaKindSet::FRAMES,
69            MemoryDomainSet::ALL,
70        ))
71    }
72
73    fn consume(&mut self, buf: MediaBuffer) -> Result<()> {
74        if let MediaBuffer::Video(_) | MediaBuffer::Audio(_) = buf {
75            self.count.fetch_add(1, Ordering::Relaxed);
76        }
77        Ok(())
78    }
79
80    fn control(&mut self, _msg: ControlMsg) -> Result<()> {
81        // Terminal, nothing to flush or forward.
82        Ok(())
83    }
84}
media_pp/elements/sink/frame_counter.rs

media_pp/elements/sink/
frame_counter.rs
