request/calloop: Improved handling of disconnect from server or client

.github/workflows/ci.yaml

@@ -8,7 +8,7 @@ jobs:
   check-and-build:
     strategy:
       matrix:
-        rust_version: ['1.70.0', stable]
+        rust_version: ['1.71.0', stable]
     runs-on: ubuntu-latest
     outputs:
       libei_version: ${{ steps.libei_version.outputs.libei_version }}
@@ -21,12 +21,6 @@ jobs:
     - run: sudo apt-get -qq update
     - run: sudo apt-get install -y libxkbcommon-dev
 
-    - run: |
-        cargo update -p tokio --precise 1.47.0
-        cargo update -p mio --precise 1.0.4
-        cargo update -p polling --precise 3.10.0
-      if: matrix.rust_version != 'stable'
-
     - run: cargo fmt --all -- --check
 
     # Make sure there are no broken intra-doc links with and without features

Cargo.toml

@@ -1,12 +1,12 @@
 [package]
 name = "reis"
-version = "0.5.0"
+version = "0.6.0"
 license = "MIT"
 description = "Pure Rust implementation of libei/libeis protocol."
 repository = "https://github.com/ids1024/reis"
 keywords = ["libei", "libeis", "wayland"]
 edition = "2021"
-rust-version = "1.70.0"
+rust-version = "1.71.0"
 
 [dependencies]
 calloop = { version = "0.14.0", optional = true }
@@ -39,6 +39,10 @@ string_to_string = "warn"
 cast_possible_wrap = { level = "allow", priority = 1 }
 cast_possible_truncation = { level = "allow", priority = 1 }
 
+[package.metadata.docs.rs]
+all-features = true
+rustdoc-args = ["--cfg", "docsrs"]
+
 [features]
 tokio = ["dep:tokio", "futures"]
 # Experimental and somewhat incomplete

src/calloop.rs

@@ -86,6 +86,19 @@ 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_sent_disconnected() {
+            // TODO express if server or client requested disconnect?
+            handle_result(
+                Ok(EisRequestSourceEvent::Request(
+                    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);
@@ -113,11 +126,15 @@ impl ConnectedContextState {
                 return Ok(calloop::PostAction::Remove);
             }
             while let Some(request) = self.request_converter.next_request() {
+                let disconnected = matches!(request, request::EisRequest::Disconnect);
                 let res = handle_result(
                     Ok(EisRequestSourceEvent::Request(request)),
                     &mut self.handle,
                     &mut cb,
                 )?;
+                if disconnected {
+                    return Ok(calloop::PostAction::Remove);
+                }
                 if res != calloop::PostAction::Continue {
                     return Ok(res);
                 }

src/lib.rs

@@ -15,6 +15,7 @@
 //!   incomplete.
 
 #![forbid(unsafe_code)]
+#![cfg_attr(docsrs, feature(doc_cfg))]
 
 // TODO split up
 
@@ -34,6 +35,8 @@ pub use object::Object;
 mod util;
 mod wire;
 
+pub use enumflags2;
+
 pub use wire::Interface;
 pub use wire::ParseError;
 

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 {