example work in progress

Gsantomaggio · Gsantomaggio · commit 6dc32c05df43 · 2024-11-18T15:43:54.000+01:00
Signed-off-by: Gabriele Santomaggio &lt;G.santomaggio@gmail.com&gt;
diff --git a/Cargo.toml b/Cargo.toml
@@ -16,7 +16,8 @@ readme = "README.md"
 members = [
     ".",
     "protocol",
-    "benchmark"
+    "benchmark",
+    "examples/super_stream_single_active_consumer", "examples/test_ss",
 ]
 
 
diff --git a/examples/super_stream_single_active_consumer/Cargo.toml b/examples/super_stream_single_active_consumer/Cargo.toml
@@ -0,0 +1,9 @@
+[package]
+name = "single_active_consumer"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+futures = "0.3.31"
+tokio = "1.41.1"
+rabbitmq-stream-client = { path = "../../" }
diff --git a/examples/super_stream_single_active_consumer/README.md b/examples/super_stream_single_active_consumer/README.md
@@ -0,0 +1,83 @@
+Super stream example
+---
+
+[Super Streams Documentation](https://www.rabbitmq.com/streams.html#super-streams) for more details.
+[Super Streams blog post](https://blog.rabbitmq.com/posts/2022/07/rabbitmq-3-11-feature-preview-super-streams)
+
+
+This example shows how to use the Super Stream feature in RabbitMQ 3.11.0.
+
+Then run the producer in one terminal:
+
+    $  cargo run -- --producer
+
+
+And the consumer in another terminal:
+
+    $ cargo run -- --consumer
+
+You should see the consumer receiving the messages from the producer.
+
+It would be something like:
+```bash
+$ cargo run -- --producer
+Starting SuperStream Producer
+Super Stream Producer connected to RabbitMQ
+Super Stream Producer sent 0 messages to invoices
+Super Stream Producer sent 1 messages to invoices
+Super Stream Producer sent 2 messages to invoices
+Super Stream Producer sent 3 messages to invoices
+```
+
+```bash
+$ dotnet run --consumer my_first_consumer
+Starting SuperStream Consumer my_first_consumer
+Super Stream Consumer connected to RabbitMQ. ConsumerName my_first_consumer
+Consumer Name my_first_consumer -Received message id: hello0 body: hello0, Stream invoices-2
+Consumer Name my_first_consumer -Received message id: hello1 body: hello1, Stream invoices-1
+Consumer Name my_first_consumer -Received message id: hello2 body: hello2, Stream invoices-0
+Consumer Name my_first_consumer -Received message id: hello3 body: hello3, Stream invoices-1
+Consumer Name my_first_consumer -Received message id: hello4 body: hello4, Stream invoices-2
+Consumer Name my_first_consumer -Received message id: hello5 body: hello5, Stream invoices-0
+Consumer Name my_first_consumer -Received message id: hello6 body: hello6, Stream invoices-2
+Consumer Name my_first_consumer -Received message id: hello7 body: hello7, Stream invoices-0
+Consumer Name my_first_consumer -Received message id: hello8 body: hello8, Stream invoices-1
+Consumer Name my_first_consumer -Received message id: hello9 body: hello9, Stream invoices-0
+```
+
+
+To see the Single active consumer in action, run another consumer:
+
+    $ dotnet run --consumer my_second_consumer
+
+You should see the second consumer receiving the part of the messages from the producer. In thi case only the messages coming from the `invoices-1`.
+
+It should be something like:
+```bash
+$ dotnet run --consumer my_second_consumer
+Starting SuperStream Consumer my_second_consumer
+Super Stream Consumer connected to RabbitMQ. ConsumerName my_second_consumer
+Consumer Name my_second_consumer -Received message id: hello1 body: hello1, Stream invoices-1
+Consumer Name my_second_consumer -Received message id: hello3 body: hello3, Stream invoices-1
+Consumer Name my_second_consumer -Received message id: hello8 body: hello8, Stream invoices-1
+Consumer Name my_second_consumer -Received message id: hello14 body: hello14, Stream invoices-1
+Consumer Name my_second_consumer -Received message id: hello15 body: hello15, Stream invoices-1
+Consumer Name my_second_consumer -Received message id: hello16 body: hello16, Stream invoices-1
+Consumer Name my_second_consumer -Received message id: hello19 body: hello19, Stream invoices-1
+```
+and the first consumer should be receiving the rest of the messages:
+```bash
+Consumer Name my_first_consumer -Received message id: hello0 body: hello0, Stream invoices-2
+Consumer Name my_first_consumer -Received message id: hello2 body: hello2, Stream invoices-0
+Consumer Name my_first_consumer -Received message id: hello4 body: hello4, Stream invoices-2
+Consumer Name my_first_consumer -Received message id: hello5 body: hello5, Stream invoices-0
+Consumer Name my_first_consumer -Received message id: hello6 body: hello6, Stream invoices-2
+Consumer Name my_first_consumer -Received message id: hello7 body: hello7, Stream invoices-0
+Consumer Name my_first_consumer -Received message id: hello9 body: hello9, Stream invoices-0
+Consumer Name my_first_consumer -Received message id: hello10 body: hello10, Stream invoices-2
+Consumer Name my_first_consumer -Received message id: hello11 body: hello11, Stream invoices-0
+```
+
+
+
+
diff --git a/examples/super_stream_single_active_consumer/src/main.rs b/examples/super_stream_single_active_consumer/src/main.rs
@@ -0,0 +1,35 @@
+mod send_super_stream;
+mod single_active_consumer_super_stream;
+
+use std::env;
+
+static SUPER_STREAM: &str = "invoices";
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let mut args = env::args().skip(1);
+    while let Some(arg) = args.next() {
+        match &arg[..] {
+            "-h" | "--help" => help(),
+            "--consumer" => {
+                single_active_consumer_super_stream::start_consumer().await?;
+            }
+
+            "--producer" => send_super_stream::start_producer().await?,
+
+            arg if arg.starts_with("-") => {
+                eprintln!("Unknown argument: {}", arg);
+            }
+
+            _ => {
+                eprintln!("Unknown argument: {}", arg);
+                help();
+            }
+        }
+    }
+    Ok(())
+}
+
+fn help() {
+    println!("--consumer or --producer")
+}
diff --git a/examples/super_stream_single_active_consumer/src/send_super_stream.rs b/examples/super_stream_single_active_consumer/src/send_super_stream.rs
@@ -0,0 +1,114 @@
+use rabbitmq_stream_client::error::StreamCreateError;
+use rabbitmq_stream_client::types::{
+    ByteCapacity, HashRoutingMurmurStrategy, Message, ResponseCode, RoutingStrategy,
+};
+use std::convert::TryInto;
+use std::sync::atomic::{AtomicU32, Ordering};
+use std::sync::Arc;
+use tokio::sync::Notify;
+use tokio::time;
+
+fn hash_strategy_value_extractor(message: &Message) -> String {
+    message
+        .application_properties()
+        .unwrap()
+        .get("id")
+        .unwrap()
+        .clone()
+        .try_into()
+        .unwrap()
+}
+
+pub async fn start_producer() -> Result<(), Box<dyn std::error::Error>> {
+    use rabbitmq_stream_client::Environment;
+    let environment = Environment::builder().build().await?;
+    let confirmed_messages = Arc::new(AtomicU32::new(0));
+    let notify_on_send = Arc::new(Notify::new());
+    let _ = environment
+        .stream_creator()
+        .max_length(ByteCapacity::GB(5))
+        .create_super_stream(crate::SUPER_STREAM, 3, None)
+        .await;
+
+    let create_response = environment
+        .stream_creator()
+        .max_length(ByteCapacity::GB(5))
+        .create_super_stream(crate::SUPER_STREAM, 3, None)
+        .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!(
+                        "[Super Stream Producer] Error creating stream: {:?} {:?}",
+                        stream, err
+                    );
+                }
+            }
+        }
+    }
+    println!(
+        "[Super Stream Producer] Sending messages to the super stream: {}",
+        crate::SUPER_STREAM
+    );
+    let super_stream_producer = environment
+        .super_stream_producer(RoutingStrategy::HashRoutingStrategy(
+            HashRoutingMurmurStrategy {
+                routing_extractor: &hash_strategy_value_extractor,
+            },
+        ))
+        .client_provided_name("rust stream producer - sac example")
+        .build(crate::SUPER_STREAM)
+        .await;
+
+    match super_stream_producer {
+        Ok(mut producer) => {
+            println!("[Super Stream Producer] Successfully created super stream producer");
+            let mut idx = 0;
+            loop {
+                let counter = confirmed_messages.clone();
+                let notifier = notify_on_send.clone();
+                let msg = Message::builder()
+                    .body(format!("super stream message_{}", idx))
+                    .application_properties()
+                    .insert("id", idx.to_string())
+                    .message_builder()
+                    .build();
+                println!("[Super Stream Producer] Sending message {}", idx);
+                let send_result = producer
+                    .send(msg, move |_| {
+                        let inner_counter = counter.clone();
+                        let inner_notifier = notifier.clone();
+                        async move {
+                            if inner_counter.fetch_add(1, Ordering::Relaxed) == idx - 1 {
+                                inner_notifier.notify_one();
+                            }
+                        }
+                    })
+                    .await;
+
+                match send_result {
+                    Ok(_) => {
+                        idx += 1;
+                        println!("[Super Stream Producer] Message {} sent", idx);
+                    }
+                    Err(err) => {
+                        println!(
+                            "[Super Stream Producer] Failed to send message. error: {}",
+                            err
+                        );
+                    }
+                }
+
+                time::sleep(time::Duration::from_millis(1_000)).await;
+            }
+        }
+        Err(err) => {
+            println!("Failed to create super stream producer. error {}", err);
+            Ok(())
+        }
+    }
+}
diff --git a/examples/super_stream_single_active_consumer/src/single_active_consumer_super_stream.rs b/examples/super_stream_single_active_consumer/src/single_active_consumer_super_stream.rs
@@ -3,19 +3,15 @@ use rabbitmq_stream_client::error::StreamCreateError;
 use rabbitmq_stream_client::types::{
     ByteCapacity, OffsetSpecification, ResponseCode, SuperStreamConsumer,
 };
