rtc_interceptor/
noop.rs

1//! NoOp Interceptor - A pass-through terminal for interceptor chains.
2
3use crate::stream_info::StreamInfo;
4use crate::{Interceptor, Packet, TaggedPacket};
5use shared::error::Error;
6use std::collections::VecDeque;
7use std::time::Instant;
8
9/// A no-operation interceptor that simply queues messages for pass-through.
10///
11/// `NoopInterceptor` serves as the innermost layer of an interceptor chain.
12/// It accepts messages via `handle_read`/`handle_write`/etc and returns them
13/// unchanged via `poll_read`/`poll_write`/etc.
14///
15/// # Example
16///
17/// ```
18/// use rtc_interceptor::{NoopInterceptor, Packet, TaggedPacket};
19/// use sansio::Protocol;
20/// use std::time::Instant;
21///
22/// let mut noop = NoopInterceptor::new();
23/// noop.handle_read(TaggedPacket {
24///     now: Instant::now(),
25///     transport: Default::default(),
26///     message: Packet::Rtp(rtp::Packet::default()),
27/// })
28/// .unwrap();
29/// assert!(noop.poll_read().is_some());
30/// ```
31pub struct NoopInterceptor {
32    read_queue: VecDeque<TaggedPacket>,
33    write_queue: VecDeque<TaggedPacket>,
34}
35
36impl NoopInterceptor {
37    /// Create a new NoopInterceptor.
38    pub fn new() -> Self {
39        Self {
40            read_queue: VecDeque::new(),
41            write_queue: VecDeque::new(),
42        }
43    }
44}
45
46impl Default for NoopInterceptor {
47    fn default() -> Self {
48        Self::new()
49    }
50}
51
52impl sansio::Protocol<TaggedPacket, TaggedPacket, ()> for NoopInterceptor {
53    type Rout = TaggedPacket;
54    type Wout = TaggedPacket;
55    type Eout = ();
56    type Error = Error;
57    type Time = Instant;
58
59    fn handle_read(&mut self, msg: TaggedPacket) -> Result<(), Self::Error> {
60        if let Packet::Rtp(_) = &msg.message {
61            self.read_queue.push_back(msg);
62        }
63        // RTCP message read must end here. If any rtcp packet needs to be forwarded to PeerConnection,
64        // just add a new interceptor to forward it by using self.interceptor.poll_read()
65        Ok(())
66    }
67
68    fn poll_read(&mut self) -> Option<Self::Rout> {
69        self.read_queue.pop_front()
70    }
71
72    fn handle_write(&mut self, msg: TaggedPacket) -> Result<(), Self::Error> {
73        self.write_queue.push_back(msg);
74        Ok(())
75    }
76
77    fn poll_write(&mut self) -> Option<Self::Wout> {
78        self.write_queue.pop_front()
79    }
80
81    fn handle_event(&mut self, _evt: ()) -> Result<(), Self::Error> {
82        Ok(())
83    }
84
85    fn poll_event(&mut self) -> Option<Self::Eout> {
86        None
87    }
88
89    fn handle_timeout(&mut self, _now: Self::Time) -> Result<(), Self::Error> {
90        Ok(())
91    }
92
93    fn poll_timeout(&mut self) -> Option<Self::Time> {
94        None
95    }
96
97    fn close(&mut self) -> Result<(), Self::Error> {
98        self.read_queue.clear();
99        self.write_queue.clear();
100        Ok(())
101    }
102}
103
104impl Interceptor for NoopInterceptor {
105    fn bind_local_stream(&mut self, _info: &StreamInfo) {}
106    fn unbind_local_stream(&mut self, _info: &StreamInfo) {}
107    fn bind_remote_stream(&mut self, _info: &StreamInfo) {}
108    fn unbind_remote_stream(&mut self, _info: &StreamInfo) {}
109}
110
111#[cfg(test)]
112mod tests {
113    use super::*;
114    use sansio::Protocol;
115
116    fn dummy_rtp_packet() -> TaggedPacket {
117        TaggedPacket {
118            now: Instant::now(),
119            transport: Default::default(),
120            message: crate::Packet::Rtp(rtp::Packet::default()),
121        }
122    }
123
124    fn dummy_rtcp_packet() -> TaggedPacket {
125        TaggedPacket {
126            now: Instant::now(),
127            transport: Default::default(),
128            message: crate::Packet::Rtcp(vec![Box::new(rtcp::raw_packet::RawPacket::default())]),
129        }
130    }
131
132    #[test]
133    fn test_noop_read_write() {
134        let mut noop = NoopInterceptor::new();
135
136        // Test read
137        let pkt1 = dummy_rtp_packet();
138        let pkt1_message = pkt1.message.clone();
139        let pkt2 = dummy_rtcp_packet();
140        noop.handle_read(pkt1).unwrap();
141        noop.handle_read(pkt2).unwrap();
142        assert_eq!(noop.poll_read().unwrap().message, pkt1_message);
143        assert!(noop.poll_read().is_none());
144
145        // Test write
146        let pkt3 = dummy_rtp_packet();
147        let pkt4 = dummy_rtp_packet();
148        let pkt3_message = pkt3.message.clone();
149        let pkt4_message = pkt4.message.clone();
150        noop.handle_write(pkt3).unwrap();
151        noop.handle_write(pkt4).unwrap();
152        assert_eq!(noop.poll_write().unwrap().message, pkt3_message);
153        assert_eq!(noop.poll_write().unwrap().message, pkt4_message);
154        assert!(noop.poll_write().is_none());
155    }
156
157    #[test]
158    fn test_noop_close_clears_queues() {
159        let mut noop = NoopInterceptor::new();
160
161        noop.handle_read(dummy_rtp_packet()).unwrap();
162        noop.handle_write(dummy_rtp_packet()).unwrap();
163
164        noop.close().unwrap();
165
166        assert!(noop.poll_read().is_none());
167        assert!(noop.poll_write().is_none());
168    }
169}
rtc_interceptor/noop.rs

rtc_interceptor/
noop.rs
