making Delivery export client in order to use store_offset and review super_stream example

Author: DanielePalaia

examples/single_active_consumer/single_active_consumer_super_stream.rs

@@ -41,9 +41,24 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         .offset(OffsetSpecification::First)
         .enable_single_active_consumer(true)
         .client_provided_name("my super stream consumer for hello rust")
-        .consumer_update(move |active, message_context| {
-            println!("single active consumer: is active: {} on stream {}", active, message_context.get_stream());
-            OffsetSpecification::First
+        .consumer_update(move |active, message_context|  async move {
+            let name = message_context.name();
+            let stream = message_context.stream();
+            let client = message_context.client();
+
+            println!(
+                "single active consumer: is active: {} on stream: {} with consumer_name: {}",
+                active, stream, name
+            );
+            let stored_offset = client.query_offset(name, stream.as_str()).await;
+
+            if let Err(e) = stored_offset {
+                return OffsetSpecification::First;
+            }
+            let stored_offset_u = stored_offset.unwrap();
+            println!("stored_offset_u {}", stored_offset_u.clone());
+            OffsetSpecification::Offset(stored_offset_u)
+
         })
         .build(super_stream)
         .await
@@ -63,6 +78,12 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
                 delivery.stream(),
                 delivery.offset()
             );
+
+            // Store an offset for every consumer
+            if delivery.consumer_name().is_some() && delivery.offset() == 1000   {
+                super_stream_consumer.client().store_offset(delivery.consumer_name().unwrap().as_str(), delivery.stream().as_str(), delivery.offset()).await;
+            }
+
         }
     }
 

src/consumer.rs

@@ -427,6 +427,7 @@ impl MessageHandler for ConsumerMessageHandler {
                             .0
                             .sender
                             .send(Ok(Delivery {
+                                name: self.0.name.clone(),
                                 stream: self.0.stream.clone(),
                                 subscription_id: self.0.subscription_id,
                                 message,
@@ -493,6 +494,7 @@ impl MessageHandler for ConsumerMessageHandler {
 /// Envelope from incoming message
 #[derive(Debug)]
 pub struct Delivery {
+    name: Option<String>,
     stream: String,
     subscription_id: u8,
     message: Message,
@@ -519,4 +521,8 @@ impl Delivery {
     pub fn offset(&self) -> u64 {
         self.offset
     }
+
+    pub fn consumer_name(&self) -> Option<String> {
+        self.name.clone()
+    }
 }

src/superstream_consumer.rs

@@ -178,6 +178,10 @@ impl SuperStreamConsumer {
     pub fn handle(&self) -> SuperStreamConsumerHandle {
         SuperStreamConsumerHandle(self.internal.clone())
     }
+
+    pub fn client(&self) -> Client {
+        self.internal.client.clone()
+    }
 }
 
 impl SuperStreamConsumerInternal {