Skip to main content

rtc_interceptor/
registry.rs

1//! Building a chain.
2
3use crate::chain::Chain;
4use crate::noop::NoopInterceptor;
5use crate::{BoxedInterceptor, Interceptor};
6use log::warn;
7use std::collections::BTreeMap;
8
9/// Where an interceptor belongs in the chain, measured by **distance from the wire**.
10///
11/// This is the chain contract's ordering table expressed as data, so that one place decides it and
12/// a test can check a builder against it. The doc comments carry that table's indices; the gaps are
13/// slots nothing fills yet.
14///
15/// Read walks the list forwards and write walks it in reverse, so a smaller slot is closer to the
16/// network in both directions.
17#[derive(Copy, Clone, Debug)]
18#[non_exhaustive]
19#[repr(usize)]
20pub enum Slot {
21    /// `CongestionControlInterceptor` — the send history, and the ingest of returning feedback.
22    ///
23    /// **Both legs.** *Read:* ingests inbound TWCC or CCFB and hands the reports to the estimator.
24    /// *Write:* records every departing RTP packet in the send history.
25    ///
26    /// **Why here — wire-most.** The write walk runs from the application down to the wire, so the
27    /// lowest slot is the *last* thing a departing packet meets. Two facts depend on that. The
28    /// departure instant it records is the moment the packet actually left, after the pacer held
29    /// it; recording at enqueue instead would charge the pacer's own queueing delay to the network
30    /// and drive the estimate down for a delay this endpoint created. And the transport-wide
31    /// sequence number the history keys on has already been assigned at 2_000, so a report naming
32    /// that number can be matched to the packet that carried it.
33    CongestionControl = 1_000,
34    /// `TwccSenderInterceptor` — stamps each departing RTP packet with the transport-wide sequence
35    /// number the remote will report against.
36    ///
37    /// **Outbound only.** Inbound packets pass through untouched.
38    ///
39    /// **Why here — below every generator, above the history.** The write walk must reach it
40    /// *before* [`Slot::CongestionControl`], or the history would key on a number that does not
41    /// exist yet. It must sit below everything that produces a packet — the NACK responder at
42    /// 4_000, the FEC encoder at 5_000, every RTCP generator above them — because a packet that
43    /// never passes this slot is never numbered, and the remote cannot report on what it cannot
44    /// name. A retransmission is exactly that case.
45    TwccSender = 2_000,
46    /// `PacerInterceptor` — gates departures, releasing at the estimated rate.
47    ///
48    /// **Outbound, with one inbound read.** *Write:* queues RTP and releases it on a timer; RTCP
49    /// passes straight through, because feedback is only useful while it is fresh. *Read:* observes
50    /// [`Attribute::TargetBitrateChanged`](crate::Attribute) going past.
51    ///
52    /// **Why here — the meter every generated byte must cross.** Everything that produces a packet
53    /// sits application-ward of this slot, so retransmissions, FEC repair and generated RTCP are all
54    /// metered rather than bursting past the estimate. It is above the TWCC sender so that numbering
55    /// happens at release rather than at enqueue, which keeps the numbers in the order the packets
56    /// actually reach the wire.
57    ///
58    /// The estimate reaches it on the *read* leg because that is the only leg it can: the
59    /// controller is wire-ward of here, so on the write leg it sees packets after this interceptor
60    /// — too late to inform it.
61    Pacer = 3_000,
62    /// `NackResponderInterceptor` — answers a NACK by resending from its own buffer.
63    ///
64    /// **Both legs.** *Write:* buffers each departing RTP packet against a later request.
65    /// *Read:* watches for inbound NACK and queues the retransmissions it asks for.
66    ///
67    /// **Why here — the lowest of the generators.** A retransmission it emits re-enters the belt at
68    /// this slot and continues down, so it is still paced (3_000), numbered (2_000) and recorded
69    /// (1_000). It has to be: a retransmission is new bytes on the wire, and an estimator that does
70    /// not see them believes the path is carrying less than it is — then raises the rate during
71    /// loss, which is the worst moment to do it.
72    NackResponder = 4_000,
73    /// `FlexFec03SendInterceptor` — generates repair packets for the media it sees leaving.
74    ///
75    /// **Outbound only.**
76    ///
77    /// **Why here — a generator, so above the pacer.** Its repair packets are real bytes and are
78    /// metered and recorded like any other. Being above the NACK responder also means the media it
79    /// protects has already been buffered for retransmission, so the two recovery mechanisms cover
80    /// the same packets rather than racing to protect different ones.
81    FecEncoder = 5_000,
82    /// `FlexFec03ReceiveInterceptor` — rebuilds packets the path dropped.
83    ///
84    /// **Inbound only.** Nothing on the write leg.
85    ///
86    /// **Why here — before anything reads a sequence number.** The read walk runs wire to
87    /// application, so this recovers a packet before the NACK generator at 7_000 can notice it was
88    /// missing. Placed the other way round, this endpoint would ask the remote to retransmit
89    /// packets it was about to rebuild locally — paying for the same data twice, and adding a round
90    /// trip to data it already had.
91    FecDecoder = 6_000,
92    /// `NackGeneratorInterceptor` — asks the remote for what did not arrive.
93    ///
94    /// **Both legs.** *Read:* detects gaps in the inbound sequence space. *Write:* emits NACK on a
95    /// timer.
96    ///
97    /// **Why here — after recovery, before re-timing.** After the FEC decoder (6_000), so a rebuilt
98    /// packet counts as arrived and is not requested again. Before the jitter buffer (13_000), so
99    /// it judges loss from arrival order rather than from playout order. And, being a generator,
100    /// application-ward of the pacer so its NACKs are metered.
101    NackGenerator = 7_000,
102    /// `TwccReceiverInterceptor` — reports arrival times to the **remote** sender's congestion
103    /// controller.
104    ///
105    /// **Both legs.** *Read:* records when each inbound packet arrived. *Write:* emits
106    /// `TransportLayerCC` on a timer.
107    ///
108    /// **Why here — the write leg is what pins it.** It reads as a receive-side interceptor, and
109    /// moving it wire-ward looks harmless because nothing it does affects what this endpoint sends.
110    /// It does not work: it *generates*, and below the pacer its feedback would leave unpaced and
111    /// unrecorded by the send history. It is also an arrival recorder, so it must precede the
112    /// jitter buffer — see [`Slot::JitterBuffer`].
113    TwccReceiver = 8_000,
114    /// `Rfc8888Interceptor` — the same job as [`Slot::TwccReceiver`] in a different format.
115    ///
116    /// **Both legs**, and pinned by the same two constraints: a generator above the pacer, an
117    /// arrival recorder before the jitter buffer. It sits next to the TWCC receiver because the two
118    /// are alternatives — registering both reports every packet to the remote twice, and its
119    /// estimator cannot tell the two formats apart, so it reads the path as carrying double.
120    Rfc8888 = 9_000,
121    /// `ReceiverReportInterceptor` — RFC 3550 reception quality, not congestion-control feedback.
122    ///
123    /// **Both legs.** *Read:* accumulates loss, jitter and the extended sequence number from
124    /// inbound RTP. *Write:* emits RR on a timer.
125    ///
126    /// **Why here — the same pair of constraints as the arrival recorders above.** Above the pacer
127    /// because it generates; before the jitter buffer because the jitter it measures must be the
128    /// path's, not this endpoint's buffering.
129    ReceiverReport = 10_000,
130    /// `SenderReportInterceptor` — emits SR on a timer, describing what this endpoint has sent.
131    ///
132    /// **Outbound only.**
133    ///
134    /// **Why here — generator, and nothing else constrains it.** Above the pacer so its reports are
135    /// metered. Nothing inbound informs it, so it has no read-side ordering requirement and its
136    /// exact position among the generators does not matter.
137    SenderReport = 11_000,
138    /// `IntervalPliInterceptor` — asks the remote for a keyframe on a timer.
139    ///
140    /// **Outbound only.**
141    ///
142    /// **Why here — generator, and nothing else constrains it**, exactly as for
143    /// [`Slot::SenderReport`].
144    IntervalPli = 12_000,
145    /// `JitterBufferInterceptor` — holds inbound packets to smooth arrival jitter, then releases
146    /// them in order on a timer.
147    ///
148    /// **Inbound only.**
149    ///
150    /// **Why here — application-most, because it re-times what passes through it.** Every arrival
151    /// recorder must precede it. One placed after would read a packet's *playout* instant and
152    /// report it to the remote as its arrival time; the remote's congestion controller would then
153    /// see this endpoint's own buffering depth as network delay variation — a delay signal
154    /// manufactured locally and indistinguishable, at the far end, from a congested path.
155    JitterBuffer = 13_000,
156    /// Anywhere else, for an interceptor this crate knows nothing about.
157    ///
158    /// The named slots are spaced a thousand apart so one of your own fits between any two of them
159    /// without renumbering anything: `Slot::from(6_500)` sits after the FEC decoder and before the
160    /// NACK generator. Reach it through [`From<usize>`](#impl-From<usize>-for-Slot) rather than by
161    /// naming the variant, so the spelling survives this gaining a richer representation.
162    ///
163    /// **Choosing a number.** Work out which legs your interceptor uses, then apply the same rules
164    /// the named slots obey:
165    ///
166    /// * *It produces packets* — retransmissions, repair, RTCP, anything the wire has not seen yet.
167    ///   Put it **above [`Slot::Pacer`]** (> 3_000), or its output leaves unpaced and the send
168    ///   history never counts the bytes. This is the constraint people miss, because an interceptor
169    ///   that only *reports* on what it received still produces packets to report with.
170    /// * *It reads inbound sequence numbers or arrival times* — loss detection, arrival recording,
171    ///   reception statistics. Put it **below [`Slot::JitterBuffer`]** (< 13_000), so it sees the
172    ///   order and timing the path produced rather than the order this endpoint replays.
173    /// * *It repairs or recovers inbound packets.* Put it **below anything that would otherwise ask
174    ///   for them again** — below [`Slot::NackGenerator`] (< 7_000), as the FEC decoder is.
175    /// * *It only observes, and emits nothing.* Nothing pins it; pick a slot that reads well next to
176    ///   its neighbours.
177    ///
178    /// Both legs walk this one list — read from low to high, write from high to low — so a slot is
179    /// a position in *both* directions at once. An interceptor that acts on each leg is subject to
180    /// the constraints of each, and those can pull in opposite directions:
181    /// [`Slot::TwccReceiver`] is the worked example.
182    Custom(usize),
183}
184
185impl Slot {
186    /// Where this sits, as a distance from the wire.
187    ///
188    /// The named slots are the thousands; a custom one is whatever it was built from.
189    pub const fn slot(self) -> usize {
190        match self {
191            Self::CongestionControl => 1_000,
192            Self::TwccSender => 2_000,
193            Self::Pacer => 3_000,
194            Self::NackResponder => 4_000,
195            Self::FecEncoder => 5_000,
196            Self::FecDecoder => 6_000,
197            Self::NackGenerator => 7_000,
198            Self::TwccReceiver => 8_000,
199            Self::Rfc8888 => 9_000,
200            Self::ReceiverReport => 10_000,
201            Self::SenderReport => 11_000,
202            Self::IntervalPli => 12_000,
203            Self::JitterBuffer => 13_000,
204            Self::Custom(position) => position,
205        }
206    }
207}
208
209/// A position of your own. See [`Slot::Custom`].
210impl From<usize> for Slot {
211    fn from(position: usize) -> Self {
212        Self::Custom(position)
213    }
214}
215
216impl From<Slot> for usize {
217    fn from(slot: Slot) -> Self {
218        slot.slot()
219    }
220}
221
222// Equality and ordering are both by position, and they are written out rather than derived because
223// deriving them would disagree with each other. A derived `PartialEq` compares variants, so
224// `Slot::from(2_000) != Slot::TwccSender` even though both name the same distance from the wire; a
225// derived `Ord` compares *declaration* order, so `Slot::Custom(1_500)` would sort after
226// `JitterBuffer` rather than between `CongestionControl` and `TwccSender` — which is the whole
227// point of allowing a custom one. Two values that compare `Equal` must also be `==`, and a sort by
228// slot must put a custom position where its number says, so both come from [`Slot::slot`].
229impl PartialEq for Slot {
230    fn eq(&self, other: &Self) -> bool {
231        self.slot() == other.slot()
232    }
233}
234
235impl Eq for Slot {}
236
237impl PartialOrd for Slot {
238    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
239        Some(self.cmp(other))
240    }
241}
242
243impl Ord for Slot {
244    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
245        self.slot().cmp(&other.slot())
246    }
247}
248
249impl std::hash::Hash for Slot {
250    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
251        self.slot().hash(state);
252    }
253}
254
255/// Collects interceptors and assembles them into a chain.
256///
257/// # Order
258///
259/// Every interceptor is added at a [`Slot`], and [`build`](Self::build) sorts by it. A slot is a
260/// **distance from the wire**: the smallest is closest to the network, the largest closest to the
261/// application. Read walks that order, write walks it in reverse, so one list serves both
262/// directions and "closest to the wire" means one thing rather than opposite things per direction.
263///
264/// ```text
265/// Registry::new()
266///     .with(Slot::TwccSender, a)     // 2_000: closest to the wire
267///     .with(Slot::NackGenerator, b)  // 7_000
268///     .with(Slot::JitterBuffer, c)   // 13_000: closest to the application
269///     .build()
270///
271/// read:   a → b → c → application
272/// write:  application → c → b → a → wire
273/// ```
274///
275/// Declaring the position rather than relying on call order is what makes the helpers in
276/// `rtc` composable: `configure_twcc` places interceptors at 2_000 and 8_000, `configure_nack` at
277/// 4_000 and 7_000, and the two interleave correctly however the caller sequences them. With order taken
278/// from insertion, calling them in either sequence produced a chain that was wrong in a different
279/// way each time, and nothing caught it — the nested registry that preceded this added *innermost*
280/// first, so `register_default_interceptors` assembled TWCC receiver → RTCP reports → NACK
281/// generator, the reverse of what the chain contract documented.
282///
283/// A slot holds one interceptor. Two of your own go at two custom positions — the named slots are
284/// spaced a thousand apart so there is room between any two of them.
285///
286/// # Example
287///
288/// ```
289/// use rtc_interceptor::{NackGeneratorBuilder, Registry, Slot, TwccSenderBuilder};
290///
291/// let chain = Registry::new()
292///     .with(Slot::TwccSender, TwccSenderBuilder::new().build())        // closest to the wire
293///     .with(Slot::NackGenerator, NackGeneratorBuilder::new().build())  // sees arrivals after it
294///     .build();                                                        // terminus appended here
295/// # let _ = chain;
296/// ```
297#[derive(Default)]
298pub struct Registry {
299    interceptors: BTreeMap<Slot, BoxedInterceptor>,
300    /// What each interceptor is called, keyed the same way as `interceptors`.
301    ///
302    /// Kept beside the chain rather than asked of it: `Interceptor` is a trait object by the time
303    /// it is stored, and a trait object cannot say what it used to be. Recording the name at the
304    /// one moment the concrete type is still in hand is the only way to have it later, which is
305    /// also why [`with`](Self::with) takes a concrete interceptor rather than a boxed one.
306    names: BTreeMap<Slot, String>,
307}
308
309/// A type's name without its module paths — `TwccSenderInterceptor`, not
310/// `rtc_interceptor::twcc::sender::TwccSenderInterceptor`.
311///
312/// Every path is shortened, not just the outermost one, so a congestion controller reads as
313/// `CongestionControlInterceptor<Gcc>`. Splitting the whole string on its last `::` would be
314/// simpler and wrong: on a generic type that separator sits inside the *argument*, and
315/// `CongestionControlInterceptor<rtc_interceptor::cc::estimator::ConstantBitrate>` comes back as
316/// `ConstantBitrate>` — the interceptor's own name gone, and a stray bracket left behind.
317///
318/// The generic argument is kept because it is often the only thing telling two interceptors apart:
319/// which estimator a congestion controller carries is the interesting half of its name.
320fn short_type_name<T: ?Sized>() -> String {
321    let full = std::any::type_name::<T>();
322    let mut out = String::with_capacity(full.len());
323    let mut segment = String::new();
324
325    let flush = |segment: &mut String, out: &mut String| {
326        out.push_str(segment.rsplit("::").next().unwrap_or(segment));
327        segment.clear();
328    };
329
330    for ch in full.chars() {
331        // A path segment runs until punctuation that cannot appear in one: `<`, `>`, `,`, a space.
332        if ch.is_alphanumeric() || ch == '_' || ch == ':' {
333            segment.push(ch);
334        } else {
335            flush(&mut segment, &mut out);
336            out.push(ch);
337        }
338    }
339    flush(&mut segment, &mut out);
340
341    out
342}
343
344impl Registry {
345    /// An empty registry.
346    pub fn new() -> Self {
347        Self::default()
348    }
349
350    /// Add an interceptor at `slot`.
351    ///
352    /// Call order does not matter: the slot decides the position. A slot holds one interceptor, so
353    /// adding a second at the same position replaces the first and says so in the log.
354    pub fn with<T: Interceptor + 'static>(mut self, slot: Slot, interceptor: T) -> Self {
355        let name = short_type_name::<T>();
356
357        // One interceptor per slot: the map key is the position. Replacing rather than stacking is
358        // what a map gives, and it is announced rather than done quietly — an interceptor that
359        // vanished because something else claimed its slot is the kind of fault that shows up much
360        // later as "the chain does not do what I configured".
361        if let Some(displaced) = self.names.insert(slot, name.clone()) {
362            warn!("{slot:?} already held {displaced}; {name} replaced it");
363        }
364        self.interceptors.insert(slot, Box::new(interceptor));
365        self
366    }
367
368    /// What this registry holds, wire-to-application: each interceptor's slot and its type name,
369    /// in the order [`build`](Self::build) will compose them.
370    ///
371    /// Present so a caller assembling a chain from several helpers can assert what it got. Each
372    /// helper places interceptors at its own landmarks and none of them sees the whole, so the
373    /// composition is precisely the thing no single helper can check.
374    pub fn slots(&self) -> Vec<(Slot, String)> {
375        // Already wire-to-application: a `BTreeMap` iterates in key order, and `Slot` orders by
376        // position. This is the order `build` will compose them in, for the same reason.
377        self.names
378            .iter()
379            .map(|(slot, name)| (*slot, name.clone()))
380            .collect()
381    }
382
383    /// Assemble the interceptor chain.
384    ///
385    /// [`NoopInterceptor`] is appended last, so every chain ends the inbound RTCP path. That is a
386    /// property of a chain rather than something a caller opts into: left out, an application would
387    /// get a stream of control traffic it never asked for, and the omission would look like working
388    /// code.
389    ///
390    /// What gets past it is decided per packet, by an interceptor attaching
391    /// [`Attribute::DeliverToApplication`](crate::Attribute::DeliverToApplication) to the ones it
392    /// vouches for — the component that knows which packets an application can act on is the one
393    /// that makes the call, rather than a switch here that could only say "all of it or none".
394    pub fn build(self) -> impl Interceptor {
395        // No sort: a `BTreeMap` is already in key order, and `Slot` orders by distance from the
396        // wire, which is the order the chain runs in.
397        let mut interceptors: Vec<BoxedInterceptor> = self.interceptors.into_values().collect();
398
399        interceptors.push(Box::new(NoopInterceptor::new()));
400
401        Chain::new(interceptors)
402    }
403}
404
405#[cfg(test)]
406mod tests {
407    use super::*;
408    use crate::StreamInfo;
409    use crate::{AttributedPacket, Packet, TaggedPacket};
410    use sansio::Protocol;
411    use shared::TransportContext;
412    use shared::error::Error;
413    use std::collections::VecDeque;
414    use std::sync::{Arc, Mutex};
415    use std::time::Instant;
416
417    #[derive(Clone, Default)]
418    struct Log(Arc<Mutex<Vec<&'static str>>>);
419
420    struct Marker {
421        name: &'static str,
422        log: Log,
423        read_queue: VecDeque<TaggedPacket>,
424        write_queue: VecDeque<TaggedPacket>,
425    }
426
427    impl Marker {
428        fn new(name: &'static str, log: Log) -> Self {
429            Self {
430                name,
431                log,
432                read_queue: VecDeque::new(),
433                write_queue: VecDeque::new(),
434            }
435        }
436    }
437
438    impl Protocol<TaggedPacket, TaggedPacket, ()> for Marker {
439        type Rout = TaggedPacket;
440        type Wout = TaggedPacket;
441        type Eout = ();
442        type Error = Error;
443        type Time = Instant;
444
445        fn handle_read(&mut self, msg: TaggedPacket) -> Result<(), Self::Error> {
446            self.log.0.lock().unwrap().push(self.name);
447            self.read_queue.push_back(msg);
448            Ok(())
449        }
450
451        fn poll_read(&mut self) -> Option<Self::Rout> {
452            self.read_queue.pop_front()
453        }
454
455        fn handle_write(&mut self, msg: TaggedPacket) -> Result<(), Self::Error> {
456            self.log.0.lock().unwrap().push(self.name);
457            self.write_queue.push_back(msg);
458            Ok(())
459        }
460
461        fn poll_write(&mut self) -> Option<Self::Wout> {
462            self.write_queue.pop_front()
463        }
464
465        fn handle_timeout(&mut self, _now: Instant) -> Result<(), Self::Error> {
466            Ok(())
467        }
468
469        fn poll_timeout(&mut self) -> Option<Self::Time> {
470            None
471        }
472    }
473
474    impl Interceptor for Marker {
475        fn bind_local_stream(&mut self, _info: &StreamInfo) {}
476        fn unbind_local_stream(&mut self, _info: &StreamInfo) {}
477        fn bind_remote_stream(&mut self, _info: &StreamInfo) {}
478        fn unbind_remote_stream(&mut self, _info: &StreamInfo) {}
479    }
480
481    fn packet() -> TaggedPacket {
482        TaggedPacket {
483            now: Instant::now(),
484            transport: TransportContext::default(),
485            message: AttributedPacket::new(Packet::Rtp(rtp::Packet::default())),
486        }
487    }
488
489    fn chain(log: &Log) -> impl Interceptor {
490        Registry::new()
491            .with(Slot::TwccSender, Marker::new("wire", log.clone()))
492            .with(Slot::NackGenerator, Marker::new("middle", log.clone()))
493            .with(Slot::JitterBuffer, Marker::new("app", log.clone()))
494            .build()
495    }
496
497    /// Slots decide the order, not the sequence of calls. Adding application-most first must
498    /// compose the same chain as adding wire-most first — the property the helpers in `rtc` rely
499    /// on to be callable in any sequence.
500    #[test]
501    fn call_order_does_not_decide_chain_order() {
502        let log = Log::default();
503        let mut chain = Registry::new()
504            .with(Slot::JitterBuffer, Marker::new("app", log.clone()))
505            .with(Slot::TwccSender, Marker::new("wire", log.clone()))
506            .with(Slot::NackGenerator, Marker::new("middle", log.clone()))
507            .build();
508
509        chain.handle_read(packet()).unwrap();
510        while chain.poll_read().is_some() {}
511
512        assert_eq!(vec!["wire", "middle", "app"], *log.0.lock().unwrap());
513    }
514
515    /// A slot holds one interceptor: adding a second at the same position replaces the first
516    /// rather than stacking with it. Two of your own go at two custom positions, which is what the
517    /// thousand-apart spacing leaves room for.
518    #[test]
519    fn a_slot_holds_one_interceptor() {
520        let log = Log::default();
521        let mut chain = Registry::new()
522            .with(Slot::NackGenerator, Marker::new("first", log.clone()))
523            .with(Slot::NackGenerator, Marker::new("second", log.clone()))
524            .build();
525
526        chain.handle_read(packet()).unwrap();
527        while chain.poll_read().is_some() {}
528
529        assert_eq!(
530            vec!["second"],
531            *log.0.lock().unwrap(),
532            "the later one claimed the slot; the earlier one is not in the chain"
533        );
534    }
535
536    /// Read runs the list forwards: the first interceptor added is closest to the wire.
537    #[test]
538    fn read_runs_in_the_order_stages_were_added() {
539        let log = Log::default();
540        let mut chain = chain(&log);
541
542        chain.handle_read(packet()).unwrap();
543        while chain.poll_read().is_some() {}
544
545        assert_eq!(vec!["wire", "middle", "app"], *log.0.lock().unwrap());
546    }
547
548    /// Write runs it backwards, so the same list describes both directions.
549    #[test]
550    fn write_runs_in_reverse() {
551        let log = Log::default();
552        let mut chain = chain(&log);
553
554        chain.handle_write(packet()).unwrap();
555        while chain.poll_write().is_some() {}
556
557        assert_eq!(vec!["app", "middle", "wire"], *log.0.lock().unwrap());
558    }
559
560    /// Ending the inbound RTCP path is a property of every chain, not something a caller adds.
561    #[test]
562    fn a_registry_with_nothing_added_still_has_the_terminus() {
563        let mut chain = Registry::new().build();
564
565        chain
566            .handle_read(TaggedPacket {
567                now: Instant::now(),
568                transport: TransportContext::default(),
569                message: AttributedPacket::new(Packet::Rtcp(vec![])),
570            })
571            .unwrap();
572        assert!(
573            chain.poll_read().is_none(),
574            "inbound RTCP stops before the application"
575        );
576    }
577
578    /// The terminus goes last, so every interceptor sees inbound RTCP before it is dropped.
579    #[test]
580    fn the_terminus_is_application_most() {
581        let log = Log::default();
582        let mut chain = Registry::new()
583            .with(Slot::TwccSender, Marker::new("wire", log.clone()))
584            .build();
585
586        chain
587            .handle_read(TaggedPacket {
588                now: Instant::now(),
589                transport: TransportContext::default(),
590                message: AttributedPacket::new(Packet::Rtcp(vec![])),
591            })
592            .unwrap();
593
594        assert_eq!(
595            vec!["wire"],
596            *log.0.lock().unwrap(),
597            "the stage saw the RTCP packet; the terminus dropped it afterwards"
598        );
599        assert!(chain.poll_read().is_none());
600    }
601
602    /// An application's own interceptor goes between two named slots, which is what the spacing is
603    /// for: nothing has to be renumbered to make room.
604    #[test]
605    fn a_custom_slot_sits_where_its_number_says() {
606        let log = Log::default();
607        let mut chain = Registry::new()
608            .with(Slot::FecDecoder, Marker::new("fec", log.clone()))
609            .with(Slot::NackGenerator, Marker::new("nack", log.clone()))
610            .with(Slot::from(6_500), Marker::new("mine", log.clone()))
611            .build();
612
613        chain.handle_read(packet()).unwrap();
614        while chain.poll_read().is_some() {}
615
616        assert_eq!(
617            vec!["fec", "mine", "nack"],
618            *log.0.lock().unwrap(),
619            "6_500 belongs after the FEC decoder at 6_000 and before the NACK generator at 7_000"
620        );
621    }
622
623    /// A custom slot naming a named slot's position *is* that slot. Equality and ordering both come
624    /// from the position, and they have to agree: a pair that compares `Equal` must also be `==`,
625    /// or a sort or a `BTreeMap` keyed on this behaves differently depending on which spelling the
626    /// caller reached for.
627    #[test]
628    fn equality_and_ordering_both_follow_the_position() {
629        assert_eq!(Slot::TwccSender, Slot::from(2_000));
630        assert_eq!(
631            std::cmp::Ordering::Equal,
632            Slot::TwccSender.cmp(&Slot::from(2_000))
633        );
634        assert!(Slot::from(1_500) > Slot::CongestionControl);
635        assert!(Slot::from(1_500) < Slot::TwccSender);
636        assert!(
637            Slot::from(20_000) > Slot::JitterBuffer,
638            "a position past every named slot sorts past them, not by declaration order"
639        );
640    }
641
642    /// The named slots keep the spacing the doc promises, so `Slot::from` has room to aim at.
643    #[test]
644    fn the_named_slots_are_spaced_by_a_thousand() {
645        let named = [
646            Slot::CongestionControl,
647            Slot::TwccSender,
648            Slot::Pacer,
649            Slot::NackResponder,
650            Slot::FecEncoder,
651            Slot::FecDecoder,
652            Slot::NackGenerator,
653            Slot::TwccReceiver,
654            Slot::Rfc8888,
655            Slot::ReceiverReport,
656            Slot::SenderReport,
657            Slot::IntervalPli,
658            Slot::JitterBuffer,
659        ];
660
661        for pair in named.windows(2) {
662            assert_eq!(
663                1_000,
664                pair[1].slot() - pair[0].slot(),
665                "{:?} and {:?} must stay a thousand apart",
666                pair[0],
667                pair[1]
668            );
669        }
670    }
671
672    /// A registry records what each interceptor is called, which a chain of trait objects could not
673    /// tell you afterwards. It is what makes a composed chain inspectable — several helpers each
674    /// place interceptors at their own landmarks, and this is the only view of the result.
675    #[test]
676    fn slots_carry_the_interceptor_names() {
677        let log = Log::default();
678        let registry = Registry::new()
679            .with(Slot::JitterBuffer, Marker::new("app", log.clone()))
680            .with(Slot::TwccSender, crate::TwccSenderBuilder::new().build());
681
682        assert_eq!(
683            vec![
684                (Slot::TwccSender, "TwccSenderInterceptor".to_owned()),
685                (Slot::JitterBuffer, "Marker".to_owned()),
686            ],
687            registry.slots(),
688            "names come back with their slots, sorted wire-to-application"
689        );
690    }
691
692    /// The module path is dropped: a name is for reading, and the full path is mostly the crate's
693    /// own directory layout.
694    #[test]
695    fn names_are_stripped_of_their_module_path() {
696        let registry = Registry::new().with(
697            Slot::CongestionControl,
698            crate::CongestionControlBuilder::new(crate::ConstantBitrate::new(1_000_000.0)).build(),
699        );
700
701        let (_, name) = &registry.slots()[0];
702        assert!(
703            !name.contains("::"),
704            "a module path leaked into the name: {name}"
705        );
706        assert_eq!(
707            "CongestionControlInterceptor<ConstantBitrate>", name,
708            "the generic argument is shortened too, and kept — it is what tells two \
709             congestion controllers apart"
710        );
711    }
712}