-use std::collections::HashMap;
 
-#[tokio::main]
-async fn main() -> Result<(), Box<dyn std::error::Error>> {
+pub async fn start_consumer() -> Result<(), Box<dyn std::error::Error>> {
     use rabbitmq_stream_client::Environment;
     let environment = Environment::builder().build().await?;
-    let message_count = 1000000;
-    let super_stream = "hello-rust-super-stream";
 
     let create_response = environment
         .stream_creator()
         .max_length(ByteCapacity::GB(5))
-        .create_super_stream(super_stream, 3, None)
+        .create_super_stream(crate::SUPER_STREAM, 3, None)
         .await;
 
     if let Err(e) = create_response {
@@ -31,7 +27,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     }
     println!(
         "Super stream consumer example, consuming messages from the super stream {}",
-        super_stream
+        crate::SUPER_STREAM
     );
 
     let mut super_stream_consumer: SuperStreamConsumer = environment
@@ -41,7 +37,7 @@ 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|  async move {
+        .consumer_update(move |active, message_context| async move {
             let name = message_context.name();
             let stream = message_context.stream();
             let client = message_context.client();
@@ -52,19 +48,18 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             );
             let stored_offset = client.query_offset(name, stream.as_str()).await;
 
-            if let Err(e) = stored_offset {
+            if let Err(_) = stored_offset {
                 return OffsetSpecification::First;
             }
             let stored_offset_u = stored_offset.unwrap();
-            println!("stored_offset_u {}", stored_offset_u.clone());
+            println!("offset: {} stored", stored_offset_u.clone());
             OffsetSpecification::Offset(stored_offset_u)
-
         })
-        .build(super_stream)
+        .build(crate::SUPER_STREAM)
         .await
         .unwrap();
 
-    for _ in 0..message_count {
+    loop {
         let delivery = super_stream_consumer.next().await.unwrap();
         {
             let delivery = delivery.unwrap();
@@ -80,15 +75,16 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             );
 
             // 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;
-            }
-
+            // store the offset each time a message is consumed
+            // that is not a best practice, but it is done here for demonstration purposes
+            super_stream_consumer
+                .client()
+                .store_offset(
+                    delivery.consumer_name().unwrap().as_str(),
+                    delivery.stream().as_str(),
+                    delivery.offset(),
+                )
+                .await?;
         }
     }
-
-    println!("Stopping super stream consumer...");
-    let _ = super_stream_consumer.handle().close().await;
-    println!("Super stream consumer stopped");
-    Ok(())
-}
+}
diff --git a/src/superstream_producer.rs b/src/superstream_producer.rs
@@ -45,13 +45,13 @@ impl SuperStreamProducer<NoDedup> {
         &mut self,
         message: Message,
         cb: impl Fn(Result<ConfirmationStatus, ProducerPublishError>) -> Fut
-            + Send
-            + Sync
-            + 'static
-            + Clone,
+        + Send
+        + Sync
+        + 'static
+        + Clone,
     ) -> Result<(), SuperStreamProducerPublishError>
     where
-        Fut: Future<Output = ()> + Send + Sync + 'static,
+        Fut: Future<Output=()> + Send + Sync + 'static,
     {
         let routes = match &self.4 {
             RoutingStrategy::HashRoutingStrategy(routing_strategy) => {
@@ -68,6 +68,7 @@ impl SuperStreamProducer<NoDedup> {
 
         for route in routes.into_iter() {
             if !self.1.contains_key(route.as_str()) {
+                println!("Sending message to route: {}", route);
                 let producer = self
                     .0
                     .environment

Original file line number	Diff line number	Diff line change
`@@ -16,7 +16,8 @@ readme = "README.md"`
`16`	`16`	`members = [`
`17`	`17`	`".",`
`18`	`18`	`"protocol",`
`19`		`- "benchmark"`
	`19`	`+ "benchmark",`
	`20`	`+ "examples/super_stream_single_active_consumer", "examples/test_ss",`
`20`	`21`	`]`
`21`	`22`
`22`	`23`