rtc_interceptor/nack/
responder.rs

1//! NACK Responder Interceptor - Responds to NACK requests by retransmitting packets.
2
3use super::send_buffer::SendBuffer;
4use super::stream_supports_nack;
5use crate::Interceptor;
6use crate::stream_info::StreamInfo;
7use crate::{AttributedPacket, Packet, TaggedPacket};
8use sansio::Protocol;
9use shared::TransportContext;
10use shared::error::Error;
11use std::collections::{HashMap, VecDeque};
12use std::time::Instant;
13
14/// Builder for the NackResponderInterceptor.
15///
16/// # Example
17///
18/// ```
19/// use rtc_interceptor::{Registry, NackResponderBuilder};
20///
21/// let chain = Registry::new()
22///     .with(NackResponderBuilder::new()
23///         .with_size(1024)
24///         .build())
25///     .build();
26/// ```
27pub struct NackResponderBuilder {
28    /// Size of the send buffer (must be power of 2: 1, 2, 4, ..., 32768).
29    size: u16,
30}
31
32impl Default for NackResponderBuilder {
33    fn default() -> Self {
34        Self { size: 1024 }
35    }
36}
37
38impl NackResponderBuilder {
39    /// Create a new builder with default settings.
40    pub fn new() -> Self {
41        Self::default()
42    }
43
44    /// Set the size of the send buffer.
45    ///
46    /// Size must be a power of 2 between 1 and 32768 (inclusive).
47    /// Larger buffers can retransmit older packets but use more memory.
48    pub fn with_size(mut self, size: u16) -> Self {
49        self.size = size;
50        self
51    }
52
53    /// Build the interceptor.
54    pub fn build(self) -> NackResponderInterceptor {
55        NackResponderInterceptor::new(self.size)
56    }
57}
58
59/// Per-stream state for the responder.
60struct LocalStream {
61    /// Buffer of sent packets for retransmission.
62    send_buffer: SendBuffer,
63    /// RTX SSRC for RFC4588 retransmission (if configured).
64    ssrc_rtx: Option<u32>,
65    /// RTX payload type for RFC4588 retransmission (if configured).
66    payload_type_rtx: Option<u8>,
67    /// Sequence number counter for RTX packets.
68    rtx_sequence_number: u16,
69}
70
71/// Interceptor that responds to NACK requests by retransmitting packets.
72///
73/// This interceptor buffers outgoing RTP packets on local streams and
74/// retransmits them when RTCP TransportLayerNack packets are received.
75pub struct NackResponderInterceptor {
76    /// Configuration
77    size: u16,
78
79    /// Send buffers per local stream SSRC
80    streams: HashMap<u32, LocalStream>,
81
82    /// Queue for retransmitted packets
83    write_queue: VecDeque<TaggedPacket>,
84    /// Inbound packets ready for the next interceptor.
85    read_queue: VecDeque<TaggedPacket>,
86}
87
88impl NackResponderInterceptor {
89    fn new(size: u16) -> Self {
90        Self {
91            read_queue: VecDeque::new(),
92            size,
93            streams: HashMap::new(),
94            write_queue: VecDeque::new(),
95        }
96    }
97
98    /// Handle a NACK request by queuing retransmissions.
99    fn handle_nack(
100        &mut self,
101        now: Instant,
102        nack: &rtcp::transport_feedbacks::transport_layer_nack::TransportLayerNack,
103    ) {
104        // Collect sequence numbers to retransmit
105        let mut seqs_to_retransmit = Vec::new();
106
107        for nack_pair in &nack.nacks {
108            // Check the base packet ID
109            seqs_to_retransmit.push(nack_pair.packet_id);
110
111            // Check each bit in lost_packets bitmap
112            for i in 0..16 {
113                if nack_pair.lost_packets & (1 << i) != 0 {
114                    let seq = nack_pair.packet_id.wrapping_add(i + 1);
115                    seqs_to_retransmit.push(seq);
116                }
117            }
118        }
119
120        let Some(stream) = self.streams.get_mut(&nack.media_ssrc) else {
121            return;
122        };
123
124        // Queue retransmissions
125        for seq in seqs_to_retransmit {
126            let Some(original_packet) = stream.send_buffer.get(seq) else {
127                continue;
128            };
129
130            let packet = if let (Some(ssrc_rtx), Some(pt_rtx)) =
131                (stream.ssrc_rtx, stream.payload_type_rtx)
132            {
133                // RFC4588: Create RTX packet
134                // - Use RTX SSRC and payload type
135                // - Prepend original sequence number (2 bytes big-endian) to payload
136                // - Use separate RTX sequence number counter
137                let original_seq = original_packet.header.sequence_number;
138                let mut rtx_payload = Vec::with_capacity(2 + original_packet.payload.len());
139                rtx_payload.extend_from_slice(&original_seq.to_be_bytes());
140                rtx_payload.extend_from_slice(&original_packet.payload);
141
142                let rtx_seq = stream.rtx_sequence_number;
143                stream.rtx_sequence_number = stream.rtx_sequence_number.wrapping_add(1);
144
145                rtp::Packet {
146                    header: rtp::header::Header {
147                        ssrc: ssrc_rtx,
148                        payload_type: pt_rtx,
149                        sequence_number: rtx_seq,
150                        timestamp: original_packet.header.timestamp,
151                        marker: original_packet.header.marker,
152                        ..Default::default()
153                    },
154                    payload: rtx_payload.into(),
155                }
156            } else {
157                // No RTX: retransmit original packet as-is
158                original_packet.clone()
159            };
160
161            self.write_queue.push_back(TaggedPacket {
162                now,
163                transport: TransportContext::default(),
164                message: AttributedPacket::new(Packet::Rtp(packet)),
165            });
166        }
167    }
168}
169
170impl Protocol<TaggedPacket, TaggedPacket, ()> for NackResponderInterceptor {
171    type Rout = TaggedPacket;
172    type Wout = TaggedPacket;
173    type Eout = ();
174    type Error = Error;
175    type Time = Instant;
176
177    fn handle_read(&mut self, msg: TaggedPacket) -> Result<(), Self::Error> {
178        // Process NACK packets
179        if let Packet::Rtcp(ref rtcp_packets) = msg.message.packet {
180            for rtcp_packet in rtcp_packets {
181                if let Some(nack) = rtcp_packet
182                    .as_any()
183                    .downcast_ref::<rtcp::transport_feedbacks::transport_layer_nack::TransportLayerNack>()
184                {
185                    self.handle_nack(msg.now, nack);
186                }
187            }
188        }
189
190        self.read_queue.push_back(msg);
191
192        Ok(())
193    }
194
195    fn poll_read(&mut self) -> Option<Self::Rout> {
196        self.read_queue.pop_front()
197    }
198
199    fn handle_write(&mut self, msg: TaggedPacket) -> Result<(), Self::Error> {
200        // Buffer outgoing RTP packets
201        if let Packet::Rtp(ref rtp_packet) = msg.message.packet
202            && let Some(stream) = self.streams.get_mut(&rtp_packet.header.ssrc)
203        {
204            stream.send_buffer.add(rtp_packet.clone());
205        }
206
207        self.write_queue.push_back(msg);
208
209        Ok(())
210    }
211
212    fn poll_write(&mut self) -> Option<TaggedPacket> {
213        // First drain retransmitted packets
214        self.write_queue.pop_front()
215    }
216
217    fn handle_timeout(&mut self, _now: Instant) -> Result<(), Self::Error> {
218        Ok(())
219    }
220
221    fn poll_timeout(&mut self) -> Option<Self::Time> {
222        None
223    }
224}
225
226impl Interceptor for NackResponderInterceptor {
227    fn bind_local_stream(&mut self, info: &StreamInfo) {
228        if stream_supports_nack(info)
229            && let Some(send_buffer) = SendBuffer::new(self.size)
230        {
231            self.streams.insert(
232                info.ssrc,
233                LocalStream {
234                    send_buffer,
235                    ssrc_rtx: info.ssrc_rtx,
236                    payload_type_rtx: info.payload_type_rtx,
237                    rtx_sequence_number: 0,
238                },
239            );
240        }
241    }
242
243    fn unbind_local_stream(&mut self, info: &StreamInfo) {
244        self.streams.remove(&info.ssrc);
245    }
246
247    fn bind_remote_stream(&mut self, _info: &StreamInfo) {}
248
249    fn unbind_remote_stream(&mut self, _info: &StreamInfo) {}
250}
rtc_interceptor/nack/responder.rs

rtc_interceptor/nack/
responder.rs
