kacrab_protocol/
frame.rs

1//! TCP wire framing.
2//!
3//! Every Kafka TCP message is wrapped in a 4-byte big-endian `i32` length
4//! prefix. This module provides the encode/decode helpers for that envelope
5//! plus a [`MAX_FRAME_LENGTH`] cap to prevent OOM on hostile or corrupt input.
6
7mod codec;
8pub mod error;
9
10use bytes::{Buf, BufMut, Bytes, BytesMut};
11
12pub use self::{
13    codec::{
14        RequestFrameSpec, ResponseEnvelope, decode_response_envelope, encode_request_frame,
15        encode_request_frame_body, encode_request_frame_body_with_buffer,
16        encode_request_frame_with_buffer, request_frame_capacity_hint,
17    },
18    error::{FrameError, FrameErrorKind},
19};
20use crate::primitives::check_remaining;
21
22/// Result alias for frame operations.
23pub type Result<T> = core::result::Result<T, FrameError>;
24
25/// Maximum accepted frame payload length (header + body, excluding the
26/// 4-byte length prefix itself). Defaults to 100 MiB, matching the Kafka
27/// broker's `socket.request.max.bytes` default.
28pub const MAX_FRAME_LENGTH: i32 = 100 * 1024 * 1024;
29
30/// Read the 4-byte big-endian frame length prefix.
31///
32/// Consumes exactly 4 bytes. Rejects negative lengths and lengths above
33/// [`MAX_FRAME_LENGTH`].
34pub fn read_frame_length(buf: &mut Bytes) -> Result<i32> {
35    check_remaining(buf, 4)?;
36    let len = buf.get_i32();
37    if len < 0 {
38        return Err(FrameErrorKind::NegativeLength { length: len }.into());
39    }
40    if len > MAX_FRAME_LENGTH {
41        return Err(FrameErrorKind::TooLarge {
42            length: len,
43            max: MAX_FRAME_LENGTH,
44        }
45        .into());
46    }
47    Ok(len)
48}
49
50/// Decode one length-prefixed response frame.
51///
52/// Reads the length via [`read_frame_length`] then splits exactly that many
53/// bytes off `buf` as the payload (header + body).
54pub fn decode_response_frame(buf: &mut Bytes) -> Result<Bytes> {
55    let frame_len = read_frame_length(buf)?;
56    let len = usize::try_from(frame_len)
57        .map_err(|_| FrameErrorKind::NegativeLength { length: frame_len })?;
58    let available = buf.remaining();
59    if available < len {
60        return Err(FrameErrorKind::Truncated {
61            needed: len,
62            available,
63        }
64        .into());
65    }
66    Ok(buf.split_to(len))
67}
68
69/// Encode a request as `[i32-BE length][header bytes][body bytes]`.
70///
71/// The length covers `header + body` only — it does NOT include the 4-byte
72/// prefix itself (matching the Kafka wire format).
73///
74/// Header serialization is the caller's responsibility; this helper takes
75/// the already-encoded header bytes so it does not depend on
76/// `crate::generated::request_header`.
77pub fn encode_request(header: &[u8], body: &[u8]) -> Result<BytesMut> {
78    let payload_len = header
79        .len()
80        .checked_add(body.len())
81        .ok_or(FrameErrorKind::TooLarge {
82            length: i32::MAX,
83            max: MAX_FRAME_LENGTH,
84        })?;
85    let max_frame_length = usize::try_from(MAX_FRAME_LENGTH).unwrap_or(usize::MAX);
86    if payload_len > max_frame_length {
87        return Err(FrameErrorKind::TooLarge {
88            length: i32::try_from(payload_len).unwrap_or(i32::MAX),
89            max: MAX_FRAME_LENGTH,
90        }
91        .into());
92    }
93    let capacity = 4usize
94        .checked_add(payload_len)
95        .ok_or(FrameErrorKind::TooLarge {
96            length: i32::MAX,
97            max: MAX_FRAME_LENGTH,
98        })?;
99    let payload_len_i32 = i32::try_from(payload_len).map_err(|_| FrameErrorKind::TooLarge {
100        length: i32::MAX,
101        max: MAX_FRAME_LENGTH,
102    })?;
103    let mut out = BytesMut::with_capacity(capacity);
104    out.put_i32(payload_len_i32);
105    out.extend_from_slice(header);
106    out.extend_from_slice(body);
107    Ok(out)
108}
kacrab_protocol/frame.rs

kacrab_protocol/
frame.rs
