media_pp/elements/sink/
packet_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, PortContract},
11    control::ControlMsg,
12    element::{Element, ElementType, Sink, element_pp_log},
13    error::Result,
14};
15
16/// Terminal sink that just counts packets. Backed by an `Arc<AtomicUsize>`
17/// so the count can be read from outside the pipeline even when this sink
18/// ends up running on a `Queue` worker thread.
19pub struct PacketCounter {
20    pp_log: PpLog,
21    name: Arc<str>,
22    count: Arc<AtomicUsize>,
23}
24
25impl PacketCounter {
26    /// Creates a sink and a shared counter that increments for each packet.
27    pub fn new(name: impl Into<String>) -> (Self, Arc<AtomicUsize>) {
28        let count = Arc::new(AtomicUsize::new(0));
29        let name: Arc<str> = name.into().into();
30        let pp_log = element_pp_log(ElementType::PacketCounter, &name, None);
31        pp_info!(pp_log: &pp_log, "created");
32        (
33            Self {
34                name,
35                pp_log,
36                count: count.clone(),
37            },
38            count,
39        )
40    }
41}
42
43impl Element for PacketCounter {
44    fn name(&self) -> Arc<str> {
45        self.name.clone()
46    }
47
48    fn element_type(&self) -> ElementType {
49        ElementType::PacketCounter
50    }
51
52    fn pp_log(&self) -> &PpLog {
53        &self.pp_log
54    }
55
56    fn pp_log_mut(&mut self) -> &mut PpLog {
57        &mut self.pp_log
58    }
59}
60
61impl Sink for PacketCounter {
62    /// Counts encoded packets specifically — FrameCounter is the decoded-side counterpart.
63    fn input_contract(&self) -> InputContract {
64        InputContract::Fixed(PortContract::Packets(MediaKindSet::PACKETS))
65    }
66
67    fn consume(&mut self, buf: MediaBuffer) -> Result<()> {
68        if let MediaBuffer::Packet(_) = buf {
69            self.count.fetch_add(1, Ordering::Relaxed);
70        }
71        Ok(())
72    }
73
74    fn control(&mut self, _msg: ControlMsg) -> Result<()> {
75        // Terminal, nothing to flush or forward.
76        Ok(())
77    }
78}
media_pp/elements/sink/packet_counter.rs

media_pp/elements/sink/
packet_counter.rs
