ed_journals/modules/fs/models/common_async/
async_log_file.rs

1use crate::fs::{FileWatcher, LogFSError, Unblocker};
2use crate::io::{AsyncIter, LogIOError};
3use crate::logs::LogEvent;
4use async_fs::File;
5use futures::io::BufReader;
6use futures::Stream;
7use serde::de::DeserializeOwned;
8use std::path::Path;
9use std::pin::{pin, Pin};
10use std::sync::Arc;
11use std::task::{Context, Poll};
12
13/// Async variant of [LogFile](crate::fs::common::LogFile).
14/// Holds both a watcher and an async iterator over for the given path.
15#[derive(Debug)]
16pub struct AsyncLogFile<R = LogEvent>
17where
18    R: DeserializeOwned + Unpin,
19{
20    iter: AsyncIter<BufReader<File>, R>,
21    _w: FileWatcher,
22}
23
24impl AsyncLogFile {
25    pub async fn new<P: AsRef<Path>>(
26        path: P,
27        blocker: impl Into<Arc<dyn Unblocker>>,
28    ) -> Result<AsyncLogFile<LogEvent>, LogFSError> {
29        AsyncLogFile::new_typed::<LogEvent, _>(path, blocker).await
30    }
31
32    pub async fn new_raw<P: AsRef<Path>>(
33        path: P,
34        blocker: impl Into<Arc<dyn Unblocker>>,
35    ) -> Result<AsyncLogFile<serde_json::Value>, LogFSError> {
36        AsyncLogFile::new_typed::<serde_json::Value, _>(path, blocker).await
37    }
38
39    pub async fn new_typed<R, P>(
40        path: P,
41        blocker: impl Into<Arc<dyn Unblocker>>,
42    ) -> Result<AsyncLogFile<R>, LogFSError>
43    where
44        R: DeserializeOwned + Unpin,
45        P: AsRef<Path>,
46    {
47        let path = path.as_ref();
48        let watcher = FileWatcher::new(path, blocker)?;
49        let file = File::open(path).await?;
50        let reader = BufReader::new(file);
51        let iter = AsyncIter::from(reader);
52
53        Ok(AsyncLogFile { _w: watcher, iter })
54    }
55}
56
57impl<R> Stream for AsyncLogFile<R>
58where
59    R: DeserializeOwned + Unpin,
60{
61    type Item = Result<R, LogIOError>;
62
63    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
64        pin!(&mut self.iter).poll_next(cx)
65    }
66}
ed_journals/modules/fs/models/common_async/async_log_file.rs

ed_journals/modules/fs/models/common_async/
async_log_file.rs
