o_sfu_rfc/
stream_decoder.rs

1//! Incremental decoding of frames from a byte stream, shared by the stream framings: the
2//! decoder buffers the bytes, a [`FrameCodec`] finds the frame boundaries.
3
4use std::{collections::VecDeque, io::Result as IoResult, marker::PhantomData};
5
6pub(crate) const DECODER_CHUNK_SIZE: usize = 4096;
7
8pub type Decoded<Frame, Error> = Option<Result<(Frame, usize), Error>>;
9
10/// How a stream framing finds the boundary of the next frame.
11pub trait FrameCodec {
12    type Frame;
13    type Error;
14
15    /// Decodes the first frame of `unread`.
16    fn try_decode(unread: &[u8]) -> Decoded<Self::Frame, Self::Error>;
17}
18
19/// Incremental decoder of frames: bytes go in through [`StreamDecoder::feed_with`], frames
20/// come out by iterating.
21///
22/// Drain the iterator after each feed, otherwise decoded frames accumulate in the buffer.
23///
24/// On error the stream is desynchronized and cannot be recovered: no byte is consumed, so
25/// every further call returns the same error. Close the stream.
26pub struct StreamDecoder<C> {
27    bytes: VecDeque<u8>,
28    codec: PhantomData<C>,
29}
30
31impl<C> Default for StreamDecoder<C> {
32    fn default() -> Self {
33        Self {
34            bytes: VecDeque::new(),
35            codec: PhantomData,
36        }
37    }
38}
39
40impl<C: FrameCodec> StreamDecoder<C> {
41    #[must_use]
42    pub fn new() -> Self {
43        Self::default()
44    }
45
46    /// Lends buffer space to `f` to **directly** feed bytes into the decoder. The bytes
47    /// are parsed into frames by [`StreamDecoder::next`].
48    ///
49    /// Returns what `f` returned, `Ok(0)` means EOF.
50    ///
51    /// # Errors
52    ///
53    /// Returns the error of `f` unchanged, and no bytes are kept.
54    pub fn feed_with(&mut self, f: impl FnOnce(&mut [u8]) -> IoResult<usize>) -> IoResult<usize> {
55        let slice_start = self.bytes.len();
56        self.bytes.resize(slice_start + DECODER_CHUNK_SIZE, 0);
57        let slice = self
58            .bytes
59            .make_contiguous()
60            .get_mut(slice_start..)
61            .unwrap_or_default();
62        let result = f(slice);
63        self.bytes
64            .truncate(slice_start + result.as_ref().map_or(0, |&count| count));
65        result
66    }
67}
68
69impl<C: FrameCodec> Iterator for StreamDecoder<C> {
70    type Item = Result<C::Frame, C::Error>;
71
72    fn next(&mut self) -> Option<Self::Item> {
73        let decoded = C::try_decode(self.bytes.make_contiguous())?;
74        Some(decoded.map(|(frame, length)| {
75            self.bytes.drain(..length);
76            frame
77        }))
78    }
79}
o_sfu_rfc/stream_decoder.rs

o_sfu_rfc/
stream_decoder.rs
