rustfs-kafka 0.22.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
119
120
121
122
123
124
125
126
127
128

//! TLS abstraction layer for Kafka connections.
//!
//! This module provides TLS connections via rustls (pure-Rust).

use std::io;
use std::net::TcpStream;

#[cfg(feature = "security")]
pub mod rustls_connector;
#[cfg(feature = "security")]
pub use rustls_connector::RustlsConnector;

/// Configuration for TLS connections
#[derive(Debug, Clone)]
pub struct TlsConfig {
    /// Whether to verify the server's hostname
    pub verify_hostname: bool,
    /// Optional path to a CA certificate file
    pub ca_cert_path: Option<String>,
    /// Optional path to a client certificate file
    pub client_cert_path: Option<String>,
    /// Optional path to a client private key file
    pub client_key_path: Option<String>,
}

impl Default for TlsConfig {
    fn default() -> Self {
        TlsConfig {
            verify_hostname: true,
            ca_cert_path: None,
            client_cert_path: None,
            client_key_path: None,
        }
    }
}

impl TlsConfig {
    /// Create a new TLS configuration with default settings
    pub fn new() -> Self {
        Self::default()
    }

    /// Set whether to verify the server's hostname
    #[must_use]
    pub fn with_hostname_verification(mut self, verify: bool) -> Self {
        self.verify_hostname = verify;
        self
    }

    /// Set the CA certificate path
    #[must_use]
    pub fn with_ca_cert(mut self, path: String) -> Self {
        self.ca_cert_path = Some(path);
        self
    }

    /// Set the client certificate and key paths
    #[must_use]
    pub fn with_client_cert(mut self, cert_path: String, key_path: String) -> Self {
        self.client_cert_path = Some(cert_path);
        self.client_key_path = Some(key_path);
        self
    }
}

/// Trait for TLS stream implementations
#[allow(dead_code)] // Methods may not be used in all configurations
pub trait TlsStream: io::Read + io::Write + Send {
    /// Returns true if this is a secured (TLS) connection
    fn is_secured(&self) -> bool;

    /// Set the read timeout
    fn set_read_timeout(&mut self, dur: Option<std::time::Duration>) -> io::Result<()>;

    /// Set the write timeout
    fn set_write_timeout(&mut self, dur: Option<std::time::Duration>) -> io::Result<()>;

    /// Shutdown the connection
    fn shutdown(&mut self) -> io::Result<()>;
}

/// Plain TCP stream wrapper implementing `TlsStream`
#[allow(dead_code)] // May not be used in all configurations
pub struct PlainStream {
    inner: TcpStream,
}

#[allow(dead_code)] // May not be used in all configurations
impl PlainStream {
    pub fn new(stream: TcpStream) -> Self {
        PlainStream { inner: stream }
    }
}

impl TlsStream for PlainStream {
    fn is_secured(&self) -> bool {
        false
    }

    fn set_read_timeout(&mut self, dur: Option<std::time::Duration>) -> io::Result<()> {
        self.inner.set_read_timeout(dur)
    }

    fn set_write_timeout(&mut self, dur: Option<std::time::Duration>) -> io::Result<()> {
        self.inner.set_write_timeout(dur)
    }

    fn shutdown(&mut self) -> io::Result<()> {
        use std::net::Shutdown;
        self.inner.shutdown(Shutdown::Both)
    }
}

impl io::Read for PlainStream {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        self.inner.read(buf)
    }
}

impl io::Write for PlainStream {
    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
        self.inner.write(buf)
    }

    fn flush(&mut self) -> io::Result<()> {
        self.inner.flush()
    }
}