netget/server/rtp/
media.rs

1//! RTP media synthesis and packetization (RFC 3550).
2//!
3//! This is the VNC-of-audio: the model never produces samples or bytes. It answers an event
4//! with a *structured description* of what a stream should carry — a tone at a frequency, a
5//! run of DTMF digits, silence — and this module turns that description into G.711 (PCMU/PCMA)
6//! samples and frames them into correct RTP packets. The one escape hatch for genuinely
7//! arbitrary payloads is an explicit `encoding: "hex"` field the caller decodes here, mirroring
8//! the `send_tcp_data` reference fix — never sniffed, always declared.
9//!
10//! Shared by both the bare `rtp` server and the `rtsp` control server (whose PLAY streams RTP).
11
12use anyhow::{bail, Context, Result};
13
14/// G.711 sample rate. Both PCMU (PT 0) and PCMA (PT 8) are 8 kHz, 8 bits/sample, mono.
15pub const G711_CLOCK_HZ: u32 = 8000;
16
17/// Samples per 20 ms RTP frame at 8 kHz — the near-universal ptime for G.711.
18pub const G711_SAMPLES_PER_FRAME: usize = 160;
19
20/// Upper bound on synthesized duration, so a single action cannot enqueue an unbounded burst
21/// of datagrams onto the (unbounded, backpressure-free) socket.
22pub const MAX_DURATION_MS: u64 = 30_000;
23
24/// A static RTP payload type from the RTP/AVP profile (RFC 3551) that this engine can synthesize.
25#[derive(Debug, Clone, Copy, PartialEq, Eq)]
26pub enum AudioCodec {
27    /// PCMU / G.711 µ-law, payload type 0.
28    Pcmu,
29    /// PCMA / G.711 A-law, payload type 8.
30    Pcma,
31}
32
33impl AudioCodec {
34    /// Parse the `payload_type` field. Accepts the codec name or its numeric PT.
35    pub fn parse(value: &str) -> Result<Self> {
36        match value.trim().to_ascii_lowercase().as_str() {
37            "pcmu" | "mulaw" | "ulaw" | "g711u" | "0" => Ok(Self::Pcmu),
38            "pcma" | "alaw" | "g711a" | "8" => Ok(Self::Pcma),
39            other => bail!(
40                "unsupported payload_type {other:?}; this engine synthesizes only \"pcmu\" (PT 0) \
41                 or \"pcma\" (PT 8)"
42            ),
43        }
44    }
45
46    /// The static RTP/AVP payload type number.
47    pub fn payload_type(self) -> u8 {
48        match self {
49            Self::Pcmu => 0,
50            Self::Pcma => 8,
51        }
52    }
53
54    /// The `a=rtpmap` encoding name used in SDP.
55    pub fn rtpmap_name(self) -> &'static str {
56        match self {
57            Self::Pcmu => "PCMU",
58            Self::Pcma => "PCMA",
59        }
60    }
61
62    /// Encode one 16-bit linear PCM sample to this codec's single byte.
63    pub fn encode_sample(self, pcm: i16) -> u8 {
64        match self {
65            Self::Pcmu => linear_to_ulaw(pcm),
66            Self::Pcma => linear_to_alaw(pcm),
67        }
68    }
69}
70
71/// What a stream should carry, as decided by the model. This is the structured description that
72/// stands in for raw bytes.
73#[derive(Debug, Clone)]
74pub enum AudioContent {
75    /// A pure sine tone at the given frequency (Hz).
76    Tone { hz: f64 },
77    /// A run of DTMF digits (0-9, *, #, A-D), 150 ms tone + 50 ms gap each, classic dual-tone.
78    Dtmf { digits: String },
79    /// Digital silence (all-zero PCM).
80    Silence,
81    /// Caller-supplied, already-encoded codec bytes. `encoding` was `"hex"`; decoded here.
82    /// Used only when the content is genuinely not describable as tone/dtmf/silence.
83    Raw { encoded: Vec<u8> },
84}
85
86/// Parse an `AudioContent` from an action object's fields.
87///
88/// Recognised shapes (checked in order):
89/// - `{"content":"tone","tone_hz":440}`
90/// - `{"content":"dtmf","digits":"123#"}`
91/// - `{"content":"silence"}`
92/// - `{"content":"raw","encoding":"hex","samples":"ff7e..."}`
93///
94/// A bare `{"tone_hz":440}` is also accepted as a tone for convenience.
95pub fn parse_audio_content(action: &serde_json::Value) -> Result<AudioContent> {
96    let kind = action
97        .get("content")
98        .and_then(|v| v.as_str())
99        .map(|s| s.trim().to_ascii_lowercase());
100
101    match kind.as_deref() {
102        Some("tone") | None => {
103            let hz = action
104                .get("tone_hz")
105                .and_then(|v| v.as_f64())
106                .unwrap_or(440.0);
107            if !(20.0..=3800.0).contains(&hz) {
108                bail!("tone_hz {hz} is outside the 20-3800 Hz range representable at 8 kHz");
109            }
110            Ok(AudioContent::Tone { hz })
111        }
112        Some("dtmf") => {
113            let digits = action
114                .get("digits")
115                .and_then(|v| v.as_str())
116                .context("content \"dtmf\" requires a \"digits\" string")?
117                .to_string();
118            if digits.is_empty() {
119                bail!("\"digits\" must be non-empty for dtmf content");
120            }
121            Ok(AudioContent::Dtmf { digits })
122        }
123        Some("silence") => Ok(AudioContent::Silence),
124        Some("raw") => {
125            let encoding = action
126                .get("encoding")
127                .and_then(|v| v.as_str())
128                .unwrap_or("hex");
129            if encoding != "hex" {
130                bail!(
131                    "raw content encoding must be \"hex\" (got {encoding:?}); raw bytes are only \
132                     accepted hex-encoded and are decoded before framing"
133                );
134            }
135            let hexstr = action
136                .get("samples")
137                .and_then(|v| v.as_str())
138                .context("raw content requires a hex \"samples\" string")?;
139            let encoded = hex::decode(hexstr.trim())
140                .context("raw \"samples\" is not valid hex; the engine decodes it for real")?;
141            if encoded.is_empty() {
142                bail!("decoded raw \"samples\" is empty");
143            }
144            Ok(AudioContent::Raw { encoded })
145        }
146        Some(other) => {
147            bail!("unknown content {other:?}; valid: \"tone\", \"dtmf\", \"silence\", \"raw\"")
148        }
149    }
150}
151
152/// Synthesize codec bytes (one per sample) for a description and duration.
153///
154/// For `Raw`, the bytes are already codec-encoded and returned verbatim (duration ignored).
155pub fn synthesize(codec: AudioCodec, content: &AudioContent, duration_ms: u64) -> Result<Vec<u8>> {
156    if duration_ms == 0 {
157        bail!("duration_ms must be greater than zero");
158    }
159    if duration_ms > MAX_DURATION_MS {
160        bail!("duration_ms {duration_ms} exceeds the {MAX_DURATION_MS} ms cap");
161    }
162    let total = (G711_CLOCK_HZ as u64 * duration_ms / 1000) as usize;
163
164    let pcm: Vec<i16> = match content {
165        AudioContent::Raw { encoded } => return Ok(encoded.clone()),
166        AudioContent::Silence => vec![0i16; total],
167        AudioContent::Tone { hz } => (0..total)
168            .map(|n| {
169                let t = n as f64 / G711_CLOCK_HZ as f64;
170                (8000.0 * (2.0 * std::f64::consts::PI * hz * t).sin()) as i16
171            })
172            .collect(),
173        AudioContent::Dtmf { digits } => synthesize_dtmf(digits),
174    };
175    // DTMF length comes from the digit count (200 ms each), not from `duration_ms`, so the cap
176    // above does not constrain it. Without this a `digits` string a handler or a script can
177    // make arbitrarily long allocates without bound — 30 s is 150 digits, and the same 30 s
178    // ceiling should mean the same thing however the content was described.
179    let max_samples = (G711_CLOCK_HZ as u64 * MAX_DURATION_MS / 1000) as usize;
180    if pcm.len() > max_samples {
181        bail!(
182            "dtmf digits synthesize to {} ms, past the {} ms cap ({} digits at {} ms each is the \
183             most one action may stream)",
184            pcm.len() * 1000 / G711_CLOCK_HZ as usize,
185            MAX_DURATION_MS,
186            max_samples / ((G711_CLOCK_HZ as usize * DTMF_DIGIT_MS) / 1000),
187            DTMF_DIGIT_MS
188        );
189    }
190
191    Ok(pcm.into_iter().map(|s| codec.encode_sample(s)).collect())
192}
193
194/// Wall-clock length of one DTMF digit: 150 ms of tone plus a 50 ms gap.
195pub const DTMF_DIGIT_MS: usize = 200;
196
197/// DTMF: each digit is the sum of a low and a high frequency, 150 ms on + 50 ms off.
198fn synthesize_dtmf(digits: &str) -> Vec<i16> {
199    let mut out = Vec::new();
200    let on = (G711_CLOCK_HZ as usize * 150) / 1000;
201    let off = (G711_CLOCK_HZ as usize * 50) / 1000;
202    for ch in digits.chars() {
203        if let Some((low, high)) = dtmf_freqs(ch) {
204            for n in 0..on {
205                let t = n as f64 / G711_CLOCK_HZ as f64;
206                let s = 4000.0
207                    * ((2.0 * std::f64::consts::PI * low * t).sin()
208                        + (2.0 * std::f64::consts::PI * high * t).sin());
209                out.push(s as i16);
210            }
211            out.extend(std::iter::repeat(0i16).take(off));
212        }
213    }
214    out
215}
216
217fn dtmf_freqs(ch: char) -> Option<(f64, f64)> {
218    let (row, col) = match ch.to_ascii_uppercase() {
219        '1' => (697.0, 1209.0),
220        '2' => (697.0, 1336.0),
221        '3' => (697.0, 1477.0),
222        'A' => (697.0, 1633.0),
223        '4' => (770.0, 1209.0),
224        '5' => (770.0, 1336.0),
225        '6' => (770.0, 1477.0),
226        'B' => (770.0, 1633.0),
227        '7' => (852.0, 1209.0),
228        '8' => (852.0, 1336.0),
229        '9' => (852.0, 1477.0),
230        'C' => (852.0, 1633.0),
231        '*' => (941.0, 1209.0),
232        '0' => (941.0, 1336.0),
233        '#' => (941.0, 1477.0),
234        'D' => (941.0, 1633.0),
235        _ => return None,
236    };
237    Some((row, col))
238}
239
240/// Stateful RTP packetizer (RFC 3550 §5.1). Holds the running sequence number and timestamp so
241/// that successive bursts on one stream stay monotonic, as a real sender must.
242#[derive(Debug, Clone)]
243pub struct RtpPacketizer {
244    ssrc: u32,
245    payload_type: u8,
246    sequence: u16,
247    timestamp: u32,
248}
249
250impl RtpPacketizer {
251    /// New packetizer. Initial sequence and timestamp are randomized per RFC 3550 §5.1 unless a
252    /// value is supplied (tests pin them for determinism).
253    pub fn new(
254        ssrc: u32,
255        payload_type: u8,
256        initial_seq: Option<u16>,
257        initial_ts: Option<u32>,
258    ) -> Self {
259        use rand::Rng;
260        let mut rng = rand::thread_rng();
261        Self {
262            ssrc,
263            payload_type,
264            sequence: initial_seq.unwrap_or_else(|| rng.gen()),
265            timestamp: initial_ts.unwrap_or_else(|| rng.gen()),
266        }
267    }
268
269    pub fn ssrc(&self) -> u32 {
270        self.ssrc
271    }
272    pub fn next_sequence(&self) -> u16 {
273        self.sequence
274    }
275    pub fn next_timestamp(&self) -> u32 {
276        self.timestamp
277    }
278
279    /// Split codec `payload` into `samples_per_frame`-sized RTP packets. The first packet of the
280    /// burst carries the marker bit (RFC 3551: start of a talkspurt). Advances internal seq/ts.
281    pub fn packetize(&mut self, payload: &[u8], samples_per_frame: usize) -> Vec<Vec<u8>> {
282        let mut packets = Vec::new();
283        let mut first = true;
284        for chunk in payload.chunks(samples_per_frame.max(1)) {
285            let pkt = build_rtp_packet(
286                self.payload_type,
287                first,
288                self.sequence,
289                self.timestamp,
290                self.ssrc,
291                chunk,
292            );
293            packets.push(pkt);
294            self.sequence = self.sequence.wrapping_add(1);
295            self.timestamp = self.timestamp.wrapping_add(chunk.len() as u32);
296            first = false;
297        }
298        packets
299    }
300}
301
302/// Build one RTP packet: 12-byte fixed header (RFC 3550 §5.1) followed by the payload.
303pub fn build_rtp_packet(
304    payload_type: u8,
305    marker: bool,
306    sequence: u16,
307    timestamp: u32,
308    ssrc: u32,
309    payload: &[u8],
310) -> Vec<u8> {
311    let mut pkt = Vec::with_capacity(12 + payload.len());
312    // V=2, P=0, X=0, CC=0
313    pkt.push(0x80);
314    // M | PT
315    pkt.push(((marker as u8) << 7) | (payload_type & 0x7F));
316    pkt.extend_from_slice(&sequence.to_be_bytes());
317    pkt.extend_from_slice(&timestamp.to_be_bytes());
318    pkt.extend_from_slice(&ssrc.to_be_bytes());
319    pkt.extend_from_slice(payload);
320    pkt
321}
322
323/// Parsed view of an inbound RTP packet header.
324#[derive(Debug, Clone)]
325pub struct ParsedRtp {
326    pub version: u8,
327    pub marker: bool,
328    pub payload_type: u8,
329    pub sequence: u16,
330    pub timestamp: u32,
331    pub ssrc: u32,
332    pub payload_len: usize,
333}
334
335/// Parse the fixed RTP header. Returns None if the buffer is too short or not version 2.
336pub fn parse_rtp(data: &[u8]) -> Option<ParsedRtp> {
337    if data.len() < 12 {
338        return None;
339    }
340    let version = data[0] >> 6;
341    if version != 2 {
342        return None;
343    }
344    let cc = (data[0] & 0x0F) as usize;
345    let header_len = 12 + cc * 4;
346    if data.len() < header_len {
347        return None;
348    }
349    Some(ParsedRtp {
350        version,
351        marker: data[1] & 0x80 != 0,
352        payload_type: data[1] & 0x7F,
353        sequence: u16::from_be_bytes([data[2], data[3]]),
354        timestamp: u32::from_be_bytes([data[4], data[5], data[6], data[7]]),
355        ssrc: u32::from_be_bytes([data[8], data[9], data[10], data[11]]),
356        payload_len: data.len() - header_len,
357    })
358}
359
360/// True if a datagram is an RTCP packet rather than RTP. RTCP packet types occupy 200-204,
361/// which the RTP profile reserves from the payload-type space so the two never collide on a
362/// muxed port (RFC 5761 §4).
363pub fn is_rtcp(data: &[u8]) -> bool {
364    data.len() >= 2 && data[0] >> 6 == 2 && (200..=204).contains(&data[1])
365}
366
367/// Build a minimal RTCP Sender Report (RFC 3550 §6.4.1) with no reception report blocks.
368pub fn build_rtcp_sender_report(
369    ssrc: u32,
370    rtp_timestamp: u32,
371    packet_count: u32,
372    octet_count: u32,
373) -> Vec<u8> {
374    // NTP timestamp: seconds since 1900 + fraction. 1900->1970 is 2_208_988_800 s.
375    let now = crate::utils::clock::SystemTime::now()
376        .duration_since(crate::utils::clock::UNIX_EPOCH)
377        .unwrap_or_default();
378    let ntp_secs = now.as_secs() + 2_208_988_800;
379    let ntp_frac = ((now.subsec_nanos() as u64) << 32) / 1_000_000_000;
380
381    let mut pkt = Vec::with_capacity(28);
382    pkt.push(0x80); // V=2, P=0, RC=0
383    pkt.push(200); // PT = SR
384    pkt.extend_from_slice(&6u16.to_be_bytes()); // length in 32-bit words minus one (28 bytes)
385    pkt.extend_from_slice(&ssrc.to_be_bytes());
386    pkt.extend_from_slice(&(ntp_secs as u32).to_be_bytes());
387    pkt.extend_from_slice(&(ntp_frac as u32).to_be_bytes());
388    pkt.extend_from_slice(&rtp_timestamp.to_be_bytes());
389    pkt.extend_from_slice(&packet_count.to_be_bytes());
390    pkt.extend_from_slice(&octet_count.to_be_bytes());
391    pkt
392}
393
394// --- G.711 encoders (ITU-T G.711, standard reference implementation) ---
395
396/// Linear 16-bit PCM to G.711 µ-law.
397pub fn linear_to_ulaw(sample: i16) -> u8 {
398    const BIAS: i32 = 0x84;
399    const CLIP: i32 = 32635;
400    let mut pcm = sample as i32;
401    let sign = if pcm < 0 {
402        pcm = -pcm;
403        0x80
404    } else {
405        0
406    };
407    if pcm > CLIP {
408        pcm = CLIP;
409    }
410    pcm += BIAS;
411    let mut exponent = 7;
412    let mut mask = 0x4000;
413    while exponent > 0 && (pcm & mask) == 0 {
414        exponent -= 1;
415        mask >>= 1;
416    }
417    let mantissa = (pcm >> (exponent + 3)) & 0x0F;
418    let ulaw = !(sign | (exponent << 4) | mantissa);
419    (ulaw & 0xFF) as u8
420}
421
422/// Linear 16-bit PCM to G.711 A-law.
423pub fn linear_to_alaw(sample: i16) -> u8 {
424    let mut pcm = sample as i32;
425    let sign = if pcm >= 0 { 0x80 } else { 0 };
426    if sign == 0 {
427        pcm = -pcm;
428    }
429    if pcm > 32635 {
430        pcm = 32635;
431    }
432    let (exponent, mantissa) = if pcm >= 256 {
433        let mut exponent = 7;
434        let mut mask = 0x4000;
435        while exponent > 1 && (pcm & mask) == 0 {
436            exponent -= 1;
437            mask >>= 1;
438        }
439        (exponent, (pcm >> (exponent + 3)) & 0x0F)
440    } else {
441        (0, (pcm >> 4) & 0x0F)
442    };
443    let alaw = (sign | (exponent << 4) | mantissa) as u8;
444    alaw ^ 0x55
445}
netget/server/rtp/media.rs

netget/server/rtp/
media.rs
