few improvements and test for simple SAC

Author: DanielePalaia

examples/single_active_consumer/README.md

@@ -5,10 +5,10 @@ This is an example to enable single active consumer functionality for superstrea
 https://www.rabbitmq.com/blog/2022/07/05/rabbitmq-3-11-feature-preview-single-active-consumer-for-streams
 https://www.rabbitmq.com/blog/2022/07/13/rabbitmq-3-11-feature-preview-super-streams
 
-This folder contains a super-stream consumer configured to enable it.
+This folder contains a consumer and a super-stream consumer configured to enable it.
 You can use the example in the super-stream folder to produce messages for a super-stream.
 
-You can then run the consumer in this folder.
+You can then run the single_active_consumer_super_stream.rs in this folder.
 Assuming the super-stream is composed by three streams, you can see that the Consumer will consume messages from all the streams part of the superstream.
 
 You can then run another consumer in parallel. 

examples/single_active_consumer/example_for_enrico.rs

@@ -0,0 +1,96 @@
+use futures::StreamExt;
+use rabbitmq_stream_client::error::StreamCreateError;
+use rabbitmq_stream_client::types::{
+    ByteCapacity, OffsetSpecification, ResponseCode,
+};
+
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    use rabbitmq_stream_client::Environment;
+    let environment = Environment::builder().build().await?;
+    let message_count = 1000000;
+    let stream = "hello-rust-super-stream-0";
+
+    let create_response = environment
+        .stream_creator()
+        .max_length(ByteCapacity::GB(5))
+        .create(stream)
+        .await;
+
+    if let Err(e) = create_response {
+        if let StreamCreateError::Create { stream, status } = e {
+            match status {
+                // we can ignore this error because the stream already exists
+                ResponseCode::StreamAlreadyExists => {}
+                err => {
+                    println!("Error creating stream: {:?} {:?}", stream, err);
+                }
+            }
+        }
+    }
+    println!(
+        "Super stream consumer example, consuming messages from the super stream {}",
+        stream
+    );
+
+    let mut consumer = environment
+        .consumer()
+        // Mandatory if sac is enabled
+        .name("consumer-group-1")
+        .offset(OffsetSpecification::First)
+        .enable_single_active_consumer(true)
+        .client_provided_name("my super stream consumer for hello rust")
+        .consumer_update(move |active, message_context|  async {
+            let name = message_context.name();
+            let stream = message_context.stream();
+            let client = message_context.client();
+
+            let mut stored_offset = 0;
+            println!("single active consumer: is active: {} on stream: {} with consumer_name: {}", active, stream, name);
+            let my_async = async    {
+                println!("im hereXXXXXXXXXXX");
+                stored_offset = client.query_offset(name, stream.as_str()).await.unwrap();
+                println!("stored offset {}", stored_offset);
+                //OffsetSpecification::Offset(stored_offset)
+
+            };
+            println!("im hereXXXXXXXXXXX  {}", stored_offset);
+            my_async.await;
+            OffsetSpecification::Offset(stored_offset)
+
+        })
+        .build(stream)
+        .await
+        .unwrap();
+
+    for i in 0..message_count {
+        let delivery = consumer.next().await.unwrap();
+        {
+            let delivery = delivery.unwrap();
+            /*println!(
+                "Got message: {:#?} from stream: {} with offset: {}",
+                delivery
+                    .message()
+                    .data()
+                    .map(|data| String::from_utf8(data.to_vec()).unwrap())
+                    .unwrap(),
+                delivery.stream(),
+                delivery.offset()
+            );*/
+
+            //store an offset
+            if i == 10000  {
+                let _ = consumer
+                    .store_offset(delivery.offset())
+                    .await
+                    .unwrap_or_else(|e| println!("Err: {}", e));
+            }
+        }
+    }
+
+    println!("Stopping consumer...");
+    let _ = consumer.handle().close().await;
+    println!("consumer stopped");
+    Ok(())
+}

examples/single_active_consumer/single_active_consumer.rs

@@ -0,0 +1,72 @@
+use futures::StreamExt;
+use rabbitmq_stream_client::error::StreamCreateError;
+use rabbitmq_stream_client::types::{
+    ByteCapacity, OffsetSpecification, ResponseCode,
+};
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    use rabbitmq_stream_client::Environment;
+    let environment = Environment::builder().build().await?;
+    let message_count = 1000000;
+    let stream = "hello-rust-stream";
+
+    let create_response = environment
+        .stream_creator()
+        .max_length(ByteCapacity::GB(5))
+        .create(stream)
+        .await;
+
+    if let Err(e) = create_response {
+        if let StreamCreateError::Create { stream, status } = e {
+            match status {
+                // we can ignore this error because the stream already exists
+                ResponseCode::StreamAlreadyExists => {}
+                err => {
+                    println!("Error creating stream: {:?} {:?}", stream, err);
+                }
+            }
+        }
+    }
+    println!(
+        "Super stream consumer example, consuming messages from the super stream {}",
+        stream
+    );
+
+    let mut consumer = environment
+        .consumer()
+        // Mandatory if sac is enabled
+        .name("consumer-group-1")
+        .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
+        })
+        .build(stream)
+        .await
+        .unwrap();
+
+    for _ in 0..message_count {
+        let delivery = consumer.next().await.unwrap();
+        {
+            let delivery = delivery.unwrap();
+            println!(
+                "Got message: {:#?} from stream: {} with offset: {}",
+                delivery
+                    .message()
+                    .data()
+                    .map(|data| String::from_utf8(data.to_vec()).unwrap())
+                    .unwrap(),
+                delivery.stream(),
+                delivery.offset()
+            );
+        }
+    }
+
+    println!("Stopping consumer...");
+    let _ = consumer.handle().close().await;
+    println!("consumer stopped");
+    Ok(())
+}

src/consumer.rs

@@ -89,19 +89,25 @@ impl FilterConfiguration {
     }
 }
 
+#[derive(Clone)]
 pub struct MessageContext {
-    consumer_name: Option<String>,
+    name: String,
     stream: String,
+    client: Client,
 }
 
 impl MessageContext {
-    pub fn get_name(&self) -> Option<String> {
-        self.consumer_name.clone()
+    pub fn name(&self) -> String {
+        self.name.clone()
     }
 
-    pub fn get_stream(&self) -> String {
+    pub fn stream(&self) -> String {
         self.stream.clone()
     }
+
+    pub fn client(&self) -> Client {
+        self.client.clone()
+    }
 }
 
 /// Builder for [`Consumer`]
@@ -451,8 +457,9 @@ impl MessageHandler for ConsumerMessageHandler {
                         // Otherwise the Offset specification is returned by the user callback
                         let is_active = consumer_update.is_active();
                         let message_context = MessageContext {
-                            consumer_name: self.0.name.clone(),
+                            name: self.0.name.clone().unwrap(),
                             stream: self.0.stream.clone(),
+                            client: self.0.client.clone(),
                         };
                         let consumer_update_listener_callback =
                             self.0.consumer_update_listener.clone().unwrap();

src/superstream_consumer.rs

@@ -131,7 +131,7 @@ impl SuperStreamConsumerBuilder {
         self
     }
 
-    pub fn consumer_update(
+    pub fn consumer_update<Fut>(
         mut self,
         consumer_update_listener: impl Fn(u8, &MessageContext) -> OffsetSpecification
             + Send