rtc_interceptor/flexfec/draft03/
sender.rs

1//! Send-side FlexFEC draft-03: protects outgoing media with repair packets.
2
3use super::encoder::FlexFec03Encoder;
4use crate::Interceptor;
5use crate::stream_info::StreamInfo;
6use crate::{AttributedPacket, Packet, TaggedPacket};
7use sansio::Protocol;
8use shared::error::Error;
9use std::collections::{HashMap, VecDeque};
10use std::time::Instant;
11
12/// Media packets gathered before a repair block is produced.
13pub const DEFAULT_NUM_MEDIA_PACKETS: u32 = 5;
14
15/// Repair packets produced per block.
16pub const DEFAULT_NUM_FEC_PACKETS: u32 = 2;
17
18/// Builder for [`FlexFec03SendInterceptor`].
19///
20/// # Example
21///
22/// ```
23/// use rtc_interceptor::{Slot, FlexFec03SendBuilder, Registry};
24///
25/// let chain = Registry::new()
26///     .with(Slot::FecEncoder, FlexFec03SendBuilder::new().with_num_fec_packets(1).build())
27///     .build();
28/// ```
29pub struct FlexFec03SendBuilder {
30    num_media_packets: u32,
31    num_fec_packets: u32,
32}
33
34impl Default for FlexFec03SendBuilder {
35    fn default() -> Self {
36        Self {
37            num_media_packets: DEFAULT_NUM_MEDIA_PACKETS,
38            num_fec_packets: DEFAULT_NUM_FEC_PACKETS,
39        }
40    }
41}
42
43impl FlexFec03SendBuilder {
44    /// Create a builder with the default block shape.
45    pub fn new() -> Self {
46        Self::default()
47    }
48
49    /// How many media packets one repair block protects.
50    ///
51    /// Larger blocks cost less bandwidth per protected packet and recover later, since the block
52    /// is only sent once it is full.
53    pub fn with_num_media_packets(mut self, num_media_packets: u32) -> Self {
54        self.num_media_packets = num_media_packets;
55        self
56    }
57
58    /// How many repair packets each block produces.
59    ///
60    /// This is what the block can survive: *n* repair packets recover up to *n* losses, spread
61    /// across the block by the interleaving.
62    pub fn with_num_fec_packets(mut self, num_fec_packets: u32) -> Self {
63        self.num_fec_packets = num_fec_packets;
64        self
65    }
66
67    /// Build the interceptor.
68    pub fn build(self) -> FlexFec03SendInterceptor {
69        FlexFec03SendInterceptor::new(self.num_media_packets, self.num_fec_packets)
70    }
71}
72
73/// One protected media stream.
74struct ProtectedStream {
75    encoder: FlexFec03Encoder,
76    /// Media packets awaiting a full block.
77    block: Vec<rtp::Packet>,
78}
79
80/// Produces FlexFEC draft-03 repair packets for outgoing media.
81///
82/// Binds only when the stream carries both a FEC SSRC and a FEC payload type: a repair stream
83/// needs its own SSRC to send on and its own payload type to be recognised by, and half an
84/// association is not usable. Those come from the negotiated `a=ssrc-group:FEC-FR`.
85pub struct FlexFec03SendInterceptor {
86    num_media_packets: u32,
87    num_fec_packets: u32,
88    /// Keyed by the **media** SSRC being protected.
89    streams: HashMap<u32, ProtectedStream>,
90    /// Repair packets the encoder produced, waiting to join the belt.
91    /// Inbound packets ready for the next interceptor.
92    read_queue: VecDeque<TaggedPacket>,
93    /// Outbound packets ready for the next interceptor: what passed through, plus
94    /// anything this one generated.
95    write_queue: VecDeque<TaggedPacket>,
96}
97
98impl FlexFec03SendInterceptor {
99    fn new(num_media_packets: u32, num_fec_packets: u32) -> Self {
100        Self {
101            read_queue: VecDeque::new(),
102            write_queue: VecDeque::new(),
103            num_media_packets: num_media_packets.max(1),
104            num_fec_packets,
105            streams: HashMap::new(),
106        }
107    }
108
109    /// The media SSRCs currently being protected.
110    pub fn protected_streams(&self) -> impl Iterator<Item = u32> + '_ {
111        self.streams.keys().copied()
112    }
113}
114
115impl Protocol<TaggedPacket, TaggedPacket, ()> for FlexFec03SendInterceptor {
116    type Rout = TaggedPacket;
117    type Wout = TaggedPacket;
118    type Eout = ();
119    type Error = Error;
120    type Time = Instant;
121
122    fn handle_read(&mut self, msg: TaggedPacket) -> Result<(), Self::Error> {
123        self.read_queue.push_back(msg);
124        Ok(())
125    }
126
127    fn poll_read(&mut self) -> Option<Self::Rout> {
128        self.read_queue.pop_front()
129    }
130
131    fn handle_write(&mut self, msg: TaggedPacket) -> Result<(), Self::Error> {
132        let Packet::Rtp(rtp_packet) = &msg.message.packet else {
133            self.write_queue.push_back(msg);
134            return Ok(());
135        };
136
137        let ssrc = rtp_packet.header.ssrc;
138        let now = msg.now;
139        let transport = msg.transport;
140
141        let Some(stream) = self.streams.get_mut(&ssrc) else {
142            self.write_queue.push_back(msg);
143            return Ok(());
144        };
145
146        stream.block.push(rtp_packet.clone());
147        let repair_packets = if stream.block.len() as u32 >= self.num_media_packets {
148            let repair = stream.encoder.encode(&stream.block, self.num_fec_packets);
149            // Cleared either way: a block the encoder refused — a gap, or one longer than the
150            // masks describe — must not be retried packet by packet as more arrive.
151            stream.block.clear();
152            repair
153        } else {
154            Vec::new()
155        };
156
157        // The media packet goes out first; the repair packets protect what has already left.
158        self.write_queue.push_back(msg);
159
160        for packet in repair_packets {
161            // Queued for `poll_write`, which puts them back on the belt: a repair packet is a real
162            // outgoing RTP packet and still needs every interceptor ahead — a transport-wide sequence
163            // number, the pacer, and a place in the send history congestion control reads.
164            self.write_queue.push_back(TaggedPacket {
165                now,
166                transport,
167                message: AttributedPacket::new(Packet::Rtp(packet)),
168            });
169        }
170        Ok(())
171    }
172
173    fn poll_write(&mut self) -> Option<TaggedPacket> {
174        self.write_queue.pop_front()
175    }
176
177    fn handle_timeout(&mut self, _now: Instant) -> Result<(), Self::Error> {
178        Ok(())
179    }
180
181    fn poll_timeout(&mut self) -> Option<Self::Time> {
182        None
183    }
184}
185
186impl Interceptor for FlexFec03SendInterceptor {
187    fn bind_local_stream(&mut self, info: &StreamInfo) {
188        // The gate FEC-PRE-01 exists to open. Both halves or neither: an SSRC with no payload
189        // type has nothing to mark its packets with, and a payload type with no SSRC has nowhere
190        // to send them.
191        if let (Some(ssrc_fec), Some(payload_type_fec)) = (info.ssrc_fec, info.payload_type_fec) {
192            self.streams.insert(
193                info.ssrc,
194                ProtectedStream {
195                    encoder: FlexFec03Encoder::new(payload_type_fec, ssrc_fec),
196                    block: Vec::new(),
197                },
198            );
199        }
200    }
201
202    fn unbind_local_stream(&mut self, info: &StreamInfo) {
203        // Any partly-filled block goes with it: those packets have already been sent unprotected,
204        // and a repair packet for a stream that has stopped has nothing left to repair.
205        self.streams.remove(&info.ssrc);
206    }
207
208    fn bind_remote_stream(&mut self, _info: &StreamInfo) {}
209
210    fn unbind_remote_stream(&mut self, _info: &StreamInfo) {}
211}
rtc_interceptor/flexfec/draft03/sender.rs

rtc_interceptor/flexfec/draft03/
sender.rs
