crabshell_services/
sway.rs

1//! Sway's workspaces, over Sway's IPC socket.
2
3mod ipc;
4#[cfg(test)]
5mod tests;
6
7use std::cell::RefCell;
8use std::collections::VecDeque;
9use std::fmt::Display;
10use std::io::{ErrorKind, Read, Write};
11use std::os::fd::AsFd;
12use std::os::unix::net::UnixStream;
13use std::path::PathBuf;
14use std::rc::Rc;
15use std::sync::Once;
16use std::time::{Duration, Instant};
17
18use calloop::timer::TimeoutAction;
19use calloop::{Interest, RegistrationToken};
20use crabshell_reactive::{ReadSignal, Signal};
21use crabshell_ui::Cx;
22use serde::Deserialize;
23
24use crate::Shared;
25
26/// How long the service waits before each reconnect.
27pub(crate) const RETRY: Duration = Duration::from_secs(2);
28
29/// One of Sway's workspaces, as `GET_WORKSPACES` describes it.
30#[derive(Clone, Debug, PartialEq, Eq, Deserialize)]
31pub struct Workspace {
32  /// Sway's id for it, which stays the same while it exists: the key to
33  /// give `children_keyed`.
34  pub id: i64,
35  /// Its name, such as `1` or `3:web`.
36  pub name: String,
37  /// The output it is on, such as `DP-1`.
38  pub output: String,
39  /// Whether it has the keyboard focus.
40  pub focused: bool,
41  /// Whether it is shown on its output.
42  pub visible: bool,
43  /// Whether a window on it asks for attention.
44  pub urgent: bool,
45}
46
47/// Where Sway listens.
48pub(crate) struct Socket {
49  pub(crate) path: PathBuf,
50  /// How long to wait before each reconnect: [`RETRY`], less in tests.
51  pub(crate) retry: Duration,
52}
53
54/// The Sway service, from `cx.sway()`. A clone is cheap and shares the
55/// one connection.
56#[derive(Clone)]
57pub struct Sway {
58  /// `None` when the shell provided no services.
59  service: Option<Rc<Service>>,
60  workspaces: ReadSignal<Vec<Workspace>>,
61}
62
63impl Sway {
64  /// The service if a config has started it with `cx.sway()`, without
65  /// starting it, so the shell can read [`changed_at`](Self::changed_at)
66  /// and a bar that shows no workspaces never connects.
67  #[doc(hidden)]
68  pub fn started(cx: &Cx) -> Option<Sway> {
69    let service = cx.service::<Service>()?;
70    let workspaces = service.state.borrow().as_ref()?.workspaces.read_only();
71    Some(Sway {
72      service: Some(service),
73      workspaces,
74    })
75  }
76
77  /// Every workspace on every output, in Sway's order. Empty while Sway
78  /// isn't connected.
79  pub fn workspaces(&self) -> ReadSignal<Vec<Workspace>> {
80    self.workspaces
81  }
82
83  /// When the workspace event that led to the current list arrived, so
84  /// the shell can log how long a switch took to reach the screen.
85  /// `None` for a list that followed no event, such as the first.
86  pub fn changed_at(&self) -> Option<Instant> {
87    let state = self.service.as_ref()?.state.borrow();
88    state.as_ref()?.changed_at
89  }
90
91  /// Switches to the workspace called `name`, as clicking it in swaybar
92  /// does. A name Sway reads as a keyword, such as `next`, is logged and
93  /// not sent, and so is a call while Sway isn't connected. Sway's
94  /// refusal is logged too.
95  pub fn focus_workspace(&self, name: &str) {
96    let Some(command) = ipc::workspace_command(name) else {
97      tracing::warn!(name, "Sway can't switch to this workspace by name");
98      return;
99    };
100    let service = self.service.as_ref();
101    if !service.is_some_and(|service| service.send(ipc::RUN_COMMAND, command.as_bytes())) {
102      tracing::warn!(name, "can't switch workspaces while Sway isn't connected");
103    }
104  }
105}
106
107/// `cx.sway()`.
108pub(crate) fn sway(cx: &Cx) -> Sway {
109  let Some(service) = cx.service::<Service>() else {
110    static LOGGED: Once = Once::new();
111    LOGGED.call_once(|| {
112      tracing::warn!("cx.sway() needs the services the shell provides; it shows no workspaces");
113    });
114    return Sway {
115      service: None,
116      workspaces: cx.signal(Vec::new()).read_only(),
117    };
118  };
119  let workspaces = service.start();
120  Sway {
121    service: Some(service),
122    workspaces,
123  }
124}
125
126/// The one Sway connection of a shell, started on first use.
127pub(crate) struct Service {
128  shared: Rc<Shared>,
129  state: RefCell<Option<State>>,
130}
131
132struct State {
133  workspaces: Signal<Vec<Workspace>>,
134  changed_at: Option<Instant>,
135  connection: Option<Connection>,
136  /// Whether the current outage was logged, so retries stay quiet.
137  outage_logged: bool,
138}
139
140struct Connection {
141  stream: UnixStream,
142  /// The loop source that waits for something to read.
143  reader: RegistrationToken,
144  /// The loop source that waits for room to write, while the socket
145  /// holds back part of `outbox`.
146  writer: Option<RegistrationToken>,
147  /// Bytes read that don't make a whole message yet.
148  inbox: Vec<u8>,
149  /// Bytes not written yet.
150  outbox: Vec<u8>,
151  /// One entry per `GET_WORKSPACES` sent and not yet answered, in order:
152  /// when the event behind it arrived, or `None` for the first.
153  fetches: VecDeque<Option<Instant>>,
154}
155
156/// Sway's answer to a `SUBSCRIBE` or to one command.
157#[derive(Deserialize)]
158struct Outcome {
159  success: bool,
160  /// Why not, when `success` is false.
161  #[serde(default)]
162  error: String,
163}
164
165impl Service {
166  pub(crate) fn new(shared: Rc<Shared>) -> Service {
167    Service {
168      shared,
169      state: RefCell::new(None),
170    }
171  }
172
173  /// The list, connecting to Sway if this is the first ask.
174  fn start(self: &Rc<Self>) -> ReadSignal<Vec<Workspace>> {
175    if let Some(state) = &*self.state.borrow() {
176      return state.workspaces.read_only();
177    }
178    let workspaces = self.shared.root.signal(Vec::new());
179    *self.state.borrow_mut() = Some(State {
180      workspaces,
181      changed_at: None,
182      connection: None,
183      outage_logged: false,
184    });
185    if self.shared.config.sway.is_none() {
186      tracing::info!("SWAYSOCK is unset, so the bar shows no workspaces");
187    } else {
188      // A failed first connect isn't retried: SWAYSOCK is stale, or
189      // another compositor runs, and retries would wake the shell.
190      self.connect();
191    }
192    workspaces.read_only()
193  }
194
195  /// Connects and asks for the list and its changes. `false` if Sway
196  /// isn't there, logged unless a drop already was.
197  fn connect(self: &Rc<Self>) -> bool {
198    let (Some(socket), Some(event_loop)) = (&self.shared.config.sway, self.shared.event_loop())
199    else {
200      return false;
201    };
202    let opened = UnixStream::connect(&socket.path).and_then(|stream| {
203      stream.set_nonblocking(true)?;
204      let fd = stream.as_fd().try_clone_to_owned()?;
205      Ok((stream, fd))
206    });
207    let mut state = self.state.borrow_mut();
208    let state = state
209      .as_mut()
210      .expect("invariant: started before connecting");
211    let (stream, fd) = match opened {
212      Ok(opened) => opened,
213      Err(error) => {
214        if !state.outage_logged {
215          let path = socket.path.display();
216          tracing::warn!(
217              %path,
218              %error,
219              "can't connect to Sway, so the bar shows no workspaces"
220          );
221        }
222        return false;
223      }
224    };
225    let service = Rc::clone(self);
226    let read = Box::new(move || service.on_readable());
227    let Some(reader) = event_loop.insert_fd(fd, Interest::READ, read) else {
228      return false;
229    };
230    if std::mem::take(&mut state.outage_logged) {
231      tracing::info!("connected to Sway again");
232    }
233    let mut connection = Connection {
234      stream,
235      reader,
236      writer: None,
237      inbox: Vec::new(),
238      outbox: Vec::new(),
239      fetches: VecDeque::from([None]),
240    };
241    // Subscribe first, so no change between the two goes unseen.
242    ipc::encode(ipc::SUBSCRIBE, br#"["workspace"]"#, &mut connection.outbox);
243    ipc::encode(ipc::GET_WORKSPACES, b"", &mut connection.outbox);
244    state.connection = Some(connection);
245    self.flush_with(&mut *state);
246    true
247  }
248
249  /// Tries to connect every `retry` until it works or the list is gone.
250  fn retry_later(self: &Rc<Self>) {
251    let (Some(socket), Some(event_loop)) = (&self.shared.config.sway, self.shared.event_loop())
252    else {
253      return;
254    };
255    let retry = socket.retry;
256    let service = Rc::clone(self);
257    event_loop.insert_timer(
258      Instant::now() + retry,
259      Box::new(move || {
260        if !service.alive() || service.connect() {
261          TimeoutAction::Drop
262        } else {
263          TimeoutAction::ToDuration(retry)
264        }
265      }),
266    );
267  }
268
269  /// Whether the list is still there: its owner, the root, isn't disposed.
270  fn alive(&self) -> bool {
271    let state = self.state.borrow();
272    let alive = state
273      .as_ref()
274      .and_then(|state| state.workspaces.try_with(|_| ()));
275    alive.is_some()
276  }
277
278  /// Reads what Sway sent and acts on each whole message.
279  fn on_readable(self: &Rc<Self>) {
280    if !self.alive() {
281      return self.close();
282    }
283    let mut chunk = [0; 64 * 1024];
284    let mut state = self.state.borrow_mut();
285    let state = state
286      .as_mut()
287      .expect("invariant: started before connecting");
288    let Some(connection) = state.connection.as_mut() else {
289      return;
290    };
291    let broken = match connection.stream.read(&mut chunk) {
292      Ok(0) => Some("Sway closed the socket".to_owned()),
293      Ok(read) => {
294        connection.inbox.extend_from_slice(&chunk[..read]);
295        connection
296          .take_messages(state.workspaces, &mut state.changed_at)
297          .err()
298      }
299      Err(error) if matches!(error.kind(), ErrorKind::WouldBlock | ErrorKind::Interrupted) => None,
300      Err(error) => Some(error.to_string()),
301    };
302    match broken {
303      None => self.flush_with(state),
304      Some(why) => self.disconnect(state, &why),
305    }
306  }
307
308  /// Writes what the socket takes, and waits for room for the rest.
309  fn flush_with(self: &Rc<Self>, state: &mut State) {
310    let (Some(connection), Some(event_loop)) =
311      (state.connection.as_mut(), self.shared.event_loop())
312    else {
313      return;
314    };
315    while !connection.outbox.is_empty() {
316      match connection.stream.write(&connection.outbox) {
317        Ok(written) => {
318          connection.outbox.drain(..written);
319        }
320        Err(error) if error.kind() == ErrorKind::Interrupted => {}
321        Err(error) if error.kind() == ErrorKind::WouldBlock => break,
322        Err(error) => return self.disconnect(state, &error),
323      }
324    }
325    if connection.outbox.len() > ipc::MAX_PAYLOAD {
326      // Each event queues a fetch, so a Sway that never reads would
327      // grow the outbox and `fetches` without end.
328      return self.disconnect(state, &"Sway reads nothing the bar sends");
329    }
330    if connection.outbox.is_empty() {
331      if let Some(writer) = connection.writer.take() {
332        event_loop.remove(writer);
333      }
334    } else if connection.writer.is_none() {
335      let fd = match connection.stream.as_fd().try_clone_to_owned() {
336        Ok(fd) => fd,
337        Err(error) => return self.disconnect(state, &error),
338      };
339      let service = Rc::clone(self);
340      let write = Box::new(move || service.flush());
341      connection.writer = event_loop.insert_fd(fd, Interest::WRITE, write);
342    }
343  }
344
345  /// Sends one message. `false` while Sway isn't connected.
346  fn send(self: &Rc<Self>, kind: u32, payload: &[u8]) -> bool {
347    let mut state = self.state.borrow_mut();
348    let Some(state) = state.as_mut() else {
349      return false;
350    };
351    let Some(connection) = state.connection.as_mut() else {
352      return false;
353    };
354    ipc::encode(kind, payload, &mut connection.outbox);
355    self.flush_with(state);
356    true
357  }
358
359  fn flush(self: &Rc<Self>) {
360    let mut state = self.state.borrow_mut();
361    if let Some(state) = state.as_mut() {
362      self.flush_with(state);
363    }
364  }
365
366  /// Drops the connection, empties the list and reconnects later.
367  fn disconnect(self: &Rc<Self>, state: &mut State, why: &dyn Display) {
368    tracing::warn!(%why, "lost the connection to Sway; reconnecting");
369    state.outage_logged = true;
370    self.drop_connection(state);
371    state.changed_at = None;
372    if state.workspaces.try_with(Vec::is_empty) == Some(false) {
373      state.workspaces.set(Vec::new());
374    }
375    self.retry_later();
376  }
377
378  /// Closes the connection for good, once the list is gone.
379  fn close(&self) {
380    if let Some(state) = self.state.borrow_mut().as_mut() {
381      self.drop_connection(state);
382    }
383  }
384
385  fn drop_connection(&self, state: &mut State) {
386    let connection = state.connection.take();
387    if let (Some(connection), Some(event_loop)) = (connection, self.shared.event_loop()) {
388      event_loop.remove(connection.reader);
389      if let Some(writer) = connection.writer {
390        event_loop.remove(writer);
391      }
392    }
393  }
394}
395
396impl Connection {
397  /// Acts on each whole message in the inbox. An error means the stream
398  /// can't be read any further.
399  fn take_messages(
400    &mut self,
401    workspaces: Signal<Vec<Workspace>>,
402    changed_at: &mut Option<Instant>,
403  ) -> Result<(), String> {
404    let now = Instant::now();
405    while let Some((kind, payload)) = ipc::decode(&mut self.inbox)? {
406      match kind {
407        ipc::GET_WORKSPACES => {
408          let event = self.fetches.pop_front().flatten();
409          match serde_json::from_slice::<Vec<Workspace>>(&payload) {
410            Ok(list) if workspaces.with_untracked(|old| *old != list) => {
411              workspaces.set(list);
412              *changed_at = event;
413            }
414            Ok(_) => {}
415            Err(error) => tracing::warn!(%error, "a bad workspace list from Sway"),
416          }
417        }
418        ipc::WORKSPACE_EVENT => {
419          self.fetches.push_back(Some(now));
420          ipc::encode(ipc::GET_WORKSPACES, b"", &mut self.outbox);
421        }
422        ipc::SUBSCRIBE => match serde_json::from_slice::<Outcome>(&payload) {
423          Ok(Outcome { success: true, .. }) => {}
424          Ok(Outcome { error, .. }) => {
425            tracing::warn!(
426                %error,
427                "Sway refused the subscription; the list won't update"
428            );
429          }
430          Err(error) => tracing::warn!(%error, "a bad subscribe reply from Sway"),
431        },
432        ipc::RUN_COMMAND => match serde_json::from_slice::<Vec<Outcome>>(&payload) {
433          Ok(outcomes) => {
434            for Outcome { error, .. } in outcomes.into_iter().filter(|outcome| !outcome.success) {
435              tracing::warn!(%error, "Sway refused to switch workspaces");
436            }
437          }
438          Err(error) => tracing::warn!(%error, "a bad command reply from Sway"),
439        },
440        kind => tracing::debug!(kind, "an unexpected message from Sway"),
441      }
442    }
443    Ok(())
444  }
445}
crabshell_services/sway.rs

crabshell_services/
sway.rs
