krabka_replicator/mm2/
offset_sync.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 OffsetSync {
9    pub topic: String,
10    pub partition: PartitionIndex,
11    pub upstream: UpstreamOffset,
12    pub downstream: DownstreamOffset,
13}
14
15impl OffsetSync {
16    #[must_use]
17    pub fn topic_name(source_alias: &str) -> String {
18        format!("mm2-offset-syncs.{source_alias}.internal")
19    }
20
21    #[must_use]
22    pub fn key_bytes(&self) -> Vec<u8> {
23        Writer::keyless()
24            .string(&self.topic)
25            .i32(self.partition.0)
26            .finish()
27    }
28
29    #[must_use]
30    pub fn value_bytes(&self) -> Vec<u8> {
31        // OffsetSync has no HEADER_SCHEMA / version field at all in the JVM MM2
32        // codec — both key and value are versionless (unlike Heartbeat and
33        // Checkpoint, whose *value* carries a version header).
34        Writer::keyless()
35            .i64(self.upstream.0)
36            .i64(self.downstream.0)
37            .finish()
38    }
39
40    /// # Errors
41    /// Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.
42    pub fn from_bytes(key: &[u8], val: &[u8]) -> Result<Self, ReplicatorError> {
43        let mut k = Reader::keyless(key);
44        let mut v = Reader::keyless(val);
45        Ok(Self {
46            topic: k.string()?,
47            partition: PartitionIndex(k.i32()?),
48            upstream: UpstreamOffset(v.i64()?),
49            downstream: DownstreamOffset(v.i64()?),
50        })
51    }
52}
krabka_replicator/mm2/offset_sync.rs

krabka_replicator/mm2/
offset_sync.rs
