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
use serde::{Deserialize, Serialize};
/// One materialized-view advancement on the change feed: after committing a
/// projector batch for a binding that opted in (`ProjectionBinding.notify`),
/// the plane publishes one record naming the index, the partition, the offset
/// window the batch covered, and how many rows landed. A consumer awaits the
/// record then queries, instead of sleeping and retrying: change notification
/// as records on a topic, consumed by offset, never a server-push watch (the
/// substrate is a log). v1 scope is materialized-view advancement. Key-value
/// and run-registry change records are candidate extensions of the same shape,
/// not implied.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct ChangeRecord {
pub v: u32,
/// The materialized index the batch advanced.
pub index: String,
/// The source partition the batch covered.
pub partition_id: u32,
/// First source offset in the committed batch.
pub from_offset: u64,
/// Last source offset in the committed batch (the new watermark).
pub to_offset: u64,
/// Rows the batch landed in the view.
pub rows: u32,
/// The stream whose source the batch read, when the deployment publishes
/// change records per stream.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub stream: Option<String>,
}
#[cfg(all(test, feature = "cbor"))]
mod tests {
use super::*;
use crate::codes::CHANGE_OP_VERSION;
use crate::framing::{decode_named, encode_named};
#[test]
fn given_a_change_record_when_round_tripped_then_should_decode_unchanged() {
let record = ChangeRecord {
v: CHANGE_OP_VERSION,
index: "readings_v1".to_owned(),
partition_id: 3,
from_offset: 100,
to_offset: 141,
rows: 42,
stream: None,
};
let bytes = encode_named(&record).expect("encodes");
let back: ChangeRecord = decode_named(&bytes).expect("decodes");
assert_eq!(back, record);
}
}