From a0e98e43fcdb069e489ca5bad93121970d93b3ce Mon Sep 17 00:00:00 2001 From: Jan Trefil <8711792+htrefil@users.noreply.github.com> Date: Sat, 26 Aug 2023 15:27:27 +0200 Subject: [PATCH] Refactor device discovery --- rkvm-input/src/monitor.rs | 74 ++++++++++++++++++--------------------- 1 file changed, 34 insertions(+), 40 deletions(-) diff --git a/rkvm-input/src/monitor.rs b/rkvm-input/src/monitor.rs index 7dba403..6dc419c 100644 --- a/rkvm-input/src/monitor.rs +++ b/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, 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)) -}