engineioxide 0.17.4

Engine IO server implementation as a Tower Service.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122

//! Payload encoder and decoder for polling transport.

use crate::{
    errors::Error, packet::Packet, peekable::PeekableReceiver, service::ProtocolVersion,
    socket::PacketBuf,
};
use bytes::Bytes;
use futures_core::Stream;
use http::Request;
use tokio::sync::MutexGuard;

mod buf;
mod decoder;
mod encoder;

const PACKET_SEPARATOR_V4: u8 = b'\x1e';
#[cfg(feature = "v3")]
const STRING_PACKET_SEPARATOR_V3: u8 = b':';
#[cfg(feature = "v3")]
const BINARY_PACKET_SEPARATOR_V3: u8 = 0xff;
#[cfg(feature = "v3")]
const STRING_PACKET_IDENTIFIER_V3: u8 = 0x00;
#[cfg(feature = "v3")]
const BINARY_PACKET_IDENTIFIER_V3: u8 = 0x01;

pub fn decoder(
    body: Request<impl http_body::Body<Error = impl std::fmt::Debug> + Unpin>,
    #[allow(unused_variables)] protocol: ProtocolVersion,
    max_payload: u64,
) -> impl Stream<Item = Result<Packet, Error>> {
    #[cfg(feature = "v3")]
    {
        use futures_util::future::Either;
        use http::header::CONTENT_TYPE;
        #[cfg(feature = "tracing")]
        tracing::debug!("decoding payload {:?}", body.headers().get(CONTENT_TYPE));
        let is_binary =
            body.headers().get(CONTENT_TYPE) == Some(&"application/octet-stream".parse().unwrap());
        match protocol {
            ProtocolVersion::V4 => Either::Left(decoder::v4_decoder(body, max_payload)),
            ProtocolVersion::V3 if is_binary => {
                Either::Right(Either::Left(decoder::v3_binary_decoder(body, max_payload)))
            }
            ProtocolVersion::V3 => {
                Either::Right(Either::Right(decoder::v3_string_decoder(body, max_payload)))
            }
        }
    }

    #[cfg(not(feature = "v3"))]
    {
        decoder::v4_decoder(body, max_payload)
    }
}

/// A payload to transmit to the client through http polling
#[non_exhaustive]
pub struct Payload {
    pub data: Bytes,
    pub has_binary: bool,
}
impl Payload {
    pub fn new(data: Bytes, has_binary: bool) -> Self {
        Self { data, has_binary }
    }
}

pub async fn encoder(
    rx: MutexGuard<'_, PeekableReceiver<PacketBuf>>,
    #[allow(unused_variables)] protocol: ProtocolVersion,
    #[cfg(feature = "v3")] supports_binary: bool,
    max_payload: u64,
) -> Result<Payload, Error> {
    #[cfg(feature = "v3")]
    {
        match protocol {
            ProtocolVersion::V4 => encoder::v4_encoder(rx, max_payload).await,
            ProtocolVersion::V3 if supports_binary => {
                encoder::v3_binary_encoder(rx, max_payload).await
            }
            ProtocolVersion::V3 => encoder::v3_string_encoder(rx, max_payload).await,
        }
    }

    #[cfg(not(feature = "v3"))]
    {
        encoder::v4_encoder(rx, max_payload).await
    }
}

/// Encodes a single packet into a byte array.
pub fn packet_encoder(
    packet: Packet,
    #[allow(unused_variables)] protocol: ProtocolVersion,
    #[cfg(feature = "v3")] supports_binary: bool,
) -> Bytes {
    #[cfg(feature = "v3")]
    {
        match protocol {
            ProtocolVersion::V4 => packet.into(),
            ProtocolVersion::V3 if supports_binary && packet.is_binary() => {
                use bytes::BytesMut;

                let size_hint = packet.get_size_hint(!supports_binary) + 2;
                let mut bytes = BytesMut::with_capacity(size_hint);
                encoder::v3_bin_packet_encoder(packet, &mut bytes);
                bytes.freeze()
            }
            ProtocolVersion::V3 => {
                use bytes::BytesMut;

                let size_hint = packet.get_size_hint(!supports_binary) + 2;
                let mut bytes = BytesMut::with_capacity(size_hint);
                encoder::v3_string_packet_encoder(packet, &mut bytes);
                bytes.freeze()
            }
        }
    }

    #[cfg(not(feature = "v3"))]
    packet.into()
}