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

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

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