playit_agent_proto/
control_feed.rs

1use std::fmt::{Debug, Formatter};
2use std::io::{Error, ErrorKind, Read, Write};
3use std::net::SocketAddr;
4
5use byteorder::{BigEndian, ReadBytesExt};
6use message_encoding::MessageEncoding;
7
8use crate::control_messages::ControlResponse;
9use crate::rpc::ControlRpcMessage;
10
11#[derive(Debug, Eq, PartialEq, Clone)]
12pub enum ControlFeed {
13    Response(ControlRpcMessage<ControlResponse>),
14    NewClient(NewClient),
15    NewClientOld(NewClientOld),
16}
17
18#[derive(Debug, Eq, PartialEq, Clone)]
19pub struct NewClient {
20    pub connect_addr: SocketAddr,
21    pub peer_addr: SocketAddr,
22    pub data_center_id: u32,
23    pub tunnel_id: u64,
24    pub port_offset: u16,
25    pub claim_instructions: ClaimInstructions,
26}
27
28#[derive(Debug, Eq, PartialEq, Clone)]
29pub struct NewClientOld {
30    pub connect_addr: SocketAddr,
31    pub peer_addr: SocketAddr,
32    pub claim_instructions: ClaimInstructions,
33    pub tunnel_server_id: u64,
34    pub data_center_id: u32,
35}
36
37#[derive(Eq, PartialEq, Clone)]
38pub struct ClaimInstructions {
39    pub address: SocketAddr,
40    pub token: Vec<u8>,
41}
42
43impl Debug for ClaimInstructions {
44    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
45        write!(f, "ClaimInstructions {{ address: {}, token: {} }}", self.address, hex::encode(&self.token))
46    }
47}
48
49impl MessageEncoding for ControlFeed {
50    fn write_to<T: Write>(&self, out: &mut T) -> std::io::Result<usize> {
51        let mut sum = 0;
52
53        match self {
54            ControlFeed::Response(res) => {
55                sum += 1u32.write_to(out)?;
56                sum += res.write_to(out)?;
57            }
58            ControlFeed::NewClientOld(client) => {
59                sum += 2u32.write_to(out)?;
60                sum += client.write_to(out)?;
61            }
62            ControlFeed::NewClient(client) => {
63                sum += 3u32.write_to(out)?;
64                sum += client.write_to(out)?;
65            }
66        }
67
68        Ok(sum)
69    }
70
71    fn read_from<T: Read>(read: &mut T) -> std::io::Result<Self> {
72        match read.read_u32::<BigEndian>()? {
73            1 => Ok(ControlFeed::Response(ControlRpcMessage::read_from(read)?)),
74            2 => Ok(ControlFeed::NewClient(NewClientOld::read_from(read)?.into())),
75            3 => Ok(ControlFeed::NewClient(NewClient::read_from(read)?)),
76            _ => Err(Error::new(ErrorKind::Other, "invalid ControlFeed id")),
77        }
78    }
79}
80
81impl MessageEncoding for NewClient {
82    fn write_to<T: Write>(&self, out: &mut T) -> std::io::Result<usize> {
83        let mut sum = 0;
84        sum += self.connect_addr.write_to(out)?;
85        sum += self.peer_addr.write_to(out)?;
86        sum += self.data_center_id.write_to(out)?;
87        sum += self.tunnel_id.write_to(out)?;
88        sum += self.port_offset.write_to(out)?;
89        sum += self.claim_instructions.write_to(out)?;
90        Ok(sum)
91    }
92
93    fn read_from<T: Read>(read: &mut T) -> std::io::Result<Self> {
94        Ok(NewClient {
95            connect_addr: SocketAddr::read_from(read)?,
96            peer_addr: SocketAddr::read_from(read)?,
97            data_center_id: u32::read_from(read)?,
98            tunnel_id: u64::read_from(read)?,
99            port_offset: u16::read_from(read)?,
100            claim_instructions: ClaimInstructions::read_from(read)?,
101        })
102    }
103}
104
105impl From<NewClientOld> for NewClient {
106    fn from(value: NewClientOld) -> Self {
107        NewClient {
108            connect_addr: value.connect_addr,
109            peer_addr: value.peer_addr,
110            data_center_id: value.data_center_id,
111            tunnel_id: 0,
112            port_offset: 0,
113            claim_instructions: value.claim_instructions,
114        }
115    }
116}
117
118impl MessageEncoding for NewClientOld {
119    fn write_to<T: Write>(&self, out: &mut T) -> std::io::Result<usize> {
120        let mut sum = 0;
121        sum += self.connect_addr.write_to(out)?;
122        sum += self.peer_addr.write_to(out)?;
123        sum += self.claim_instructions.write_to(out)?;
124        sum += self.tunnel_server_id.write_to(out)?;
125        sum += self.data_center_id.write_to(out)?;
126        Ok(sum)
127    }
128
129    fn read_from<T: Read>(read: &mut T) -> std::io::Result<Self> {
130        Ok(NewClientOld {
131            connect_addr: SocketAddr::read_from(read)?,
132            peer_addr: SocketAddr::read_from(read)?,
133            claim_instructions: ClaimInstructions::read_from(read)?,
134            tunnel_server_id: read.read_u64::<BigEndian>()?,
135            data_center_id: read.read_u32::<BigEndian>()?,
136        })
137    }
138}
139
140impl MessageEncoding for ClaimInstructions {
141    fn write_to<T: Write>(&self, out: &mut T) -> std::io::Result<usize> {
142        let mut sum = 0;
143        sum += self.address.write_to(out)?;
144        sum += self.token.write_to(out)?;
145        Ok(sum)
146    }
147
148    fn read_from<T: Read>(read: &mut T) -> std::io::Result<Self> {
149        Ok(ClaimInstructions {
150            address: SocketAddr::read_from(read)?,
151            token: Vec::read_from(read)?,
152        })
153    }
154}
155
156#[cfg(test)]
157mod test {
158    use super::*;
159
160    #[test]
161    fn parse_control() {
162        let data = "0000000204d1198d10046804d053c766cc4904d1198c029306000000000000004c2\
163        c003cd1198d100468d053c766cc49cba8329c930664e9431200000000000000010000000000298c05779c\
164        9306000000000e00000000000065b2000171012de0fdb1b6d5de58be82911b07bc00000000000065b20000000e";
165
166        let bytes = hex::decode(data).unwrap();
167        let mut reader = &bytes[..];
168        let req = ControlFeed::read_from(&mut reader).unwrap();
169        println!("{:?}", req);
170    }
171}
172
playit_agent_proto/control_feed.rs

playit_agent_proto/
control_feed.rs
