headroom

public

Star0

Watch0

6 commits 1 branch 0 tags 0 releases521.6 KiB

//! blocking [`Client`].

use std::collections::VecDeque;
use std::io::{BufReader, BufWriter};
use std::os::unix::net::UnixStream;
use std::path::{Path, PathBuf};

use serde::de::DeserializeOwned;

use headroom_ipc::{
    default_socket_path, Codec, Event, HelloData, Op, ProtoError, Request, Response,
    ResponsePayload, Route, ServerFrame, Status, Topic,
};

/// errors from the blocking client.
#[derive(Debug, thiserror::Error)]
pub enum ClientError {
    #[error("ipc: {0}")]
    Ipc(#[from] headroom_ipc::Error),

    /// server's first frame was not the `hello` event.
    #[error("expected hello event from server, got {0}")]
    BadHello(String),

    /// response id we never issued.
    #[error("response with unknown id {0}")]
    UnknownResponseId(u64),

    /// protocol-level error for an op.
    #[error("server error: {0}")]
    Protocol(#[from] ProtoError),

    #[error("no default socket path (XDG_RUNTIME_DIR unset and /proc/self/status unreadable)")]
    NoDefaultPath,

    /// typed-helper response failed to deserialize into the expected shape.
    #[error("response shape mismatch: {0}")]
    DecodeResult(serde_json::Error),
}

/// blocking client for the headroom control protocol.
///
/// single-threaded by construction; for concurrent request/response and event
/// consumption, open two connections.
pub struct Client {
    reader: BufReader<UnixStream>,
    writer: BufWriter<UnixStream>,
    codec: Codec,
    next_id: u64,
    pending_events: VecDeque<Event>,
    hello: HelloData,
    socket_path: PathBuf,
}

impl Client {
    /// connect at the default socket path.
    pub fn connect() -> Result<Self, ClientError> {
        let path = default_socket_path().ok_or(ClientError::NoDefaultPath)?;
        Self::connect_at(&path)
    }

    /// connect at the given socket path.
    pub fn connect_at(path: &Path) -> Result<Self, ClientError> {
        let stream = UnixStream::connect(path).map_err(|e| ClientError::Ipc(e.into()))?;
        let reader_half = stream.try_clone().map_err(|e| ClientError::Ipc(e.into()))?;
        let writer_half = stream;

        let mut me = Self {
            reader: BufReader::new(reader_half),
            writer: BufWriter::new(writer_half),
            codec: Codec::new(),
            next_id: 1,
            pending_events: VecDeque::new(),
            // placeholder; populated by handshake below
            hello: HelloData {
                daemon: String::new(),
                version: String::new(),
                protocol: 0,
            },
            socket_path: path.to_path_buf(),
        };

        me.handshake()?;
        Ok(me)
    }

    fn handshake(&mut self) -> Result<(), ClientError> {
        let frame: ServerFrame = self.codec.read(&mut self.reader)?;
        match frame {
            ServerFrame::Event(ev)
                if ev.topic == Topic::Control && ev.event.as_str() == "hello" =>
            {
                let hello: HelloData =
                    serde_json::from_value(ev.data).map_err(ClientError::DecodeResult)?;
                self.hello = hello;
            }
            ServerFrame::Event(ev) => {
                return Err(ClientError::BadHello(format!(
                    "{} event on {}",
                    ev.event, ev.topic
                )))
            }
            ServerFrame::Response(r) => {
                return Err(ClientError::BadHello(format!("response id={}", r.id)))
            }
        }

        // best effort for older daemons
        self.send_raw(Op::Hello {
            protocol: headroom_ipc::PROTOCOL_VERSION,
        })?;
        Ok(())
    }

    /// `hello` payload received on connect.
    #[must_use]
    pub fn hello(&self) -> &HelloData {
        &self.hello
    }

    /// socket path this client is connected to.
    #[must_use]
    pub fn socket_path(&self) -> &Path {
        &self.socket_path
    }

    fn alloc_id(&mut self) -> u64 {
        let id = self.next_id;
        // wrap: u64::MAX requests on one connection won't happen, but be correct
        self.next_id = self.next_id.wrapping_add(1);
        id
    }

    /// send a request, block until the paired response arrives.
    ///
    /// stray events received meanwhile are queued for [`next_event`](Self::next_event).
    pub fn send(&mut self, op: Op) -> Result<serde_json::Value, ClientError> {
        let payload = self.send_raw(op)?;
        match payload {
            ResponsePayload::Ok { result } => Ok(result),
            ResponsePayload::Err { error } => Err(ClientError::Protocol(error)),
        }
    }

    /// like [`send`](Self::send) but returns the raw [`ResponsePayload`] —
    /// keeps a protocol-level error in-band instead of as [`ClientError::Protocol`].
    pub fn send_raw(&mut self, op: Op) -> Result<ResponsePayload, ClientError> {
        let id = self.alloc_id();
        let req = Request::new(id, op);
        self.codec.write(&mut self.writer, &req)?;

        loop {
            let frame: ServerFrame = self.codec.read(&mut self.reader)?;
            match frame {
                ServerFrame::Response(Response {
                    id: rid,
                    payload: _,
                }) if rid != id => {
                    return Err(ClientError::UnknownResponseId(rid));
                }
                ServerFrame::Response(Response { payload, .. }) => return Ok(payload),
                ServerFrame::Event(ev) => {
                    self.pending_events.push_back(ev);
                }
            }
        }
    }

    /// block until the next event arrives. drains the queue before reading the socket.
    ///
    /// an unsolicited response (no matching request) is rejected as
    /// [`ClientError::UnknownResponseId`].
    pub fn next_event(&mut self) -> Result<Event, ClientError> {
        if let Some(ev) = self.pending_events.pop_front() {
            return Ok(ev);
        }
        match self.codec.read::<_, ServerFrame>(&mut self.reader)? {
            ServerFrame::Event(ev) => Ok(ev),
            ServerFrame::Response(r) => Err(ClientError::UnknownResponseId(r.id)),
        }
    }

    /// pop a queued event without blocking. does **not** read from the socket.
    pub fn pending_event(&mut self) -> Option<Event> {
        self.pending_events.pop_front()
    }

    // typed convenience wrappers

    fn send_into<T: DeserializeOwned>(&mut self, op: Op) -> Result<T, ClientError> {
        let value = self.send(op)?;
        serde_json::from_value(value).map_err(ClientError::DecodeResult)
    }

    /// `status`
    pub fn status(&mut self) -> Result<Status, ClientError> {
        self.send_into(Op::Status)
    }

    /// `profile.list`
    pub fn profile_list(&mut self) -> Result<Vec<headroom_ipc::ProfileInfo>, ClientError> {
        #[derive(serde::Deserialize)]
        struct Body {
            profiles: Vec<headroom_ipc::ProfileInfo>,
        }
        let body: Body = self.send_into(Op::ProfileList)?;
        Ok(body.profiles)
    }

    /// `profile.use`
    pub fn profile_use(&mut self, name: &str) -> Result<String, ClientError> {
        #[derive(serde::Deserialize)]
        struct Body {
            name: String,
        }
        let body: Body = self.send_into(Op::ProfileUse {
            name: name.to_owned(),
        })?;
        Ok(body.name)
    }

    /// `profile.show`
    pub fn profile_show(
        &mut self,
        name: Option<&str>,
    ) -> Result<serde_json::Value, ClientError> {
        self.send(Op::ProfileShow {
            name: name.map(String::from),
        })
    }

    /// `profile.reload`
    pub fn profile_reload(&mut self) -> Result<Vec<String>, ClientError> {
        #[derive(serde::Deserialize)]
        struct Body {
            reloaded: Vec<String>,
        }
        let body: Body = self.send_into(Op::ProfileReload)?;
        Ok(body.reloaded)
    }

    /// `route.list`
    pub fn route_list(&mut self) -> Result<headroom_ipc::RouteList, ClientError> {
        self.send_into(Op::RouteList)
    }

    /// `route.set`
    pub fn route_set(&mut self, app: &str, to: Route) -> Result<(), ClientError> {
        let _: serde_json::Value = self.send(Op::RouteSet {
            app: app.to_owned(),
            to,
        })?;
        Ok(())
    }

    /// `route.unset`
    pub fn route_unset(&mut self, app: &str) -> Result<(), ClientError> {
        let _: serde_json::Value = self.send(Op::RouteUnset {
            app: app.to_owned(),
        })?;
        Ok(())
    }

    /// `route.stream`
    pub fn route_stream(&mut self, node_id: u32, to: Route) -> Result<(), ClientError> {
        let _: serde_json::Value = self.send(Op::RouteStream { node_id, to })?;
        Ok(())
    }

    /// `setting.get`
    pub fn setting_get(&mut self, key: &str) -> Result<serde_json::Value, ClientError> {
        #[derive(serde::Deserialize)]
        struct Body {
            #[allow(dead_code)]
            key: String,
            value: serde_json::Value,
        }
        let body: Body = self.send_into(Op::SettingGet {
            key: key.to_owned(),
        })?;
        Ok(body.value)
    }

    /// `setting.set`
    pub fn setting_set(
        &mut self,
        key: &str,
        value: serde_json::Value,
    ) -> Result<(), ClientError> {
        let _: serde_json::Value = self.send(Op::SettingSet {
            key: key.to_owned(),
            value,
        })?;
        Ok(())
    }

    /// `setting.clear` — returns whether an override was present.
    pub fn setting_clear(&mut self, key: &str) -> Result<bool, ClientError> {
        #[derive(serde::Deserialize)]
        struct Body {
            cleared: bool,
        }
        let body: Body = self.send_into(Op::SettingClear {
            key: key.to_owned(),
        })?;
        Ok(body.cleared)
    }

    /// `setting.reset` — returns how many overrides were cleared.
    pub fn setting_reset(&mut self) -> Result<usize, ClientError> {
        #[derive(serde::Deserialize)]
        struct Body {
            cleared: usize,
        }
        let body: Body = self.send_into(Op::SettingReset)?;
        Ok(body.cleared)
    }

    /// `bypass.set`
    pub fn bypass_set(&mut self, enabled: bool) -> Result<(), ClientError> {
        let _: serde_json::Value = self.send(Op::BypassSet { enabled })?;
        Ok(())
    }

    /// `per-app.list`
    pub fn layer_a_list(
        &mut self,
    ) -> Result<Vec<headroom_ipc::LayerASnapshot>, ClientError> {
        #[derive(serde::Deserialize)]
        struct Body {
            layer_a: Vec<headroom_ipc::LayerASnapshot>,
        }
        let body: Body = self.send_into(Op::LayerAList)?;
        Ok(body.layer_a)
    }

    /// `per-app.set`
    pub fn per_app_set(&mut self, app: &str, enabled: bool) -> Result<(), ClientError> {
        let _: serde_json::Value = self.send(Op::PerAppSet {
            app: app.to_owned(),
            enabled,
        })?;
        Ok(())
    }

    /// `per-app.master`
    pub fn per_app_master(&mut self, enabled: bool) -> Result<(), ClientError> {
        let _: serde_json::Value = self.send(Op::PerAppMaster { enabled })?;
        Ok(())
    }

    /// `per-app.reset`
    pub fn layer_a_reset(&mut self, node_id: u32) -> Result<(), ClientError> {
        let _: serde_json::Value = self.send(Op::LayerAReset { node_id })?;
        Ok(())
    }

    /// `subscribe`
    pub fn subscribe(&mut self, topics: &[Topic]) -> Result<Vec<Topic>, ClientError> {
        #[derive(serde::Deserialize)]
        struct Body {
            subscribed: Vec<Topic>,
        }
        let body: Body = self.send_into(Op::Subscribe {
            topics: topics.to_vec(),
        })?;
        Ok(body.subscribed)
    }

    /// `unsubscribe`
    pub fn unsubscribe(&mut self, topics: &[Topic]) -> Result<Vec<Topic>, ClientError> {
        #[derive(serde::Deserialize)]
        struct Body {
            unsubscribed: Vec<Topic>,
        }
        let body: Body = self.send_into(Op::Unsubscribe {
            topics: topics.to_vec(),
        })?;
        Ok(body.unsubscribed)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::io::{BufReader, BufWriter};
    use std::os::unix::net::UnixStream;
    use std::thread;

    use headroom_ipc::{Codec, Event, HelloData, Op, Request, Response, ServerFrame, Topic};

    /// A tiny in-process server that runs on the other end of a
    /// `UnixStream::pair`. Knows just enough to exercise the client.
    fn spawn_test_server() -> (UnixStream, thread::JoinHandle<()>) {
        let (a, b) = UnixStream::pair().unwrap();
        let server_handle = thread::spawn(move || {
            let codec = Codec::new();
            let read_side = b.try_clone().unwrap();
            let mut reader = BufReader::new(read_side);
            let mut writer = BufWriter::new(b);

            // Send hello.
            let hello = Event::new(
                Topic::Control,
                "hello",
                &HelloData {
                    daemon: "headroom".into(),
                    version: "0.1.0-test".into(),
                    protocol: headroom_ipc::PROTOCOL_VERSION,
                },
            )
            .unwrap();
            codec
                .write(&mut writer, &ServerFrame::Event(hello))
                .unwrap();

            // Serve one round.
            loop {
                let req: Request = match codec.read(&mut reader) {
                    Ok(r) => r,
                    Err(_) => return,
                };
                let resp = match req.op {
                    Op::Status => Response::ok(
                        req.id,
                        &serde_json::json!({
                            "version": "0.1.0-test",
                            "protocol": headroom_ipc::PROTOCOL_VERSION,
                            "uptime_s": 0u64,
                            "profile": "default",
                            "bypass": false,
                            "sinks": {
                                "processed": {"ready": false},
                                "real":      {"ready": false},
                            },
                            "streams": []
                        }),
                    )
                    .unwrap(),
                    Op::ProfileUse { name } => {
                        Response::ok(req.id, &serde_json::json!({ "name": name })).unwrap()
                    }
                    Op::Subscribe { topics } => {
                        // Acknowledge, then push one event of each
                        // subscribed topic so the client can demonstrate
                        // event handling.
                        let body = serde_json::json!({ "subscribed": &topics });
                        let resp = Response::ok(req.id, &body).unwrap();
                        codec
                            .write(&mut writer, &ServerFrame::Response(resp.clone()))
                            .unwrap();

                        for t in topics {
                            let ev = Event::new(t, "tick", &serde_json::json!({})).unwrap();
                            codec.write(&mut writer, &ServerFrame::Event(ev)).unwrap();
                        }
                        continue;
                    }
                    _ => Response::ok(req.id, &serde_json::Value::Null).unwrap(),
                };
                codec
                    .write(&mut writer, &ServerFrame::Response(resp))
                    .unwrap();
            }
        });
        (a, server_handle)
    }

    fn client_on(stream: UnixStream) -> Client {
        let reader_half = stream.try_clone().unwrap();
        let writer_half = stream;
        let mut me = Client {
            reader: BufReader::new(reader_half),
            writer: BufWriter::new(writer_half),
            codec: Codec::new(),
            next_id: 1,
            pending_events: VecDeque::new(),
            hello: HelloData {
                daemon: String::new(),
                version: String::new(),
                protocol: 0,
            },
            socket_path: PathBuf::from("<test>"),
        };
        me.handshake().unwrap();
        me
    }

    #[test]
    fn handshake_then_status() {
        let (client_sock, _server) = spawn_test_server();
        let mut client = client_on(client_sock);
        assert_eq!(client.hello().daemon, "headroom");
        assert_eq!(client.hello().protocol, headroom_ipc::PROTOCOL_VERSION);

        let status = client.status().unwrap();
        assert_eq!(status.profile, "default");
        assert!(!status.bypass);
    }

    #[test]
    fn profile_use_returns_name() {
        let (client_sock, _server) = spawn_test_server();
        let mut client = client_on(client_sock);
        let name = client.profile_use("night").unwrap();
        assert_eq!(name, "night");
    }

    #[test]
    fn subscribe_then_consume_event() {
        let (client_sock, _server) = spawn_test_server();
        let mut client = client_on(client_sock);
        let acked = client.subscribe(&[Topic::Meters]).unwrap();
        assert_eq!(acked, vec![Topic::Meters]);

        let ev = client.next_event().unwrap();
        assert_eq!(ev.topic, Topic::Meters);
        assert_eq!(ev.event, "tick");
    }

    #[test]
    fn events_interleaved_during_request_are_queued() {
        // The test server pushes events *after* the subscribe response,
        // so let's check that requesting another op afterwards drains
        // them through the queue.
        let (client_sock, _server) = spawn_test_server();
        let mut client = client_on(client_sock);
        client.subscribe(&[Topic::Meters, Topic::Profile]).unwrap();

        // Now issue another request. The server hasn't sent the events
        // until we read more, but our client will keep reading.
        let status = client.status().unwrap();
        assert_eq!(status.profile, "default");

        // We may have buffered events from the prior subscribe and from
        // any in flight; drain them.
        let mut topics = Vec::new();
        while let Some(ev) = client.pending_event() {
            topics.push(ev.topic);
        }
        // The events arrived between the subscribe-ack and the status
        // response; both should be queued.
        assert!(topics.contains(&Topic::Meters));
        assert!(topics.contains(&Topic::Profile));
    }
}

1	//! blocking [`Client`].
2
3	use std::collections::VecDeque;
4	use std::io::{BufReader, BufWriter};
5	use std::os::unix::net::UnixStream;
6	use std::path::{Path, PathBuf};
7
8	use serde::de::DeserializeOwned;
9
10	use headroom_ipc::{
11	default_socket_path, Codec, Event, HelloData, Op, ProtoError, Request, Response,
12	ResponsePayload, Route, ServerFrame, Status, Topic,
13	};
14
15	/// errors from the blocking client.
16	#[derive(Debug, thiserror::Error)]
17	pub enum ClientError {
18	#[error("ipc: {0}")]
19	Ipc(#[from] headroom_ipc::Error),
20
21	/// server's first frame was not the `hello` event.
22	#[error("expected hello event from server, got {0}")]
23	BadHello(String),
24
25	/// response id we never issued.
26	#[error("response with unknown id {0}")]
27	UnknownResponseId(u64),
28
29	/// protocol-level error for an op.
30	#[error("server error: {0}")]
31	Protocol(#[from] ProtoError),
32
33	#[error("no default socket path (XDG_RUNTIME_DIR unset and /proc/self/status unreadable)")]
34	NoDefaultPath,
35
36	/// typed-helper response failed to deserialize into the expected shape.
37	#[error("response shape mismatch: {0}")]
38	DecodeResult(serde_json::Error),
39	}
40
41	/// blocking client for the headroom control protocol.
42	///
43	/// single-threaded by construction; for concurrent request/response and event
44	/// consumption, open two connections.
45	pub struct Client {
46	reader: BufReader<UnixStream>,
47	writer: BufWriter<UnixStream>,
48	codec: Codec,
49	next_id: u64,
50	pending_events: VecDeque<Event>,
51	hello: HelloData,
52	socket_path: PathBuf,
53	}
54
55	impl Client {
56	/// connect at the default socket path.
57	pub fn connect() -> Result<Self, ClientError> {
58	let path = default_socket_path().ok_or(ClientError::NoDefaultPath)?;
59	Self::connect_at(&path)
60	}
61
62	/// connect at the given socket path.
63	pub fn connect_at(path: &Path) -> Result<Self, ClientError> {
64	let stream = UnixStream::connect(path).map_err(\|e\| ClientError::Ipc(e.into()))?;
65	let reader_half = stream.try_clone().map_err(\|e\| ClientError::Ipc(e.into()))?;
66	let writer_half = stream;
67
68	let mut me = Self {
69	reader: BufReader::new(reader_half),
70	writer: BufWriter::new(writer_half),
71	codec: Codec::new(),
72	next_id: 1,
73	pending_events: VecDeque::new(),
74	// placeholder; populated by handshake below
75	hello: HelloData {
76	daemon: String::new(),
77	version: String::new(),
78	protocol: 0,
79	},
80	socket_path: path.to_path_buf(),
81	};
82
83	me.handshake()?;
84	Ok(me)
85	}
86
87	fn handshake(&mut self) -> Result<(), ClientError> {
88	let frame: ServerFrame = self.codec.read(&mut self.reader)?;
89	match frame {
90	ServerFrame::Event(ev)
91	if ev.topic == Topic::Control && ev.event.as_str() == "hello" =>
92	{
93	let hello: HelloData =
94	serde_json::from_value(ev.data).map_err(ClientError::DecodeResult)?;
95	self.hello = hello;
96	}
97	ServerFrame::Event(ev) => {
98	return Err(ClientError::BadHello(format!(
99	"{} event on {}",
100	ev.event, ev.topic
101	)))
102	}
103	ServerFrame::Response(r) => {
104	return Err(ClientError::BadHello(format!("response id={}", r.id)))
105	}
106	}
107
108	// best effort for older daemons
109	self.send_raw(Op::Hello {
110	protocol: headroom_ipc::PROTOCOL_VERSION,
111	})?;
112	Ok(())
113	}
114
115	/// `hello` payload received on connect.
116	#[must_use]
117	pub fn hello(&self) -> &HelloData {
118	&self.hello
119	}
120
121	/// socket path this client is connected to.
122	#[must_use]
123	pub fn socket_path(&self) -> &Path {
124	&self.socket_path
125	}
126
127	fn alloc_id(&mut self) -> u64 {
128	let id = self.next_id;
129	// wrap: u64::MAX requests on one connection won't happen, but be correct
130	self.next_id = self.next_id.wrapping_add(1);
131	id
132	}
133
134	/// send a request, block until the paired response arrives.
135	///
136	/// stray events received meanwhile are queued for [`next_event`](Self::next_event).
137	pub fn send(&mut self, op: Op) -> Result<serde_json::Value, ClientError> {
138	let payload = self.send_raw(op)?;
139	match payload {
140	ResponsePayload::Ok { result } => Ok(result),
141	ResponsePayload::Err { error } => Err(ClientError::Protocol(error)),
142	}
143	}
144
145	/// like [`send`](Self::send) but returns the raw [`ResponsePayload`] —
146	/// keeps a protocol-level error in-band instead of as [`ClientError::Protocol`].
147	pub fn send_raw(&mut self, op: Op) -> Result<ResponsePayload, ClientError> {
148	let id = self.alloc_id();
149	let req = Request::new(id, op);
150	self.codec.write(&mut self.writer, &req)?;
151
152	loop {
153	let frame: ServerFrame = self.codec.read(&mut self.reader)?;
154	match frame {
155	ServerFrame::Response(Response {
156	id: rid,
157	payload: _,
158	}) if rid != id => {
159	return Err(ClientError::UnknownResponseId(rid));
160	}
161	ServerFrame::Response(Response { payload, .. }) => return Ok(payload),
162	ServerFrame::Event(ev) => {
163	self.pending_events.push_back(ev);
164	}
165	}
166	}
167	}
168
169	/// block until the next event arrives. drains the queue before reading the socket.
170	///
171	/// an unsolicited response (no matching request) is rejected as
172	/// [`ClientError::UnknownResponseId`].
173	pub fn next_event(&mut self) -> Result<Event, ClientError> {
174	if let Some(ev) = self.pending_events.pop_front() {
175	return Ok(ev);
176	}
177	match self.codec.read::<_, ServerFrame>(&mut self.reader)? {
178	ServerFrame::Event(ev) => Ok(ev),
179	ServerFrame::Response(r) => Err(ClientError::UnknownResponseId(r.id)),
180	}
181	}
182
183	/// pop a queued event without blocking. does not read from the socket.
184	pub fn pending_event(&mut self) -> Option<Event> {
185	self.pending_events.pop_front()
186	}
187
188	// typed convenience wrappers
189
190	fn send_into<T: DeserializeOwned>(&mut self, op: Op) -> Result<T, ClientError> {
191	let value = self.send(op)?;
192	serde_json::from_value(value).map_err(ClientError::DecodeResult)
193	}
194
195	/// `status`
196	pub fn status(&mut self) -> Result<Status, ClientError> {
197	self.send_into(Op::Status)
198	}
199
200	/// `profile.list`
201	pub fn profile_list(&mut self) -> Result<Vec<headroom_ipc::ProfileInfo>, ClientError> {
202	#[derive(serde::Deserialize)]
203	struct Body {
204	profiles: Vec<headroom_ipc::ProfileInfo>,
205	}
206	let body: Body = self.send_into(Op::ProfileList)?;
207	Ok(body.profiles)
208	}
209
210	/// `profile.use`
211	pub fn profile_use(&mut self, name: &str) -> Result<String, ClientError> {
212	#[derive(serde::Deserialize)]
213	struct Body {
214	name: String,
215	}
216	let body: Body = self.send_into(Op::ProfileUse {
217	name: name.to_owned(),
218	})?;
219	Ok(body.name)
220	}
221
222	/// `profile.show`
223	pub fn profile_show(
224	&mut self,
225	name: Option<&str>,
226	) -> Result<serde_json::Value, ClientError> {
227	self.send(Op::ProfileShow {
228	name: name.map(String::from),
229	})
230	}
231
232	/// `profile.reload`
233	pub fn profile_reload(&mut self) -> Result<Vec<String>, ClientError> {
234	#[derive(serde::Deserialize)]
235	struct Body {
236	reloaded: Vec<String>,
237	}
238	let body: Body = self.send_into(Op::ProfileReload)?;
239	Ok(body.reloaded)
240	}
241
242	/// `route.list`
243	pub fn route_list(&mut self) -> Result<headroom_ipc::RouteList, ClientError> {
244	self.send_into(Op::RouteList)
245	}
246
247	/// `route.set`
248	pub fn route_set(&mut self, app: &str, to: Route) -> Result<(), ClientError> {
249	let _: serde_json::Value = self.send(Op::RouteSet {
250	app: app.to_owned(),
251	to,
252	})?;
253	Ok(())
254	}
255
256	/// `route.unset`
257	pub fn route_unset(&mut self, app: &str) -> Result<(), ClientError> {
258	let _: serde_json::Value = self.send(Op::RouteUnset {
259	app: app.to_owned(),
260	})?;
261	Ok(())
262	}
263
264	/// `route.stream`
265	pub fn route_stream(&mut self, node_id: u32, to: Route) -> Result<(), ClientError> {
266	let _: serde_json::Value = self.send(Op::RouteStream { node_id, to })?;
267	Ok(())
268	}
269
270	/// `setting.get`
271	pub fn setting_get(&mut self, key: &str) -> Result<serde_json::Value, ClientError> {
272	#[derive(serde::Deserialize)]
273	struct Body {
274	#[allow(dead_code)]
275	key: String,
276	value: serde_json::Value,
277	}
278	let body: Body = self.send_into(Op::SettingGet {
279	key: key.to_owned(),
280	})?;
281	Ok(body.value)
282	}
283
284	/// `setting.set`
285	pub fn setting_set(
286	&mut self,
287	key: &str,
288	value: serde_json::Value,
289	) -> Result<(), ClientError> {
290	let _: serde_json::Value = self.send(Op::SettingSet {
291	key: key.to_owned(),
292	value,
293	})?;
294	Ok(())
295	}
296
297	/// `setting.clear` — returns whether an override was present.
298	pub fn setting_clear(&mut self, key: &str) -> Result<bool, ClientError> {
299	#[derive(serde::Deserialize)]
300	struct Body {
301	cleared: bool,
302	}
303	let body: Body = self.send_into(Op::SettingClear {
304	key: key.to_owned(),
305	})?;
306	Ok(body.cleared)
307	}
308
309	/// `setting.reset` — returns how many overrides were cleared.
310	pub fn setting_reset(&mut self) -> Result<usize, ClientError> {
311	#[derive(serde::Deserialize)]
312	struct Body {
313	cleared: usize,
314	}
315	let body: Body = self.send_into(Op::SettingReset)?;
316	Ok(body.cleared)
317	}
318
319	/// `bypass.set`
320	pub fn bypass_set(&mut self, enabled: bool) -> Result<(), ClientError> {
321	let _: serde_json::Value = self.send(Op::BypassSet { enabled })?;
322	Ok(())
323	}
324
325	/// `per-app.list`
326	pub fn layer_a_list(
327	&mut self,
328	) -> Result<Vec<headroom_ipc::LayerASnapshot>, ClientError> {
329	#[derive(serde::Deserialize)]
330	struct Body {
331	layer_a: Vec<headroom_ipc::LayerASnapshot>,
332	}
333	let body: Body = self.send_into(Op::LayerAList)?;
334	Ok(body.layer_a)
335	}
336
337	/// `per-app.set`
338	pub fn per_app_set(&mut self, app: &str, enabled: bool) -> Result<(), ClientError> {
339	let _: serde_json::Value = self.send(Op::PerAppSet {
340	app: app.to_owned(),
341	enabled,
342	})?;
343	Ok(())
344	}
345
346	/// `per-app.master`
347	pub fn per_app_master(&mut self, enabled: bool) -> Result<(), ClientError> {
348	let _: serde_json::Value = self.send(Op::PerAppMaster { enabled })?;
349	Ok(())
350	}
351
352	/// `per-app.reset`
353	pub fn layer_a_reset(&mut self, node_id: u32) -> Result<(), ClientError> {
354	let _: serde_json::Value = self.send(Op::LayerAReset { node_id })?;
355	Ok(())
356	}
357
358	/// `subscribe`
359	pub fn subscribe(&mut self, topics: &[Topic]) -> Result<Vec<Topic>, ClientError> {
360	#[derive(serde::Deserialize)]
361	struct Body {
362	subscribed: Vec<Topic>,
363	}
364	let body: Body = self.send_into(Op::Subscribe {
365	topics: topics.to_vec(),
366	})?;
367	Ok(body.subscribed)
368	}
369
370	/// `unsubscribe`
371	pub fn unsubscribe(&mut self, topics: &[Topic]) -> Result<Vec<Topic>, ClientError> {
372	#[derive(serde::Deserialize)]
373	struct Body {
374	unsubscribed: Vec<Topic>,
375	}
376	let body: Body = self.send_into(Op::Unsubscribe {
377	topics: topics.to_vec(),
378	})?;
379	Ok(body.unsubscribed)
380	}
381	}
382
383	#[cfg(test)]
384	mod tests {
385	use super::*;
386	use std::io::{BufReader, BufWriter};
387	use std::os::unix::net::UnixStream;
388	use std::thread;
389
390	use headroom_ipc::{Codec, Event, HelloData, Op, Request, Response, ServerFrame, Topic};
391
392	/// A tiny in-process server that runs on the other end of a
393	/// `UnixStream::pair`. Knows just enough to exercise the client.
394	fn spawn_test_server() -> (UnixStream, thread::JoinHandle<()>) {
395	let (a, b) = UnixStream::pair().unwrap();
396	let server_handle = thread::spawn(move \|\| {
397	let codec = Codec::new();
398	let read_side = b.try_clone().unwrap();
399	let mut reader = BufReader::new(read_side);
400	let mut writer = BufWriter::new(b);
401
402	// Send hello.
403	let hello = Event::new(
404	Topic::Control,
405	"hello",
406	&HelloData {
407	daemon: "headroom".into(),
408	version: "0.1.0-test".into(),
409	protocol: headroom_ipc::PROTOCOL_VERSION,
410	},
411	)
412	.unwrap();
413	codec
414	.write(&mut writer, &ServerFrame::Event(hello))
415	.unwrap();
416
417	// Serve one round.
418	loop {
419	let req: Request = match codec.read(&mut reader) {
420	Ok(r) => r,
421	Err(_) => return,
422	};
423	let resp = match req.op {
424	Op::Status => Response::ok(
425	req.id,
426	&serde_json::json!({
427	"version": "0.1.0-test",
428	"protocol": headroom_ipc::PROTOCOL_VERSION,
429	"uptime_s": 0u64,
430	"profile": "default",
431	"bypass": false,
432	"sinks": {
433	"processed": {"ready": false},
434	"real": {"ready": false},
435	},
436	"streams": []
437	}),
438	)
439	.unwrap(),
440	Op::ProfileUse { name } => {
441	Response::ok(req.id, &serde_json::json!({ "name": name })).unwrap()
442	}
443	Op::Subscribe { topics } => {
444	// Acknowledge, then push one event of each
445	// subscribed topic so the client can demonstrate
446	// event handling.
447	let body = serde_json::json!({ "subscribed": &topics });
448	let resp = Response::ok(req.id, &body).unwrap();
449	codec
450	.write(&mut writer, &ServerFrame::Response(resp.clone()))
451	.unwrap();
452
453	for t in topics {
454	let ev = Event::new(t, "tick", &serde_json::json!({})).unwrap();
455	codec.write(&mut writer, &ServerFrame::Event(ev)).unwrap();
456	}
457	continue;
458	}
459	_ => Response::ok(req.id, &serde_json::Value::Null).unwrap(),
460	};
461	codec
462	.write(&mut writer, &ServerFrame::Response(resp))
463	.unwrap();
464	}
465	});
466	(a, server_handle)
467	}
468
469	fn client_on(stream: UnixStream) -> Client {
470	let reader_half = stream.try_clone().unwrap();
471	let writer_half = stream;
472	let mut me = Client {
473	reader: BufReader::new(reader_half),
474	writer: BufWriter::new(writer_half),
475	codec: Codec::new(),
476	next_id: 1,
477	pending_events: VecDeque::new(),
478	hello: HelloData {
479	daemon: String::new(),
480	version: String::new(),
481	protocol: 0,
482	},
483	socket_path: PathBuf::from("<test>"),
484	};
485	me.handshake().unwrap();
486	me
487	}
488
489	#[test]
490	fn handshake_then_status() {
491	let (client_sock, _server) = spawn_test_server();
492	let mut client = client_on(client_sock);
493	assert_eq!(client.hello().daemon, "headroom");
494	assert_eq!(client.hello().protocol, headroom_ipc::PROTOCOL_VERSION);
495
496	let status = client.status().unwrap();
497	assert_eq!(status.profile, "default");
498	assert!(!status.bypass);
499	}
500
501	#[test]
502	fn profile_use_returns_name() {
503	let (client_sock, _server) = spawn_test_server();
504	let mut client = client_on(client_sock);
505	let name = client.profile_use("night").unwrap();
506	assert_eq!(name, "night");
507	}
508
509	#[test]
510	fn subscribe_then_consume_event() {
511	let (client_sock, _server) = spawn_test_server();
512	let mut client = client_on(client_sock);
513	let acked = client.subscribe(&[Topic::Meters]).unwrap();
514	assert_eq!(acked, vec![Topic::Meters]);
515
516	let ev = client.next_event().unwrap();
517	assert_eq!(ev.topic, Topic::Meters);
518	assert_eq!(ev.event, "tick");
519	}
520
521	#[test]
522	fn events_interleaved_during_request_are_queued() {
523	// The test server pushes events after the subscribe response,
524	// so let's check that requesting another op afterwards drains
525	// them through the queue.
526	let (client_sock, _server) = spawn_test_server();
527	let mut client = client_on(client_sock);
528	client.subscribe(&[Topic::Meters, Topic::Profile]).unwrap();
529
530	// Now issue another request. The server hasn't sent the events
531	// until we read more, but our client will keep reading.
532	let status = client.status().unwrap();
533	assert_eq!(status.profile, "default");
534
535	// We may have buffered events from the prior subscribe and from
536	// any in flight; drain them.
537	let mut topics = Vec::new();
538	while let Some(ev) = client.pending_event() {
539	topics.push(ev.topic);
540	}
541	// The events arrived between the subscribe-ack and the status
542	// response; both should be queued.
543	assert!(topics.contains(&Topic::Meters));
544	assert!(topics.contains(&Topic::Profile));
545	}
546	}