Merge pull request #37 from kpn-dsh/fix/improve-example

Fix/improve example

Author: toelo3

Cargo.toml

@@ -1,6 +1,7 @@
 [workspace]
 members = [
-    "dsh_sdk"
+    "dsh_sdk",
+    "example_dsh_service",
 ]
 readme = "README.md"
 resolver = "2"

dsh_sdk/examples/produce_consume.rs

@@ -1,4 +1,5 @@
 use dsh_sdk::dsh::Properties;
+use dsh_sdk::rdkafka::consumer::CommitMode;
 use dsh_sdk::rdkafka::consumer::{Consumer, StreamConsumer};
 use dsh_sdk::rdkafka::producer::{FutureProducer, FutureRecord};
 use dsh_sdk::rdkafka::Message;
@@ -31,6 +32,7 @@ async fn consume(consumer: &mut StreamConsumer, topic: &str) {
         let payload = String::from_utf8_lossy(msg.payload().unwrap());
         let key = usize::from_be_bytes(msg.key().unwrap().try_into().unwrap());
         println!("Received message: key: {}, payload: {}", key, payload);
+        consumer.commit_message(&msg, CommitMode::Async).unwrap();
         i += 1;
     }
 }

example_dsh_service/src/main.rs

@@ -3,13 +3,15 @@ use dsh_sdk::graceful_shutdown::Shutdown;
 use dsh_sdk::rdkafka::consumer::{CommitMode, Consumer, StreamConsumer};
 use dsh_sdk::rdkafka::message::{BorrowedMessage, Message};
 
+use log::{info, error};
+
 mod custom_metrics;
 
 fn deserialize_and_print(msg: &BorrowedMessage) {
     let payload = String::from_utf8_lossy(msg.payload().unwrap_or(b""));
     let key = String::from_utf8_lossy(msg.key().unwrap_or(b""));
 
-    println!(
+    info!(
         "Received message from topic {} partition {} offset {} with key {:?} and payload {}",
         msg.topic(),
         msg.partition(),
@@ -28,14 +30,12 @@ async fn consume(consumer: StreamConsumer, shutdown: Shutdown) {
                     // Deserialize and print the message
                     deserialize_and_print(&msg);
                     // Commit the message
-                    match consumer.commit_message(&msg, CommitMode::Sync)
-                    {
-                        Ok(_) => {}
-                        Err(e) => println!("Error while committing message: {:?}", e),
+                    if let Err(e) = consumer.commit_message(&msg, CommitMode::Sync) {
+                        error!("Error while committing message: {:?}", e);
                     }
             },
             _ = shutdown.recv() => {
-                println!("Shutdown requested, breaking out of consumer");
+                info!("Shutdown requested, breaking out of consumer");
                 consumer.unsubscribe();
                 break;
              }
@@ -76,14 +76,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     consumer_client_config.set("auto.offset.reset", "latest");
 
     // Create a new consumer instance
-    let consumer: StreamConsumer = consumer_client_config
-        .create()
-        .expect("Consumer creation failed");
+    let consumer: StreamConsumer = consumer_client_config.create()?;
 
     // Subscribe to the configured topics
-    consumer
-        .subscribe(&topics)
-        .expect("Can't subscribe to specified topics");
+    consumer.subscribe(&topics)?;
 
     // Create handle for consuming messages,
     let shutdown_clone = shutdown.clone();
@@ -94,10 +90,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     // Wait for shutdown signal or that the consumer has stopped
     tokio::select! {
         _ = shutdown.signal_listener() => {
-            println!("Shutdown signal received");
+            info!("Shutdown signal received");
         }
         _ = consumer_handle => {
-            println!("Consumer stopped");
+            info!("Consumer stopped");
             shutdown.start(); // Start the shutdown process (this will stop other potential running tasks that implemented the shutdown listener)
         }
     }