oneshot connections

niri-ipc/src/socket.rs

@@ -55,7 +55,7 @@ impl NiriSocket {
     /// Ok(Ok([Response](crate::Response))) corresponds to a successful response from the running
     /// niri instance. Ok(Err([String])) corresponds to an error received from the running niri
     /// instance. Err([std::io::Error]) corresponds to an error in the IPC communication.
-    pub fn send(&mut self, request: Request) -> io::Result<Reply> {
+    pub fn send(mut self, request: Request) -> io::Result<Reply> {
         let mut buf = serde_json::to_vec(&request).unwrap();
         writeln!(buf).unwrap();
         self.stream.write_all(&buf)?; // .context("error writing IPC request")?;

src/ipc/client.rs

@@ -6,7 +6,7 @@ use crate::cli::Msg;
 use crate::utils::version;
 
 pub fn handle_msg(msg: Msg, json: bool) -> anyhow::Result<()> {
-    let mut client = NiriSocket::new()
+    let client = NiriSocket::new()
         .context("a communication error occured while trying to initialize the socket")?;
 
     // Default SIGPIPE so that our prints don't panic on stdout closing.
@@ -38,10 +38,7 @@ pub fn handle_msg(msg: Msg, json: bool) -> anyhow::Result<()> {
                 eprintln!("Note: unable to get the compositor's version.");
                 eprintln!("Did you forget to restart niri after an update?");
             } else {
-                // We're making a new client here just for some vague notion of
-                // backwards compatibility.
-                // It is in general not necessary to do so.
-                match NiriSocket::new().and_then(|mut client| client.send(Request::Version)) {
+                match NiriSocket::new().and_then(|client| client.send(Request::Version)) {
                     Ok(Ok(Response::Version(server_version))) => {
                         let my_version = version();
                         if my_version != server_version {

src/ipc/server.rs

@@ -115,26 +115,29 @@ async fn handle_client(ctx: ClientCtx, stream: Async<'_, UnixStream>) -> anyhow:
 
     let mut lines = BufReader::new(read).lines();
 
-    while let Some(line) = lines.next().await {
-        let reply: Reply = serde_json::from_str(&match line {
-            Ok(line) => line,
-            // ConnectionReset is expected when the client disconnects
-            Err(err) if err.kind() == io::ErrorKind::ConnectionReset => break,
-            Err(err) => return Err(err).context("error reading line"),
-        })
+    let line = match lines.next().await.unwrap_or(Err(io::Error::new(io::ErrorKind::UnexpectedEof, "Unreachable; BufReader returned None but when the stream ends, the connection should be reset"))) {
+        Ok(line) => line,
+        Err(err) if err.kind() == io::ErrorKind::ConnectionReset => return Ok(()),
+        Err(err) => return Err(err).context("error reading line"),
+    };
+
+    let reply: Reply = serde_json::from_str(&line)
         .map_err(|err| format!("error parsing request: {err}"))
         .and_then(|req| process(&ctx, req));
 
-        if let Err(err) = &reply {
-            warn!("error processing IPC request: {err:?}");
-        }
-
-        let mut buf = serde_json::to_vec(&reply).context("error formatting reply")?;
-        writeln!(buf).unwrap();
-        write.write_all(&buf).await.context("error writing reply")?;
-        write.flush().await.context("error flushing reply")?;
+    if let Err(err) = &reply {
+        warn!("error processing IPC request: {err:?}");
     }
 
+    let mut buf = serde_json::to_vec(&reply).context("error formatting reply")?;
+    writeln!(buf).unwrap();
+    write.write_all(&buf).await.context("error writing reply")?;
+    write.flush().await.context("error flushing reply")?;
+
+    // We do not check for more lines at this moment.
+    // Dropping the stream will reset the connection before we read them.
+    // For now, a client should not be sending more than one request per connection.
+
     Ok(())
 }