omni_cli/transport/
tcp.rs

1//! Direct bare-NDJSON-over-TCP transport (config `Transport::Tcp`, bd-bfb732).
2//!
3//! For receivers that host an inbound TCP listener (e.g. a Wear OS TCP-push
4//! foreground service — real Android can host an inbound server). Unlike the
5//! relay (which speaks the register/cast frame protocol), this sends **bare**
6//! [`InjectionCommand`] NDJSON lines, one per line — exactly what such a
7//! receiver drains and applies.
8
9use std::io::Write;
10use std::net::TcpStream;
11
12use crate::AppError;
13use crate::wire::InjectionCommand;
14
15/// Connect to `host:port` and send each command as a bare NDJSON line.
16///
17/// # Errors
18/// [`AppError::internal`] on connect / write / flush failure, or
19/// [`AppError::internal`] from command serialization.
20pub fn cast_bare(host: &str, port: u16, cmds: &[InjectionCommand]) -> Result<(), AppError> {
21    let addr = format!("{host}:{port}");
22    let mut stream = super::connect_bounded(&addr, super::CONNECT_TIMEOUT)
23        .map_err(|e| AppError::internal("tcp_connect", format!("{addr}: {e}")))?;
24    for cmd in cmds {
25        let line = cmd.json_line()?;
26        stream
27            .write_all(line.as_bytes())
28            .and_then(|()| stream.write_all(b"\n"))
29            .map_err(|e| AppError::internal("tcp_write", format!("{addr}: {e}")))?;
30    }
31    stream
32        .flush()
33        .map_err(|e| AppError::internal("tcp_flush", format!("{addr}: {e}")))?;
34    Ok(())
35}
36
37/// Render the bare NDJSON lines that would be sent to `host:port` (for dry-run).
38///
39/// # Errors
40/// [`AppError::internal`] if command serialization fails.
41pub fn render_bare(
42    host: &str,
43    port: u16,
44    cmds: &[InjectionCommand],
45) -> Result<Vec<String>, AppError> {
46    let mut out = Vec::with_capacity(cmds.len());
47    for cmd in cmds {
48        out.push(format!("tcp {host}:{port} <- {}", cmd.json_line()?));
49    }
50    Ok(out)
51}
52
53/// A hold-open bare-NDJSON-over-TCP sender for STREAMING dispatch (bd-772cb2):
54/// one connection is kept open across many commands instead of reconnecting per
55/// command. Connects lazily on the first cast and reconnects on the next cast
56/// after a write error (so a transient device drop does not abort the stream).
57pub struct TcpSender {
58    addr: String,
59    stream: Option<TcpStream>,
60}
61
62impl TcpSender {
63    /// Build a sender for `host:port` (no connection is made until the first cast).
64    #[must_use]
65    pub fn new(host: &str, port: u16) -> Self {
66        Self {
67            addr: format!("{host}:{port}"),
68            stream: None,
69        }
70    }
71
72    /// Send one command as a bare NDJSON line over the held-open connection,
73    /// connecting first if needed. On any I/O error the connection is dropped so
74    /// the next call reconnects.
75    ///
76    /// # Errors
77    /// [`AppError::internal`] on connect / write / flush failure (the stream is
78    /// reset for a fresh reconnect next time).
79    pub fn cast_one(&mut self, cmd: &InjectionCommand) -> Result<(), AppError> {
80        if self.stream.is_none() {
81            let s = super::connect_bounded(&self.addr, super::CONNECT_TIMEOUT)
82                .map_err(|e| AppError::internal("tcp_connect", format!("{}: {e}", self.addr)))?;
83            self.stream = Some(s);
84        }
85        let line = cmd.json_line()?;
86        let stream = self.stream.as_mut().expect("stream just set");
87        let result = stream
88            .write_all(line.as_bytes())
89            .and_then(|()| stream.write_all(b"\n"))
90            .and_then(|()| stream.flush());
91        if let Err(e) = result {
92            self.stream = None; // force reconnect next time
93            return Err(AppError::internal(
94                "tcp_write",
95                format!("{}: {e}", self.addr),
96            ));
97        }
98        Ok(())
99    }
100}
101
102#[cfg(test)]
103mod tests {
104    use super::*;
105    use std::io::{BufRead, BufReader};
106    use std::net::TcpListener;
107    use std::time::Duration;
108
109    #[test]
110    fn cast_bare_sends_ndjson_lines() {
111        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
112        let addr = listener.local_addr().unwrap();
113        let handle = std::thread::spawn(move || {
114            let (stream, _) = listener.accept().unwrap();
115            let mut reader = BufReader::new(stream);
116            let mut got = Vec::new();
117            for _ in 0..2 {
118                let mut line = String::new();
119                reader.read_line(&mut line).unwrap();
120                got.push(line.trim().to_owned());
121            }
122            got
123        });
124        // Give the listener a moment.
125        std::thread::sleep(Duration::from_millis(50));
126        cast_bare(
127            "127.0.0.1",
128            addr.port(),
129            &[InjectionCommand::text("hi"), InjectionCommand::key("enter")],
130        )
131        .unwrap();
132        let got = handle.join().unwrap();
133        assert_eq!(
134            got,
135            vec![
136                r#"{"v":1,"type":"text","text":"hi"}"#.to_owned(),
137                r#"{"v":1,"type":"key","key":"enter"}"#.to_owned(),
138            ]
139        );
140    }
141
142    #[test]
143    fn render_bare_previews_lines() {
144        let lines = render_bare("h", 9, &[InjectionCommand::scroll(3)]).unwrap();
145        assert_eq!(
146            lines,
147            vec![r#"tcp h:9 <- {"v":1,"type":"scroll","amount":3}"#.to_owned()]
148        );
149    }
150
151    #[test]
152    fn tcp_sender_holds_one_connection_for_many_casts() {
153        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
154        let addr = listener.local_addr().unwrap();
155        let handle = std::thread::spawn(move || {
156            // Accept exactly ONE connection and read 3 lines over it. A second
157            // accept would only happen if the sender reconnected per command.
158            let (stream, _) = listener.accept().unwrap();
159            let mut reader = BufReader::new(stream);
160            let mut got = Vec::new();
161            for _ in 0..3 {
162                let mut line = String::new();
163                reader.read_line(&mut line).unwrap();
164                got.push(line.trim().to_owned());
165            }
166            got
167        });
168        std::thread::sleep(Duration::from_millis(50));
169        let mut sender = TcpSender::new("127.0.0.1", addr.port());
170        sender.cast_one(&InjectionCommand::text("a")).unwrap();
171        sender.cast_one(&InjectionCommand::key("enter")).unwrap();
172        sender.cast_one(&InjectionCommand::scroll(2)).unwrap();
173        let got = handle.join().unwrap();
174        assert_eq!(
175            got,
176            vec![
177                r#"{"v":1,"type":"text","text":"a"}"#.to_owned(),
178                r#"{"v":1,"type":"key","key":"enter"}"#.to_owned(),
179                r#"{"v":1,"type":"scroll","amount":2}"#.to_owned(),
180            ]
181        );
182    }
183}
omni_cli/transport/tcp.rs

omni_cli/transport/
tcp.rs
