krabka_hlc/
headers.rs

1//! Carriage of a stamp on Kafka record headers.
2
3use crate::stamp::Hlc;
4
5/// The Kafka record header key that carries a hybrid logical clock stamp.
6///
7/// The key is lower case, and a reader matches it byte for byte. Kafka header
8/// keys are case-sensitive, and this key belongs to Krabka, so there is no
9/// second spelling to accept.
10pub const HLC_HEADER: &str = "krabka.hlc";
11
12/// The stamp on a record, or `None` when the record carries none.
13///
14/// Each header is a key plus a raw byte value, so this function works with any
15/// producer or consumer header type. The caller converts its own `Header` into
16/// a `(key, value)` pair at the edge.
17///
18/// This function returns `None` for an absent header, for a value of the wrong
19/// length, and for a value it cannot read. It reports no error, because a
20/// consumer that cannot read the producer's stamp must still be able to apply
21/// the record. Use [`Hlc::decode`] where the caller wants the reason.
22///
23/// A record can carry the same header key more than once, and Kafka's own
24/// `Headers.lastHeader` reads the last one. This function does the same.
25#[must_use]
26pub fn extract_from_headers<'a, I, V>(headers: I) -> Option<Hlc>
27where
28    I: IntoIterator<Item = (&'a str, V)>,
29    V: AsRef<[u8]>,
30{
31    let value = headers
32        .into_iter()
33        .filter(|(key, _)| *key == HLC_HEADER)
34        .map(|(_, value)| value)
35        .last()?;
36    Hlc::decode(value.as_ref()).ok()
37}
krabka_hlc/headers.rs

krabka_hlc/
headers.rs
