test

Author: DanielePalaia

rust-stream/Cargo.lock

@@ -6,6 +6,6 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-rabbitmq-stream-client = "0.4.2"
+rabbitmq-stream-client = "0.4.3"
 tokio = { version = "1.39.0", features = ["rt", "rt-multi-thread", "macros"] }
 futures = "0.3.30"

rust-stream/Cargo.toml

@@ -10,7 +10,7 @@ use tokio::task;
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     use rabbitmq_stream_client::Environment;
     let environment = Environment::builder().build().await?;
-    let stream = "stream-offset-tracking-rust";
+    let stream = "pippo";
     let first_offset = Arc::new(AtomicI64::new(-1));
     let last_offset = Arc::new(AtomicI64::new(-1));
     let notify_on_close = Arc::new(Notify::new());
@@ -33,28 +33,29 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         }
     }
 
+    let stored_offset:u64 = 45;
     let mut consumer = environment
         .consumer()
         .name("consumer-1")
-        .offset(OffsetSpecification::First)
+        .offset(OffsetSpecification::Offset(stored_offset))
         .build(stream)
         .await
         .unwrap();
 
     println!("Started consuming");
 
-    let mut stored_offset: u64 = consumer.query_offset().await.unwrap_or_else(|_| 0);
+    /*let mut stored_offset: u64 = consumer.query_offset().await.unwrap_or_else(|_| 0);
 
     if stored_offset > 0 {
         stored_offset += 1;
     }
     consumer = environment
         .consumer()
         .name("consumer-1")
-        .offset(OffsetSpecification::Offset(stored_offset))
+        .offset(OffsetSpecification::Offset(42))
         .build(stream)
         .await
-        .unwrap();
+        .unwrap();*/
 
     let first_cloned_offset = first_offset.clone();
     let last_cloned_offset = last_offset.clone();
@@ -65,6 +66,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         while let Some(delivery) = consumer.next().await {
             let d = delivery.unwrap();
 
+            println!("offset {} ", d.offset());
             if first_offset.load(Ordering::Relaxed) == -1 {
                 println!("First message received");
                 _ = first_offset.compare_exchange(
@@ -74,19 +76,19 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
                     Ordering::Relaxed,
                 );
             }
-            received_messages = received_messages + 1;
-            if received_messages % 10 == 0
-                || String::from_utf8_lossy(d.message().data().unwrap()).contains("marker")
+            //received_messages = received_messages + 1;
+            if String::from_utf8_lossy(d.message().data().unwrap()).contains("marker")
             {
-                let _ = consumer
+                /*let _ = consumer
                     .store_offset(d.offset())
                     .await
-                    .unwrap_or_else(|e| println!("Err: {}", e));
+                    .unwrap_or_else(|e| println!("Err: {}", e));*/
                 if String::from_utf8_lossy(d.message().data().unwrap()).contains("marker") {
                     last_offset.store(d.offset() as i64, Ordering::Relaxed);
                     let handle = consumer.handle();
                     _ = handle.close().await;
                     notify_on_close_cloned.notify_one();
+                    break;
                 }
             }
         }

rust-stream/src/bin/receive_offset_tracking.rs

@@ -8,7 +8,7 @@ use tokio::sync::Notify;
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     use rabbitmq_stream_client::Environment;
     let environment = Environment::builder().build().await?;
-    let stream = "stream-offset-tracking-rust";
+    let stream = "pippo";
     let message_count = 100;
     let confirmed_messages = Arc::new(AtomicU32::new(0));
     let notify_on_send = Arc::new(Notify::new());