tawasuyu/brahman

1//! `mirada-link` — el transporte Cerebro↔Cuerpo del compositor.
2//!
3//! [`mirada_protocol`] define *qué* se dice (los enums y el marco de
4//! cable); este crate define *cómo viaja*: un socket Unix con un hilo
5//! lector de fondo que entrega los mensajes recibidos por un canal, para
6//! que el dueño del [`Link`] sólo tenga que sondear sin bloquearse.
7//!
8//! Los dos procesos usan el mismo tipo, parametrizado al revés:
9//!
10//! - El Cerebro tiene un [`BrainLink`]: envía [`BrainCommand`], recibe
11//!   [`BodyEvent`].
12//! - El Cuerpo tiene un [`BodyLink`]: envía [`BodyEvent`], recibe
13//!   [`BrainCommand`].
14//!
15//! Para arrancar el par hay tres caminos: [`connected_pair`] (un
16//! `socketpair`, ideal para heredar un fd al lanzar al hijo o para
17//! tests), [`Link::connect`] (conectar a una ruta) y [`Link::listen`]
18//! (escuchar en una ruta y aceptar una conexión).
19
20#![forbid(unsafe_code)]
21
22use std::io::{self, BufReader};
23use std::marker::PhantomData;
24use std::net::Shutdown;
25use std::os::unix::net::{UnixListener, UnixStream};
26use std::path::Path;
27use std::sync::mpsc::{self, Receiver};
28use std::thread;
29
30use serde::de::DeserializeOwned;
31use serde::Serialize;
32
33use mirada_protocol::{read_frame, write_frame, BodyEvent, BrainCommand};
34
35/// El extremo del Cerebro: envía [`BrainCommand`], recibe [`BodyEvent`].
36pub type BrainLink = Link<BrainCommand, BodyEvent>;
37
38/// El extremo del Cuerpo: envía [`BodyEvent`], recibe [`BrainCommand`].
39pub type BodyLink = Link<BodyEvent, BrainCommand>;
40
41/// Un extremo del canal: envía mensajes de tipo `Out` y recibe `In`.
42///
43/// La escritura es síncrona sobre el socket; la lectura la hace un hilo
44/// de fondo que deposita lo recibido en un canal interno. Al soltar el
45/// `Link` se cierra el socket, lo que termina el hilo lector propio y le
46/// señala EOF al otro extremo.
47pub struct Link<Out, In> {
48    writer: UnixStream,
49    incoming: Receiver<In>,
50    _out: PhantomData<fn(Out)>,
51}
52
53impl<Out, In> Link<Out, In>
54where
55    Out: Serialize,
56    In: DeserializeOwned + Send + 'static,
57{
58    /// Construye un `Link` sobre un socket ya conectado.
59    pub fn from_stream(stream: UnixStream) -> io::Result<Self> {
60        let reader = stream.try_clone()?;
61        let (tx, rx) = mpsc::channel();
62        thread::spawn(move || {
63            let mut r = BufReader::new(reader);
64            // Lee marcos hasta EOF limpio o error de socket.
65            while let Ok(Some(msg)) = read_frame::<_, In>(&mut r) {
66                if tx.send(msg).is_err() {
67                    break; // el dueño soltó el Link
68                }
69            }
70        });
71        Ok(Self { writer: stream, incoming: rx, _out: PhantomData })
72    }
73
74    /// Conecta a un socket Unix en `path` (lado cliente).
75    pub fn connect<P: AsRef<Path>>(path: P) -> io::Result<Self> {
76        Self::from_stream(UnixStream::connect(path)?)
77    }
78
79    /// Escucha en `path` y bloquea hasta aceptar una conexión (lado
80    /// servidor). El socket de escucha se cierra tras el primer cliente.
81    pub fn listen<P: AsRef<Path>>(path: P) -> io::Result<Self> {
82        let listener = UnixListener::bind(path)?;
83        let (stream, _) = listener.accept()?;
84        Self::from_stream(stream)
85    }
86
87    /// Envía un mensaje. Falla si el otro extremo cerró el canal.
88    pub fn send(&mut self, msg: &Out) -> io::Result<()> {
89        write_frame(&mut self.writer, msg)
90    }
91
92    /// Recoge un mensaje si hay alguno pendiente, sin bloquear.
93    pub fn try_recv(&self) -> Option<In> {
94        self.incoming.try_recv().ok()
95    }
96
97    /// Vacía todos los mensajes pendientes — un tick del bucle de eventos.
98    pub fn drain(&self) -> Vec<In> {
99        self.incoming.try_iter().collect()
100    }
101
102    /// Bloquea hasta recibir un mensaje. Devuelve `None` si el otro
103    /// extremo cerró el canal.
104    pub fn recv(&self) -> Option<In> {
105        self.incoming.recv().ok()
106    }
107}
108
109impl<Out, In> Drop for Link<Out, In> {
110    fn drop(&mut self) {
111        // Cierra la conexión: el hilo lector propio recibe EOF y termina,
112        // y el otro extremo ve EOF en su próxima lectura.
113        let _ = self.writer.shutdown(Shutdown::Both);
114    }
115}
116
117/// Crea un par Cerebro↔Cuerpo conectado en memoria, con un `socketpair`.
118///
119/// Es el camino de los tests y también el del despliegue real cuando el
120/// Cerebro lanza al Cuerpo como proceso hijo y le hereda un extremo.
121pub fn connected_pair() -> io::Result<(BrainLink, BodyLink)> {
122    let (a, b) = UnixStream::pair()?;
123    Ok((Link::from_stream(a)?, Link::from_stream(b)?))
124}
125
126#[cfg(test)]
127mod tests {
128    use super::*;
129    use mirada_protocol::{Rect, WindowPlacement};
130    use std::time::Duration;
131
132    fn place(id: u64) -> BrainCommand {
133        BrainCommand::Place(vec![WindowPlacement {
134            id,
135            rect: Rect::new(0, 0, 800, 600),
136            visible: true,
137            focused: true,
138            floating: false,
139            fullscreen: false,
140        }])
141    }
142
143    #[test]
144    fn brain_command_reaches_the_body() {
145        let (mut brain, body) = connected_pair().unwrap();
146        brain.send(&place(1)).unwrap();
147        // Da un instante al hilo lector.
148        for _ in 0..100 {
149            if let Some(cmd) = body.try_recv() {
150                assert_eq!(cmd, place(1));
151                return;
152            }
153            thread::sleep(Duration::from_millis(2));
154        }
155        panic!("el comando no llegó al Cuerpo");
156    }
157
158    #[test]
159    fn body_event_reaches_the_brain() {
160        let (brain, mut body) = connected_pair().unwrap();
161        let ev = BodyEvent::Keybind("Super+Return".into());
162        body.send(&ev).unwrap();
163        assert_eq!(brain.recv(), Some(ev));
164    }
165
166    #[test]
167    fn many_messages_keep_their_order() {
168        let (brain, mut body) = connected_pair().unwrap();
169        for id in 0..20 {
170            body.send(&BodyEvent::WindowClosed { id }).unwrap();
171        }
172        for id in 0..20 {
173            assert_eq!(brain.recv(), Some(BodyEvent::WindowClosed { id }));
174        }
175    }
176
177    #[test]
178    fn drain_collects_everything_pending() {
179        let (mut brain, body) = connected_pair().unwrap();
180        for id in 1..=5 {
181            brain.send(&place(id)).unwrap();
182        }
183        // Espera a que el hilo lector encole los cinco.
184        let mut got = Vec::new();
185        for _ in 0..100 {
186            got.extend(body.drain());
187            if got.len() == 5 {
188                break;
189            }
190            thread::sleep(Duration::from_millis(2));
191        }
192        assert_eq!(got.len(), 5);
193    }
194
195    #[test]
196    fn dropping_one_end_closes_the_other() {
197        let (brain, body) = connected_pair().unwrap();
198        drop(body);
199        // Sin nadie al otro lado, recv termina con None en vez de colgarse.
200        assert_eq!(brain.recv(), None);
201    }
202
203    #[test]
204    fn sending_into_a_closed_link_errors() {
205        let (mut brain, body) = connected_pair().unwrap();
206        drop(body);
207        // La primera escritura puede pasar al búfer del socket; alguna
208        // de ellas acaba fallando con tubería rota.
209        let mut errored = false;
210        for id in 0..1000 {
211            if brain.send(&place(id)).is_err() {
212                errored = true;
213                break;
214            }
215        }
216        assert!(errored, "se esperaba un error de tubería rota");
217    }
218
219    #[test]
220    fn connect_and_listen_round_trip_over_a_path() {
221        let dir = std::env::temp_dir();
222        let path = dir.join(format!("mirada-link-test-{}.sock", std::process::id()));
223        let _ = std::fs::remove_file(&path);
224
225        let server_path = path.clone();
226        let server = thread::spawn(move || {
227            let mut link: BodyLink = Link::listen(&server_path).unwrap();
228            link.send(&BodyEvent::OutputAdded { id: 0, width: 1920, height: 1080 })
229                .unwrap();
230            // Mantén vivo el extremo hasta que el cliente lea.
231            link.recv()
232        });
233
234        // Espera a que el servidor publique el socket.
235        let mut brain: Option<BrainLink> = None;
236        for _ in 0..200 {
237            if let Ok(l) = Link::connect(&path) {
238                brain = Some(l);
239                break;
240            }
241            thread::sleep(Duration::from_millis(2));
242        }
243        let mut brain = brain.expect("no se pudo conectar al servidor");
244        assert_eq!(
245            brain.recv(),
246            Some(BodyEvent::OutputAdded { id: 0, width: 1920, height: 1080 })
247        );
248        brain.send(&BrainCommand::Shutdown).unwrap();
249        assert_eq!(server.join().unwrap(), Some(BrainCommand::Shutdown));
250        let _ = std::fs::remove_file(&path);
251    }
252}