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::stream_info::StreamInfo;
5use crate::{Interceptor, Packet, TaggedPacket, interceptor};
6use shared::error::Error;
7use std::collections::HashMap;
8use std::marker::PhantomData;
9
10/// Media packets gathered before a repair block is produced.
11pub const DEFAULT_NUM_MEDIA_PACKETS: u32 = 5;
12
13/// Repair packets produced per block.
14pub const DEFAULT_NUM_FEC_PACKETS: u32 = 2;
15
16/// Builder for [`FlexFec03SendInterceptor`].
17///
18/// # Example
19///
20/// ```
21/// use rtc_interceptor::{FlexFec03SendBuilder, Registry};
22///
23/// let chain = Registry::new()
24///     .with(FlexFec03SendBuilder::new().with_num_fec_packets(1).build())
25///     .build();
26/// ```
27pub struct FlexFec03SendBuilder<P> {
28    num_media_packets: u32,
29    num_fec_packets: u32,
30    _phantom: PhantomData<P>,
31}
32
33impl<P> Default for FlexFec03SendBuilder<P> {
34    fn default() -> Self {
35        Self {
36            num_media_packets: DEFAULT_NUM_MEDIA_PACKETS,
37            num_fec_packets: DEFAULT_NUM_FEC_PACKETS,
38            _phantom: PhantomData,
39        }
40    }
41}
42
43impl<P> FlexFec03SendBuilder<P> {
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 factory function.
68    pub fn build(self) -> impl FnOnce(P) -> FlexFec03SendInterceptor<P> {
69        move |inner| {
70            FlexFec03SendInterceptor::new(inner, self.num_media_packets, self.num_fec_packets)
71        }
72    }
73}
74
75/// One protected media stream.
76struct ProtectedStream {
77    encoder: FlexFec03Encoder,
78    /// Media packets awaiting a full block.
79    block: Vec<rtp::Packet>,
80}
81
82/// Produces FlexFEC draft-03 repair packets for outgoing media.
83///
84/// Binds only when the stream carries both a FEC SSRC and a FEC payload type: a repair stream
85/// needs its own SSRC to send on and its own payload type to be recognised by, and half an
86/// association is not usable. Those come from the negotiated `a=ssrc-group:FEC-FR`.
87#[derive(Interceptor)]
88pub struct FlexFec03SendInterceptor<P> {
89    #[next]
90    inner: P,
91    num_media_packets: u32,
92    num_fec_packets: u32,
93    /// Keyed by the **media** SSRC being protected.
94    streams: HashMap<u32, ProtectedStream>,
95}
96
97impl<P> FlexFec03SendInterceptor<P> {
98    fn new(inner: P, num_media_packets: u32, num_fec_packets: u32) -> Self {
99        Self {
100            inner,
101            num_media_packets: num_media_packets.max(1),
102            num_fec_packets,
103            streams: HashMap::new(),
104        }
105    }
106
107    /// The media SSRCs currently being protected.
108    pub fn protected_streams(&self) -> impl Iterator<Item = u32> + '_ {
109        self.streams.keys().copied()
110    }
111}
112
113#[interceptor]
114impl<P: Interceptor> FlexFec03SendInterceptor<P> {
115    #[overrides]
116    fn bind_local_stream(&mut self, info: &StreamInfo) {
117        // The gate FEC-PRE-01 exists to open. Both halves or neither: an SSRC with no payload
118        // type has nothing to mark its packets with, and a payload type with no SSRC has nowhere
119        // to send them.
120        if let (Some(ssrc_fec), Some(payload_type_fec)) = (info.ssrc_fec, info.payload_type_fec) {
121            self.streams.insert(
122                info.ssrc,
123                ProtectedStream {
124                    encoder: FlexFec03Encoder::new(payload_type_fec, ssrc_fec),
125                    block: Vec::new(),
126                },
127            );
128        }
129        self.inner.bind_local_stream(info);
130    }
131
132    #[overrides]
133    fn unbind_local_stream(&mut self, info: &StreamInfo) {
134        // Any partly-filled block goes with it: those packets have already been sent unprotected,
135        // and a repair packet for a stream that has stopped has nothing left to repair.
136        self.streams.remove(&info.ssrc);
137        self.inner.unbind_local_stream(info);
138    }
139
140    #[overrides]
141    fn handle_write(&mut self, msg: TaggedPacket) -> Result<(), Self::Error> {
142        let Packet::Rtp(rtp_packet) = &msg.message else {
143            return self.inner.handle_write(msg);
144        };
145
146        let ssrc = rtp_packet.header.ssrc;
147        let now = msg.now;
148        let transport = msg.transport;
149
150        let Some(stream) = self.streams.get_mut(&ssrc) else {
151            return self.inner.handle_write(msg);
152        };
153
154        stream.block.push(rtp_packet.clone());
155        let repair_packets = if stream.block.len() as u32 >= self.num_media_packets {
156            let repair = stream.encoder.encode(&stream.block, self.num_fec_packets);
157            // Cleared either way: a block the encoder refused — a gap, or one longer than the
158            // masks describe — must not be retried packet by packet as more arrive.
159            stream.block.clear();
160            repair
161        } else {
162            Vec::new()
163        };
164
165        // The media packet goes out first; the repair packets protect what has already left.
166        self.inner.handle_write(msg)?;
167
168        for packet in repair_packets {
169            // Re-injected rather than queued locally (chain contract rule 2): a repair packet is
170            // a real outgoing RTP packet and still needs the layers below — a transport-wide
171            // sequence number, and a place in the send history congestion control reads.
172            self.inner.handle_write(TaggedPacket {
173                now,
174                transport,
175                message: Packet::Rtp(packet),
176            })?;
177        }
178
179        Ok(())
180    }
181}
rtc_interceptor/flexfec/draft03/sender.rs

rtc_interceptor/flexfec/draft03/
sender.rs
