indi/client/
tcpstream.rs

1use quick_xml::{events::Event, NsReader};
2use tokio::{
3    io::{AsyncRead, AsyncWriteExt},
4    net::{
5        tcp::{OwnedReadHalf, OwnedWriteHalf},
6        TcpStream,
7    },
8};
9use tracing::error;
10
11use crate::{client::Connectable, Command};
12use tokio::io::BufReader;
13
14use super::{AsyncClientConnection, AsyncReadConnection, AsyncWriteConnection};
15
16impl AsyncClientConnection for TcpStream {
17    type Reader = AsyncIndiReader<OwnedReadHalf>;
18    type Writer = AsyncIndiWriter;
19
20    fn to_indi(self) -> (Self::Writer, Self::Reader) {
21        let (reader, writer) = self.into_split();
22        let reader = NsReader::from_reader(BufReader::new(reader));
23
24        (AsyncIndiWriter { writer }, AsyncIndiReader::new(reader))
25    }
26}
27impl Connectable for TcpStream {
28    type ConnectionError = std::io::Error;
29    async fn connect(addr: String) -> Result<Self, Self::ConnectionError> {
30        Self::connect(addr).await
31    }
32}
33
34pub struct AsyncIndiReader<T> {
35    reader: NsReader<BufReader<T>>,
36}
37
38impl<T: AsyncRead + Unpin> AsyncIndiReader<T> {
39    fn new(reader: quick_xml::reader::NsReader<BufReader<T>>) -> AsyncIndiReader<T> {
40        AsyncIndiReader { reader }
41    }
42
43    async fn read_xml_documents(&mut self) -> Option<Result<String, crate::DeError>> {
44        let mut buffer = Vec::new();
45        let mut document = Vec::new();
46        let mut depth = 0;
47        loop {
48            let event = match self.reader.read_event_into_async(&mut buffer).await {
49                Ok(e) => e,
50                Err(e) => return Some(Err(e.into())),
51            };
52            match event {
53                Event::Start(e) => {
54                    depth += 1;
55                    document.extend_from_slice(b"<");
56                    document.extend_from_slice(e.name().as_ref());
57                    for attr in e.attributes() {
58                        let attr = match attr {
59                            Ok(d) => d,
60                            Err(e) => return Some(Err(e.into())),
61                        };
62                        document.extend_from_slice(b" ");
63                        document.extend_from_slice(attr.key.as_ref());
64                        document.extend_from_slice(b"=\"");
65                        document.extend_from_slice(&attr.value);
66                        document.extend_from_slice(b"\"");
67                    }
68                    document.extend_from_slice(b">");
69                }
70                Event::End(e) => {
71                    depth -= 1;
72                    document.extend_from_slice(b"</");
73                    document.extend_from_slice(e.name().as_ref());
74                    document.extend_from_slice(b">");
75                    if depth == 0 {
76                        let doc = match String::from_utf8(document) {
77                            Ok(d) => d,
78                            Err(e) => return Some(Err(e.into())),
79                        };
80                        return Some(Ok(doc));
81                    }
82                }
83                Event::Text(e) => {
84                    document.extend_from_slice(&e.into_inner());
85                }
86                Event::Eof => return None,
87                Event::Empty(e) => {
88                    document.extend_from_slice(b"<");
89                    document.extend_from_slice(e.name().as_ref());
90                    for attr in e.attributes() {
91                        let attr = match attr {
92                            Ok(d) => d,
93                            Err(e) => return Some(Err(e.into())),
94                        };
95                        document.extend_from_slice(b" ");
96                        document.extend_from_slice(attr.key.as_ref());
97                        document.extend_from_slice(b"=\"");
98                        document.extend_from_slice(&attr.value);
99                        document.extend_from_slice(b"\"");
100                    }
101
102                    document.extend_from_slice(b">");
103                    document.extend_from_slice(b"</");
104                    document.extend_from_slice(e.name().as_ref());
105                    document.extend_from_slice(b">");
106                    if depth == 0 {
107                        let doc = match String::from_utf8(document) {
108                            Ok(d) => d,
109                            Err(e) => return Some(Err(e.into())),
110                        };
111                        return Some(Ok(doc));
112                    }
113                }
114                _ => {}
115            }
116            buffer.clear();
117        }
118    }
119}
120
121impl<T: AsyncRead + Unpin + Send> AsyncReadConnection for AsyncIndiReader<T> {
122    async fn read(&mut self) -> Option<Result<crate::Command, crate::DeError>> {
123        let doc = match self.read_xml_documents().await? {
124            Ok(doc) => doc,
125            Err(e) => return Some(Err(e.into())),
126        };
127        let cmd = quick_xml::de::from_str::<crate::Command>(&doc).map_err(|x| x.into());
128
129        if let Err(e) = &cmd {
130            error!("Failed to parse ( {:?} ):\n{}", e, &doc);
131        }
132        return Some(cmd);
133    }
134}
135
136pub struct AsyncIndiWriter {
137    writer: OwnedWriteHalf,
138}
139
140impl AsyncWriteConnection for AsyncIndiWriter {
141    async fn write(&mut self, cmd: Command) -> Result<(), crate::DeError> {
142        let buffer = quick_xml::se::to_string(&cmd)?;
143        self.writer.write(buffer.as_bytes()).await?;
144
145        self.writer.write(b"\n").await?;
146        self.writer.flush().await?;
147        Ok(())
148    }
149
150    async fn shutdown(&mut self) -> Result<(), crate::DeError> {
151        Ok(self.writer.shutdown().await?)
152    }
153}
154
155#[cfg(test)]
156mod test {
157    use super::*;
158    use tokio::time::Duration;
159
160    #[tokio::test]
161    async fn test_threads_stop_on_shutdown() {
162        let connection = TcpStream::connect("indi:7624")
163            .await
164            .expect("connecting to indi");
165        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
166        let mut client = crate::client::Client::new(Some(tx));
167        let client_task: tokio::task::JoinHandle<()> =
168            tokio::task::spawn(crate::client::start(client.devices.clone(), rx, connection));
169
170        client.shutdown();
171        tokio::time::timeout(Duration::from_secs(1), client_task)
172            .await
173            .unwrap()
174            .unwrap();
175    }
176}
indi/client/tcpstream.rs

indi/client/
tcpstream.rs
