pcvolkmer/mv64e-kafka-to-rest-gateway
1
0
Fork 0
mirror of https://github.com/pcvolkmer/mv64e-kafka-to-rest-gateway synced 2025-09-13 09:12:52 +00:00
Code Activity

refactor: extract basic client config

Browse Source
This commit is contained in:
Paul-Christian Volkmer
2025-09-06 12:55:40 +02:00
parent 1aeab5d4ef
commit d27c8276fe
1 changed files with 40 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

76
src/main.rs
View File

@@ -4,7 +4,7 @@ use clap::Parser;
use mv64e_mtb_dto::Mtb; use mv64e_mtb_dto::Mtb;
use rdkafka::config::RDKafkaLogLevel; use rdkafka::config::RDKafkaLogLevel;
use rdkafka::consumer::{CommitMode, Consumer, StreamConsumer}; use rdkafka::consumer::{CommitMode, Consumer, StreamConsumer};
use rdkafka::message::Headers; use rdkafka::message::{BorrowedMessage, Headers};
use rdkafka::producer::{FutureProducer, FutureRecord}; use rdkafka::producer::{FutureProducer, FutureRecord};
use rdkafka::{ClientConfig, Message}; use rdkafka::{ClientConfig, Message};
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
@@ -48,26 +48,29 @@ async fn handle_record(payload: Mtb) -> Result<HttpResponse, HttpClientError> {
} }
} }
#[tokio::main] fn extract_request_id(msg: &BorrowedMessage) -> Option<String> {
async fn main() -> Result<(), Box<dyn Error>> { match msg.headers() {
#[cfg(debug_assertions)] None => None,
{ Some(headers) => {
tracing_subscriber::fmt() if let Some(value) = headers
.with_max_level(tracing::Level::DEBUG) .iter().find(|header| header.key == "requestId")?.value
.init(); {
} match str::from_utf8(value) {
Ok(value) => Some(value.to_string()),
#[cfg(not(debug_assertions))] Err(_) => None,
{ }
tracing_subscriber::fmt() } else {
.with_max_level(tracing::Level::INFO) None
.init(); }
}
} }
}
fn client_config() -> ClientConfig {
let mut client_config = ClientConfig::new(); let mut client_config = ClientConfig::new();
client_config.set("bootstrap.servers", &CONFIG.bootstrap_servers.to_string()); client_config.set("bootstrap.servers", &CONFIG.bootstrap_servers.to_string());
let client_config = if CONFIG.ssl_cert_file.is_some() || CONFIG.ssl_key_file.is_some() { if CONFIG.ssl_cert_file.is_some() || CONFIG.ssl_key_file.is_some() {
client_config client_config
.set("security.protocol", "ssl") .set("security.protocol", "ssl")
.set( .set(
@@ -88,9 +91,26 @@ async fn main() -> Result<(), Box<dyn Error>> {
client_config client_config
} else { } else {
client_config client_config
}; }
}
let mut consumer_client_config = client_config.clone(); #[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
#[cfg(debug_assertions)]
{
tracing_subscriber::fmt()
.with_max_level(tracing::Level::DEBUG)
.init();
}
#[cfg(not(debug_assertions))]
{
tracing_subscriber::fmt()
.with_max_level(tracing::Level::INFO)
.init();
}
let mut consumer_client_config = client_config();
let consumer: StreamConsumer = consumer_client_config let consumer: StreamConsumer = consumer_client_config
.set("group.id", &CONFIG.group_id) .set("group.id", &CONFIG.group_id)
@@ -107,7 +127,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
info!("Kafka topic '{}' subscribed", CONFIG.topic); info!("Kafka topic '{}' subscribed", CONFIG.topic);
let mut producer_client_config = client_config.clone(); let mut producer_client_config = client_config();
let producer: &FutureProducer = &producer_client_config let producer: &FutureProducer = &producer_client_config
.set("bootstrap.servers", &CONFIG.bootstrap_servers.to_string()) .set("bootstrap.servers", &CONFIG.bootstrap_servers.to_string())
@@ -126,23 +146,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
continue; continue;
}; };
let request_id = match msg.headers() { let request_id = extract_request_id(&msg).unwrap_or_default();
None => None,
Some(headers) => {
if let Some(x) = headers
.iter()
.filter(|header| header.key == "requestId")
.next()
.unwrap()
.value
{
Some(str::from_utf8(x).unwrap().to_string())
} else {
None
}
}
}
.unwrap_or_default();
match handle_record(payload).await { match handle_record(payload).await {
Err(err) => error!("{}", err), Err(err) => error!("{}", err),