rtc_interceptor/report/
receiver.rs

1//! Receiver Report Interceptor - Generates RTCP Receiver Reports.
2
3use crate::report::receiver_stream::ReceiverStream;
4use crate::stream_info::StreamInfo;
5use crate::{Interceptor, Packet, TaggedPacket, interceptor};
6use shared::TransportContext;
7use shared::error::Error;
8use std::collections::{HashMap, VecDeque};
9use std::marker::PhantomData;
10use std::time::{Duration, Instant};
11
12/// Builder for the ReceiverReportInterceptor.
13///
14/// # Example
15///
16/// ```
17/// use rtc_interceptor::{Registry, ReceiverReportBuilder};
18/// use std::time::Duration;
19///
20/// // With default interval (1 second)
21/// let chain = Registry::new()
22///     .with(ReceiverReportBuilder::new().build())
23///     .build();
24///
25/// // With custom interval
26/// let chain = Registry::new()
27///     .with(ReceiverReportBuilder::new().with_interval(Duration::from_millis(500)).build())
28///     .build();
29/// ```
30pub struct ReceiverReportBuilder<P> {
31    /// Interval between receiver reports.
32    interval: Duration,
33    _phantom: PhantomData<P>,
34}
35
36impl<P> Default for ReceiverReportBuilder<P> {
37    fn default() -> Self {
38        Self {
39            interval: Duration::from_secs(1),
40            _phantom: PhantomData,
41        }
42    }
43}
44
45impl<P> ReceiverReportBuilder<P> {
46    /// Create a new builder with default settings.
47    ///
48    /// Default interval is 1 second.
49    pub fn new() -> Self {
50        Self::default()
51    }
52
53    /// Set a custom interval between receiver reports.
54    ///
55    /// # Example
56    ///
57    /// ```
58    /// use rtc_interceptor::{ReceiverReportBuilder, Registry};
59    /// use std::time::Duration;
60    ///
61    /// // The builder is generic over the next layer, so its type is pinned by `with`.
62    /// let registry = Registry::new().with(
63    ///     ReceiverReportBuilder::new()
64    ///         .with_interval(Duration::from_millis(500))
65    ///         .build(),
66    /// );
67    /// ```
68    pub fn with_interval(mut self, interval: Duration) -> Self {
69        self.interval = interval;
70        self
71    }
72
73    /// Create a builder function for use with Registry.
74    ///
75    /// This returns a closure that can be passed to `Registry::with()`.
76    ///
77    /// # Example
78    ///
79    /// ```
80    /// use rtc_interceptor::{Registry, ReceiverReportBuilder};
81    ///
82    /// let registry = Registry::new()
83    ///     .with(ReceiverReportBuilder::new().build());
84    /// ```
85    pub fn build(self) -> impl FnOnce(P) -> ReceiverReportInterceptor<P> {
86        move |inner| ReceiverReportInterceptor::new(inner, self.interval)
87    }
88}
89
90/// Interceptor that generates RTCP Receiver Reports.
91///
92/// This interceptor monitors incoming RTP packets, tracks statistics per stream,
93/// and periodically generates RTCP Receiver Reports.
94///
95/// # Type Parameters
96///
97/// - `P`: The inner protocol being wrapped
98///
99/// # Example
100///
101/// ```
102/// use rtc_interceptor::{Registry, ReceiverReportBuilder};
103///
104/// let chain = Registry::new()
105///     .with(ReceiverReportBuilder::new().build())
106///     .build();
107/// ```
108#[derive(Interceptor)]
109pub struct ReceiverReportInterceptor<P> {
110    #[next]
111    inner: P,
112
113    interval: Duration,
114    next_timeout: Option<Instant>,
115
116    streams: HashMap<u32, ReceiverStream>,
117
118    read_queue: VecDeque<TaggedPacket>,
119    write_queue: VecDeque<TaggedPacket>,
120}
121
122impl<P> ReceiverReportInterceptor<P> {
123    /// Create a new ReceiverReportInterceptor with default configuration.
124    fn new(inner: P, interval: Duration) -> Self {
125        Self {
126            inner,
127
128            interval,
129            next_timeout: None,
130
131            streams: HashMap::new(),
132
133            read_queue: VecDeque::new(),
134            write_queue: VecDeque::new(),
135        }
136    }
137
138    /// Process an incoming RTP packet for statistics.
139    fn process_rtp(&mut self, now: Instant, ssrc: u32, seq: u16, timestamp: u32) {
140        // Create stream if it doesn't exist
141        let stream = self.streams.entry(ssrc).or_insert_with(|| {
142            // Default clock rate, should be configured per stream in real usage
143            ReceiverStream::new(ssrc, 90000)
144        });
145
146        // Create a minimal RTP packet for processing
147        let pkt = rtp::packet::Packet {
148            header: rtp::header::Header {
149                ssrc,
150                sequence_number: seq,
151                timestamp,
152                ..Default::default()
153            },
154            ..Default::default()
155        };
156
157        stream.process_rtp(now, &pkt);
158    }
159
160    /// Process an incoming RTCP Sender Report.
161    fn process_sender_report(&mut self, now: Instant, sr: &rtcp::sender_report::SenderReport) {
162        if let Some(stream) = self.streams.get_mut(&sr.ssrc) {
163            stream.process_sender_report(now, sr);
164        }
165    }
166
167    /// Generate receiver reports for all tracked streams.
168    fn generate_reports(&mut self, now: Instant) -> Vec<rtcp::receiver_report::ReceiverReport> {
169        self.streams
170            .values_mut()
171            .map(|stream| stream.generate_report(now))
172            .collect()
173    }
174
175    /// Register a new stream with its clock rate.
176    fn register_stream(&mut self, ssrc: u32, clock_rate: u32) {
177        self.streams
178            .entry(ssrc)
179            .or_insert_with(|| ReceiverStream::new(ssrc, clock_rate));
180    }
181}
182
183#[interceptor]
184impl<P: Interceptor> ReceiverReportInterceptor<P> {
185    #[overrides]
186    fn handle_read(&mut self, msg: TaggedPacket) -> Result<(), Self::Error> {
187        if let Packet::Rtcp(rtcp_packets) = &msg.message {
188            for rtcp_packet in rtcp_packets {
189                if let Some(sr) = rtcp_packet
190                    .as_any()
191                    .downcast_ref::<rtcp::sender_report::SenderReport>()
192                    && let Some(stream) = self.streams.get_mut(&sr.ssrc)
193                {
194                    stream.process_sender_report(msg.now, sr);
195                }
196            }
197        } else if let Packet::Rtp(rtp_packet) = &msg.message
198            && let Some(stream) = self.streams.get_mut(&rtp_packet.header.ssrc)
199        {
200            stream.process_rtp(msg.now, rtp_packet);
201
202            // Arm the report timer from the first packet's instant (see nack::generator).
203            if self.next_timeout.is_none() {
204                self.next_timeout = Some(msg.now + self.interval);
205            }
206        }
207
208        self.inner.handle_read(msg)
209    }
210
211    #[overrides]
212    fn poll_write(&mut self) -> Option<Self::Wout> {
213        // First drain generated RTCP reports
214        if let Some(pkt) = self.write_queue.pop_front() {
215            return Some(pkt);
216        }
217        self.inner.poll_write()
218    }
219
220    #[overrides]
221    fn handle_timeout(&mut self, now: Self::Time) -> Result<(), Self::Error> {
222        if let Some(next_timeout) = self.next_timeout
223            && now >= next_timeout
224        {
225            self.next_timeout = Some(now + self.interval);
226
227            for stream in self.streams.values_mut() {
228                let rr = stream.generate_report(now);
229                self.write_queue.push_back(TaggedPacket {
230                    now,
231                    transport: TransportContext::default(),
232                    message: Packet::Rtcp(vec![Box::new(rr)]),
233                });
234            }
235        }
236
237        self.inner.handle_timeout(now)
238    }
239
240    #[overrides]
241    fn poll_timeout(&mut self) -> Option<Self::Time> {
242        match (self.next_timeout, self.inner.poll_timeout()) {
243            (Some(a), Some(b)) => Some(a.min(b)),
244            (Some(a), None) => Some(a),
245            (None, Some(b)) => Some(b),
246            (None, None) => None,
247        }
248    }
249
250    #[overrides]
251    fn bind_remote_stream(&mut self, info: &StreamInfo) {
252        let stream = ReceiverStream::new(info.ssrc, info.clock_rate);
253        self.streams.insert(info.ssrc, stream);
254
255        self.inner.bind_remote_stream(info);
256    }
257
258    #[overrides]
259    fn unbind_remote_stream(&mut self, info: &StreamInfo) {
260        self.streams.remove(&info.ssrc);
261
262        self.inner.unbind_remote_stream(info);
263    }
264}
265
266#[cfg(test)]
267mod tests {
268    use super::*;
269    use crate::Registry;
270    use sansio::Protocol;
271
272    fn dummy_rtp_packet() -> TaggedPacket {
273        TaggedPacket {
274            now: Instant::now(),
275            transport: Default::default(),
276            message: crate::Packet::Rtp(rtp::Packet::default()),
277        }
278    }
279
280    #[test]
281    fn test_receiver_report_builder_default() {
282        // Build with default interval (1 second)
283        let chain = Registry::new()
284            .with(ReceiverReportBuilder::default().build())
285            .build();
286
287        assert_eq!(chain.interval, Duration::from_secs(1));
288        assert!(chain.streams.is_empty());
289    }
290
291    #[test]
292    fn test_receiver_report_builder_with_custom_interval() {
293        // Build with custom interval
294        let chain = Registry::new()
295            .with(
296                ReceiverReportBuilder::default()
297                    .with_interval(Duration::from_millis(500))
298                    .build(),
299            )
300            .build();
301
302        assert_eq!(chain.interval, Duration::from_millis(500));
303    }
304
305    #[test]
306    fn test_receiver_report_chain_handle_read_write() {
307        // Build a chain and test packet flow
308        let mut chain = Registry::new()
309            .with(ReceiverReportBuilder::default().build())
310            .build();
311
312        // Test read path
313        let pkt = dummy_rtp_packet();
314        chain.handle_read(pkt).unwrap();
315        assert!(chain.poll_read().is_some());
316
317        // Test write path
318        let pkt2 = dummy_rtp_packet();
319        let pkt2_message = pkt2.message.clone();
320        chain.handle_write(pkt2).unwrap();
321        assert_eq!(chain.poll_write().unwrap().message, pkt2_message);
322    }
323
324    #[test]
325    fn test_register_stream() {
326        let mut chain = Registry::new()
327            .with(ReceiverReportBuilder::default().build())
328            .build();
329
330        chain.register_stream(12345, 48000);
331        assert!(chain.streams.contains_key(&12345));
332    }
333
334    #[test]
335    fn test_process_rtp() {
336        let mut chain = Registry::new()
337            .with(ReceiverReportBuilder::default().build())
338            .build();
339
340        let now = Instant::now();
341        chain.process_rtp(now, 12345, 1, 1000);
342
343        assert!(chain.streams.contains_key(&12345));
344    }
345
346    #[test]
347    fn test_generate_reports() {
348        let mut chain = Registry::new()
349            .with(ReceiverReportBuilder::default().build())
350            .build();
351
352        let now = Instant::now();
353        chain.process_rtp(now, 12345, 1, 1000);
354        chain.process_rtp(now, 12345, 2, 2000);
355
356        let reports = chain.generate_reports(now);
357        assert_eq!(reports.len(), 1);
358    }
359
360    #[test]
361    fn test_chained_interceptors() {
362        use crate::report::sender::SenderReportBuilder;
363
364        // Demonstrate chaining multiple interceptors
365        let mut chain = Registry::new()
366            .with(ReceiverReportBuilder::default().build())
367            .with(
368                SenderReportBuilder::default()
369                    .with_interval(Duration::from_millis(250))
370                    .build(),
371            )
372            .build();
373
374        // Test packet flow through the chain
375        let pkt = dummy_rtp_packet();
376        chain.handle_read(pkt).unwrap();
377        assert!(chain.poll_read().is_some());
378
379        let pkt2 = dummy_rtp_packet();
380        let pkt2_message = pkt2.message.clone();
381        chain.handle_write(pkt2).unwrap();
382        assert_eq!(chain.poll_write().unwrap().message, pkt2_message);
383    }
384
385    #[test]
386    fn test_receiver_report_generation_on_timeout() {
387        // Port of pion's TestReceiverInterceptor - tests full timeout/report cycle
388        // No ticker mocking needed - sans-I/O pattern lets us control time directly
389        let mut chain = Registry::new()
390            .with(
391                ReceiverReportBuilder::default()
392                    .with_interval(Duration::from_secs(1))
393                    .build(),
394            )
395            .build();
396
397        // Bind a remote stream
398        let info = StreamInfo {
399            ssrc: 123456,
400            clock_rate: 90000,
401            ..Default::default()
402        };
403        chain.bind_remote_stream(&info);
404
405        let base_time = Instant::now();
406
407        // Receive some RTP packets through the read path
408        for i in 0..10u16 {
409            let pkt = TaggedPacket {
410                now: base_time,
411                transport: Default::default(),
412                message: Packet::Rtp(rtp::Packet {
413                    header: rtp::header::Header {
414                        ssrc: 123456,
415                        sequence_number: i,
416                        timestamp: i as u32 * 3000,
417                        ..Default::default()
418                    },
419                    ..Default::default()
420                }),
421            };
422            chain.handle_read(pkt).unwrap();
423            chain.poll_read();
424        }
425
426        // First timeout triggers report generation (eto was set at construction)
427        chain.handle_timeout(base_time).unwrap();
428
429        // Drain any reports from initial timeout
430        while chain.poll_write().is_some() {}
431
432        // Advance time past the interval
433        let later_time = base_time + Duration::from_secs(2);
434        chain.handle_timeout(later_time).unwrap();
435
436        // Now a receiver report should be generated
437        let report = chain.poll_write();
438        assert!(report.is_some());
439
440        if let Some(tagged) = report {
441            if let Packet::Rtcp(rtcp_packets) = tagged.message {
442                assert_eq!(rtcp_packets.len(), 1);
443                let rr = rtcp_packets[0]
444                    .as_any()
445                    .downcast_ref::<rtcp::receiver_report::ReceiverReport>()
446                    .expect("Expected ReceiverReport");
447                assert_eq!(rr.reports.len(), 1);
448                assert_eq!(rr.reports[0].ssrc, 123456);
449                assert_eq!(rr.reports[0].last_sequence_number, 9);
450                assert_eq!(rr.reports[0].fraction_lost, 0);
451                assert_eq!(rr.reports[0].total_lost, 0);
452            } else {
453                panic!("Expected RTCP packet");
454            }
455        }
456    }
457
458    #[test]
459    fn test_receiver_report_with_packet_loss() {
460        // Test receiver report generation with packet loss
461        let mut chain = Registry::new()
462            .with(
463                ReceiverReportBuilder::default()
464                    .with_interval(Duration::from_secs(1))
465                    .build(),
466            )
467            .build();
468
469        let info = StreamInfo {
470            ssrc: 123456,
471            clock_rate: 90000,
472            ..Default::default()
473        };
474        chain.bind_remote_stream(&info);
475
476        let base_time = Instant::now();
477
478        // Receive packet 1
479        let pkt = TaggedPacket {
480            now: base_time,
481            transport: Default::default(),
482            message: Packet::Rtp(rtp::Packet {
483                header: rtp::header::Header {
484                    ssrc: 123456,
485                    sequence_number: 1,
486                    timestamp: 3000,
487                    ..Default::default()
488                },
489                ..Default::default()
490            }),
491        };
492        chain.handle_read(pkt).unwrap();
493        chain.poll_read();
494
495        // Skip packet 2, receive packet 3
496        let pkt = TaggedPacket {
497            now: base_time,
498            transport: Default::default(),
499            message: Packet::Rtp(rtp::Packet {
500                header: rtp::header::Header {
501                    ssrc: 123456,
502                    sequence_number: 3,
503                    timestamp: 9000,
504                    ..Default::default()
505                },
506                ..Default::default()
507            }),
508        };
509        chain.handle_read(pkt).unwrap();
510        chain.poll_read();
511
512        // Trigger timeout
513        let later_time = base_time + Duration::from_secs(2);
514        chain.handle_timeout(later_time).unwrap();
515
516        let report = chain.poll_write();
517        assert!(report.is_some());
518
519        if let Some(tagged) = report {
520            if let Packet::Rtcp(rtcp_packets) = tagged.message {
521                let rr = rtcp_packets[0]
522                    .as_any()
523                    .downcast_ref::<rtcp::receiver_report::ReceiverReport>()
524                    .expect("Expected ReceiverReport");
525                assert_eq!(rr.reports[0].last_sequence_number, 3);
526                // 1 packet lost out of 3 total
527                assert_eq!(rr.reports[0].total_lost, 1);
528                // fraction_lost = 256 * 1 / 3 = 85
529                assert_eq!(rr.reports[0].fraction_lost, (256u32 * 1 / 3) as u8);
530            } else {
531                panic!("Expected RTCP packet");
532            }
533        }
534    }
535
536    #[test]
537    fn test_receiver_report_with_sender_report() {
538        // Test that receiver report includes DLSR after receiving sender report
539        let mut chain = Registry::new()
540            .with(
541                ReceiverReportBuilder::default()
542                    .with_interval(Duration::from_secs(1))
543                    .build(),
544            )
545            .build();
546
547        let info = StreamInfo {
548            ssrc: 123456,
549            clock_rate: 90000,
550            ..Default::default()
551        };
552        chain.bind_remote_stream(&info);
553
554        let base_time = Instant::now();
555
556        // Receive an RTP packet first
557        let pkt = TaggedPacket {
558            now: base_time,
559            transport: Default::default(),
560            message: Packet::Rtp(rtp::Packet {
561                header: rtp::header::Header {
562                    ssrc: 123456,
563                    sequence_number: 1,
564                    timestamp: 3000,
565                    ..Default::default()
566                },
567                ..Default::default()
568            }),
569        };
570        chain.handle_read(pkt).unwrap();
571        chain.poll_read();
572
573        // Receive a sender report
574        let sr = rtcp::sender_report::SenderReport {
575            ssrc: 123456,
576            ntp_time: 0x1234_5678_0000_0000,
577            rtp_time: 3000,
578            packet_count: 100,
579            octet_count: 10000,
580            ..Default::default()
581        };
582        let sr_pkt = TaggedPacket {
583            now: base_time,
584            transport: Default::default(),
585            message: Packet::Rtcp(vec![Box::new(sr)]),
586        };
587        chain.handle_read(sr_pkt).unwrap();
588
589        // Generate receiver report 1 second later
590        let later_time = base_time + Duration::from_secs(1);
591        chain.handle_timeout(later_time).unwrap();
592
593        let report = chain.poll_write();
594        assert!(report.is_some());
595
596        if let Some(tagged) = report {
597            if let Packet::Rtcp(rtcp_packets) = tagged.message {
598                let rr = rtcp_packets[0]
599                    .as_any()
600                    .downcast_ref::<rtcp::receiver_report::ReceiverReport>()
601                    .expect("Expected ReceiverReport");
602                // DLSR should be ~65536 (1 second in 1/65536 units)
603                assert_eq!(rr.reports[0].delay, 65536);
604                // LSR is middle 32 bits of NTP time
605                assert_eq!(rr.reports[0].last_sender_report, 0x5678_0000);
606            } else {
607                panic!("Expected RTCP packet");
608            }
609        }
610    }
611
612    #[test]
613    fn test_receiver_report_multiple_streams() {
614        // Test that multiple remote streams each generate their own reports
615        let mut chain = Registry::new()
616            .with(
617                ReceiverReportBuilder::default()
618                    .with_interval(Duration::from_secs(1))
619                    .build(),
620            )
621            .build();
622
623        let info1 = StreamInfo {
624            ssrc: 111111,
625            clock_rate: 90000,
626            ..Default::default()
627        };
628        let info2 = StreamInfo {
629            ssrc: 222222,
630            clock_rate: 48000,
631            ..Default::default()
632        };
633        chain.bind_remote_stream(&info1);
634        chain.bind_remote_stream(&info2);
635
636        let base_time = Instant::now();
637
638        // Receive packets on stream 1
639        for i in 0..5u16 {
640            let pkt = TaggedPacket {
641                now: base_time,
642                transport: Default::default(),
643                message: Packet::Rtp(rtp::Packet {
644                    header: rtp::header::Header {
645                        ssrc: 111111,
646                        sequence_number: i,
647                        timestamp: i as u32 * 3000,
648                        ..Default::default()
649                    },
650                    ..Default::default()
651                }),
652            };
653            chain.handle_read(pkt).unwrap();
654            chain.poll_read();
655        }
656
657        // Receive packets on stream 2 with a gap (packet loss)
658        let pkt = TaggedPacket {
659            now: base_time,
660            transport: Default::default(),
661            message: Packet::Rtp(rtp::Packet {
662                header: rtp::header::Header {
663                    ssrc: 222222,
664                    sequence_number: 0,
665                    timestamp: 0,
666                    ..Default::default()
667                },
668                ..Default::default()
669            }),
670        };
671        chain.handle_read(pkt).unwrap();
672        chain.poll_read();
673
674        let pkt = TaggedPacket {
675            now: base_time,
676            transport: Default::default(),
677            message: Packet::Rtp(rtp::Packet {
678                header: rtp::header::Header {
679                    ssrc: 222222,
680                    sequence_number: 5, // Skip 1-4
681                    timestamp: 5 * 960,
682                    ..Default::default()
683                },
684                ..Default::default()
685            }),
686        };
687        chain.handle_read(pkt).unwrap();
688        chain.poll_read();
689
690        // Trigger timeout
691        let later_time = base_time + Duration::from_secs(2);
692        chain.handle_timeout(later_time).unwrap();
693
694        // Collect all reports
695        let mut ssrcs = vec![];
696        let mut total_lost = vec![];
697
698        while let Some(tagged) = chain.poll_write() {
699            if let Packet::Rtcp(rtcp_packets) = tagged.message {
700                for rtcp_pkt in rtcp_packets {
701                    if let Some(rr) = rtcp_pkt
702                        .as_any()
703                        .downcast_ref::<rtcp::receiver_report::ReceiverReport>()
704                    {
705                        for report in &rr.reports {
706                            ssrcs.push(report.ssrc);
707                            total_lost.push(report.total_lost);
708                        }
709                    }
710                }
711            }
712        }
713
714        assert_eq!(ssrcs.len(), 2);
715        assert!(ssrcs.contains(&111111));
716        assert!(ssrcs.contains(&222222));
717
718        // Stream 1 should have no loss
719        let idx1 = ssrcs.iter().position(|&s| s == 111111).unwrap();
720        assert_eq!(total_lost[idx1], 0);
721
722        // Stream 2 should have 4 lost packets (1-4)
723        let idx2 = ssrcs.iter().position(|&s| s == 222222).unwrap();
724        assert_eq!(total_lost[idx2], 4);
725    }
726
727    #[test]
728    fn test_receiver_report_unbind_stream() {
729        // Test that unbinding a remote stream stops generating reports for it
730        let mut chain = Registry::new()
731            .with(
732                ReceiverReportBuilder::default()
733                    .with_interval(Duration::from_secs(1))
734                    .build(),
735            )
736            .build();
737
738        let info = StreamInfo {
739            ssrc: 123456,
740            clock_rate: 90000,
741            ..Default::default()
742        };
743        chain.bind_remote_stream(&info);
744
745        let base_time = Instant::now();
746
747        // Receive some packets
748        let pkt = TaggedPacket {
749            now: base_time,
750            transport: Default::default(),
751            message: Packet::Rtp(rtp::Packet {
752                header: rtp::header::Header {
753                    ssrc: 123456,
754                    sequence_number: 0,
755                    timestamp: 0,
756                    ..Default::default()
757                },
758                ..Default::default()
759            }),
760        };
761        chain.handle_read(pkt).unwrap();
762        chain.poll_read();
763
764        // Unbind the stream
765        chain.unbind_remote_stream(&info);
766
767        // Trigger timeout
768        let later_time = base_time + Duration::from_secs(2);
769        chain.handle_timeout(later_time).unwrap();
770
771        // No report should be generated (stream was unbound)
772        assert!(chain.poll_write().is_none());
773    }
774
775    #[test]
776    fn test_receiver_report_sequence_wrap() {
777        // Test sequence number wraparound handling
778        let mut chain = Registry::new()
779            .with(
780                ReceiverReportBuilder::default()
781                    .with_interval(Duration::from_secs(1))
782                    .build(),
783            )
784            .build();
785
786        let info = StreamInfo {
787            ssrc: 123456,
788            clock_rate: 90000,
789            ..Default::default()
790        };
791        chain.bind_remote_stream(&info);
792
793        let base_time = Instant::now();
794
795        // Receive packet at 0xffff
796        let pkt = TaggedPacket {
797            now: base_time,
798            transport: Default::default(),
799            message: Packet::Rtp(rtp::Packet {
800                header: rtp::header::Header {
801                    ssrc: 123456,
802                    sequence_number: 0xffff,
803                    timestamp: 0,
804                    ..Default::default()
805                },
806                ..Default::default()
807            }),
808        };
809        chain.handle_read(pkt).unwrap();
810        chain.poll_read();
811
812        // Wrap around to 0x00
813        let pkt = TaggedPacket {
814            now: base_time,
815            transport: Default::default(),
816            message: Packet::Rtp(rtp::Packet {
817                header: rtp::header::Header {
818                    ssrc: 123456,
819                    sequence_number: 0x00,
820                    timestamp: 3000,
821                    ..Default::default()
822                },
823                ..Default::default()
824            }),
825        };
826        chain.handle_read(pkt).unwrap();
827        chain.poll_read();
828
829        // Trigger timeout
830        let later_time = base_time + Duration::from_secs(2);
831        chain.handle_timeout(later_time).unwrap();
832
833        let report = chain.poll_write();
834        assert!(report.is_some());
835
836        if let Some(tagged) = report {
837            if let Packet::Rtcp(rtcp_packets) = tagged.message {
838                let rr = rtcp_packets[0]
839                    .as_any()
840                    .downcast_ref::<rtcp::receiver_report::ReceiverReport>()
841                    .expect("Expected ReceiverReport");
842                // Extended sequence number should show 1 cycle (1 << 16)
843                assert_eq!(rr.reports[0].last_sequence_number, 1 << 16);
844                assert_eq!(rr.reports[0].fraction_lost, 0);
845                assert_eq!(rr.reports[0].total_lost, 0);
846            } else {
847                panic!("Expected RTCP packet");
848            }
849        }
850    }
851}
rtc_interceptor/report/receiver.rs

rtc_interceptor/report/
receiver.rs
