kacrab_protocol/
string.rs

1//! UTF-8 strings and Kafka's fixed/compact length-prefixed encodings.
2//!
3//! [`KafkaString`] is backed by [`Bytes`] so decoded messages can borrow from
4//! the frame buffer without copying.
5//!
6//! Kafka uses four string shapes:
7//!
8//! | Variant                        | Length encoding              |
9//! |--------------------------------|------------------------------|
10//! | `read_string` / `write_string` | `i16`                        |
11//! | `read_compact_string` / …      | unsigned varint of `len + 1` |
12//! | `read_nullable_string` / …     | `i16` (`-1` = null)          |
13//! | `read_compact_nullable_string` | varint (`0` = null)          |
14
15pub mod error;
16
17use std::fmt;
18
19use bytes::{BufMut, Bytes, BytesMut};
20
21pub use self::error::{StringError, StringErrorKind};
22use crate::primitives::{
23    check_remaining, read_i16, read_unsigned_varint, unsigned_varint_len, write_unsigned_varint,
24};
25
26/// Result alias for string read operations.
27pub type Result<T> = core::result::Result<T, StringError>;
28
29/// A Kafka protocol string backed by `Bytes` for zero-copy access.
30///
31/// UTF-8 is validated on construction so [`Self::as_str`] is infallible
32/// without `unsafe` (re-validation is cheap and `unsafe` is forbidden by
33/// the workspace lint set).
34#[derive(Debug, Clone, PartialEq, Eq, Hash)]
35pub struct KafkaString {
36    inner: Bytes,
37}
38
39impl KafkaString {
40    /// Construct from raw bytes, validating UTF-8.
41    pub fn new(b: Bytes) -> Result<Self> {
42        if let Err(source) = std::str::from_utf8(&b) {
43            return Err(StringErrorKind::InvalidUtf8 { source }.into());
44        }
45        Ok(Self { inner: b })
46    }
47
48    /// Construct from a `'static` string literal. Infallible.
49    #[must_use]
50    pub const fn from_static(s: &'static str) -> Self {
51        Self {
52            inner: Bytes::from_static(s.as_bytes()),
53        }
54    }
55
56    /// Borrow as `&str`.
57    #[must_use]
58    pub fn as_str(&self) -> &str {
59        std::str::from_utf8(&self.inner).unwrap_or_else(|_| {
60            debug_assert!(false, "KafkaString contains invalid UTF-8");
61            ""
62        })
63    }
64
65    /// Borrow the raw UTF-8 bytes.
66    #[must_use]
67    pub fn as_bytes(&self) -> &[u8] {
68        &self.inner
69    }
70
71    /// Length in bytes (not characters).
72    #[must_use]
73    pub const fn len(&self) -> usize {
74        self.inner.len()
75    }
76
77    /// `true` if the string has zero bytes.
78    #[must_use]
79    pub const fn is_empty(&self) -> bool {
80        self.inner.is_empty()
81    }
82}
83
84impl fmt::Display for KafkaString {
85    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
86        f.write_str(self.as_str())
87    }
88}
89
90impl Default for KafkaString {
91    fn default() -> Self {
92        Self {
93            inner: Bytes::new(),
94        }
95    }
96}
97
98impl From<String> for KafkaString {
99    fn from(s: String) -> Self {
100        Self {
101            inner: Bytes::from(s.into_bytes()),
102        }
103    }
104}
105
106fn i16_max_len() -> usize {
107    usize::try_from(i16::MAX).unwrap_or(usize::MAX)
108}
109
110fn compact_max_len() -> usize {
111    usize::try_from(u32::MAX)
112        .unwrap_or(usize::MAX)
113        .saturating_sub(1)
114}
115
116fn i16_len_to_usize(len: i16) -> Result<usize> {
117    usize::try_from(len).map_err(|_| {
118        StringErrorKind::TooLong {
119            length: usize::MAX,
120            max: i16_max_len(),
121        }
122        .into()
123    })
124}
125
126fn compact_len_to_usize(raw: u32) -> Result<usize> {
127    let raw_len = raw.checked_sub(1).ok_or(StringErrorKind::UnexpectedNull)?;
128    usize::try_from(raw_len).map_err(|_| {
129        StringErrorKind::TooLong {
130            length: usize::MAX,
131            max: compact_max_len(),
132        }
133        .into()
134    })
135}
136
137fn read_i16_len(buf: &mut Bytes) -> Result<usize> {
138    let len = read_i16(buf)?;
139    if len < 0 {
140        return Err(StringErrorKind::NegativeLength {
141            length: i32::from(len),
142        }
143        .into());
144    }
145    i16_len_to_usize(len)
146}
147
148fn read_nullable_i16_len(buf: &mut Bytes) -> Result<Option<usize>> {
149    let len = read_i16(buf)?;
150    if len < 0 {
151        return Ok(None);
152    }
153    i16_len_to_usize(len).map(Some)
154}
155
156fn read_compact_len(buf: &mut Bytes) -> Result<usize> {
157    let raw = read_unsigned_varint(buf)?;
158    if raw == 0 {
159        return Err(StringErrorKind::UnexpectedNull.into());
160    }
161    compact_len_to_usize(raw)
162}
163
164fn read_nullable_compact_len(buf: &mut Bytes) -> Result<Option<usize>> {
165    let raw = read_unsigned_varint(buf)?;
166    if raw == 0 {
167        return Ok(None);
168    }
169    compact_len_to_usize(raw).map(Some)
170}
171
172fn split_string(buf: &mut Bytes, len: usize) -> Result<KafkaString> {
173    check_remaining(buf, len)?;
174    KafkaString::new(buf.split_to(len))
175}
176
177fn compact_len_plus_one(len: usize) -> Result<u32> {
178    len.checked_add(1)
179        .and_then(|len| u32::try_from(len).ok())
180        .ok_or_else(|| {
181            StringError::new(StringErrorKind::TooLong {
182                length: len,
183                max: compact_max_len(),
184            })
185        })
186}
187
188/// Read a non-flexible Kafka string (`i16` length prefix).
189pub fn read_string(buf: &mut Bytes) -> Result<KafkaString> {
190    let len = read_i16_len(buf)?;
191    split_string(buf, len)
192}
193
194/// Write a non-flexible Kafka string (`i16` length prefix).
195pub fn write_string(buf: &mut BytesMut, value: &KafkaString) -> Result<()> {
196    let len = i16::try_from(value.len()).map_err(|_| {
197        StringError::new(StringErrorKind::TooLong {
198            length: value.len(),
199            max: i16_max_len(),
200        })
201    })?;
202    buf.put_i16(len);
203    buf.extend_from_slice(value.as_bytes());
204    Ok(())
205}
206
207/// Encoded length of a non-flexible Kafka string (`i16` length prefix).
208pub fn string_len(value: &KafkaString) -> Result<usize> {
209    let _len = i16::try_from(value.len()).map_err(|_| {
210        StringError::new(StringErrorKind::TooLong {
211            length: value.len(),
212            max: i16_max_len(),
213        })
214    })?;
215    value.len().checked_add(2).ok_or_else(|| {
216        StringError::new(StringErrorKind::TooLong {
217            length: value.len(),
218            max: i16_max_len(),
219        })
220    })
221}
222
223/// Read a compact Kafka string (unsigned varint of `len + 1`).
224pub fn read_compact_string(buf: &mut Bytes) -> Result<KafkaString> {
225    let len = read_compact_len(buf)?;
226    split_string(buf, len)
227}
228
229/// Write a compact Kafka string (unsigned varint of `len + 1`).
230pub fn write_compact_string(buf: &mut BytesMut, value: &KafkaString) -> Result<()> {
231    write_unsigned_varint(buf, compact_len_plus_one(value.len())?);
232    buf.extend_from_slice(value.as_bytes());
233    Ok(())
234}
235
236/// Encoded length of a compact Kafka string.
237pub fn compact_string_len(value: &KafkaString) -> Result<usize> {
238    let len_plus_one = compact_len_plus_one(value.len())?;
239    value
240        .len()
241        .checked_add(unsigned_varint_len(len_plus_one))
242        .ok_or_else(|| {
243            StringError::new(StringErrorKind::TooLong {
244                length: value.len(),
245                max: compact_max_len(),
246            })
247        })
248}
249
250/// Read a nullable non-flexible Kafka string (`i16`, `-1` = null).
251pub fn read_nullable_string(buf: &mut Bytes) -> Result<Option<KafkaString>> {
252    let Some(len) = read_nullable_i16_len(buf)? else {
253        return Ok(None);
254    };
255    split_string(buf, len).map(Some)
256}
257
258/// Write a nullable non-flexible Kafka string (`i16`, `-1` = null).
259pub fn write_nullable_string(buf: &mut BytesMut, value: Option<&KafkaString>) -> Result<()> {
260    match value {
261        None => {
262            buf.put_i16(-1);
263            Ok(())
264        },
265        Some(s) => write_string(buf, s),
266    }
267}
268
269/// Encoded length of a nullable non-flexible Kafka string.
270pub fn nullable_string_len(value: Option<&KafkaString>) -> Result<usize> {
271    value.map_or(Ok(2), string_len)
272}
273
274/// Read a nullable compact Kafka string (unsigned varint, `0` = null).
275pub fn read_compact_nullable_string(buf: &mut Bytes) -> Result<Option<KafkaString>> {
276    let Some(len) = read_nullable_compact_len(buf)? else {
277        return Ok(None);
278    };
279    split_string(buf, len).map(Some)
280}
281
282/// Write a nullable compact Kafka string (unsigned varint, `0` = null).
283pub fn write_compact_nullable_string(
284    buf: &mut BytesMut,
285    value: Option<&KafkaString>,
286) -> Result<()> {
287    match value {
288        None => {
289            write_unsigned_varint(buf, 0);
290            Ok(())
291        },
292        Some(s) => write_compact_string(buf, s),
293    }
294}
295
296/// Encoded length of a nullable compact Kafka string.
297pub fn compact_nullable_string_len(value: Option<&KafkaString>) -> Result<usize> {
298    value.map_or(Ok(1), compact_string_len)
299}
kacrab_protocol/string.rs

kacrab_protocol/
string.rs
