fcast_protocol/
lib.rs

1//! # FCast Protocol
2//!
3//! Implementation of the data models documented [here](https://gitlab.futo.org/videostreaming/fcast/-/wikis/Protocol-version-3).
4
5use std::collections::HashMap;
6
7use base64::{
8    alphabet::URL_SAFE,
9    engine::{general_purpose::GeneralPurpose, DecodePaddingMode, GeneralPurposeConfig},
10    Engine as _,
11};
12use serde::{Deserialize, Serialize};
13use serde_repr::{Deserialize_repr, Serialize_repr};
14
15pub mod companion;
16#[cfg(feature = "tokio-receiver")]
17pub mod receiver;
18#[cfg(feature = "tokio-sender")]
19pub mod sender;
20pub mod v1;
21pub mod v2;
22pub mod v3;
23pub mod v4;
24
25pub const HEADER_LENGTH: usize = 5;
26
27#[derive(Debug)]
28pub enum TryFromByteError {
29    UnknownOpcode(u8),
30}
31
32impl std::error::Error for TryFromByteError {}
33
34impl std::fmt::Display for TryFromByteError {
35    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
36        match self {
37            TryFromByteError::UnknownOpcode(opcode) => write!(f, "Unknown opcode: {opcode}"),
38        }
39    }
40}
41
42#[derive(Debug, PartialEq, Copy, Clone)]
43pub enum Opcode {
44    /// Not used
45    None = 0,
46    /// Sender message to play media content, body is [`v3::PlayMessage`]
47    Play = 1,
48    /// Sender message to pause media content, no body
49    Pause = 2,
50    /// Sender message to resume media content, no body
51    Resume = 3,
52    /// Sender message to stop media content, no body
53    Stop = 4,
54    /// Sender message to seek, body is [`SeekMessage`]
55    Seek = 5,
56    /// Receiver message to notify an updated playback state, body is [`v3::PlaybackUpdateMessage`]
57    PlaybackUpdate = 6,
58    /// Receiver message to notify when the volume has changed, body is [`VolumeUpdateMessage`]
59    VolumeUpdate = 7,
60    /// Sender message to change volume, body is [`SetVolumeMessage`]
61    SetVolume = 8,
62    /// Server message to notify the sender a playback error happened, body is [`PlaybackErrorMessage`]
63    PlaybackError = 9,
64    /// Sender message to change playback speed, body is [`SetSpeedMessage`]
65    SetSpeed = 10,
66    /// Message to notify the other of the current version, body is [`VersionMessage`]
67    Version = 11,
68    /// Message to get the other party to pong, no body
69    Ping = 12,
70    /// Message to respond to a ping from the other party, no body
71    Pong = 13,
72
73    // V3
74    /// Message to notify the other party of device information and state, body is InitialSenderMessage if receiver or
75    /// [`v3::InitialReceiverMessage`] if sender
76    Initial = 14,
77    /// Receiver message to notify all senders when any device has sent a [`v3::PlayMessage`], body is [`v3::PlayUpdateMessage`]
78    PlayUpdate = 15,
79    /// Sender message to set the item index in a playlist to play content from, body is [`v3::SetPlaylistItemMessage`]
80    SetPlaylistItem = 16,
81    /// Sender message to subscribe to a receiver event, body is [`v3::SubscribeEventMessage`]
82    SubscribeEvent = 17,
83    /// Sender message to unsubscribe to a receiver event, body is [`v3::UnsubscribeEventMessage`]
84    UnsubscribeEvent = 18,
85    /// Receiver message to notify when a sender subscribed event has occurred, body is [`v3::EventMessage`]
86    Event = 19,
87
88    // V4
89    Flatbuf = 20,
90    Resource = 21,
91}
92
93impl TryFrom<u8> for Opcode {
94    type Error = TryFromByteError;
95
96    fn try_from(value: u8) -> Result<Self, Self::Error> {
97        Ok(match value {
98            0 => Self::None,
99            1 => Self::Play,
100            2 => Self::Pause,
101            3 => Self::Resume,
102            4 => Self::Stop,
103            5 => Self::Seek,
104            6 => Self::PlaybackUpdate,
105            7 => Self::VolumeUpdate,
106            8 => Self::SetVolume,
107            9 => Self::PlaybackError,
108            10 => Self::SetSpeed,
109            11 => Self::Version,
110            12 => Self::Ping,
111            13 => Self::Pong,
112            14 => Self::Initial,
113            15 => Self::PlayUpdate,
114            16 => Self::SetPlaylistItem,
115            17 => Self::SubscribeEvent,
116            18 => Self::UnsubscribeEvent,
117            19 => Self::Event,
118            20 => Self::Flatbuf,
119            21 => Self::Resource,
120            _ => return Err(TryFromByteError::UnknownOpcode(value)),
121        })
122    }
123}
124
125#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize_repr, Deserialize_repr)]
126#[repr(u8)]
127pub enum PlaybackState {
128    Idle = 0,
129    Playing = 1,
130    Paused = 2,
131}
132
133#[allow(dead_code)]
134#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]
135pub struct PlaybackErrorMessage {
136    pub message: String,
137}
138
139#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]
140pub struct VersionMessage {
141    pub version: u64,
142}
143
144#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
145pub struct SetSpeedMessage {
146    pub speed: f64,
147}
148
149#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
150pub struct SetVolumeMessage {
151    pub volume: f64,
152}
153
154#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
155pub struct SeekMessage {
156    pub time: f64,
157}
158
159#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
160pub struct FCastService {
161    pub port: u16,
162    pub r#type: i32,
163}
164
165#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
166pub struct FCastNetworkConfig {
167    pub name: String,
168    pub addresses: Vec<String>,
169    pub services: Vec<FCastService>,
170    pub txt: Option<HashMap<String, String>>,
171}
172
173impl FCastNetworkConfig {
174    pub fn parse_url(url: &str) -> Option<Self> {
175        let connection_info = url.strip_prefix("fcast://r/")?;
176        let b64_engine = GeneralPurpose::new(
177            &URL_SAFE,
178            GeneralPurposeConfig::new().with_decode_padding_mode(DecodePaddingMode::Indifferent),
179        );
180        let json = b64_engine.decode(connection_info).ok()?;
181        serde_json::from_slice::<Self>(&json).ok()
182    }
183
184    pub fn to_url(&self) -> serde_json::Result<String> {
185        let net_config = serde_json::to_string(self)?;
186        let url = format!(
187            "fcast://r/{}",
188            base64::engine::general_purpose::URL_SAFE
189                .encode(net_config)
190                .as_str(),
191        );
192        Ok(url)
193    }
194}
195
196#[derive(Debug, PartialEq, Eq)]
197enum ReaderState {
198    MissingLength,
199    MissingBody { length: usize },
200    ShouldClear { body_length: usize },
201}
202
203#[derive(Debug, PartialEq, Eq)]
204pub enum ReadResult<'a> {
205    NeedData,
206    Read(&'a [u8]),
207    PacketTooLarge(usize),
208}
209
210#[derive(Debug)]
211pub enum PushDataError {
212    BufferTooBig,
213}
214
215pub struct PacketReader {
216    buffer: Vec<u8>,
217    state: ReaderState,
218    /// Start of unconsumed data in `buffer`.
219    pos: usize,
220    /// End of valid data in `buffer`.
221    len: usize,
222    max_packet_size: usize,
223}
224
225impl PacketReader {
226    pub fn new(max_packet_size: usize, padding: usize) -> Self {
227        Self {
228            buffer: vec![0; size_of::<u32>() + max_packet_size + padding],
229            state: ReaderState::MissingLength,
230            pos: 0,
231            len: 0,
232            max_packet_size,
233        }
234    }
235
236    /// Number of buffered bytes not yet consumed as packets.
237    fn buffered(&self) -> usize {
238        self.len - self.pos
239    }
240
241    /// Resolve a pending [`ReaderState::ShouldClear`]: advance `pos` past the packet that was
242    /// returned by the previous `get_packet` call. No bytes move.
243    fn discard_consumed(&mut self) {
244        if let ReaderState::ShouldClear { body_length } = self.state {
245            self.pos += size_of::<u32>() + body_length;
246            self.state = ReaderState::MissingLength;
247        }
248    }
249
250    /// Reclaim the space of already-consumed packets by moving the unconsumed tail to the front of
251    /// the buffer. Called once per refill (`push_data`/`spare_capacity_mut`), not per packet.
252    fn compact(&mut self) {
253        self.discard_consumed();
254        if self.pos == 0 {
255            return;
256        }
257        self.buffer.copy_within(self.pos..self.len, 0);
258        self.len -= self.pos;
259        self.pos = 0;
260    }
261
262    fn next_state(&mut self) -> ReadResult<'_> {
263        const LEN_SIZE: usize = std::mem::size_of::<u32>();
264
265        match self.state {
266            ReaderState::MissingLength => {
267                if self.buffered() >= LEN_SIZE {
268                    let length = u32::from_le_bytes(
269                        self.buffer[self.pos..self.pos + LEN_SIZE]
270                            .try_into()
271                            .expect("slice is LEN_SIZE bytes"),
272                    ) as usize;
273                    if length > self.max_packet_size {
274                        ReadResult::PacketTooLarge(length)
275                    } else {
276                        self.state = ReaderState::MissingBody { length };
277                        self.next_state()
278                    }
279                } else {
280                    ReadResult::NeedData
281                }
282            }
283            ReaderState::MissingBody { length } => {
284                if self.buffered().saturating_sub(LEN_SIZE) >= length {
285                    self.state = ReaderState::ShouldClear {
286                        body_length: length,
287                    };
288                    let start = self.pos + LEN_SIZE;
289                    ReadResult::Read(&self.buffer[start..start + length])
290                } else {
291                    ReadResult::NeedData
292                }
293            }
294            ReaderState::ShouldClear { .. } => {
295                self.discard_consumed();
296                self.next_state()
297            }
298        }
299    }
300
301    /// Push data to the reader's internal buffer.
302    ///
303    /// `get_packet()` should be called to extract packets.
304    pub fn push_data(&mut self, data: &[u8]) -> Result<(), PushDataError> {
305        if self.len + data.len() > self.buffer.len() {
306            self.compact();
307            if self.len + data.len() > self.buffer.len() {
308                return Err(PushDataError::BufferTooBig);
309            }
310        }
311        self.buffer[self.len..self.len + data.len()].copy_from_slice(data);
312        self.len += data.len();
313        Ok(())
314    }
315
316    /// Borrow the unused tail of the internal buffer to fill in place.
317    ///
318    /// This is the zero-copy counterpart to [`push_data`]: instead of reading into a scratch buffer
319    /// and copying that in, a transport can write straight into the reassembly buffer and mark the
320    /// bytes received with [`commit`]. That removes one copy of every received byte on the hot
321    /// receive path.
322    ///
323    /// Space freed by already-consumed packets is reclaimed here (at most one compaction per
324    /// refill), so the returned slice is empty only when unconsumed data fills the whole buffer. A
325    /// caller that drains to [`NeedData`] before each read never sees that: a mid-packet reader
326    /// holds fewer than `size_of::<u32>() + max_packet_size` bytes, less than the buffer's capacity
327    /// even with `padding == 0`. The slice is therefore never empty, so a read into it cannot
328    /// return `Ok(0)` and be mistaken for end-of-stream. `padding` only trades memory for fewer,
329    /// larger reads.
330    ///
331    /// # Example
332    ///
333    /// ```
334    /// use std::io::Read;
335    ///
336    /// use fcast_protocol::{PacketReader, ReadResult};
337    ///
338    /// // A transport carrying one framed packet: length prefix 3, body [1, 2, 3].
339    /// let mut stream: &[u8] = &[3, 0, 0, 0, 1, 2, 3];
340    ///
341    /// let mut reader = PacketReader::new(1024, 0);
342    /// let n = stream.read(reader.spare_capacity_mut())?;
343    /// reader.commit(n);
344    /// assert_eq!(reader.get_packet(), ReadResult::Read(&[1, 2, 3]));
345    /// assert_eq!(reader.get_packet(), ReadResult::NeedData);
346    /// # Ok::<(), std::io::Error>(())
347    /// ```
348    ///
349    /// [`push_data`]: Self::push_data
350    /// [`commit`]: Self::commit
351    /// [`NeedData`]: ReadResult::NeedData
352    pub fn spare_capacity_mut(&mut self) -> &mut [u8] {
353        self.compact();
354        &mut self.buffer[self.len..]
355    }
356
357    /// Mark `n` bytes written into the slice returned by [`spare_capacity_mut`] as
358    /// received.
359    ///
360    /// `n` must not exceed the length of that slice (a transport must never report having read more
361    /// bytes than the slice could hold). In debug builds this is asserted; in release builds an
362    /// out-of-range `n` corrupts the reader's length bookkeeping, so it is a caller bug rather than
363    /// defined behaviour.
364    ///
365    /// [`spare_capacity_mut`]: Self::spare_capacity_mut
366    pub fn commit(&mut self, n: usize) {
367        debug_assert!(
368            self.len + n <= self.buffer.len(),
369            "commit({n}) overflows reader buffer (len={}, capacity={})",
370            self.len,
371            self.buffer.len()
372        );
373        self.len += n;
374    }
375
376    /// Get a packet if it's available.
377    ///
378    /// This should be called in a loop until `None` is returned which means more data is needed.
379    pub fn get_packet(&mut self) -> ReadResult<'_> {
380        self.next_state()
381    }
382
383    /// Take all buffered bytes that are not part of an already-returned packet and reset the
384    /// reader.
385    ///
386    /// This is used when the underlying connection is handed to another protocol layer (e.g. a TLS
387    /// upgrade after the plaintext `Version` exchange): a single read may have pulled in bytes
388    /// belonging to that next layer, and those must be replayed there instead of being lost.
389    pub fn drain_unparsed(&mut self) -> Vec<u8> {
390        self.discard_consumed();
391        let data = self.buffer[self.pos..self.len].to_vec();
392        self.pos = 0;
393        self.len = 0;
394        self.state = ReaderState::MissingLength;
395        data
396    }
397}
398
399#[cfg(test)]
400mod tests {
401    use super::*;
402
403    #[test]
404    fn test_network_config_url() {
405        let samples = [
406            FCastNetworkConfig {
407                name: "Living Room".to_string(),
408                addresses: vec!["192.168.1.42".to_string()],
409                services: vec![FCastService {
410                    port: 46899,
411                    r#type: 0,
412                }],
413                txt: None,
414            },
415            FCastNetworkConfig {
416                name: "kitchen-tv".to_string(),
417                addresses: vec![
418                    "10.0.0.5".to_string(),
419                    "fe80::1ff:fe23:4567:890a".to_string(),
420                ],
421                services: vec![FCastService {
422                    port: 46899,
423                    r#type: 0,
424                }],
425                txt: Some(HashMap::from([
426                    ("version".to_string(), "3".to_string()),
427                    ("id".to_string(), "abc-123".to_string()),
428                ])),
429            },
430            FCastNetworkConfig {
431                name: "æøå".to_string(),
432                addresses: vec![],
433                services: vec![],
434                txt: Some(HashMap::new()),
435            },
436        ];
437
438        for config in samples {
439            let url = config.to_url().expect("serializing to url should succeed");
440            assert!(url.starts_with("fcast://r/"), "unexpected url: {url}");
441            let parsed = FCastNetworkConfig::parse_url(&url)
442                .unwrap_or_else(|| panic!("parsing url should succeed: {url}"));
443            assert_eq!(parsed, config);
444        }
445    }
446
447    #[test]
448    fn test_parse_url_rejects_invalid() {
449        assert!(FCastNetworkConfig::parse_url("https://example.com").is_none());
450        assert!(FCastNetworkConfig::parse_url("fcast://r/not-valid-base64-$$$").is_none());
451    }
452
453    #[test]
454    fn packet_reader_single() {
455        let mut reader = PacketReader::new(100, 0);
456        reader
457            .push_data(&[1u32.to_le_bytes().as_slice(), [0u8].as_slice()].concat())
458            .unwrap();
459        assert_eq!(reader.get_packet(), ReadResult::Read(&[0]));
460        assert_eq!(reader.state, ReaderState::ShouldClear { body_length: 1 });
461        assert_eq!(reader.get_packet(), ReadResult::NeedData);
462        assert_eq!(reader.buffered(), 0);
463        assert_eq!(reader.state, ReaderState::MissingLength);
464        assert_eq!(reader.buffered(), 0);
465    }
466
467    #[test]
468    fn packet_reader_small_push() {
469        let mut reader = PacketReader::new(100, 0);
470        let length = 1u32.to_le_bytes();
471        reader.push_data(&[length[0], length[1]]).unwrap();
472        assert_eq!(reader.get_packet(), ReadResult::NeedData);
473        reader.push_data(&[length[2]]).unwrap();
474        assert_eq!(reader.get_packet(), ReadResult::NeedData);
475        assert_eq!(reader.state, ReaderState::MissingLength);
476        reader.push_data(&[length[3]]).unwrap();
477        assert_eq!(reader.get_packet(), ReadResult::NeedData);
478        reader.push_data(&[0]).unwrap();
479        assert_eq!(reader.get_packet(), ReadResult::Read(&[0]));
480        assert_eq!(reader.state, ReaderState::ShouldClear { body_length: 1 });
481        assert_eq!(reader.get_packet(), ReadResult::NeedData);
482        assert_eq!(reader.buffered(), 0);
483    }
484
485    #[rustfmt::skip]
486    #[test]
487    fn packet_reader_many_packets_single_push() {
488        let mut reader = PacketReader::new(100, 0);
489        reader.push_data(&[
490            1u32.to_le_bytes().as_slice(), [0u8].as_slice(),
491            2u32.to_le_bytes().as_slice(), [0u8, 1].as_slice(),
492            3u32.to_le_bytes().as_slice(), [0u8, 1, 2].as_slice(),
493        ].concat()).unwrap();
494        assert_eq!(reader.get_packet(), ReadResult::Read(&[0]));
495        assert_eq!(reader.state, ReaderState::ShouldClear { body_length: 1 });
496        assert_eq!(reader.get_packet(), ReadResult::Read(&[0, 1]));
497        assert_eq!(reader.state, ReaderState::ShouldClear { body_length: 2 });
498        assert_eq!(reader.get_packet(), ReadResult::Read(&[0, 1, 2]));
499        assert_eq!(reader.state, ReaderState::ShouldClear { body_length: 3 });
500        assert_eq!(reader.get_packet(), ReadResult::NeedData);
501        assert_eq!(reader.state, ReaderState::MissingLength);
502        assert_eq!(reader.buffered(), 0);
503    }
504
505    #[test]
506    fn packet_reader_partial_body() {
507        let mut reader = PacketReader::new(100, 0);
508        reader
509            .push_data(&[4u32.to_le_bytes().as_slice(), [0u8, 1].as_slice()].concat())
510            .unwrap();
511        assert_eq!(reader.get_packet(), ReadResult::NeedData);
512        reader.push_data(&[2]).unwrap();
513        assert_eq!(reader.get_packet(), ReadResult::NeedData);
514        reader.push_data(&[3]).unwrap();
515        assert_eq!(reader.get_packet(), ReadResult::Read(&[0, 1, 2, 3]));
516        assert_eq!(reader.state, ReaderState::ShouldClear { body_length: 4 });
517        assert_eq!(reader.get_packet(), ReadResult::NeedData);
518        assert_eq!(reader.state, ReaderState::MissingLength);
519        assert_eq!(reader.buffered(), 0);
520    }
521
522    #[test]
523    fn packet_reader_large_body() {
524        let mut reader = PacketReader::new(100, 0);
525        let body = (0..10).collect::<Vec<u8>>();
526        reader
527            .push_data(&[10u32.to_le_bytes().as_slice(), body.as_slice()].concat())
528            .unwrap();
529        assert_eq!(reader.get_packet(), ReadResult::Read(&body));
530        assert_eq!(reader.state, ReaderState::ShouldClear { body_length: 10 });
531        assert_eq!(reader.get_packet(), ReadResult::NeedData);
532        assert_eq!(reader.state, ReaderState::MissingLength);
533        assert_eq!(reader.buffered(), 0);
534    }
535
536    #[test]
537    fn large_body_size() {
538        let mut reader = PacketReader::new(65280, 0);
539        let body = &[255, 255, 255, 0];
540        reader.push_data(body).unwrap();
541        assert_eq!(
542            reader.get_packet(),
543            ReadResult::PacketTooLarge(u32::from_le_bytes(*body) as usize)
544        );
545    }
546
547    #[test]
548    fn drain_unparsed_returns_bytes_after_packet() {
549        let mut reader = PacketReader::new(100, 16);
550        let trailing = [0x16u8, 0x03, 0x01, 0x02, 0x00, 0x42];
551        reader
552            .push_data(
553                &[
554                    1u32.to_le_bytes().as_slice(),
555                    [7u8].as_slice(),
556                    trailing.as_slice(),
557                ]
558                .concat(),
559            )
560            .unwrap();
561
562        assert_eq!(reader.get_packet(), ReadResult::Read(&[7]));
563        assert_eq!(reader.state, ReaderState::ShouldClear { body_length: 1 });
564        assert_eq!(reader.drain_unparsed(), trailing);
565        assert_eq!(reader.buffered(), 0);
566        assert_eq!(reader.state, ReaderState::MissingLength);
567    }
568
569    #[test]
570    fn drain_unparsed_without_reading_returns_everything() {
571        let mut reader = PacketReader::new(100, 0);
572        let data = [0x16u8, 0x03, 0x01, 0x00];
573        reader.push_data(&data).unwrap();
574        assert_eq!(reader.drain_unparsed(), data);
575        assert_eq!(reader.buffered(), 0);
576        assert_eq!(reader.state, ReaderState::MissingLength);
577    }
578
579    #[test]
580    fn drain_unparsed_with_partial_packet_returns_everything() {
581        let mut reader = PacketReader::new(100, 0);
582        let data = [4u32.to_le_bytes().as_slice(), [0u8, 1].as_slice()].concat();
583        reader.push_data(&data).unwrap();
584        assert_eq!(reader.get_packet(), ReadResult::NeedData);
585        assert_eq!(reader.state, ReaderState::MissingBody { length: 4 });
586
587        assert_eq!(reader.drain_unparsed(), data);
588        assert_eq!(reader.buffered(), 0);
589        assert_eq!(reader.state, ReaderState::MissingLength);
590    }
591
592    #[test]
593    fn drain_unparsed_when_empty_is_empty() {
594        let mut reader = PacketReader::new(100, 0);
595        assert!(reader.drain_unparsed().is_empty());
596        assert_eq!(reader.state, ReaderState::MissingLength);
597    }
598
599    #[test]
600    fn reader_is_reusable_after_drain() {
601        let mut reader = PacketReader::new(100, 16);
602        reader
603            .push_data(
604                &[
605                    1u32.to_le_bytes().as_slice(),
606                    [7u8].as_slice(),
607                    &[0xaa, 0xbb],
608                ]
609                .concat(),
610            )
611            .unwrap();
612        assert_eq!(reader.get_packet(), ReadResult::Read(&[7]));
613        assert_eq!(reader.drain_unparsed(), [0xaa, 0xbb]);
614
615        reader
616            .push_data(&[2u32.to_le_bytes().as_slice(), [8u8, 9].as_slice()].concat())
617            .unwrap();
618        assert_eq!(reader.get_packet(), ReadResult::Read(&[8, 9]));
619        assert_eq!(reader.get_packet(), ReadResult::NeedData);
620        assert_eq!(reader.buffered(), 0);
621    }
622
623    // ---- zero-copy read path: `spare_capacity_mut` + `commit` ----
624
625    const LEN_SIZE: usize = std::mem::size_of::<u32>();
626
627    fn frame(body: &[u8]) -> Vec<u8> {
628        let mut v = (body.len() as u32).to_le_bytes().to_vec();
629        v.extend_from_slice(body);
630        v
631    }
632
633    fn drain_zerocopy(reader: &mut PacketReader, data: &[u8], chunk: usize) -> Vec<Vec<u8>> {
634        let mut out = Vec::new();
635        let mut pos = 0;
636        while pos < data.len() {
637            let spare = reader.spare_capacity_mut();
638            assert!(
639                !spare.is_empty(),
640                "spare capacity empty before a read (would be read as EOF)"
641            );
642            let want = if chunk == 0 {
643                spare.len()
644            } else {
645                chunk.min(spare.len())
646            };
647            let take = want.min(data.len() - pos);
648            spare[..take].copy_from_slice(&data[pos..pos + take]);
649            reader.commit(take);
650            pos += take;
651
652            loop {
653                match reader.get_packet() {
654                    ReadResult::Read(p) => out.push(p.to_vec()),
655                    ReadResult::NeedData => break,
656                    ReadResult::PacketTooLarge(s) => panic!("unexpected PacketTooLarge({s})"),
657                }
658            }
659        }
660        out
661    }
662
663    fn drain_pushdata(reader: &mut PacketReader, data: &[u8], scratch: usize) -> Vec<Vec<u8>> {
664        let mut out = Vec::new();
665        let mut buf = vec![0u8; scratch];
666        let mut pos = 0;
667        while pos < data.len() {
668            let n = scratch.min(data.len() - pos);
669            buf[..n].copy_from_slice(&data[pos..pos + n]);
670            reader.push_data(&buf[..n]).expect("push_data overflowed");
671            pos += n;
672            loop {
673                match reader.get_packet() {
674                    ReadResult::Read(p) => out.push(p.to_vec()),
675                    ReadResult::NeedData => break,
676                    ReadResult::PacketTooLarge(s) => panic!("unexpected PacketTooLarge({s})"),
677                }
678            }
679        }
680        out
681    }
682
683    #[test]
684    fn spare_capacity_starts_at_full_buffer() {
685        let mut reader = PacketReader::new(100, 16);
686        assert_eq!(reader.spare_capacity_mut().len(), LEN_SIZE + 100 + 16);
687    }
688
689    #[test]
690    fn commit_zero_is_noop() {
691        let mut reader = PacketReader::new(100, 16);
692        let before = reader.spare_capacity_mut().len();
693        reader.commit(0);
694        assert_eq!(reader.spare_capacity_mut().len(), before);
695        assert_eq!(reader.get_packet(), ReadResult::NeedData);
696        assert_eq!(reader.buffered(), 0);
697    }
698
699    #[test]
700    fn spare_capacity_shrinks_by_commit_and_regrows_after_consume() {
701        let mut reader = PacketReader::new(100, 16);
702        let cap = LEN_SIZE + 100 + 16;
703
704        // Write a whole framed packet plus the length prefix of a second, in place.
705        let first = frame(&[0xAA, 0xBB, 0xCC]);
706        let second_prefix = 2u32.to_le_bytes();
707        let n = {
708            let spare = reader.spare_capacity_mut();
709            spare[..first.len()].copy_from_slice(&first);
710            spare[first.len()..first.len() + LEN_SIZE].copy_from_slice(&second_prefix);
711            first.len() + LEN_SIZE
712        };
713        reader.commit(n);
714        assert_eq!(reader.spare_capacity_mut().len(), cap - n);
715
716        // Consuming the first packet frees its bytes; the next `spare_capacity_mut` compacts
717        // (leftover prefix moves to the front) and the spare grows back to
718        // all-but-the-leftover.
719        assert_eq!(reader.get_packet(), ReadResult::Read(&[0xAA, 0xBB, 0xCC]));
720        assert_eq!(reader.get_packet(), ReadResult::NeedData);
721        assert_eq!(reader.buffered(), LEN_SIZE);
722        assert_eq!(reader.spare_capacity_mut().len(), cap - LEN_SIZE);
723    }
724
725    #[test]
726    fn consuming_packets_does_not_move_buffered_data() {
727        // Consumption must be a cursor advance, not a per-packet memmove of the tail -
728        // otherwise a read that batches K packets costs O(K^2) byte moves.
729        let mut reader = PacketReader::new(100, 16);
730        let stream: Vec<u8> = [frame(&[0]), frame(&[0, 1]), frame(&[0, 1, 2])].concat();
731        reader.push_data(&stream).unwrap();
732
733        assert_eq!(reader.get_packet(), ReadResult::Read(&[0]));
734        assert_eq!(reader.get_packet(), ReadResult::Read(&[0, 1]));
735        assert_eq!(reader.get_packet(), ReadResult::Read(&[0, 1, 2]));
736        assert_eq!(reader.get_packet(), ReadResult::NeedData);
737
738        // All three packets were consumed purely by advancing the cursor: the write end
739        // never moved back, so no bytes were copied while draining.
740        assert_eq!(reader.pos, stream.len());
741        assert_eq!(reader.len, stream.len());
742
743        // The next refill reclaims the whole buffer in one step - and since nothing is
744        // buffered mid-packet, it is a free cursor reset rather than a copy.
745        assert_eq!(
746            reader.spare_capacity_mut().len(),
747            LEN_SIZE + 100 + 16,
748            "refill should reclaim all consumed space"
749        );
750        assert_eq!((reader.pos, reader.len), (0, 0));
751    }
752
753    #[test]
754    fn zerocopy_single_packet() {
755        let mut reader = PacketReader::new(100, 16);
756        let body = [7u8, 8, 9];
757        let out = drain_zerocopy(&mut reader, &frame(&body), 0);
758        assert_eq!(out, vec![body.to_vec()]);
759        assert_eq!(reader.buffered(), 0);
760    }
761
762    #[test]
763    fn zerocopy_reassembles_across_all_chunk_sizes() {
764        let bodies: Vec<Vec<u8>> = vec![
765            vec![0], // opcode-only packet
766            vec![1, 2],
767            (0..37u8).collect(),
768            vec![0xFF; 90], // near max
769            vec![42],
770            (0..64u8).rev().collect(),
771        ];
772        let mut stream = Vec::new();
773        for b in &bodies {
774            stream.extend_from_slice(&frame(b));
775        }
776
777        for chunk in [1usize, 2, 3, 4, 5, 6, 7, 8, 13, 31, 64, 100, 8192, 0] {
778            let mut reader = PacketReader::new(100, 8192);
779            let out = drain_zerocopy(&mut reader, &stream, chunk);
780            assert_eq!(out, bodies, "mismatch at chunk size {chunk}");
781            assert_eq!(
782                reader.buffered(),
783                0,
784                "buffer not drained at chunk size {chunk}"
785            );
786        }
787    }
788
789    #[test]
790    fn zerocopy_large_packet_split_byte_by_byte() {
791        let mut reader = PacketReader::new(100_000, 8192);
792        let body: Vec<u8> = (0..5000).map(|i| (i % 251) as u8).collect();
793        let out = drain_zerocopy(&mut reader, &frame(&body), 1);
794        assert_eq!(out, vec![body]);
795        assert_eq!(reader.buffered(), 0);
796    }
797
798    #[test]
799    fn zerocopy_matches_push_data_path() {
800        // Same input, same chunking, two APIs - identical extracted packets.
801        let bodies: Vec<Vec<u8>> = vec![vec![1], (0..50u8).collect(), vec![9; 80], vec![2, 3]];
802        let mut stream = Vec::new();
803        for b in &bodies {
804            stream.extend_from_slice(&frame(b));
805        }
806
807        for chunk in [1usize, 3, 7, 64, 128] {
808            let mut zc = PacketReader::new(100, 8192);
809            let mut pd = PacketReader::new(100, 8192);
810            let zc_out = drain_zerocopy(&mut zc, &stream, chunk);
811            let pd_out = drain_pushdata(&mut pd, &stream, chunk);
812            assert_eq!(
813                zc_out, pd_out,
814                "zero-copy vs push_data diverged at chunk {chunk}"
815            );
816            assert_eq!(zc_out, bodies);
817        }
818    }
819
820    #[test]
821    fn zerocopy_full_buffer_still_yields_a_packet() {
822        // buffer = 4 + 64 + 16 = 84. Fill it exactly: one max-size (64B) packet plus 16
823        // bytes of the next. A full buffer must still surface the complete packet - the
824        // invariant that guarantees the spare never stays empty.
825        let max = 64usize;
826        let padding = 16usize;
827        let mut reader = PacketReader::new(max, padding);
828        let big = frame(&vec![0x5Au8; max]); // 4 + 64 = 68 bytes
829                                             // 16 trailing bytes forming the *start* of a second packet: a length prefix of 16
830                                             // but only 12 of those 16 body bytes present, so it stays incomplete (NeedData)
831                                             // rather than parsing as another packet.
832        let mut trailing = 16u32.to_le_bytes().to_vec();
833        trailing.extend_from_slice(&[0xEE; 12]);
834        assert_eq!(trailing.len(), 16);
835        let n = {
836            let spare = reader.spare_capacity_mut();
837            assert_eq!(spare.len(), LEN_SIZE + max + padding);
838            spare[..big.len()].copy_from_slice(&big);
839            spare[big.len()..big.len() + trailing.len()].copy_from_slice(&trailing);
840            big.len() + trailing.len()
841        };
842        reader.commit(n);
843        assert_eq!(
844            reader.spare_capacity_mut().len(),
845            0,
846            "buffer should be exactly full"
847        );
848
849        assert_eq!(reader.get_packet(), ReadResult::Read(&[0x5A; 64]));
850        assert_eq!(reader.get_packet(), ReadResult::NeedData);
851        // Refilling compacts the 16 leftover bytes to the front; spare recovered (no
852        // deadlock).
853        assert_eq!(reader.buffered(), 16);
854        assert_eq!(
855            reader.spare_capacity_mut().len(),
856            LEN_SIZE + max + padding - 16
857        );
858    }
859
860    #[test]
861    fn zerocopy_never_false_eof_under_back_to_back_max_packets() {
862        // Greedy reads (chunk = 0) over a stream of max-size packets: the assertion inside
863        // `drain_zerocopy` fails if the spare is ever empty before a read.
864        let max = 200usize;
865        let mut reader = PacketReader::new(max, 64);
866        let bodies: Vec<Vec<u8>> = (0..15)
867            .map(|i| vec![i as u8; max]) // each body exactly max_packet_size
868            .collect();
869        let mut stream = Vec::new();
870        for b in &bodies {
871            stream.extend_from_slice(&frame(b));
872        }
873        let out = drain_zerocopy(&mut reader, &stream, 0);
874        assert_eq!(out, bodies);
875        assert_eq!(reader.buffered(), 0);
876    }
877
878    #[test]
879    fn zerocopy_drain_unparsed_recovers_tls_prefix() {
880        // Mirrors the receiver's TLS upgrade: a single read pulls in the plaintext `Version`
881        // packet plus the first bytes of the following TLS ClientHello, committed in place.
882        let mut reader = PacketReader::new(100, 16);
883        let version = frame(&[Opcode::Version as u8, b'{', b'}']);
884        let handshake = [0x16u8, 0x03, 0x01, 0x02, 0x00, 0x42];
885        let n = {
886            let spare = reader.spare_capacity_mut();
887            spare[..version.len()].copy_from_slice(&version);
888            spare[version.len()..version.len() + handshake.len()].copy_from_slice(&handshake);
889            version.len() + handshake.len()
890        };
891        reader.commit(n);
892
893        assert_eq!(
894            reader.get_packet(),
895            ReadResult::Read(&[Opcode::Version as u8, b'{', b'}'])
896        );
897        assert_eq!(reader.drain_unparsed(), handshake);
898        assert_eq!(reader.buffered(), 0);
899        assert_eq!(reader.state, ReaderState::MissingLength);
900    }
901
902    #[test]
903    fn zerocopy_too_large_prefix_is_reported() {
904        let mut reader = PacketReader::new(64, 16);
905        // Length prefix of 65 (> max 64). Write just the prefix and commit.
906        let prefix = 65u32.to_le_bytes();
907        reader.spare_capacity_mut()[..LEN_SIZE].copy_from_slice(&prefix);
908        reader.commit(LEN_SIZE);
909        assert_eq!(reader.get_packet(), ReadResult::PacketTooLarge(65));
910    }
911
912    #[test]
913    fn zerocopy_randomized_reassembly() {
914        // Deterministic xorshift PRNG, random packet counts, sizes and read chunking. The
915        // extracted packets must always equal the framed input, regardless of segmentation.
916        let mut rng: u64 = 0x9E37_79B9_7F4A_7C15;
917        let mut next = || {
918            rng ^= rng << 13;
919            rng ^= rng >> 7;
920            rng ^= rng << 17;
921            rng
922        };
923
924        let max = 300usize;
925        for _ in 0..400 {
926            let n_packets = (next() % 12) as usize + 1;
927            let bodies: Vec<Vec<u8>> = (0..n_packets)
928                .map(|_| {
929                    let len = (next() as usize % max) + 1; // 1..=max
930                    (0..len).map(|_| next() as u8).collect()
931                })
932                .collect();
933            let mut stream = Vec::new();
934            for b in &bodies {
935                stream.extend_from_slice(&frame(b));
936            }
937            let chunk = (next() as usize % 40) + 1; // 1..=40 (also exercises tiny reads)
938            let mut reader = PacketReader::new(max, 8192);
939            let out = drain_zerocopy(&mut reader, &stream, chunk);
940            assert_eq!(
941                out, bodies,
942                "randomized reassembly mismatch (chunk={chunk})"
943            );
944            assert_eq!(reader.buffered(), 0);
945        }
946    }
947}
fcast_protocol/lib.rs

fcast_protocol/
lib.rs
