laser_sdk/
watch.rs

1use crate::cursor::Cursor;
2use crate::error::LaserError;
3use crate::laser::Laser;
4use laser_wire::change::ChangeRecord;
5use laser_wire::framing::decode_named;
6
7impl Laser {
8    /// The change-feed accessor: [`ChangeRecord`]s the managed backend publishes after
9    /// each committed projector batch for a binding that opted into `notify`.
10    /// "Query after my data landed" becomes await-then-query instead of
11    /// sleep-and-retry, and it composes with read-your-writes rather than
12    /// replacing it. Gated on the `WATCH` capability: refused typed when the
13    /// deployment does not publish the feed, so a consumer can never wait on
14    /// a topic nothing writes to.
15    pub fn watch(&self) -> Watch<'_> {
16        Watch {
17            laser: self,
18            index: None,
19        }
20    }
21}
22
23/// A fluent change-feed read. Narrow with [`index`](Self::index), then drain
24/// with [`records`](Self::records). Build it with [`Laser::watch`].
25pub struct Watch<'a> {
26    laser: &'a Laser,
27    index: Option<String>,
28}
29
30impl<'a> Watch<'a> {
31    /// Keep only advancements of this materialized index, named as
32    /// [`Laser::resource_name`] names it.
33    #[must_use]
34    pub fn index(mut self, index: impl Into<String>) -> Self {
35        self.index = Some(self.laser.resource_name(&index.into()));
36        self
37    }
38
39    /// Open the feed reader: the existing [`Cursor`] over the changes topic on
40    /// the ops stream, decoding each record and applying the index filter
41    /// client-side. No new consumption machinery, the feed is ordinary records
42    /// consumed by offset.
43    pub fn records(self) -> Result<WatchReader, LaserError> {
44        if !self.laser.current_capabilities().watch {
45            return Err(LaserError::unsupported_feature(
46                "watch",
47                "watch",
48                "the change feed is not published by this deployment",
49            ));
50        }
51        let ops_stream = self.laser.ops_stream();
52        // Under stream tenancy each stream's change feed rides its own ops
53        // topic, so a scoped handle reads its stream's topic.
54        let changes_topic = match self.laser.resource_stream() {
55            Some(stream) if self.laser.current_capabilities().stream_tenancy => {
56                laser_wire::topics::stream_ops_topic(stream, laser_wire::topics::CHANGES_TOPIC)
57            }
58            _ => self.laser.changes_topic(),
59        };
60        let cursor = self.laser.reader_on(&ops_stream, &changes_topic)?;
61        Ok(WatchReader {
62            cursor,
63            index: self.index,
64        })
65    }
66}
67
68/// A resumable change-feed reader. Each [`poll`](Self::poll) drains the
69/// records that landed since the last one. Persist [`offsets`](Self::offsets)
70/// and seed a fresh reader to resume across restarts, exactly like any
71/// [`Cursor`].
72pub struct WatchReader {
73    cursor: Cursor,
74    index: Option<String>,
75}
76
77impl WatchReader {
78    /// Resume from a previous reader's offsets.
79    #[must_use]
80    pub fn from_offsets(mut self, offsets: Vec<u64>) -> Self {
81        self.cursor = self.cursor.from_offsets(offsets);
82        self
83    }
84
85    /// Drain the change records that landed since the last poll, filtered to
86    /// the watched index when one was set. A record that does not decode as a
87    /// [`ChangeRecord`] is skipped: the changes topic defaults to
88    /// `laser_wire::topics::CHANGES_TOPIC` but is overridable
89    /// ([`Laser::with_changes_topic`]), and a misbehaving deployment publishing
90    /// other traffic on it must not wedge the reader either way.
91    pub async fn poll(&mut self) -> Result<Vec<ChangeRecord>, LaserError> {
92        let messages = self.cursor.poll().await?;
93        Ok(messages
94            .into_iter()
95            .filter_map(|message| decode_named::<ChangeRecord>(&message.payload).ok())
96            .filter(|record| {
97                self.index
98                    .as_deref()
99                    .is_none_or(|index| record.index == index)
100            })
101            .collect())
102    }
103
104    /// The per-partition offsets consumed so far, the resume state.
105    pub fn offsets(&self) -> &[u64] {
106        self.cursor.offsets()
107    }
108
109    /// This reader as a [`Stream`](futures::Stream) of change records, one at a
110    /// time, draining what landed since the last read and ending once caught up
111    /// (the same shape as the async reader in the Python binding). Drive it with
112    /// `futures::StreamExt`.
113    pub fn stream(self) -> impl futures::Stream<Item = Result<ChangeRecord, LaserError>> {
114        futures::stream::unfold(
115            (self, std::collections::VecDeque::new(), false),
116            |(mut reader, mut buffered, stop)| async move {
117                if stop {
118                    return None;
119                }
120                loop {
121                    if let Some(record) = buffered.pop_front() {
122                        return Some((Ok(record), (reader, buffered, false)));
123                    }
124                    match reader.poll().await {
125                        Ok(batch) if batch.is_empty() => return None,
126                        Ok(batch) => buffered.extend(batch),
127                        Err(error) => return Some((Err(error), (reader, buffered, true))),
128                    }
129                }
130            },
131        )
132    }
133}
134
135#[cfg(test)]
136mod tests {
137    use crate::laser::Laser;
138
139    #[test]
140    fn given_a_default_stream_when_watching_an_index_then_should_match_the_scoped_name() {
141        let laser = Laser::from_client(crate::iggy::prelude::IggyClient::default())
142            .with_default_stream("acme");
143        let watch = laser.watch().index("readings");
144        assert_eq!(watch.index.as_deref(), Some("stream:acme/readings"));
145    }
146}
laser_sdk/watch.rs

laser_sdk/
watch.rs
