Refactor device discovery

rkvm-input/src/monitor.rs

@@ -2,6 +2,7 @@ use crate::interceptor::{Interceptor, OpenError};
 
 use futures::StreamExt;
 use inotify::{Inotify, WatchMask};
+use std::ffi::OsStr;
 use std::io::{Error, ErrorKind};
 use std::path::Path;
 use tokio::fs;
@@ -16,19 +17,10 @@ pub struct Monitor {
 impl Monitor {
     pub fn new() -> Self {
         let (sender, receiver) = mpsc::channel(1);
+
         tokio::spawn(async move {
             let run = async {
                 let mut read_dir = fs::read_dir(EVENT_PATH).await?;
-                while let Some(entry) = read_dir.next_entry().await? {
-                    let interceptor = match open(&entry.path()).await? {
-                        Some(interceptor) => interceptor,
-                        None => continue,
-                    };
-
-                    if sender.send(Ok(interceptor)).await.is_err() {
-                        return Ok(());
-                    }
-                }
 
                 let mut inotify = Inotify::init()?;
                 inotify.add_watch(EVENT_PATH, WatchMask::CREATE)?;
@@ -36,18 +28,40 @@ impl Monitor {
                 // This buffer size should be OK, since we don't expect a lot of devices
                 // to be plugged in frequently.
                 let mut stream = inotify.event_stream([0; 512])?;
-                while let Some(event) = stream.next().await {
-                    let event = event?;
 
-                    if let Some(name) = event.name {
-                        let interceptor = match open(&Path::new(EVENT_PATH).join(&name)).await? {
-                            Some(interceptor) => interceptor,
-                            None => continue,
-                        };
+                loop {
+                    let path = match read_dir.next_entry().await? {
+                        Some(entry) => entry.path(),
+                        None => match stream.next().await {
+                            Some(event) => {
+                                let event = event?;
+                                let name = match event.name {
+                                    Some(name) => name,
+                                    None => continue,
+                                };
 
-                        if sender.send(Ok(interceptor)).await.is_err() {
-                            return Ok(());
-                        }
+                                Path::new(EVENT_PATH).join(&name)
+                            }
+                            None => break,
+                        },
+                    };
+
+                    if !path
+                        .file_name()
+                        .and_then(OsStr::to_str)
+                        .map_or(false, |name| name.starts_with("event"))
+                    {
+                        continue;
+                    }
+
+                    let interceptor = match Interceptor::open(&path).await {
+                        Ok(interceptor) => interceptor,
+                        Err(OpenError::Io(err)) => return Err(err),
+                        Err(OpenError::NotAppliable) => continue,
+                    };
+
+                    if sender.send(Ok(interceptor)).await.is_err() {
+                        return Ok(());
                     }
                 }
 
@@ -75,23 +89,3 @@ impl Monitor {
             .ok_or_else(|| Error::new(ErrorKind::BrokenPipe, "Monitor task exited"))?
     }
 }
-
-async fn open(path: &Path) -> Result<Option<Interceptor>, Error> {
-    // Skip non input event files.
-    if path
-        .file_name()
-        .and_then(|name| name.to_str())
-        .map(|name| !name.starts_with("event"))
-        .unwrap_or(true)
-    {
-        return Ok(None);
-    }
-
-    let interceptor = match Interceptor::open(&path).await {
-        Ok(interceptor) => interceptor,
-        Err(OpenError::Io(err)) => return Err(err),
-        Err(OpenError::NotAppliable) => return Ok(None),
-    };
-
-    Ok(Some(interceptor))
-}