ruststream_sea_file/
message.rs

1//! [`SeaMessage`]: a delivered message, shared by the file and stdio transports.
2
3use bytes::Bytes;
4use ruststream::{AckError, Headers, IncomingMessage, Positioned};
5use sea_streamer_types::{Buffer as _, Message as _, SharedMessage};
6
7use crate::wire;
8
9/// Header exposing the message's sequence number within its stream.
10pub const SEQUENCE_HEADER: &str = "stream-sequence";
11
12/// A position in a stream file's retained log, accepted by
13/// [`Seeker::seek`](ruststream::Seeker::seek).
14///
15/// Captured positions ([`Positioned::position`]) carry the pinned semantics the framework
16/// defines: seeking to one redelivers exactly that message (the transport's sequence rewind
17/// is inclusive). The other forms keep the transport's own semantics: `Beginning` replays
18/// everything retained, `End` skips to the tip, and `Timestamp` resumes at the earliest
19/// message strictly later than the instant (milliseconds since the Unix epoch).
20#[derive(Debug, Clone, Copy, PartialEq, Eq)]
21pub enum FilePosition {
22    /// Everything retained.
23    Beginning,
24    /// The tip of the stream.
25    End,
26    /// A captured message sequence (inclusive).
27    Sequence(u64),
28    /// Milliseconds since the Unix epoch (exclusive).
29    Timestamp(u64),
30}
31
32// Constructor forms of the variants: the `start_at(..)` macro clause resolves the position's
33// type from a constructor call, which a bare unit-variant path does not provide.
34impl FilePosition {
35    /// Everything retained: [`FilePosition::Beginning`].
36    #[must_use]
37    pub const fn beginning() -> Self {
38        Self::Beginning
39    }
40
41    /// The tip of the stream: [`FilePosition::End`].
42    #[must_use]
43    pub const fn end() -> Self {
44        Self::End
45    }
46
47    /// A message sequence, redelivered inclusively: [`FilePosition::Sequence`].
48    #[must_use]
49    pub const fn sequence(sequence: u64) -> Self {
50        Self::Sequence(sequence)
51    }
52
53    /// Milliseconds since the Unix epoch, resuming strictly later:
54    /// [`FilePosition::Timestamp`].
55    #[must_use]
56    pub const fn timestamp(millis: u64) -> Self {
57        Self::Timestamp(millis)
58    }
59}
60
61/// A message delivered by one of this crate's subscribers.
62///
63/// The transport keeps no consumer positions (its resumable mode is unimplemented upstream),
64/// so acknowledgement reports [`AckError::Unsupported`] rather than pretending; resume
65/// explicitly via the descriptor's start position or a captured [`FilePosition`].
66pub struct SeaMessage {
67    payload: Bytes,
68    headers: Headers,
69    stream: String,
70    sequence: u64,
71}
72
73impl std::fmt::Debug for SeaMessage {
74    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
75        f.debug_struct("SeaMessage")
76            .field("stream", &self.stream)
77            .field("sequence", &self.sequence)
78            .field("payload_len", &self.payload.len())
79            .finish_non_exhaustive()
80    }
81}
82
83impl SeaMessage {
84    pub(crate) fn new(message: &SharedMessage) -> Self {
85        let (mut headers, payload) = wire::decode(message.message().as_bytes());
86        let sequence = message.sequence();
87        headers.insert(SEQUENCE_HEADER, sequence.to_string());
88        Self {
89            payload,
90            headers,
91            stream: message.stream_key().name().to_owned(),
92            sequence,
93        }
94    }
95
96    /// The stream key this message was published to.
97    #[must_use]
98    pub fn stream(&self) -> &str {
99        &self.stream
100    }
101}
102
103impl Positioned for SeaMessage {
104    type Position = FilePosition;
105
106    fn position(&self) -> FilePosition {
107        FilePosition::Sequence(self.sequence)
108    }
109}
110
111impl IncomingMessage for SeaMessage {
112    fn payload(&self) -> &[u8] {
113        &self.payload
114    }
115
116    fn headers(&self) -> &Headers {
117        &self.headers
118    }
119
120    async fn ack(self) -> Result<(), AckError> {
121        Err(AckError::Unsupported)
122    }
123
124    async fn nack(self, _requeue: bool) -> Result<(), AckError> {
125        Err(AckError::Unsupported)
126    }
127}
ruststream_sea_file/message.rs

ruststream_sea_file/
message.rs
