ruststream_sea_file/
stream.rs

1//! [`FileStream`]: the subscription descriptor for the file transport.
2
3use ruststream::SubscriptionSource;
4
5use crate::error::SeaFileError;
6use crate::file::ConnectedFileBroker;
7use crate::subscriber::FileSubscriber;
8
9/// A subscription descriptor for one stream key in the file.
10///
11/// A plain descriptor follows the live tail; where reading begins is the framework's
12/// `start_at(..)` clause with a [`FilePosition`](crate::FilePosition) (or a live seek through
13/// the `Seek` parameter). [`replay`](Self::replay) is the one reading mode the position API
14/// cannot express: it reads the finished file and completes the stream at its end instead of
15/// following live writes.
16///
17/// Implements [`SubscriptionSource`], so it can sit inline in the `#[subscriber(..)]`
18/// decorator:
19///
20/// ```
21/// use ruststream_sea_file::FileStream;
22///
23/// let live = FileStream::new("orders");
24/// let batch = FileStream::new("orders").replay();
25/// # let _ = (live, batch);
26/// ```
27#[derive(Debug, Clone, PartialEq, Eq)]
28#[must_use]
29pub struct FileStream {
30    stream: String,
31    replay: bool,
32}
33
34impl FileStream {
35    /// Names the stream key.
36    pub fn new(stream: impl Into<String>) -> Self {
37        Self {
38            stream: stream.into(),
39            replay: false,
40        }
41    }
42
43    /// Replays the retained file from the beginning and ends at its tail instead of
44    /// following live writes; the subscription completes at the end of the file.
45    pub fn replay(mut self) -> Self {
46        self.replay = true;
47        self
48    }
49
50    /// The stream key this descriptor resolves.
51    #[must_use]
52    pub fn stream(&self) -> &str {
53        &self.stream
54    }
55
56    pub(crate) fn replay_value(&self) -> bool {
57        self.replay
58    }
59
60    /// Rejects descriptors that cannot form a subscription, before any I/O.
61    pub(crate) fn validate(&self) -> Result<(), SeaFileError> {
62        if self.stream.is_empty() {
63            return Err(SeaFileError::Invalid("stream key must be non-empty".into()));
64        }
65        Ok(())
66    }
67}
68
69impl SubscriptionSource<ConnectedFileBroker> for FileStream {
70    type Subscriber = FileSubscriber;
71
72    fn name(&self) -> &str {
73        self.stream()
74    }
75
76    async fn subscribe(
77        self,
78        connected: &ConnectedFileBroker,
79    ) -> Result<FileSubscriber, SeaFileError> {
80        connected.subscribe_stream(self).await
81    }
82}
83
84#[cfg(test)]
85mod tests {
86    use super::*;
87
88    #[test]
89    fn empty_stream_keys_are_rejected_before_io() {
90        assert!(FileStream::new("").validate().is_err());
91    }
92
93    #[test]
94    fn replay_reads_the_retained_file() {
95        assert!(FileStream::new("orders").replay().replay_value());
96        assert!(!FileStream::new("orders").replay_value());
97    }
98}
ruststream_sea_file/stream.rs

ruststream_sea_file/
stream.rs
