engineioxide_core/
packet.rs

1use std::{fmt, time::Duration};
2
3use base64::{Engine, engine::general_purpose};
4use bytes::Bytes;
5use serde::{Deserialize, Deserializer, Serialize, Serializer};
6use smallvec::SmallVec;
7
8use crate::{ProtocolVersion, Sid, Str, TransportType};
9
10/// A Packet type to use when receiving and sending data from the client
11#[derive(Debug, Clone, PartialEq)]
12pub enum Packet {
13    /// Open packet used to initiate a connection
14    Open(OpenPacket),
15    /// Close packet used to close a connection
16    Close,
17    /// Ping packet used to check if the connection is still alive
18    /// The client never sends this packet, it is only used by the server
19    Ping,
20    /// Pong packet used to respond to a Ping packet
21    /// The server never sends this packet, it is only used by the client
22    Pong,
23
24    /// Special Ping packet used to initiate a connection
25    PingUpgrade,
26    /// Special Pong packet used to respond to a PingUpgrade packet and upgrade the connection
27    PongUpgrade,
28
29    /// Message packet used to send a message to the client
30    Message(Str),
31    /// Upgrade packet to upgrade the connection from polling to websocket
32    Upgrade,
33
34    /// Noop packet used to send something to a opened polling connection so it gracefully closes to allow the client to upgrade to websocket
35    Noop,
36
37    /// Binary packet used to send binary data to the client
38    /// Converts to a String using base64 encoding when using polling connection
39    /// Or to a websocket binary frame when using websocket connection
40    ///
41    /// When receiving, it is only used with polling connection, websocket use binary frame
42    Binary(Bytes), // Not part of the protocol, used internally
43
44    /// Binary packet used to send binary data to the client
45    /// Converts to a String using base64 encoding when using polling connection
46    /// Or to a websocket binary frame when using websocket connection
47    ///
48    /// When receiving, it is only used with polling connection, websocket use binary frame
49    ///
50    /// This is a special packet, excepionally specific to the V3 protocol.
51    BinaryV3(Bytes), // Not part of the protocol, used internally
52}
53
54/// An error that occurs when parsing a packet.
55#[derive(Debug)]
56pub enum PacketParseError {
57    /// Invalid connect packet
58    InvalidConnectPacket(serde_json::Error),
59    /// The packet type is invalid.
60    InvalidPacketType(Option<char>),
61    /// The packet payload is invalid.
62    InvalidPacketPayload,
63    /// The packet length is invalid.
64    InvalidPacketLen,
65    /// The packet chunk is invalid
66    InvalidUtf8Boundary(std::str::Utf8Error),
67    /// The base64 decoding failed.
68    Base64Decode(base64::DecodeError),
69    /// The payload is too large.
70    PayloadTooLarge {
71        /// The maximum allowed payload size.
72        max: u64,
73    },
74}
75impl fmt::Display for PacketParseError {
76    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
77        match self {
78            PacketParseError::InvalidConnectPacket(e) => write!(f, "invalid connect packet: {e}"),
79            PacketParseError::InvalidPacketType(c) => write!(f, "invalid packet type: {c:?}"),
80            PacketParseError::InvalidPacketPayload => write!(f, "invalid packet payload"),
81            PacketParseError::InvalidPacketLen => write!(f, "invalid packet length"),
82            PacketParseError::InvalidUtf8Boundary(err) => write!(
83                f,
84                "invalid utf8 boundary when parsing payload into packet chunks: {err}"
85            ),
86            PacketParseError::Base64Decode(err) => write!(f, "base64 decode error: {err}"),
87            PacketParseError::PayloadTooLarge { max } => {
88                write!(f, "payload too large: max {max}")
89            }
90        }
91    }
92}
93impl From<base64::DecodeError> for PacketParseError {
94    fn from(err: base64::DecodeError) -> Self {
95        PacketParseError::Base64Decode(err)
96    }
97}
98impl From<std::string::FromUtf8Error> for PacketParseError {
99    fn from(err: std::string::FromUtf8Error) -> Self {
100        PacketParseError::InvalidUtf8Boundary(err.utf8_error())
101    }
102}
103impl From<std::str::Utf8Error> for PacketParseError {
104    fn from(err: std::str::Utf8Error) -> Self {
105        PacketParseError::InvalidUtf8Boundary(err)
106    }
107}
108impl From<serde_json::Error> for PacketParseError {
109    fn from(err: serde_json::Error) -> Self {
110        PacketParseError::InvalidConnectPacket(err)
111    }
112}
113impl std::error::Error for PacketParseError {}
114
115impl Packet {
116    /// Check if the packet is a binary packet
117    pub fn is_binary(&self) -> bool {
118        matches!(self, Packet::Binary(_) | Packet::BinaryV3(_))
119    }
120
121    /// If the packet is a message packet (text), it returns the message
122    pub fn into_message(self) -> Str {
123        match self {
124            Packet::Message(msg) => msg,
125            _ => panic!("Packet is not a message"),
126        }
127    }
128
129    /// If the packet is a binary packet, it returns the binary data
130    pub fn into_binary(self) -> Bytes {
131        match self {
132            Packet::Binary(data) => data,
133            Packet::BinaryV3(data) => data,
134            _ => panic!("Packet is not a binary"),
135        }
136    }
137
138    /// Get the max size the packet could have when serialized
139    ///
140    ///  If b64 is true, it returns the max size when serialized to base64
141    ///
142    /// The base64 max size factor is `ceil(n / 3) * 4`
143    pub fn get_size_hint(&self, b64: bool) -> usize {
144        match self {
145            Packet::Open(_) => 156, // max possible size for the open packet serialized
146            Packet::Close => 1,
147            Packet::Ping => 1,
148            Packet::Pong => 1,
149            Packet::PingUpgrade => 6,
150            Packet::PongUpgrade => 6,
151            Packet::Message(msg) => 1 + msg.len(),
152            Packet::Upgrade => 1,
153            Packet::Noop => 1,
154            Packet::Binary(data) => {
155                if b64 {
156                    1 + base64::encoded_len(data.len(), true).unwrap_or(usize::MAX - 1)
157                } else {
158                    1 + data.len()
159                }
160            }
161            Packet::BinaryV3(data) => {
162                if b64 {
163                    2 + base64::encoded_len(data.len(), true).unwrap_or(usize::MAX - 2)
164                } else {
165                    1 + data.len()
166                }
167            }
168        }
169    }
170}
171
172impl From<Packet> for Bytes {
173    fn from(value: Packet) -> Self {
174        String::from(value).into()
175    }
176}
177
178/// Serialize a [Packet] to a [String] according to the Engine.IO protocol
179impl From<Packet> for String {
180    fn from(packet: Packet) -> String {
181        let len = packet.get_size_hint(true);
182        let mut buffer = String::with_capacity(len);
183        match packet {
184            Packet::Open(open) => {
185                buffer.push('0');
186                buffer.push_str(&serde_json::to_string(&open).unwrap());
187            }
188            Packet::Close => buffer.push('1'),
189            Packet::Ping => buffer.push('2'),
190            Packet::Pong => buffer.push('3'),
191            Packet::PingUpgrade => buffer.push_str("2probe"),
192            Packet::PongUpgrade => buffer.push_str("3probe"),
193            Packet::Message(msg) => {
194                buffer.push('4');
195                buffer.push_str(&msg);
196            }
197            Packet::Upgrade => buffer.push('5'),
198            Packet::Noop => buffer.push('6'),
199            Packet::Binary(data) => {
200                buffer.push('b');
201                general_purpose::STANDARD.encode_string(data, &mut buffer);
202            }
203            Packet::BinaryV3(data) => {
204                buffer.push_str("b4");
205                general_purpose::STANDARD.encode_string(data, &mut buffer);
206            }
207        };
208        buffer
209    }
210}
211
212/// Deserialize a [Packet] from a [String] according to the Engine.IO protocol
213impl Packet {
214    /// Parses a packet from a string value using the specified protocol version.
215    pub fn parse(
216        protocol: ProtocolVersion,
217        value: impl Into<Str>,
218    ) -> Result<Self, PacketParseError> {
219        let value = value.into();
220        let packet_type = value
221            .as_bytes()
222            .first()
223            .ok_or(PacketParseError::InvalidPacketType(None))?;
224        let is_upgrade = value.len() == 6 && &value[1..6] == "probe";
225        let res = match packet_type {
226            b'1' => Packet::Close,
227            b'2' if is_upgrade => Packet::PingUpgrade,
228            b'2' => Packet::Ping,
229            b'3' if is_upgrade => Packet::PongUpgrade,
230            b'3' => Packet::Pong,
231            b'4' => Packet::Message(value.slice(1..)),
232            b'5' => Packet::Upgrade,
233            b'6' => Packet::Noop,
234            b'b' if protocol == ProtocolVersion::V3 => Packet::BinaryV3(
235                general_purpose::STANDARD
236                    .decode(value.slice(2..).as_bytes())?
237                    .into(),
238            ),
239            b'b' => Packet::Binary(
240                general_purpose::STANDARD
241                    .decode(value.slice(1..).as_bytes())?
242                    .into(),
243            ),
244            c => Err(PacketParseError::InvalidPacketType(Some(*c as char)))?,
245        };
246        Ok(res)
247    }
248}
249
250/// An OpenPacket is used to initiate a connection
251#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)]
252#[serde(rename_all = "camelCase")]
253pub struct OpenPacket {
254    /// The session ID.
255    pub sid: Sid,
256
257    /// The list of available transport upgrades.
258    pub upgrades: SmallVec<[TransportType; 1]>,
259
260    /// The ping interval, used in the heartbeat mechanism.
261    #[serde(
262        serialize_with = "serialize_duration_millis",
263        deserialize_with = "deserialize_duration_from_millis"
264    )]
265    pub ping_interval: Duration,
266
267    /// The ping timeout, used in the heartbeat mechanism.
268    #[serde(
269        serialize_with = "serialize_duration_millis",
270        deserialize_with = "deserialize_duration_from_millis"
271    )]
272    pub ping_timeout: Duration,
273
274    /// The maximum number of bytes per chunk, used by the client to
275    /// aggregate packets into payloads.
276    pub max_payload: u64,
277}
278
279/// Helper to serialize a duration as milliseconds
280pub fn serialize_duration_millis<S>(duration: &Duration, serializer: S) -> Result<S::Ok, S::Error>
281where
282    S: Serializer,
283{
284    // serialize_u128 is not supported so we need to cast it to u64, see https://github.com/serde-rs/json/issues/846
285    serializer.serialize_u64(duration.as_millis() as u64)
286}
287
288/// Helper to deserialize a duration from milliseconds
289pub fn deserialize_duration_from_millis<'de, D>(deserializer: D) -> Result<Duration, D::Error>
290where
291    D: Deserializer<'de>,
292{
293    let millis = u64::deserialize(deserializer)?;
294    Ok(Duration::from_millis(millis))
295}
296
297/// This default implementation should only be used for testing purposes.
298impl Default for OpenPacket {
299    fn default() -> Self {
300        Self {
301            sid: Sid::ZERO,
302            upgrades: smallvec::smallvec![TransportType::Websocket],
303            ping_interval: Duration::from_millis(25000),
304            ping_timeout: Duration::from_millis(20000),
305            max_payload: 100000,
306        }
307    }
308}
309
310/// Buffered packets to send to the client.
311/// It is used to ensure atomicity when sending multiple packets to the client.
312///
313/// The [`PacketBuf`] stack size will impact the dynamically allocated buffer
314/// of the internal mpsc channel.
315pub type PacketBuf = SmallVec<[Packet; 2]>;
316
317#[cfg(test)]
318mod tests {
319
320    use super::*;
321    use std::time::Duration;
322
323    #[test]
324    fn test_open_packet() {
325        let sid = Sid::new();
326        let packet = Packet::Open(OpenPacket {
327            sid,
328            upgrades: smallvec::smallvec![TransportType::Websocket],
329            ping_interval: Duration::from_millis(25000),
330            ping_timeout: Duration::from_millis(20000),
331            max_payload: 100000,
332        });
333        let packet_str: String = packet.into();
334        assert_eq!(
335            packet_str,
336            format!(
337                "0{{\"sid\":\"{sid}\",\"upgrades\":[\"websocket\"],\"pingInterval\":25000,\"pingTimeout\":20000,\"maxPayload\":100000}}"
338            )
339        );
340    }
341
342    #[test]
343    fn test_message_packet() {
344        let packet = Packet::Message("hello".into());
345        let packet_str: String = packet.into();
346        assert_eq!(packet_str, "4hello");
347    }
348
349    #[test]
350    fn test_binary_packet() {
351        let packet = Packet::Binary(vec![1, 2, 3].into());
352        let packet_str: String = packet.into();
353        assert_eq!(packet_str, "bAQID");
354    }
355
356    #[test]
357    fn test_binary_packet_v4_deserialize_payload_starting_with_4() {
358        let data = vec![0xE0, 0xE1, 0xE2];
359        // Sanity check: the base64 encoding indeed starts with '4'.
360        let packet_str: String = Packet::Binary(data.clone().into()).into();
361        assert_eq!(packet_str, "b4OHi");
362
363        let packet = Packet::parse(ProtocolVersion::V4, packet_str).unwrap();
364        assert_eq!(packet, Packet::Binary(data.into()));
365    }
366
367    #[test]
368    fn test_binary_packet_v3() {
369        let packet = Packet::BinaryV3(vec![1, 2, 3].into());
370        let packet_str: String = packet.into();
371        assert_eq!(packet_str, "b4AQID");
372    }
373
374    #[test]
375    fn test_packet_get_size_hint() {
376        // Max serialized packet
377        let open = OpenPacket {
378            sid: Sid::new(),
379            ping_interval: Duration::MAX,
380            ping_timeout: Duration::MAX,
381            max_payload: u64::MAX,
382            upgrades: smallvec::smallvec![TransportType::Websocket],
383        };
384        let size = serde_json::to_string(&open).unwrap().len();
385        let packet = Packet::Open(open);
386        assert_eq!(packet.get_size_hint(false), size);
387
388        let packet = Packet::Close;
389        assert_eq!(packet.get_size_hint(false), 1);
390
391        let packet = Packet::Ping;
392        assert_eq!(packet.get_size_hint(false), 1);
393
394        let packet = Packet::Pong;
395        assert_eq!(packet.get_size_hint(false), 1);
396
397        let packet = Packet::PingUpgrade;
398        assert_eq!(packet.get_size_hint(false), 6);
399
400        let packet = Packet::PongUpgrade;
401        assert_eq!(packet.get_size_hint(false), 6);
402
403        let packet = Packet::Message("hello".into());
404        assert_eq!(packet.get_size_hint(false), 6);
405
406        let packet = Packet::Upgrade;
407        assert_eq!(packet.get_size_hint(false), 1);
408
409        let packet = Packet::Noop;
410        assert_eq!(packet.get_size_hint(false), 1);
411
412        let packet = Packet::Binary(vec![1, 2, 3].into());
413        assert_eq!(packet.get_size_hint(false), 4);
414        assert_eq!(packet.get_size_hint(true), 5);
415
416        let packet = Packet::BinaryV3(vec![1, 2, 3].into());
417        assert_eq!(packet.get_size_hint(false), 4);
418        assert_eq!(packet.get_size_hint(true), 6);
419    }
420}
engineioxide_core/packet.rs

engineioxide_core/
packet.rs
