krabka_client_core/
request.rs

1use bytes::BufMut;
2pub use krabka_protocol::ProtocolRequest;
3use krabka_protocol::{
4    Encode, ProtocolError,
5    owned::{leave_group_request, offset_commit_request, offset_fetch_request},
6};
7
8/// A protocol request with a narrower version range than its wire type.
9///
10/// The lower bound is raised to at least the wrapped request's minimum, and
11/// the upper bound is capped at its maximum. The stable version follows the
12/// wrapped request unless the cap lowers it.
13#[derive(Clone, Debug, PartialEq, Eq)]
14pub struct VersionBoundedRequest<Request, const MIN_VERSION: i16, const MAX_VERSION: i16> {
15    request: Request,
16}
17
18impl<Request, const MIN_VERSION: i16, const MAX_VERSION: i16>
19    VersionBoundedRequest<Request, MIN_VERSION, MAX_VERSION>
20{
21    /// Wrap a request in the configured version range.
22    pub fn new(request: Request) -> Self {
23        Self { request }
24    }
25}
26
27impl<Request, const MIN_VERSION: i16, const MAX_VERSION: i16> Encode
28    for VersionBoundedRequest<Request, MIN_VERSION, MAX_VERSION>
29where
30    Request: ProtocolRequest,
31{
32    fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
33        self.request.encode(buf, version)
34    }
35
36    fn encoded_len(&self, version: i16) -> usize {
37        self.request.encoded_len(version)
38    }
39}
40
41impl<Request, const MIN_VERSION: i16, const MAX_VERSION: i16> ProtocolRequest
42    for VersionBoundedRequest<Request, MIN_VERSION, MAX_VERSION>
43where
44    Request: ProtocolRequest,
45{
46    const API_KEY: i16 = Request::API_KEY;
47    const MIN_VERSION: i16 = if Request::MIN_VERSION > MIN_VERSION {
48        Request::MIN_VERSION
49    } else {
50        MIN_VERSION
51    };
52    const MAX_VERSION: i16 = if Request::MAX_VERSION < MAX_VERSION {
53        Request::MAX_VERSION
54    } else {
55        MAX_VERSION
56    };
57    const LATEST_STABLE_VERSION: i16 = if Request::LATEST_STABLE_VERSION < Self::MAX_VERSION {
58        Request::LATEST_STABLE_VERSION
59    } else {
60        Self::MAX_VERSION
61    };
62    const FLEXIBLE_MIN: i16 = Request::FLEXIBLE_MIN;
63    type Response = Request::Response;
64}
65
66/// An `OffsetCommit` request that names its topics, capped at v9.
67pub type TopicNameOffsetCommit = VersionBoundedRequest<
68    offset_commit_request::OffsetCommitRequest,
69    { offset_commit_request::MIN_VERSION },
70    9,
71>;
72
73/// An `OffsetFetch` request that names its topics, capped at v9.
74pub type TopicNameOffsetFetch = VersionBoundedRequest<
75    offset_fetch_request::OffsetFetchRequest,
76    { offset_fetch_request::MIN_VERSION },
77    9,
78>;
79
80/// An admin `OffsetFetch` request for all topics, which needs v2 through v9.
81pub type AllTopicsOffsetFetch =
82    VersionBoundedRequest<offset_fetch_request::OffsetFetchRequest, 2, 9>;
83
84/// A `LeaveGroup` request that carries its member list, needing v3 or newer.
85pub type MembersOnlyLeaveGroup = VersionBoundedRequest<
86    leave_group_request::LeaveGroupRequest,
87    3,
88    { leave_group_request::MAX_VERSION },
89>;
krabka_client_core/request.rs

krabka_client_core/
request.rs
