Start9/start-os
2
0
Fork 0
mirror of https://github.com/Start9Labs/start-os.git synced 2026-03-30 20:14:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix file_stream and remove non-terminating test

Browse Source
This commit is contained in:
Aiden McClelland
2025-08-29 11:48:30 -06:00
parent 8163db7ac3
commit 369e559518
1 changed files with 42 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

72
core/startos/src/util/io.rs
View File

@@ -1532,18 +1532,36 @@ pub trait ReadWriter: AsyncRead + AsyncWrite {}
impl<T: AsyncRead + AsyncWrite> ReadWriter for T {} impl<T: AsyncRead + AsyncWrite> ReadWriter for T {}
#[instrument(skip_all)] #[instrument(skip_all)]
async fn wait_for_created(stream: &mut EventStream<[u8; 1024]>, path: &Path) -> Result<(), Error> { fn wait_for_created<'a>(
let parent = stream stream: &'a mut EventStream<[u8; 1024]>,
.watches() path: &'a Path,
.add(path.parent().unwrap_or("/".as_ref()), WatchMask::CREATE)?; ) -> BoxFuture<'a, Result<(), Error>> {
while let Some(e) = stream.try_next().await? { async {
if e.mask & EventMask::CREATE != EventMask::empty() && e.name.as_deref() == path.file_name() let parent_path = path.parent().unwrap_or("/".as_ref());
{ let parent;
break; loop {
match stream.watches().add(parent_path, WatchMask::CREATE) {
Ok(desc) => {
parent = desc;
break;
}
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
wait_for_created(stream, parent_path).await?;
}
Err(e) => Err(e)?,
}
} }
while let Some(e) = stream.try_next().await? {
if e.mask & EventMask::CREATE != EventMask::empty()
&& e.name.as_deref() == path.file_name()
{
break;
}
}
stream.watches().remove(parent)?;
Ok(())
} }
stream.watches().remove(parent)?; .boxed()
Ok(())
} }
#[instrument(skip_all)] #[instrument(skip_all)]
@@ -1554,32 +1572,26 @@ pub fn file_string_stream(
async_stream::try_stream! { async_stream::try_stream! {
let mut stream = Inotify::init()?.into_event_stream([0; 1024])?; let mut stream = Inotify::init()?.into_event_stream([0; 1024])?;
loop { loop {
stream.watches().add( loop {
&path, match stream.watches().add(
WatchMask::MODIFY | WatchMask::MOVE_SELF | WatchMask::MOVED_TO | WatchMask::DELETE_SELF, &path,
)?; WatchMask::MODIFY | WatchMask::MOVE_SELF | WatchMask::MOVED_TO | WatchMask::DELETE_SELF,
if let Some(contents) = maybe_read_file_to_string(&path).await? { ) {
yield Some(contents); Ok(_) => break,
} else { Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
wait_for_created(&mut stream, &path).await?; yield None;
wait_for_created(&mut stream, &path).await?;
}
Err(e) => Err(e)?,
}
} }
yield maybe_read_file_to_string(&path).await?;
while let Some(e) = stream.try_next().await? { while let Some(e) = stream.try_next().await? {
yield maybe_read_file_to_string(&path).await?;
if e.mask & EventMask::DELETE_SELF != EventMask::empty() { if e.mask & EventMask::DELETE_SELF != EventMask::empty() {
break; break;
} }
yield maybe_read_file_to_string(&path).await?;
} }
} }
} }
} }
#[tokio::test]
async fn test_wait_for_created() {
let mut stream = Inotify::init()
.unwrap()
.into_event_stream([0; 1024])
.unwrap();
wait_for_created(&mut stream, Path::new("/tmp/wait-for-me"))
.await
.unwrap();
}