request: Shutdown socket when server sends disconnected

This allows the reading thread to cleanly exit without waiting for the
client.

Author: ids1024

src/calloop.rs

@@ -86,6 +86,17 @@ impl ConnectedContextState {
     where
         F: FnMut(Result<EisRequestSourceEvent, Error>, &mut Connection) -> io::Result<PostAction>,
     {
+        // If server has sent `disconected`, return `Disconnect` event and stop polling.
+        if self.handle.has_send_disconnected() {
+            // TODO express if server or client requested disconnect?
+            handle_result(
+                Ok(request::EisRequest::Disconnect),
+                &mut self.handle,
+                &mut cb,
+            )?;
+            return Ok(calloop::PostAction::Remove);
+        }
+
         if let Err(err) = self.context.read() {
             handle_result(Err(Error::Io(err)), &mut self.handle, &mut cb)?;
             return Ok(calloop::PostAction::Remove);

src/request.rs

@@ -13,7 +13,10 @@ use crate::{
 use std::{
     collections::{HashMap, HashSet, VecDeque},
     fmt,
-    sync::{Arc, Mutex, Weak},
+    sync::{
+        atomic::{AtomicBool, Ordering},
+        Arc, Mutex, Weak,
+    },
 };
 
 pub use crate::event::DeviceCapability;
@@ -49,6 +52,7 @@ struct ConnectionInner {
     devices: Mutex<HashMap<eis::Device, Device>>,
     device_for_interface: Mutex<HashMap<Object, Device>>,
     last_serial: Mutex<u32>,
+    disconnected: AtomicBool,
 }
 
 /// High-level server-side wrapper for `ei_connection`.
@@ -89,6 +93,18 @@ impl Connection {
         }
         self.connection()
             .disconnected(self.last_serial(), reason, explanation);
+        // If flush fails because buffer is full, client can just get an EOF without
+        // a message.
+        let _ = self.flush();
+        self.0.disconnected.store(true, Ordering::SeqCst);
+        // Shutdown read end of socket, so anything reading/polling it will get EOF,
+        // without waiting for client to disconnect first.
+        self.0.context.0.shutdown_read();
+    }
+
+    #[cfg(feature = "calloop")]
+    pub(crate) fn has_sent_disconnected(&self) -> bool {
+        self.0.disconnected.load(Ordering::SeqCst)
     }
 
     /// Sends buffered messages. Call after you're finished with sending events.
@@ -247,6 +263,7 @@ impl EisRequestConverter {
                 devices: Mutex::default(),
                 device_for_interface: Mutex::default(),
                 last_serial: Mutex::new(initial_serial),
+                disconnected: AtomicBool::new(false),
             })),
         }
     }

src/wire/backend.rs

@@ -326,6 +326,11 @@ impl Backend {
     pub fn flush(&self) -> rustix::io::Result<()> {
         self.0.write.lock().unwrap().flush_write(&self.0.socket)
     }
+
+    /// Shutdown read end of socket, so all future reads will return EOF
+    pub(crate) fn shutdown_read(&self) {
+        let _ = self.0.socket.shutdown(std::net::Shutdown::Read);
+    }
 }
 
 fn is_reis_debug() -> bool {