krabka_replicator/mm2/
checkpoint.rs

1use super::{Reader, Writer};
2use crate::{
3    error::ReplicatorError,
4    ids::{DownstreamOffset, PartitionIndex, UpstreamOffset},
5};
6
7#[derive(Debug, Clone, PartialEq, Eq)]
8pub struct Checkpoint {
9    pub group: String,
10    pub topic: String,
11    pub partition: PartitionIndex,
12    pub upstream: UpstreamOffset,
13    pub downstream: DownstreamOffset,
14    pub metadata: String,
15}
16
17impl Checkpoint {
18    #[must_use]
19    pub fn topic_name(source_alias: &str) -> String {
20        format!("{source_alias}.checkpoints.internal")
21    }
22
23    #[must_use]
24    pub fn key_bytes(&self) -> Vec<u8> {
25        Writer::keyless()
26            .string(&self.group)
27            .string(&self.topic)
28            .i32(self.partition.0)
29            .finish()
30    }
31
32    #[must_use]
33    pub fn value_bytes(&self) -> Vec<u8> {
34        Writer::new()
35            .i64(self.upstream.0)
36            .i64(self.downstream.0)
37            .string(&self.metadata)
38            .finish()
39    }
40
41    /// # Errors
42    /// Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.
43    pub fn from_bytes(key: &[u8], val: &[u8]) -> Result<Self, ReplicatorError> {
44        let mut k = Reader::keyless(key);
45        let mut v = Reader::new(val)?;
46        Ok(Self {
47            group: k.string()?,
48            topic: k.string()?,
49            partition: PartitionIndex(k.i32()?),
50            upstream: UpstreamOffset(v.i64()?),
51            downstream: DownstreamOffset(v.i64()?),
52            metadata: v.string()?,
53        })
54    }
55}
krabka_replicator/mm2/checkpoint.rs

krabka_replicator/mm2/
checkpoint.rs
