rtc_interceptor/intervalpli/
generator.rs

1//! Periodic Picture Loss Indication for bound remote streams.
2
3use super::stream_supports_pli;
4use crate::Interceptor;
5use crate::stream_info::StreamInfo;
6use crate::{AttributedPacket, Packet, TaggedPacket};
7use sansio::Protocol;
8use shared::TransportContext;
9use shared::error::Error;
10use std::collections::{BTreeSet, VecDeque};
11use std::time::{Duration, Instant};
12
13/// How often a keyframe is requested when no interval is configured.
14pub const DEFAULT_INTERVAL: Duration = Duration::from_secs(3);
15
16/// Requests a keyframe from every bound remote stream on a fixed interval.
17///
18/// # Where this belongs in the chain
19///
20/// It only generates, and what it generates is RTCP, so its position is fixed by nothing except
21/// being on the application side of the pacer — as everything that generates is, so that what it
22/// produces is still metered on its way out.
23///
24/// Sans-I/O has no clock of its own, so the interval is measured from the first `Instant` the
25/// interceptor is handed, whether that arrives via `handle_read` or `handle_timeout`.
26pub struct IntervalPliInterceptor {
27    interval: Duration,
28    /// Bound remote streams that negotiated PLI. Ordered so a tick emits deterministically.
29    streams: BTreeSet<u32>,
30    /// Streams bound but not yet sent their first request.
31    ///
32    /// Upstream asks immediately on bind, which needs an instant this interceptor does not have until
33    /// one is handed to it; these are flushed at the first opportunity.
34    pending_immediate: BTreeSet<u32>,
35    next_timeout: Option<Instant>,
36    write_queue: VecDeque<TaggedPacket>,
37    /// Inbound packets ready for the next interceptor.
38    read_queue: VecDeque<TaggedPacket>,
39}
40
41impl Default for IntervalPliInterceptor {
42    fn default() -> Self {
43        Self::new(DEFAULT_INTERVAL)
44    }
45}
46
47impl IntervalPliInterceptor {
48    /// A generator asking every bound stream for a keyframe every `interval`.
49    ///
50    /// A zero interval disables periodic requests entirely, matching upstream, which creates no
51    /// ticker when its interval is not positive.
52    pub fn new(interval: Duration) -> Self {
53        Self {
54            read_queue: VecDeque::new(),
55            interval,
56            streams: BTreeSet::new(),
57            pending_immediate: BTreeSet::new(),
58            next_timeout: None,
59            write_queue: VecDeque::new(),
60        }
61    }
62
63    /// The streams currently being asked for keyframes.
64    pub fn bound_streams(&self) -> impl Iterator<Item = u32> + '_ {
65        self.streams.iter().copied()
66    }
67
68    /// Queue one RTCP packet carrying a PLI per SSRC.
69    ///
70    /// One compound packet rather than one datagram each, as upstream does: they are all going to
71    /// the same peer at the same instant.
72    fn queue_plis(&mut self, now: Instant, ssrcs: &[u32]) {
73        if ssrcs.is_empty() {
74            return;
75        }
76
77        // Asking now satisfies the ask-on-bind these streams were still waiting for, so a forced
78        // request does not arrive alongside a duplicate of itself.
79        for ssrc in ssrcs {
80            self.pending_immediate.remove(ssrc);
81        }
82
83        let plis: Vec<Box<dyn rtcp::Packet>> = ssrcs
84            .iter()
85            .map(|&ssrc| {
86                Box::new(
87                    rtcp::payload_feedbacks::picture_loss_indication::PictureLossIndication {
88                        sender_ssrc: 0,
89                        media_ssrc: ssrc,
90                    },
91                ) as Box<dyn rtcp::Packet>
92            })
93            .collect();
94
95        self.write_queue.push_back(TaggedPacket {
96            now,
97            transport: TransportContext::default(),
98            message: AttributedPacket::new(Packet::Rtcp(plis)),
99        });
100    }
101
102    /// Arm the interval from the first instant this interceptor is handed, and send the first request
103    /// for any stream bound since.
104    fn observe(&mut self, now: Instant) {
105        if !self.pending_immediate.is_empty() {
106            let ssrcs: Vec<u32> = self.pending_immediate.iter().copied().collect();
107            self.queue_plis(now, &ssrcs);
108        }
109        self.arm(now);
110    }
111
112    /// Start the interval running, if it is not already and there is anything to ask.
113    fn arm(&mut self, now: Instant) {
114        if self.next_timeout.is_none() && self.is_periodic() && !self.streams.is_empty() {
115            self.next_timeout = Some(now + self.interval);
116        }
117    }
118
119    fn is_periodic(&self) -> bool {
120        !self.interval.is_zero()
121    }
122
123    /// SSRCs that are bound, from a request naming some or all of them.
124    ///
125    /// A PLI for a stream nobody is receiving has no destination, so unbound SSRCs are dropped.
126    fn targets(&self, requested: Option<&Vec<u32>>) -> Vec<u32> {
127        match requested {
128            None => self.streams.iter().copied().collect(),
129            Some(ssrcs) => ssrcs
130                .iter()
131                .copied()
132                .filter(|ssrc| self.streams.contains(ssrc))
133                .collect(),
134        }
135    }
136}
137
138impl Protocol<TaggedPacket, TaggedPacket, ()> for IntervalPliInterceptor {
139    type Rout = TaggedPacket;
140    type Wout = TaggedPacket;
141    type Eout = ();
142    type Error = Error;
143    type Time = Instant;
144
145    fn handle_read(&mut self, msg: TaggedPacket) -> Result<(), Self::Error> {
146        self.observe(msg.now);
147
148        // A keyframe request arrives as an attribute on a packet rather than out of band: with no
149        // event channel, that is how one interceptor tells another something.
150
151        self.read_queue.push_back(msg);
152        Ok(())
153    }
154
155    fn poll_read(&mut self) -> Option<Self::Rout> {
156        self.read_queue.pop_front()
157    }
158
159    fn handle_write(&mut self, msg: TaggedPacket) -> Result<(), Self::Error> {
160        self.observe(msg.now);
161
162        // The leg an application's request arrives on.
163
164        self.write_queue.push_back(msg);
165        Ok(())
166    }
167
168    fn poll_write(&mut self) -> Option<TaggedPacket> {
169        self.write_queue.pop_front()
170    }
171
172    fn handle_timeout(&mut self, now: Instant) -> Result<(), Error> {
173        self.observe(now);
174
175        if let Some(next_timeout) = self.next_timeout
176            && now >= next_timeout
177        {
178            self.next_timeout = Some(now + self.interval);
179            let ssrcs: Vec<u32> = self.streams.iter().copied().collect();
180            self.queue_plis(now, &ssrcs);
181        }
182        Ok(())
183    }
184
185    fn poll_timeout(&mut self) -> Option<Instant> {
186        self.next_timeout
187    }
188}
189
190impl Interceptor for IntervalPliInterceptor {
191    fn bind_remote_stream(&mut self, info: &StreamInfo) {
192        if stream_supports_pli(info) {
193            self.streams.insert(info.ssrc);
194            self.pending_immediate.insert(info.ssrc);
195        }
196    }
197
198    fn unbind_remote_stream(&mut self, info: &StreamInfo) {
199        self.streams.remove(&info.ssrc);
200        self.pending_immediate.remove(&info.ssrc);
201        if self.streams.is_empty() {
202            // Nothing left to ask: stop asking to be woken.
203            self.next_timeout = None;
204        }
205    }
206
207    fn bind_local_stream(&mut self, _info: &StreamInfo) {}
208
209    fn unbind_local_stream(&mut self, _info: &StreamInfo) {}
210}
211
212#[cfg(test)]
213mod tests {
214    use super::*;
215    use crate::chain::Chain;
216    use crate::stream_info::RTCPFeedback;
217    use sansio::Protocol;
218
219    fn stream_info(ssrc: u32) -> StreamInfo {
220        StreamInfo {
221            ssrc,
222            rtcp_feedback: vec![RTCPFeedback {
223                typ: "nack".to_owned(),
224                parameter: "pli".to_owned(),
225            }],
226            ..Default::default()
227        }
228    }
229
230    fn plis(chain: &mut Chain) -> Vec<u32> {
231        let mut out = Vec::new();
232        while let Some(pkt) = chain.poll_write() {
233            if let Packet::Rtcp(packets) = &pkt.message.packet {
234                for p in packets {
235                    if let Some(pli) = p
236                        .as_any()
237                        .downcast_ref::<rtcp::payload_feedbacks::picture_loss_indication::PictureLossIndication>(
238                        ) {
239                        out.push(pli.media_ssrc);
240                    }
241                }
242            }
243        }
244        out
245    }
246
247    fn chain(interval: Duration) -> Chain {
248        Chain::new(vec![Box::new(IntervalPliInterceptor::new(interval))])
249    }
250
251    #[test]
252    fn a_bound_stream_is_asked_immediately() {
253        let now = Instant::now();
254        let mut chain = chain(Duration::from_secs(3));
255        chain.bind_remote_stream(&stream_info(7));
256
257        chain.handle_timeout(now).unwrap();
258        assert_eq!(vec![7], plis(&mut chain));
259    }
260
261    #[test]
262    fn requests_repeat_on_the_interval() {
263        let now = Instant::now();
264        let mut chain = chain(Duration::from_secs(1));
265        chain.bind_remote_stream(&stream_info(7));
266
267        chain.handle_timeout(now).unwrap();
268        assert_eq!(vec![7], plis(&mut chain), "the immediate one");
269
270        chain
271            .handle_timeout(now + Duration::from_millis(999))
272            .unwrap();
273        assert!(plis(&mut chain).is_empty(), "not due yet");
274
275        chain.handle_timeout(now + Duration::from_secs(1)).unwrap();
276        assert_eq!(vec![7], plis(&mut chain), "due");
277    }
278
279    #[test]
280    fn an_unbound_stream_stops_being_asked() {
281        let now = Instant::now();
282        let mut chain = chain(Duration::from_secs(1));
283        chain.bind_remote_stream(&stream_info(7));
284        chain.handle_timeout(now).unwrap();
285        let _ = plis(&mut chain);
286
287        chain.unbind_remote_stream(&stream_info(7));
288        chain.handle_timeout(now + Duration::from_secs(5)).unwrap();
289
290        assert!(plis(&mut chain).is_empty());
291        assert_eq!(None, chain.poll_timeout(), "and stops asking to be woken");
292    }
293
294    /// A zero interval means the one immediate request and nothing after it.
295    #[test]
296    fn a_zero_interval_disables_the_periodic_requests() {
297        let now = Instant::now();
298        let mut chain = chain(Duration::ZERO);
299        chain.bind_remote_stream(&stream_info(7));
300
301        chain.handle_timeout(now).unwrap();
302        assert_eq!(vec![7], plis(&mut chain), "the immediate one still happens");
303        assert_eq!(None, chain.poll_timeout(), "but no interval is armed");
304
305        chain.handle_timeout(now + Duration::from_secs(60)).unwrap();
306        assert!(plis(&mut chain).is_empty());
307    }
308}
rtc_interceptor/intervalpli/generator.rs

rtc_interceptor/intervalpli/
generator.rs
