kafkang 0.3.0

Rust client for Apache Kafka
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

use std::io::{Read, Write};

use crate::codecs::{FromByte, ToByte};
use crate::error::{Error, Result};

use super::{API_KEY_SASL_HANDSHAKE, HeaderRequest, HeaderResponse};

// v1 indicates that the SASL tokens will be wrapped in `SaslAuthenticate` requests (KIP-86).
const SASL_HANDSHAKE_REQUEST_VERSION: i16 = 1;

#[derive(Debug)]
pub struct SaslHandshakeRequest<'a> {
    pub header: HeaderRequest<'a>,
    pub mechanism: &'a str,
}

impl<'a> SaslHandshakeRequest<'a> {
    pub fn new(correlation_id: i32, client_id: &'a str, mechanism: &'a str) -> Self {
        SaslHandshakeRequest {
            header: HeaderRequest::new(
                API_KEY_SASL_HANDSHAKE,
                SASL_HANDSHAKE_REQUEST_VERSION,
                correlation_id,
                client_id,
            ),
            mechanism,
        }
    }
}

impl ToByte for SaslHandshakeRequest<'_> {
    fn encode<W: Write>(&self, buffer: &mut W) -> Result<()> {
        try_multi!(self.header.encode(buffer), self.mechanism.encode(buffer))
    }
}

// --------------------------------------------------------------------

#[derive(Default, Debug)]
pub struct SaslHandshakeResponse {
    pub header: HeaderResponse,
    pub error_code: i16,
    pub enabled_mechanisms: Vec<String>,
}

impl SaslHandshakeResponse {
    pub fn check_error(&self) -> Result<()> {
        match Error::from_protocol(self.error_code) {
            Some(err) => Err(err),
            None => Ok(()),
        }
    }
}

impl FromByte for SaslHandshakeResponse {
    type R = SaslHandshakeResponse;

    #[allow(unused_must_use)]
    fn decode<T: Read>(&mut self, buffer: &mut T) -> Result<()> {
        try_multi!(
            self.header.decode(buffer),
            self.error_code.decode(buffer),
            self.enabled_mechanisms.decode(buffer)
        )
    }
}

// --------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use std::io::Cursor;

    use crate::codecs::{FromByte, ToByte};

    use super::{SaslHandshakeRequest, SaslHandshakeResponse};

    #[test]
    fn test_sasl_handshake_request_v1_encoding() {
        let correlation_id: i32 = 123;
        let client_id = "test-client";
        let mechanism = "PLAIN";

        let req = SaslHandshakeRequest::new(correlation_id, client_id, mechanism);

        let mut buf = Vec::new();
        req.encode(&mut buf).unwrap();

        let mut expected = Vec::new();
        (17i16).encode(&mut expected).unwrap(); // api_key
        (1i16).encode(&mut expected).unwrap(); // api_version
        correlation_id.encode(&mut expected).unwrap();
        client_id.encode(&mut expected).unwrap();
        mechanism.encode(&mut expected).unwrap();

        assert_eq!(buf, expected);
    }

    #[test]
    fn test_sasl_handshake_response_v0_decoding() {
        let correlation_id: i32 = 456;
        let error_code: i16 = 0;
        let mechanisms = ["PLAIN", "SCRAM-SHA-256"];

        let mut buf = Vec::new();
        correlation_id.encode(&mut buf).unwrap();
        error_code.encode(&mut buf).unwrap();
        (mechanisms.len() as i32).encode(&mut buf).unwrap();
        for m in mechanisms {
            m.encode(&mut buf).unwrap();
        }

        let resp = SaslHandshakeResponse::decode_new(&mut Cursor::new(buf)).unwrap();
        assert_eq!(resp.header.correlation, correlation_id);
        assert_eq!(resp.error_code, 0);
        assert_eq!(resp.enabled_mechanisms, ["PLAIN", "SCRAM-SHA-256"]);
    }
}