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