kafrust-protocol 0.2.9

Kafka wire protocol primitives for kafrust.
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

use crate::codec::{Decoder, Encoder};
use crate::error::Result;
use crate::header::RequestHeader;

pub const API_KEY: i16 = 30;

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CreateAclsRequestV1 {
    pub correlation_id: i32,
    pub client_id: Option<String>,
    pub creations: Vec<CreateAclsCreationV1>,
}

impl CreateAclsRequestV1 {
    pub fn encode(&self) -> Result<Vec<u8>> {
        let mut encoder = Encoder::new();
        RequestHeader {
            api_key: API_KEY,
            api_version: 1,
            correlation_id: self.correlation_id,
            client_id: self.client_id.clone(),
        }
        .encode_v1(&mut encoder)?;
        encoder.write_array(Some(&self.creations), |encoder, creation| {
            encoder.write_i8(creation.resource_type);
            encoder.write_string(&creation.resource_name)?;
            encoder.write_i8(creation.resource_pattern_type);
            encoder.write_string(&creation.principal)?;
            encoder.write_string(&creation.host)?;
            encoder.write_i8(creation.operation);
            encoder.write_i8(creation.permission_type);
            Ok(())
        })?;
        Ok(encoder.into_bytes())
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CreateAclsCreationV1 {
    pub resource_type: i8,
    pub resource_name: String,
    pub resource_pattern_type: i8,
    pub principal: String,
    pub host: String,
    pub operation: i8,
    pub permission_type: i8,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CreateAclsResponseV1 {
    pub throttle_time_ms: i32,
    pub results: Vec<CreateAclsResultV1>,
}

impl CreateAclsResponseV1 {
    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
        Ok(Self {
            throttle_time_ms: decoder.read_i32()?,
            results: decoder
                .read_array("create ACL results", CreateAclsResultV1::decode)?
                .unwrap_or_default(),
        })
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CreateAclsResultV1 {
    pub error_code: i16,
    pub error_message: Option<String>,
}

impl CreateAclsResultV1 {
    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
        Ok(Self {
            error_code: decoder.read_i16()?,
            error_message: decoder.read_nullable_string()?,
        })
    }
}

#[cfg(test)]
#[allow(clippy::unwrap_used)]
mod tests {
    use super::{CreateAclsCreationV1, CreateAclsRequestV1, CreateAclsResponseV1, API_KEY};
    use crate::codec::{Decoder, Encoder};

    #[test]
    fn encodes_create_acls_v1_request() {
        let request = CreateAclsRequestV1 {
            correlation_id: 9,
            client_id: None,
            creations: vec![CreateAclsCreationV1 {
                resource_type: 2,
                resource_name: "orders".to_owned(),
                resource_pattern_type: 3,
                principal: "User:alice".to_owned(),
                host: "*".to_owned(),
                operation: 3,
                permission_type: 1,
            }],
        };

        let bytes = request.encode().unwrap();
        assert_eq!(&bytes[0..4], &[0, API_KEY as u8, 0, 1]);
        assert_eq!(&bytes[4..8], &[0, 0, 0, 9]);
        assert_eq!(bytes.last(), Some(&1));
    }

    #[test]
    fn decodes_create_acls_v1_response() {
        let mut bytes = Encoder::new();
        bytes.write_i32(5);
        bytes.write_i32(2);
        bytes.write_i16(0);
        bytes.write_nullable_string(None).unwrap();
        bytes.write_i16(29);
        bytes.write_nullable_string(Some("denied")).unwrap();
        let bytes = bytes.into_bytes();
        let mut decoder = Decoder::new(&bytes);

        let response = CreateAclsResponseV1::decode_body(&mut decoder).unwrap();

        assert_eq!(response.throttle_time_ms, 5);
        assert_eq!(response.results.len(), 2);
        assert_eq!(response.results[0].error_code, 0);
        assert_eq!(response.results[1].error_code, 29);
        assert_eq!(response.results[1].error_message.as_deref(), Some("denied"));
        assert!(decoder.is_empty());
    }
}