alleviate CPU producer utilization

README.md

@@ -38,13 +38,14 @@ bmp:
   port: 4000
 
 kafka:
-  host: kafka.example.com
-  port: 9092
+  enable: true
+  brokers: "kafka.example.com:9092"
+  auth_protocol: PLAINTEXT
   topic: bgp-updates
 
 state:
   enable: true
-  path: /app/dump.txt
+  path: /app/dump.json
   save_interval: 10
 ```
 

src/config.rs

@@ -98,9 +98,13 @@ pub struct KafkaConfig {
     /// Default: 1048576
     pub message_max_bytes: usize,
 
-    /// Kafka producer max wait time
+    /// Kafka producer batch wait time
     /// Default: 1000
-    pub max_wait_time: u64,
+    pub batch_wait_time: u64,
+
+    /// Kafka producer batch wait interval
+    /// Default: 100
+    pub batch_wait_interval: u64,
 }
 
 pub fn get_kafka_config(config: &Config) -> KafkaConfig {
@@ -125,7 +129,8 @@ pub fn get_kafka_config(config: &Config) -> KafkaConfig {
             .get_string("kafka.topic")
             .unwrap_or("risotto-updates".to_string()),
         message_max_bytes: config.get_int("kafka.message_max_bytes").unwrap_or(1048576) as usize,
-        max_wait_time: config.get_int("kafka.max_wait_time").unwrap_or(1000) as u64,
+        batch_wait_time: config.get_int("kafka.batch_wait_time").unwrap_or(1000) as u64,
+        batch_wait_interval: config.get_int("kafka.batch_wait_interval").unwrap_or(100) as u64,
     }
 }
 

src/producer.rs

@@ -77,15 +77,15 @@ pub async fn handle(config: &KafkaConfig, rx: Receiver<String>) -> Result<()> {
         }
 
         loop {
-            let now = std::time::Instant::now();
-            if now.duration_since(start_time)
-                > std::time::Duration::from_millis(config.max_wait_time)
+            if std::time::Instant::now().duration_since(start_time)
+                > std::time::Duration::from_millis(config.batch_wait_time)
             {
                 break;
             }
 
             let message = rx.try_recv();
             if message.is_err() {
+                tokio::time::sleep(Duration::from_millis(config.batch_wait_interval)).await;
                 continue;
             }