cast_sender/
client.rs

1use std::net::SocketAddr;
2use std::sync::Arc;
3
4use async_native_tls::{TlsConnector, TlsStream};
5use async_net::TcpStream;
6use futures_util::io::{ReadHalf, WriteHalf};
7use futures_util::{AsyncReadExt, AsyncWriteExt};
8use prost::Message;
9use smol::lock::Mutex;
10
11use super::proto;
12use super::{namespace::NamespaceUrn, Error, Payload};
13
14#[derive(Debug, Clone)]
15pub struct Response {
16    pub source_id: String,
17    pub destination_id: String,
18    // Probably not strictly necessary, since the namespace can be derived
19    // using the payload, but this may not have any guarantee of correctness,
20    // since the namespace may differ from the deserialized enum variant.
21    pub namespace: NamespaceUrn,
22    pub payload: Payload,
23    // Part of the payload
24    pub request_id: Option<u32>,
25}
26
27#[derive(Debug, Clone)]
28pub struct Client {
29    read_stream: Arc<Mutex<ReadHalf<TlsStream<TcpStream>>>>,
30    write_stream: Arc<Mutex<WriteHalf<TlsStream<TcpStream>>>>,
31}
32
33impl Client {
34    pub async fn connect(addr: &str) -> Result<Self, Error> {
35        let addr = SocketAddr::new(addr.parse()?, 8009);
36
37        // Casts devices are using self signed certs
38        let tls_connector = TlsConnector::new().danger_accept_invalid_certs(true);
39        let tcp_stream = TcpStream::connect(&addr).await?;
40
41        let tls_stream = tls_connector
42            .connect(addr.to_string(), tcp_stream.clone())
43            .await?;
44
45        let (read_stream, write_stream) = tls_stream.split();
46
47        Ok(Self {
48            read_stream: Arc::new(Mutex::new(read_stream)),
49            write_stream: Arc::new(Mutex::new(write_stream)),
50        })
51    }
52
53    pub async fn receive(&self) -> Result<Response, Error> {
54        let mut read_stream = self.read_stream.lock().await;
55
56        // The first package is a u32 specifying the packet length....
57        let mut buf: [u8; 4] = [0; 4];
58        read_stream.read_exact(&mut buf).await?;
59        let len = u32::from_be_bytes(buf);
60
61        // ... then get the actual package with the specified length
62        let mut buf: Vec<u8> = vec![0; len as usize];
63        read_stream.read_exact(&mut buf).await?;
64
65        let msg: proto::CastMessage = proto::CastMessage::decode(&buf[..])?;
66        let ns: NamespaceUrn = msg.namespace.parse().unwrap();
67        let mut pl: PayloadData = serde_json::from_str(msg.payload_utf8())?;
68
69        if let Payload::Custom(u) = &mut pl.data {
70            u.namespace = ns.clone();
71        };
72
73        debug!(
74            "[RECV] {} -> {} | Namespace: {:?} | Request: {:?}",
75            msg.source_id, msg.destination_id, ns, pl.request_id
76        );
77        debug!("       {:#?}", pl);
78        Ok(Response {
79            source_id: msg.source_id,
80            destination_id: msg.destination_id,
81            namespace: ns,
82            payload: pl.data,
83            request_id: pl.request_id,
84        })
85    }
86
87    pub async fn send<P: Into<Payload>>(
88        &self,
89        destination_id: String,
90        payload: P,
91        request_id: Option<u32>,
92    ) -> Result<(), Error> {
93        let payload: Payload = payload.into();
94        let payload_data = PayloadData {
95            request_id,
96            data: payload.clone(),
97        };
98
99        let payload_json = serde_json::to_string(&payload_data).unwrap();
100        let msg = proto::CastMessage {
101            protocol_version: proto::cast_message::ProtocolVersion::Castv210.into(),
102            source_id: "sender-0".into(),
103            destination_id,
104            namespace: payload.namespace().to_string(),
105            payload_type: proto::cast_message::PayloadType::String.into(),
106            payload_utf8: Some(payload_json.clone()),
107            payload_binary: None,
108            continued: None,
109            remaining_length: None,
110        };
111
112        debug!(
113            "[SEND] {} -> {} | Namespace: {:?} | Request: {:?}",
114            msg.source_id,
115            msg.destination_id,
116            payload.namespace(),
117            request_id,
118        );
119        debug!("       {}", payload_json);
120
121        let mut write_stream = self.write_stream.lock().await;
122        let len: u32 = msg.encoded_len().try_into().unwrap();
123
124        // First send package length
125        write_stream.write_all(&len.to_be_bytes()).await?;
126
127        // Then the actual package
128        write_stream.write_all(&msg.encode_to_vec()).await?;
129
130        Ok(())
131    }
132}
133
134#[derive(Serialize, Deserialize, Debug)]
135#[serde(rename_all = "camelCase")]
136struct PayloadData {
137    #[serde(skip_serializing_if = "Option::is_none")]
138    request_id: Option<u32>,
139    #[serde(flatten)]
140    data: Payload,
141}
cast_sender/client.rs

cast_sender/
client.rs
