Howl send vc00

Cargo.toml

@@ -10,7 +10,7 @@ log = "0.4.29"
 uuid = { version = "1.22.0", features = ["v4"] }
 env_logger = "0.11.9"
 clap-verbosity-flag = "3.0.4"
-isis_streaming_data_types = "0.1.1"
+isis_streaming_data_types = "0.1.3"
 rand = {  version = "0.10.1", features = ["thread_rng"]}
 rand_distr = "0.6.0"
 flatbuffers = "25.12.19"

src/cli_utils.rs

@@ -1,4 +1,5 @@
 use anyhow::{Context, Result, bail};
+use rdkafka::ClientConfig;
 use std::str::FromStr;
 
 pub(crate) fn parse_broker_spec(s: &str) -> Result<BrokerAndTopic> {
@@ -68,6 +69,21 @@ pub struct KafkaOption {
     pub value: String,
 }
 
+pub fn set_kafka_options(
+    client_config: &mut ClientConfig,
+    kafka_config: &Option<Vec<KafkaOption>>,
+) {
+    if let Some(kafka_options) = kafka_config {
+        for option in kafka_options {
+            println!(
+                "Setting Kafka config option {}={}",
+                option.key, option.value
+            );
+            client_config.set(&option.key, &option.value);
+        }
+    }
+}
+
 impl FromStr for KafkaOption {
     type Err = String;
 

src/consume.rs

@@ -1,5 +1,5 @@
 use crate::KafkaOption;
-use crate::cli_utils::BrokerAndTopic;
+use crate::cli_utils::{BrokerAndTopic, set_kafka_options};
 
 use isis_streaming_data_types::{deserialize_message, get_schema_id};
 use log::{debug, error, info};
@@ -35,16 +35,7 @@ pub fn consume(config: &ConsumeConfig) {
     let mut client_config = ClientConfig::new();
     client_config.set("group.id", Uuid::new_v4().to_string());
     client_config.set("bootstrap.servers", config.topic.broker());
-
-    if let Some(kafka_options) = &config.kafka_config {
-        for option in kafka_options {
-            println!(
-                "Setting Kafka config option {}={}",
-                option.key, option.value
-            );
-            client_config.set(&option.key, &option.value);
-        }
-    }
+    set_kafka_options(&mut client_config, &config.kafka_config);
 
     let consumer: BaseConsumer = client_config.create().expect("Base creation failed");
 

src/count.rs

@@ -1,5 +1,5 @@
 use crate::KafkaOption;
-use crate::cli_utils::BrokerAndTopic;
+use crate::cli_utils::{BrokerAndTopic, set_kafka_options};
 use futures::stream::StreamExt;
 use log::error;
 use rdkafka::consumer::{Consumer, DefaultConsumerContext, StreamConsumer};
@@ -15,16 +15,7 @@ pub async fn count(
     let mut config = ClientConfig::new();
     config.set("group.id", Uuid::new_v4().to_string());
     config.set("bootstrap.servers", topic.broker());
-
-    if let Some(kafka_options) = kafka_config {
-        for option in kafka_options {
-            println!(
-                "Setting Kafka config option {}={}",
-                option.key, option.value
-            );
-            config.set(&option.key, &option.value);
-        }
-    }
+    set_kafka_options(&mut config, &kafka_config);
 
     let consumer: StreamConsumer<DefaultConsumerContext> =
         config.create().expect("Consumer creation failed");