planarian_gym/
serve.rs

1//! `planarian-gym serve`: the daemon core over the fake machine, answering
2//! the real socket protocol, with a short history worth looking at. For
3//! trying the console (and the CLI) without a NixOS machine, and for
4//! checking that the console draws what the daemon reports.
5
6use crate::sim::{Sim, peer, unit};
7use planarian_ipc::{Request, Response};
8use planarian_schema::{ChangeId, Value};
9use std::io::{BufRead, BufReader, Write};
10use std::os::unix::net::UnixListener;
11use std::sync::mpsc;
12use std::time::{Duration, Instant};
13
14fn apply(sim: &mut Sim, change: &str, args: &[(&str, &str)]) -> Response {
15    let args = args.iter().map(|(k, v)| (k.to_string(), Value::Str(v.to_string()))).collect();
16    sim.core.as_mut().expect("core").handle(peer("root"), Request::Apply { change: ChangeId(change.into()), args })
17}
18
19pub fn serve(socket: &str) {
20    let mut sim = Sim::new();
21    for _ in 0..3 {
22        sim.tick();
23    }
24    // A change that commits, one that is reverted, a user's stop, a service
25    // that keeps failing its health check, and one that is killed and repaired.
26    for r in [
27        apply(&mut sim, "font-scale", &[("scale", "1.25")]),
28        apply(&mut sim, "listen-address", &[("service", "gym-web"), ("addr", "192.0.2.10")]),
29    ] {
30        eprintln!("{r:?}");
31    }
32    let core = sim.core.as_mut().expect("core");
33    eprintln!("{:?}", core.handle(peer("root"), Request::Stop { unit: unit("gym-cache") }));
34    sim.fake.w().blocked.insert(unit("gym-metrics"));
35    sim.fake.w().signal(&unit("gym-db"), 11);
36    for _ in 0..12 {
37        sim.tick();
38    }
39
40    let _ = std::fs::remove_file(socket);
41    let listener = UnixListener::bind(socket).unwrap_or_else(|e| panic!("{socket}: {e}"));
42    eprintln!("planarian-gym: serving the fake machine on {socket}");
43    type Job = (Request, mpsc::Sender<Response>);
44    let (tx, rx) = mpsc::channel::<Job>();
45    std::thread::spawn(move || {
46        for conn in listener.incoming().flatten() {
47            let tx = tx.clone();
48            std::thread::spawn(move || {
49                let mut line = String::new();
50                if BufReader::new(&conn).read_line(&mut line).is_err() {
51                    return;
52                }
53                let resp = match serde_json::from_str::<Request>(&line) {
54                    Ok(req) => {
55                        let (rtx, rrx) = mpsc::channel();
56                        if tx.send((req, rtx)).is_err() {
57                            return;
58                        }
59                        rrx.recv().unwrap_or(Response::Error { message: "stopped".into() })
60                    }
61                    Err(e) => Response::Error { message: format!("bad request: {e}") },
62                };
63                let _ = (&conn).write_all(resp.encode().as_bytes());
64            });
65        }
66    });
67    // One fake tick (5 s) per real second.
68    let mut next = Instant::now();
69    loop {
70        match rx.recv_timeout(next.saturating_duration_since(Instant::now())) {
71            Ok((req, reply)) => {
72                let resp = sim.core.as_mut().expect("core").handle(peer("root"), req);
73                let _ = reply.send(resp);
74            }
75            Err(mpsc::RecvTimeoutError::Timeout) => {
76                sim.tick();
77                next = Instant::now() + Duration::from_secs(1);
78            }
79            Err(mpsc::RecvTimeoutError::Disconnected) => break,
80        }
81    }
82}
planarian_gym/serve.rs

planarian_gym/
serve.rs
