RTC Interceptor - Sans-IO interceptor framework for RTP/RTCP processing.

This crate provides a composable interceptor framework built on top of the [sansio::Protocol] trait. Interceptors can process, modify, or generate RTP/RTCP packets as they flow through the pipeline.

Available Interceptors

RTCP Reports

NACK (Negative Acknowledgement)

TWCC (Transport Wide Congestion Control)

Congestion control

Utility

Design

A chain is a flat list of interceptors driven over a shared belt. Each one can:

• transform a packet passing through, or swallow it to drop or delay it
• emit packets it generated or was holding, which rejoin the belt and carry on
• act on timeouts, for periodic work like report generation
• track stream statistics and state

All interceptors work with [TaggedPacket] — an RTP or RTCP packet with transport metadata, carrying [Attribute]s that say what happened to it on the way. No interceptor holds a reference to another; [Registry] assembles the list and walks it. [Registry::build] appends [NoopInterceptor] last, so inbound RTCP stops before the application — control traffic the interceptors act on is not media the caller asked for. An interceptor that wants a particular packet delivered anyway attaches [Attribute::DeliverToApplication] to it.

Direction

A chain is a flat list ordered by distance from the wire: the first interceptor is closest to the network, the last closest to the application. Direction is a property of the walk, not of the structure:

read   (network → application)   forward:  first → … → last
write  (application → network)   reverse:  last  → … → first

Each interceptor is fed from a shared belt and its output is collected back onto it, so what a interceptor emits is seen by every interceptor still ahead of it in the walk. A retransmission emitted mid-chain still gets paced, numbered and recorded, because there is no way out of the chain except through the interceptors that follow.

One list serves both directions, so "closest to the wire" means one thing rather than opposite things per direction — which is why the send history and the FEC decoder sit next to each other, one being the last thing on the way out and the other the first on the way in.

Quick Start

use rtc_interceptor::{
    NackGeneratorBuilder, NackResponderBuilder, ReceiverReportBuilder,
    Registry, SenderReportBuilder, Slot, TwccReceiverBuilder, TwccSenderBuilder,
};
use std::time::Duration;

// The slot decides the position, not the order of these calls; they are listed
// wire-to-application here only because that reads the way the chain runs — forwards on the
// read path, and in reverse on the write path.
let chain = Registry::new()
    .with(Slot::TwccSender, TwccSenderBuilder::new().build())
    .with(Slot::NackResponder, NackResponderBuilder::new().build())
    .with(Slot::NackGenerator, NackGeneratorBuilder::new().build())
    .with(Slot::TwccReceiver, TwccReceiverBuilder::new().build())
    .with(Slot::ReceiverReport, ReceiverReportBuilder::new().build())
    .with(Slot::SenderReport, SenderReportBuilder::new().with_interval(Duration::from_secs(1)).build())
    .build();

// `build` appends [`NoopInterceptor`] last, so inbound RTCP — control traffic the interceptors
// above act on — stops there rather than arriving mixed in with the application's media. To
// receive some of it, add an interceptor that marks those packets `DeliverToApplication`.
# let _ = chain;

One chain type

[Registry::build] returns a single concrete type whatever it was built from, so a struct can hold one without a type parameter and two connections with different chains share a collection:

use rtc_interceptor::{Slot, NackGeneratorBuilder, Registry, SenderReportBuilder};

# let nack_enabled = true; // e.g. from configuration, negotiated SDP, …
let chain = if nack_enabled {
    Registry::new().with(Slot::NackGenerator, NackGeneratorBuilder::new().build()).build()
} else {
    Registry::new().with(Slot::SenderReport, SenderReportBuilder::new().build()).build()
};

The cost is one virtual call per interceptor per packet, which is nothing beside SRTP.

Stream Binding

Before interceptors can process packets for a stream, the stream must be bound:

use rtc_interceptor::{Slot, Interceptor, RTCPFeedback, RTPHeaderExtension, Registry, StreamInfo};

let mut chain = Registry::new().build();

// Create stream info with NACK and TWCC support
let stream_info = StreamInfo {
    ssrc: 0x12345678,
    clock_rate: 90000,
    mime_type: "video/VP8".to_string(),
    payload_type: 96,
    rtcp_feedback: vec![RTCPFeedback {
        typ: "nack".to_string(),
        parameter: String::new(),
    }],
    rtp_header_extensions: vec![RTPHeaderExtension {
        uri: "http://www.ietf.org/id/draft-holmer-rmcat-transport-wide-cc-extensions-01".to_string(),
        id: 5,
    }],
    ..Default::default()
};

// Bind for outgoing streams (sender side)
chain.bind_local_stream(&stream_info);

// Bind for incoming streams (receiver side)
chain.bind_remote_stream(&stream_info);

Writing your own

Implement [sansio::Protocol] and [Interceptor], then add it wherever it belongs in the list. What handle_* takes in, poll_* gives back — so even a pass-through needs a queue, because the queue is what the next interceptor is fed from:

use rtc_interceptor::{Slot, Interceptor, Registry, StreamInfo, TaggedPacket};
use sansio::Protocol;
use std::collections::VecDeque;
use std::time::Instant;

/// Counts packets on their way out.
#[derive(Default)]
struct Counter {
    sent: u64,
    read_queue: VecDeque<TaggedPacket>,
    write_queue: VecDeque<TaggedPacket>,
}

impl Protocol<TaggedPacket, TaggedPacket, ()> for Counter {
    type Rout = TaggedPacket;
    type Wout = TaggedPacket;
    type Eout = ();
    type Error = shared::error::Error;
    type Time = Instant;

    fn handle_read(&mut self, msg: TaggedPacket) -> Result<(), Self::Error> {
        self.read_queue.push_back(msg);
        Ok(())
    }

    fn poll_read(&mut self) -> Option<Self::Rout> {
        self.read_queue.pop_front()
    }

    fn handle_write(&mut self, msg: TaggedPacket) -> Result<(), Self::Error> {
        self.sent += 1;
        self.write_queue.push_back(msg); // queueing nothing would swallow it
        Ok(())
    }

    fn poll_write(&mut self) -> Option<Self::Wout> {
        self.write_queue.pop_front()
    }
}

impl Interceptor for Counter {
    fn bind_local_stream(&mut self, _info: &StreamInfo) {}
    fn unbind_local_stream(&mut self, _info: &StreamInfo) {}
    fn bind_remote_stream(&mut self, _info: &StreamInfo) {}
    fn unbind_remote_stream(&mut self, _info: &StreamInfo) {}
}

let chain = Registry::new().with(Slot::NackGenerator, Counter::default()).build();
# let _ = chain;

Queue nothing to drop or delay a packet, and queue delayed or generated ones whenever they are ready — from handle_timeout, say. They leave through poll_* and continue through every interceptor ahead.