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

1use crate::fs::models::common_async::AsyncLogFile;
2use crate::fs::{LogFSError, Unblocker};
3use crate::io::{LogIOError, LogPath};
4use crate::logs::LogEvent;
5use futures::Stream;
6use serde::de::DeserializeOwned;
7use std::pin::{pin, Pin};
8use std::sync::Arc;
9use std::task::{Context, Poll};
10
11/// Async variant of [NewestFile](crate::fs::common::NewestFile).
12/// Holds an [AsyncLogFile] which is changed when [AsyncNewestFile::maybe_next] is called with a
13/// [LogPath] which is newer than the current one.
14pub struct AsyncNewestFile<R = LogEvent>
15where
16    R: DeserializeOwned + Unpin,
17{
18    current_path: Option<LogPath>,
19    current_file: Option<AsyncLogFile<R>>,
20    unblocker: Arc<dyn Unblocker>,
21}
22
23impl AsyncNewestFile {
24    pub fn new(blocker: impl Into<Arc<dyn Unblocker>>) -> AsyncNewestFile<LogEvent> {
25        AsyncNewestFile::new_typed::<LogEvent>(blocker)
26    }
27
28    pub fn new_raw(blocker: impl Into<Arc<dyn Unblocker>>) -> AsyncNewestFile<serde_json::Value> {
29        AsyncNewestFile::new_typed::<serde_json::Value>(blocker)
30    }
31
32    pub fn new_typed<R>(blocker: impl Into<Arc<dyn Unblocker>>) -> AsyncNewestFile<R>
33    where
34        R: DeserializeOwned + Unpin,
35    {
36        AsyncNewestFile {
37            current_path: None,
38            current_file: None,
39            unblocker: blocker.into(),
40        }
41    }
42}
43
44impl<R> AsyncNewestFile<R>
45where
46    R: DeserializeOwned + Unpin,
47{
48    pub async fn maybe_next(&mut self, path: &LogPath) -> Result<bool, LogFSError> {
49        if self.current_path.is_none()
50            || self
51                .current_path
52                .as_ref()
53                .is_some_and(|current| path > current)
54        {
55            self.current_path = Some(path.clone());
56            self.current_file =
57                Some(AsyncLogFile::new_typed::<R, _>(path, self.unblocker.clone()).await?);
58
59            return Ok(true);
60        }
61
62        Ok(false)
63    }
64}
65
66impl<R> Stream for AsyncNewestFile<R>
67where
68    R: DeserializeOwned + Unpin,
69{
70    type Item = Result<R, LogIOError>;
71
72    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
73        match self.current_file.as_mut() {
74            Some(file) => pin!(file).poll_next(cx),
75            None => Poll::Ready(None),
76        }
77    }
78}
ed_journals/modules/fs/models/common_async/async_newest_file.rs

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