classicube_relay/
stream.rs

1use std::{collections::HashSet, io::Cursor, sync::Mutex};
2
3use lazy_static::lazy_static;
4
5use crate::{
6    Packet,
7    packet::{ContinuePacket, ContinuePacketError, Scope, StartPacket, StartPacketError},
8};
9
10pub const PLUGIN_MESSAGE_DATA_LENGTH: usize = 64;
11
12lazy_static! {
13    static ref OUTGOING_PACKET_ID: Mutex<HashSet<u8>> = Default::default();
14}
15
16#[derive(Debug, thiserror::Error)]
17pub enum StreamError {
18    #[error("{0}")]
19    LengthOverflow(String),
20
21    #[error("can't find free outgoing packet id")]
22    PacketIdLimit,
23
24    #[error(transparent)]
25    StartPacket(#[from] StartPacketError),
26
27    #[error(transparent)]
28    ContinuePacket(#[from] ContinuePacketError),
29}
30type Result<T> = std::result::Result<T, StreamError>;
31
32// [u8; 64]
33#[derive(Debug)]
34pub struct Stream {
35    stream_id: u8,
36    pub data: Vec<u8>,
37    pub scope: Scope,
38}
39impl Stream {
40    pub fn new<S: Into<Scope>>(data: Vec<u8>, scope: S) -> Result<Self> {
41        if data.len() > u16::MAX as usize {
42            return Err(StreamError::LengthOverflow(
43                "data.len() > u16::MAX".to_string(),
44            ));
45        }
46
47        let stream_id = Self::new_outgoing_packet_id()?;
48        Ok(Self {
49            stream_id,
50            data,
51            scope: scope.into(),
52        })
53    }
54
55    pub fn packets(&self) -> Result<Vec<Packet>> {
56        if self.data.len() > u16::MAX as usize {
57            return Err(StreamError::LengthOverflow(
58                "data.len() > u16::MAX".to_string(),
59            ));
60        }
61
62        let mut packets = vec![];
63
64        let mut cursor = Cursor::new(&self.data);
65        packets.push(Packet::Start(StartPacket::new_reader(
66            self.stream_id,
67            self.scope.clone(),
68            self.data.len() as u16,
69            &mut cursor,
70        )?));
71
72        while cursor.position() < self.data.len() as u64 {
73            packets.push(Packet::Continue(ContinuePacket::new_reader(
74                self.stream_id,
75                &mut cursor,
76            )?));
77        }
78
79        Ok(packets)
80    }
81
82    fn new_outgoing_packet_id() -> Result<u8> {
83        let mut guard = OUTGOING_PACKET_ID.lock().unwrap();
84
85        let maybe_id = (0..2u8.pow(7)).find(|id| !guard.contains(id));
86        if let Some(id) = maybe_id {
87            guard.insert(id);
88            Ok(id)
89        } else {
90            Err(StreamError::PacketIdLimit)
91        }
92    }
93
94    fn free_outgoing_packet_id(id: u8) {
95        let mut guard = OUTGOING_PACKET_ID.lock().unwrap();
96        guard.remove(&id);
97    }
98}
99impl Drop for Stream {
100    fn drop(&mut self) {
101        Self::free_outgoing_packet_id(self.stream_id);
102    }
103}
classicube_relay/stream.rs

classicube_relay/
stream.rs
