flv_core/
demux.rs

1//! FLV file-header + tag reader — incremental, byte-chunk push/poll.
2
3#![forbid(unsafe_code)]
4
5use bytes::Bytes;
6
7use crate::error::Error;
8use crate::types::{Tag, TagType};
9
10const TAG_HEADER_LEN: usize = 11;
11const TRAILER_LEN: usize = 4; // PreviousTagSize
12
13/// Reads an FLV file header followed by tags from pushed byte chunks.
14#[derive(Debug, Clone, Default)]
15pub struct Demuxer {
16    buf: Vec<u8>,
17    header_parsed: bool,
18    has_audio: bool,
19    has_video: bool,
20}
21
22impl Demuxer {
23    /// New, empty demux session.
24    #[must_use]
25    pub fn new() -> Self {
26        Self::default()
27    }
28
29    /// Append incoming bytes.
30    pub fn push_bytes(&mut self, data: &[u8]) {
31        self.buf.extend_from_slice(data);
32    }
33
34    /// Whether the file header (parsed) declares audio tags present.
35    #[must_use]
36    pub const fn has_audio(&self) -> Option<bool> {
37        if self.header_parsed {
38            Some(self.has_audio)
39        } else {
40            None
41        }
42    }
43
44    /// Whether the file header (parsed) declares video tags present.
45    #[must_use]
46    pub const fn has_video(&self) -> Option<bool> {
47        if self.header_parsed {
48            Some(self.has_video)
49        } else {
50            None
51        }
52    }
53
54    /// Pop the next complete tag, or `Ok(None)` if not enough bytes are buffered
55    /// yet — call again after more `push_bytes`.
56    pub fn poll_tag(&mut self) -> Result<Option<Tag>, Error> {
57        if !self.header_parsed && !self.parse_header()? {
58            return Ok(None);
59        }
60        self.parse_tag()
61    }
62
63    fn parse_header(&mut self) -> Result<bool, Error> {
64        if self.buf.len() < 9 {
65            return Ok(false);
66        }
67        if &self.buf[0..3] != b"FLV" {
68            return Err(Error::BadSignature);
69        }
70        let flags = self.buf[4];
71        let data_offset =
72            u32::from_be_bytes(self.buf[5..9].try_into().unwrap_or_default()) as usize;
73        let total_header = data_offset + TRAILER_LEN;
74        if self.buf.len() < total_header {
75            return Ok(false);
76        }
77        self.has_video = flags & 0x01 != 0;
78        self.has_audio = flags & 0x04 != 0;
79        self.header_parsed = true;
80        self.buf.drain(0..total_header);
81        Ok(true)
82    }
83
84    fn parse_tag(&mut self) -> Result<Option<Tag>, Error> {
85        if self.buf.len() < TAG_HEADER_LEN {
86            return Ok(None);
87        }
88        let tag_type =
89            TagType::from_value(self.buf[0]).ok_or(Error::UnknownTagType(self.buf[0]))?;
90        let data_size = (usize::from(self.buf[1]) << 16)
91            | (usize::from(self.buf[2]) << 8)
92            | usize::from(self.buf[3]);
93        let ts_low =
94            u32::from(self.buf[4]) << 16 | u32::from(self.buf[5]) << 8 | u32::from(self.buf[6]);
95        let ts_ext = u32::from(self.buf[7]);
96        let timestamp_ms = (ts_ext << 24) | ts_low;
97
98        let total = TAG_HEADER_LEN + data_size + TRAILER_LEN;
99        if self.buf.len() < total {
100            return Ok(None);
101        }
102        let data = Bytes::copy_from_slice(&self.buf[TAG_HEADER_LEN..TAG_HEADER_LEN + data_size]);
103        self.buf.drain(0..total);
104        Ok(Some(Tag {
105            tag_type,
106            timestamp_ms,
107            data,
108        }))
109    }
110}
111
112#[cfg(test)]
113#[path = "demux_tests.rs"]
114mod tests;
flv_core/demux.rs

flv_core/
demux.rs
