rtc_media/io/ogg_reader/
mod.rs

1//! Reading Opus audio from an Ogg container.
2//!
3//! An Ogg stream is a sequence of pages; the first two carry the Opus headers (`OpusHead`, then
4//! `OpusTags`), and the rest carry audio packets. [`OggReader`](crate::io::ogg_reader::OggReader) walks the pages and hands back
5//! one Opus packet at a time, which is exactly the unit RTP carries.
6//!
7//! The `OpusHead` fields worth noting are [`pre_skip`](crate::io::ogg_reader::OggHeader::pre_skip) — decoder warm-up
8//! samples to discard — and [`sample_rate`](crate::io::ogg_reader::OggHeader::sample_rate), which records the *original*
9//! input rate; Opus itself always decodes at 48 kHz.
10#[cfg(test)]
11mod ogg_reader_test;
12
13use std::io::{Cursor, Read};
14
15use byteorder::{LittleEndian, ReadBytesExt};
16use bytes::BytesMut;
17
18use crate::io::ResetFn;
19use shared::error::{Error, Result};
20
21/// Page header flag: this page continues a packet from the previous page.
22pub const PAGE_HEADER_TYPE_CONTINUATION_OF_STREAM: u8 = 0x00;
23/// Page header flag: the first page of a logical stream.
24pub const PAGE_HEADER_TYPE_BEGINNING_OF_STREAM: u8 = 0x02;
25/// Page header flag: the last page of a logical stream.
26pub const PAGE_HEADER_TYPE_END_OF_STREAM: u8 = 0x04;
27/// The recommended Opus pre-skip: 3840 samples (80 ms at 48 kHz) of decoder warm-up to
28/// discard.
29pub const DEFAULT_PRE_SKIP: u16 = 3840; // 3840 recommended in the RFC
30/// The four-byte signature that begins every Ogg page.
31pub const PAGE_HEADER_SIGNATURE: &[u8] = b"OggS";
32/// The signature of the Opus identification header.
33pub const ID_PAGE_SIGNATURE: &[u8] = b"OpusHead";
34/// The signature of the Opus comment header.
35pub const COMMENT_PAGE_SIGNATURE: &[u8] = b"OpusTags";
36/// The fixed part of an Ogg page header, in bytes, before the segment table.
37pub const PAGE_HEADER_SIZE: usize = 27;
38/// The size of the `OpusHead` payload in bytes.
39pub const ID_PAGE_PAYLOAD_SIZE: usize = 19;
40
41/// Header type classification for Opus pages
42#[derive(Debug, Clone, Copy, PartialEq, Eq)]
43pub enum OggHeaderType {
44    /// OpusHead - Opus ID page
45    OpusHead,
46    /// OpusTags - Opus comment/metadata page
47    OpusTags,
48}
49
50/// OggReader is used to read Ogg files and return page payloads
51pub struct OggReader<R: Read> {
52    reader: R,
53    bytes_read: usize,
54    checksum_table: [u32; 256],
55    do_checksum: bool,
56}
57
58/// OggHeader is the metadata from the first two pages
59/// in the file (ID and Comment)
60/// <https://tools.ietf.org/html/rfc7845.html#section-3>
61#[derive(Debug, Clone)]
62pub struct OggHeader {
63    /// The channel mapping family, which says how channels map to speakers.
64    pub channel_map: u8,
65    /// The channel count.
66    pub channels: u8,
67    /// A gain in Q7.8 dB to apply when decoding.
68    pub output_gain: u16,
69    /// Samples to discard from the start of the stream — decoder warm-up.
70    pub pre_skip: u16,
71    /// The original input sample rate. Opus always decodes at 48 kHz regardless.
72    pub sample_rate: u32,
73    /// The `OpusHead` version, currently 1.
74    pub version: u8,
75    /// The number of Opus streams, for mapping families above 0.
76    pub stream_count: u8,
77    /// How many of those streams are coupled stereo pairs.
78    pub coupled_count: u8,
79    /// Which stream channel feeds each output channel.
80    pub channel_mapping: Vec<u8>,
81}
82
83/// OpusTags contains Vorbis comment metadata from an OpusTags page
84/// <https://www.xiph.org/vorbis/doc/v-comment.html>
85#[derive(Debug, Clone, Default)]
86pub struct OpusTags {
87    /// The encoder that produced the file.
88    pub vendor: String,
89    /// Metadata tags from the `OpusTags` header.
90    pub user_comments: Vec<UserComment>,
91}
92
93/// A key-value pair from Vorbis comments
94#[derive(Debug, Clone)]
95pub struct UserComment {
96    /// The tag name, such as `TITLE` or `ARTIST`.
97    pub comment: String,
98    /// The tag value.
99    pub value: String,
100}
101
102/// OggPageHeader is the metadata for a Page
103/// Pages are the fundamental unit of multiplexing in an Ogg stream
104/// <https://tools.ietf.org/html/rfc7845.html#section-1>
105#[derive(Debug, Clone)]
106pub struct OggPageHeader {
107    /// The page's granule position: total decoded samples at 48 kHz through this page.
108    pub granule_position: u64,
109    /// Serial number of the logical bitstream (track)
110    pub serial: u32,
111    /// Page header type flags
112    pub header_type: u8,
113
114    sig: [u8; 4],
115    version: u8,
116    index: u32,
117    segments_count: u8,
118}
119
120impl OggPageHeader {
121    /// Classify the page payload as OpusHead or OpusTags header
122    pub fn opus_header_type(&self, payload: &[u8]) -> Option<OggHeaderType> {
123        if payload.len() < 8 {
124            return None;
125        }
126
127        let sig = &payload[..8];
128        if sig == ID_PAGE_SIGNATURE {
129            // OpusHead must be beginning of stream
130            if self.header_type == PAGE_HEADER_TYPE_BEGINNING_OF_STREAM {
131                return Some(OggHeaderType::OpusHead);
132            }
133            return None;
134        }
135        if sig == COMMENT_PAGE_SIGNATURE {
136            return Some(OggHeaderType::OpusTags);
137        }
138
139        None
140    }
141
142    /// Check if this is the beginning of a stream
143    pub fn is_beginning_of_stream(&self) -> bool {
144        self.header_type == PAGE_HEADER_TYPE_BEGINNING_OF_STREAM
145    }
146
147    /// Check if this is the end of a stream
148    pub fn is_end_of_stream(&self) -> bool {
149        self.header_type == PAGE_HEADER_TYPE_END_OF_STREAM
150    }
151}
152
153/// Parse an OpusHead from a page payload
154/// <https://tools.ietf.org/html/rfc7845.html#section-5.1>
155pub fn parse_opus_head(payload: &[u8]) -> Result<OggHeader> {
156    if payload.len() < ID_PAGE_PAYLOAD_SIZE {
157        return Err(Error::ErrBadIDPageLength);
158    }
159
160    if &payload[..8] != ID_PAGE_SIGNATURE {
161        return Err(Error::ErrBadIDPagePayloadSignature);
162    }
163
164    let mut reader = Cursor::new(&payload[8..]);
165    let version = reader.read_u8()?;
166    let channels = reader.read_u8()?;
167    let pre_skip = reader.read_u16::<LittleEndian>()?;
168    let sample_rate = reader.read_u32::<LittleEndian>()?;
169    let output_gain = reader.read_u16::<LittleEndian>()?;
170    let channel_map = reader.read_u8()?;
171
172    let (stream_count, coupled_count, channel_mapping) = match channel_map {
173        0 => {
174            // Family 0: mono or stereo, no mapping table
175            if payload.len() != ID_PAGE_PAYLOAD_SIZE {
176                return Err(Error::ErrBadIDPageLength);
177            }
178            (0, 0, vec![])
179        }
180        1 | 2 | 255 => {
181            // Extended channel mapping
182            let expected_len = 21 + channels as usize;
183            if payload.len() < expected_len {
184                return Err(Error::ErrBadIDPageLength);
185            }
186            let stream_count = payload[19];
187            let coupled_count = payload[20];
188            let channel_mapping = payload[21..expected_len].to_vec();
189            (stream_count, coupled_count, channel_mapping)
190        }
191        3 => {
192            return Err(Error::ErrUnsupportedChannelMappingFamily);
193        }
194        _ => {
195            return Err(Error::ErrUnsupportedChannelMappingFamily);
196        }
197    };
198
199    Ok(OggHeader {
200        channel_map,
201        channels,
202        output_gain,
203        pre_skip,
204        sample_rate,
205        version,
206        stream_count,
207        coupled_count,
208        channel_mapping,
209    })
210}
211
212/// Parse OpusTags from a page payload
213/// <https://tools.ietf.org/html/rfc7845.html#section-5.2>
214pub fn parse_opus_tags(payload: &[u8]) -> Result<OpusTags> {
215    const HEADER_MAGIC_LEN: usize = 8;
216    const U32_SIZE: usize = 4;
217    const MIN_HEADER_LEN: usize = HEADER_MAGIC_LEN + U32_SIZE + U32_SIZE;
218
219    if payload.len() < MIN_HEADER_LEN {
220        return Err(Error::ErrBadOpusTagsSignature);
221    }
222
223    if &payload[..8] != COMMENT_PAGE_SIGNATURE {
224        return Err(Error::ErrBadOpusTagsSignature);
225    }
226
227    // Parse vendor string
228    let vendor_len = u32::from_le_bytes([
229        payload[HEADER_MAGIC_LEN],
230        payload[HEADER_MAGIC_LEN + 1],
231        payload[HEADER_MAGIC_LEN + 2],
232        payload[HEADER_MAGIC_LEN + 3],
233    ]) as usize;
234
235    let vendor_start = HEADER_MAGIC_LEN + U32_SIZE;
236    let vendor_end = vendor_start + vendor_len;
237
238    if vendor_end + U32_SIZE > payload.len() {
239        return Err(Error::ErrBadOpusTagsSignature);
240    }
241
242    let vendor = String::from_utf8_lossy(&payload[vendor_start..vendor_end]).to_string();
243
244    // Parse user comments
245    let comment_count = u32::from_le_bytes([
246        payload[vendor_end],
247        payload[vendor_end + 1],
248        payload[vendor_end + 2],
249        payload[vendor_end + 3],
250    ]) as usize;
251
252    let mut pos = vendor_end + U32_SIZE;
253    let mut user_comments = Vec::with_capacity(comment_count);
254
255    for _ in 0..comment_count {
256        if pos + U32_SIZE > payload.len() {
257            return Err(Error::ErrBadOpusTagsSignature);
258        }
259
260        let comment_len = u32::from_le_bytes([
261            payload[pos],
262            payload[pos + 1],
263            payload[pos + 2],
264            payload[pos + 3],
265        ]) as usize;
266        pos += U32_SIZE;
267
268        if pos + comment_len > payload.len() {
269            return Err(Error::ErrBadOpusTagsSignature);
270        }
271
272        let comment_str = String::from_utf8_lossy(&payload[pos..pos + comment_len]).to_string();
273        pos += comment_len;
274
275        // Split on first '=' to get key=value pair
276        if let Some(eq_pos) = comment_str.find('=') {
277            user_comments.push(UserComment {
278                comment: comment_str[..eq_pos].to_string(),
279                value: comment_str[eq_pos + 1..].to_string(),
280            });
281        }
282    }
283
284    Ok(OpusTags {
285        vendor,
286        user_comments,
287    })
288}
289
290impl<R: Read> OggReader<R> {
291    /// new returns a new Ogg reader and Ogg header
292    /// with an io.Reader input
293    ///
294    /// Warning: This only parses the first OpusHead (a single logical bitstream/track)
295    /// and returns a single OggHeader. If you need to handle Ogg containers with multiple
296    /// Opus headers/tracks, use new_with_options and scan pages via parse_next_page
297    /// to find and parse each OpusHead.
298    pub fn new(reader: R, do_checksum: bool) -> Result<(OggReader<R>, OggHeader)> {
299        let mut r = OggReader {
300            reader,
301            bytes_read: 0,
302            checksum_table: generate_checksum_table(),
303            do_checksum,
304        };
305
306        let header = r.read_headers()?;
307
308        Ok((r, header))
309    }
310
311    /// Create a new OggReader without consuming headers
312    ///
313    /// Use this when you need to handle Ogg containers with multiple
314    /// logical bitstreams (tracks). You can then use parse_next_page
315    /// to iterate through pages and parse_opus_head/parse_opus_tags
316    /// to parse the header pages for each track.
317    pub fn new_with_options(reader: R, do_checksum: bool) -> OggReader<R> {
318        OggReader {
319            reader,
320            bytes_read: 0,
321            checksum_table: generate_checksum_table(),
322            do_checksum,
323        }
324    }
325
326    fn read_headers(&mut self) -> Result<OggHeader> {
327        let (payload, page_header) = self.parse_next_page()?;
328
329        if page_header.sig != PAGE_HEADER_SIGNATURE {
330            return Err(Error::ErrBadIDPageSignature);
331        }
332
333        if page_header.header_type != PAGE_HEADER_TYPE_BEGINNING_OF_STREAM {
334            return Err(Error::ErrBadIDPageType);
335        }
336
337        parse_opus_head(&payload)
338    }
339
340    // parse_next_page reads from stream and returns Ogg page payload, header,
341    // and an error if there is incomplete page data.
342    /// Reads the next Ogg page, returning its payload and header.
343    ///
344    /// # Errors
345    ///
346    /// Fails on an I/O error, at end of stream, or if the page signature or checksum is wrong.
347    pub fn parse_next_page(&mut self) -> Result<(BytesMut, OggPageHeader)> {
348        let mut h = [0u8; PAGE_HEADER_SIZE];
349        self.reader.read_exact(&mut h)?;
350
351        let mut head_reader = Cursor::new(h);
352        let mut sig = [0u8; 4]; //0-3
353        head_reader.read_exact(&mut sig)?;
354        let version = head_reader.read_u8()?; //4
355        let header_type = head_reader.read_u8()?; //5
356        let granule_position = head_reader.read_u64::<LittleEndian>()?; //6-13
357        let serial = head_reader.read_u32::<LittleEndian>()?; //14-17
358        let index = head_reader.read_u32::<LittleEndian>()?; //18-21
359        let checksum = head_reader.read_u32::<LittleEndian>()?; //22-25
360        let segments_count = head_reader.read_u8()?; //26
361
362        let mut size_buffer = vec![0u8; segments_count as usize];
363        self.reader.read_exact(&mut size_buffer)?;
364
365        let mut payload_size = 0usize;
366        for s in &size_buffer {
367            payload_size += *s as usize;
368        }
369
370        let mut payload = BytesMut::with_capacity(payload_size);
371        payload.resize(payload_size, 0);
372        self.reader.read_exact(&mut payload)?;
373
374        if self.do_checksum {
375            let mut sum = 0;
376
377            for (index, v) in h.iter().enumerate() {
378                // Don't include expected checksum in our generation
379                if index > 21 && index < 26 {
380                    sum = self.update_checksum(0, sum);
381                    continue;
382                }
383                sum = self.update_checksum(*v, sum);
384            }
385
386            for v in &size_buffer {
387                sum = self.update_checksum(*v, sum);
388            }
389            for v in &payload[..] {
390                sum = self.update_checksum(*v, sum);
391            }
392
393            if sum != checksum {
394                return Err(Error::ErrChecksumMismatch);
395            }
396        }
397
398        let page_header = OggPageHeader {
399            granule_position,
400            sig,
401            version,
402            header_type,
403            serial,
404            index,
405            segments_count,
406        };
407
408        Ok((payload, page_header))
409    }
410
411    /// reset_reader resets the internal stream of OggReader. This is useful
412    /// for live streams, where the end of the file might be read without the
413    /// data being finished.
414    pub fn reset_reader(&mut self, mut reset: ResetFn<R>) {
415        self.reader = reset(self.bytes_read);
416    }
417
418    fn update_checksum(&self, v: u8, sum: u32) -> u32 {
419        (sum << 8) ^ self.checksum_table[(((sum >> 24) as u8) ^ v) as usize]
420    }
421}
422
423pub(crate) fn generate_checksum_table() -> [u32; 256] {
424    let mut table = [0u32; 256];
425    const POLY: u32 = 0x04c11db7;
426
427    for (i, t) in table.iter_mut().enumerate() {
428        let mut r = (i as u32) << 24;
429        for _ in 0..8 {
430            if (r & 0x80000000) != 0 {
431                r = (r << 1) ^ POLY;
432            } else {
433                r <<= 1;
434            }
435        }
436        *t = r;
437    }
438    table
439}
rtc_media/io/ogg_reader/mod.rs

rtc_media/io/ogg_reader/
mod.rs
