Skip to content

Commit bc3019d

Browse files
committed
refactored: used serde
1 parent 05b9965 commit bc3019d

File tree

2 files changed

+10
-29
lines changed

2 files changed

+10
-29
lines changed

src/cli.rs

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@ use std::path::PathBuf;
2222
use url::Url;
2323

2424
use crate::{
25+
kafka::SslProtocol,
2526
oidc::{self, OpenidConfig},
2627
option::{validation, Compression, Mode},
2728
};
@@ -125,7 +126,7 @@ pub struct Cli {
125126
pub kafka_host: Option<String>,
126127
pub kafka_group: Option<String>,
127128
pub kafka_client_id: Option<String>,
128-
pub kafka_security_protocol: Option<String>,
129+
pub kafka_security_protocol: Option<SslProtocol>,
129130
pub kafka_partitions: Option<String>,
130131
}
131132

@@ -582,7 +583,9 @@ impl FromArgMatches for Cli {
582583
self.kafka_host = m.get_one::<String>(Self::KAFKA_HOST).cloned();
583584
self.kafka_group = m.get_one::<String>(Self::KAFKA_GROUP).cloned();
584585
self.kafka_client_id = m.get_one::<String>(Self::KAFKA_CLIENT_ID).cloned();
585-
self.kafka_security_protocol = m.get_one::<String>(Self::KAFKA_SECURITY_PROTOCOL).cloned();
586+
self.kafka_security_protocol = m
587+
.get_one::<SslProtocol>(Self::KAFKA_SECURITY_PROTOCOL)
588+
.cloned();
586589
self.kafka_partitions = m.get_one::<String>(Self::KAFKA_PARTITIONS).cloned();
587590

588591
self.local_cache_path = m.get_one::<PathBuf>(Self::CACHE).cloned();

src/kafka.rs

Lines changed: 5 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@ use rdkafka::error::{KafkaError as NativeKafkaError, RDKafkaError};
88
use rdkafka::message::BorrowedMessage;
99
use rdkafka::util::Timeout;
1010
use rdkafka::{Message, TopicPartitionList};
11-
use std::fmt::Display;
11+
use serde::{Deserialize, Serialize};
1212
use std::num::ParseIntError;
1313
use std::sync::Arc;
1414
use std::{collections::HashMap, fmt::Debug, str::FromStr};
@@ -26,34 +26,13 @@ use crate::{
2626
storage::StreamType,
2727
};
2828

29-
enum SslProtocol {
29+
#[derive(Debug, Deserialize, Serialize, Clone, Copy)]
30+
pub enum SslProtocol {
3031
Plaintext,
3132
Ssl,
3233
SaslPlaintext,
3334
SaslSsl,
3435
}
35-
impl Display for SslProtocol {
36-
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
37-
f.write_str(match self {
38-
SslProtocol::Plaintext => "plaintext",
39-
SslProtocol::Ssl => "ssl",
40-
SslProtocol::SaslPlaintext => "sasl_plaintext",
41-
SslProtocol::SaslSsl => "sasl_ssl",
42-
})
43-
}
44-
}
45-
impl FromStr for SslProtocol {
46-
type Err = KafkaError;
47-
fn from_str(s: &str) -> Result<Self, Self::Err> {
48-
match s.to_ascii_lowercase().as_str() {
49-
"plaintext" => Ok(SslProtocol::Plaintext),
50-
"ssl" => Ok(SslProtocol::Ssl),
51-
"sasl_plaintext" => Ok(SslProtocol::SaslPlaintext),
52-
"sasl_ssl" => Ok(SslProtocol::SaslSsl),
53-
_ => Err(KafkaError::InvalidSslProtocolError(s.to_string())),
54-
}
55-
}
56-
}
5736

5837
#[allow(dead_code)]
5938
#[derive(Debug, thiserror::Error)]
@@ -150,9 +129,8 @@ fn setup_consumer() -> Result<(StreamConsumer, String), KafkaError> {
150129
// conf.set("api.version.request", val.to_string());
151130
// }
152131

153-
if let Some(val) = CONFIG.parseable.kafka_security_protocol.as_ref() {
154-
let mapped: SslProtocol = val.parse()?;
155-
conf.set("security.protocol", mapped.to_string());
132+
if let Some(ssl_protocol) = CONFIG.parseable.kafka_security_protocol.as_ref() {
133+
conf.set("security.protocol", serde_json::to_string(&ssl_protocol)?);
156134
}
157135

158136
let consumer: StreamConsumer = conf.create()?;

0 commit comments

Comments
 (0)