krafka 0.9.2

A pure Rust, async-native Apache Kafka client
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151

use bytes::{Buf, BufMut};

use super::{VersionedDecode, VersionedEncode, non_nullable_string};
use crate::error::{ErrorCode, Result};
use crate::protocol::check_compact_array_len;
use crate::protocol::primitives::{Decode, KafkaString, TaggedFields, TryEncode};

// ============================================================================
// ListClientMetricsResources API (Key 74)
// ============================================================================

/// ListClientMetricsResources request (API Key 74). Flexible from v0.
#[derive(Debug, Clone)]
pub struct ListClientMetricsResourcesRequest;

impl ListClientMetricsResourcesRequest {
    /// Encode for version 0 (flexible from v0, empty body).
    pub fn encode_v0(&self, buf: &mut impl BufMut) -> Result<()> {
        TaggedFields::default().try_encode(buf)?;
        Ok(())
    }
}

/// A client metrics resource name.
#[derive(Debug, Clone)]
pub struct ClientMetricsResource {
    /// Resource name.
    pub name: String,
}

/// ListClientMetricsResources response (API Key 74). Flexible from v0.
#[derive(Debug, Clone)]
pub struct ListClientMetricsResourcesResponse {
    /// Throttle time in milliseconds.
    pub throttle_time_ms: i32,
    /// Error code.
    pub error_code: ErrorCode,
    /// Client metrics resource names.
    pub client_metrics_resources: Vec<ClientMetricsResource>,
}

impl ListClientMetricsResourcesResponse {
    /// Decode from version 0 (flexible from v0).
    pub fn decode_v0(buf: &mut impl Buf) -> Result<Self> {
        let throttle_time_ms = i32::decode(buf)?;
        let error_code = ErrorCode::from_i16(i16::decode(buf)?);
        let resource_count =
            check_compact_array_len(crate::util::varint::decode_unsigned_varint(buf)?)?;
        let mut client_metrics_resources = Vec::with_capacity(resource_count);
        for _ in 0..resource_count {
            let name =
                non_nullable_string("metric resource name", KafkaString::decode_compact(buf)?.0)?;
            let _ = TaggedFields::decode(buf)?;
            client_metrics_resources.push(ClientMetricsResource { name });
        }
        let _ = TaggedFields::decode(buf)?;
        Ok(Self {
            throttle_time_ms,
            error_code,
            client_metrics_resources,
        })
    }
}

impl VersionedEncode for ListClientMetricsResourcesRequest {
    fn encode_versioned(&self, version: i16, buf: &mut impl BufMut) -> Result<()> {
        match version {
            0 => self.encode_v0(buf)?,
            _ => return unsupported_encode!("ListClientMetricsResourcesRequest", version),
        }
        Ok(())
    }
}

impl VersionedDecode for ListClientMetricsResourcesResponse {
    fn decode_versioned(version: i16, buf: &mut impl Buf) -> Result<Self> {
        match version {
            0 => Self::decode_v0(buf),
            _ => unsupported_decode!("ListClientMetricsResourcesResponse", version),
        }
    }
}

#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
mod tests {
    use super::*;

    use crate::util::varint;
    use bytes::BytesMut;

    /// Helper: encode a compact string into `buf`.
    fn put_compact_string(buf: &mut BytesMut, s: Option<&str>) {
        match s {
            Some(val) => {
                buf.put_u8((val.len() + 1) as u8);
                buf.put_slice(val.as_bytes());
            }
            None => buf.put_u8(0),
        }
    }

    /// Helper: write empty tagged fields (varint 0).
    fn put_tagged_fields(buf: &mut BytesMut) {
        buf.put_u8(0);
    }

    #[test]
    fn test_list_client_metrics_resources_request_encode_v0() {
        let req = ListClientMetricsResourcesRequest;
        let mut buf = BytesMut::new();
        req.encode_v0(&mut buf).unwrap();

        let mut cur = &buf[..];
        assert_eq!(cur.get_u8(), 0); // tagged fields only
        assert!(cur.is_empty());
    }

    #[test]
    fn test_list_client_metrics_resources_response_decode_v0() {
        let mut buf = BytesMut::new();
        buf.put_i32(0); // throttle_time_ms
        buf.put_i16(0); // error_code
        varint::encode_unsigned_varint(3, &mut buf); // 2 resources
        put_compact_string(&mut buf, Some("metric-a"));
        put_tagged_fields(&mut buf); // resource tagged fields
        put_compact_string(&mut buf, Some("metric-b"));
        put_tagged_fields(&mut buf); // resource tagged fields
        put_tagged_fields(&mut buf); // top-level tagged fields

        let resp = ListClientMetricsResourcesResponse::decode_v0(&mut buf.freeze()).unwrap();
        assert_eq!(resp.throttle_time_ms, 0);
        assert!(resp.error_code.is_ok());
        assert_eq!(resp.client_metrics_resources.len(), 2);
        assert_eq!(resp.client_metrics_resources[0].name, "metric-a");
        assert_eq!(resp.client_metrics_resources[1].name, "metric-b");
    }

    #[test]
    fn test_list_client_metrics_resources_response_decode_v0_empty() {
        let mut buf = BytesMut::new();
        buf.put_i32(10); // throttle_time_ms
        buf.put_i16(0); // error_code
        varint::encode_unsigned_varint(1, &mut buf); // 0 resources
        put_tagged_fields(&mut buf); // top-level tagged fields

        let resp = ListClientMetricsResourcesResponse::decode_v0(&mut buf.freeze()).unwrap();
        assert_eq!(resp.throttle_time_ms, 10);
        assert!(resp.client_metrics_resources.is_empty());
    }
}