krabka_raft/kraft/transport/wire/
codec.rs

1//! Constants and integer conversions shared by the peer request and response
2//! codecs.
3//!
4//! The metadata topic identity, the captured wire versions, the `Epoch`,
5//! `NodeId`, uuid and cluster id conversions to and from their wire forms, and the
6//! `encode_body` helper are all needed by both [`request`](super::request) and
7//! [`response`](super::response), so they sit in one place instead of being
8//! duplicated.
9
10use {
11    crate::kraft::types::{Epoch, NodeId},
12    base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD},
13    bytes::{Bytes, BytesMut},
14    krabka_protocol::{Encode, primitives::uuid::Uuid as MetaUuid, records::RecordsPayload},
15};
16
17/// `KRaft` metadata log topic name.
18pub const METADATA_TOPIC: &str = "__cluster_metadata";
19/// The single metadata partition.
20pub const METADATA_PARTITION: i32 = 0;
21/// The fixed `KRaft` `__cluster_metadata` topic id (KIP-595). Fetch v13 and
22/// above key the topic by this id, not by name.
23pub const METADATA_TOPIC_ID: MetaUuid = MetaUuid([0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1]);
24
25/// Captured flexible wire versions, byte-validated against fixture frames.
26pub const VOTE_VERSION: i16 = 2;
27pub const QUORUM_EPOCH_VERSION: i16 = 1;
28pub const FETCH_VERSION: i16 = 18;
29pub const FETCH_SNAPSHOT_VERSION: i16 = 1;
30
31pub fn records_payload_to_bytes(payload: &RecordsPayload) -> Bytes {
32    match payload {
33        RecordsPayload::Raw(bytes) => bytes.clone(),
34        other => {
35            let mut out = BytesMut::new();
36            let _ = other.encode_to(&mut out);
37            out.freeze()
38        }
39    }
40}
41
42/// Converts between the consensus `Epoch` (u32) and the wire `i32`.
43/// `KRaft` uses an i32 `leaderEpoch`. The KIP-595 wire carries the leader
44/// epoch as a raw `int32` and stays raw here. The consensus epoch is always
45/// non-negative.
46pub fn epoch_to_wire(e: Epoch) -> i32 { i32::try_from(e).unwrap_or(i32::MAX) }
47pub fn epoch_from_wire(e: i32) -> Epoch { u32::try_from(e).unwrap_or(0) }
48/// Converts between the `NodeId` (u64) and the wire `i32` replica id.
49pub fn node_to_wire(n: NodeId) -> i32 { i32::try_from(n.0).unwrap_or(i32::MAX) }
50pub fn node_from_wire(n: i32) -> NodeId { NodeId(u64::try_from(n).unwrap_or(0)) }
51/// Converts between a `uuid::Uuid` and the wire `uuid` of the generated codecs.
52pub fn uuid_to_wire(id: uuid::Uuid) -> MetaUuid { MetaUuid(id.into_bytes()) }
53pub fn uuid_from_wire(id: MetaUuid) -> uuid::Uuid { uuid::Uuid::from_bytes(id.0) }
54/// A request `ClusterId` in Kafka's base64 form.
55pub fn cluster_id_to_wire(id: Option<uuid::Uuid>) -> Option<String> {
56    id.map(|id| URL_SAFE_NO_PAD.encode(id.as_bytes()))
57}
58
59pub fn encode_body<T: Encode>(msg: &T, version: i16) -> Bytes {
60    let mut buf = BytesMut::new();
61    // Generated codecs only error on out-of-range version, which is fixed
62    // here, so encode is infallible in practice.
63    let _ = msg.encode(&mut buf, version);
64    buf.freeze()
65}
krabka_raft/kraft/transport/wire/codec.rs

krabka_raft/kraft/transport/wire/
codec.rs
