Merge branch 'main' into implement_super_stream

Author: DanielePalaia

src/client/dispatcher.rs

@@ -344,7 +344,7 @@ mod tests {
 
     #[tokio::test]
     async fn should_reject_requests_after_closing() {
-        let mock_source = MockIO::push();
+        let _mock_source = MockIO::push();
 
         let dispatcher = Dispatcher::with_handler(|_| async { Ok(()) });
 

src/consumer.rs

@@ -120,6 +120,7 @@ impl ConsumerBuilder {
                         }
                     }
                 } else {
+                    client.close().await?;
                     client = Client::connect(ClientOptions {
                         host: replica.host.clone(),
                         port: replica.port as u16,
@@ -271,6 +272,7 @@ impl ConsumerHandle {
                 let response = self.0.client.unsubscribe(self.0.subscription_id).await?;
                 if response.is_ok() {
                     self.0.waker.wake();
+                    self.0.client.close().await?;
                     Ok(())
                 } else {
                     Err(ConsumerCloseError::Close {

src/producer.rs

@@ -181,7 +181,10 @@ impl<T> ProducerBuilder<T> {
 
         let publish_sequence = if let Some(name) = self.name {
             let sequence = client.query_publisher_sequence(&name, stream).await?;
-            Arc::new(AtomicU64::new(sequence))
+
+            let first_sequence = if sequence == 0 { 0 } else { sequence + 1 };
+
+            Arc::new(AtomicU64::new(first_sequence))
         } else {
             Arc::new(AtomicU64::new(0))
         };

tests/integration/client_test.rs

@@ -63,6 +63,8 @@ async fn client_create_super_stream_error_test() {
 
     assert_eq!(&ResponseCode::StreamAlreadyExists, response.code());
 }
+
+#[tokio::test(flavor = "multi_thread")]
 async fn client_delete_stream_test() {
     let test = TestClient::create().await;
 

tests/integration/common.rs

@@ -1,9 +1,10 @@
-use std::collections::HashMap;
+use std::{collections::HashMap, future::Future, sync::Arc};
 
 use fake::{Fake, Faker};
 use rabbitmq_stream_client::{Client, ClientOptions, Environment};
 use rabbitmq_stream_protocol::commands::generic::GenericResponse;
 use rabbitmq_stream_protocol::ResponseCode;
+use tokio::sync::Semaphore;
 
 pub struct TestClient {
     pub client: Client,
@@ -12,6 +13,28 @@ pub struct TestClient {
     pub partitions: Vec<String>,
 }
 
+#[derive(Clone)]
+pub struct Countdown(Arc<Semaphore>);
+
+impl Drop for Countdown {
+    fn drop(&mut self) {
+        self.0.add_permits(1);
+    }
+}
+
+impl Countdown {
+    pub fn new(n: u32) -> (Self, impl Future + Send) {
+        let sem = Arc::new(Semaphore::new(0));
+        let latch = Self(sem.clone());
+
+        let wait = async move {
+            let _ = sem.acquire_many(n).await;
+        };
+
+        (latch, wait)
+    }
+}
+
 pub struct TestEnvironment {
     pub env: Environment,
     pub stream: String,

tests/integration/producer_test.rs

@@ -1,6 +1,8 @@
+use std::{collections::HashSet, sync::Arc};
+
 use chrono::Utc;
 use fake::{Fake, Faker};
-use futures::StreamExt;
+use futures::{lock::Mutex, StreamExt};
 use tokio::sync::mpsc::channel;
 
 use rabbitmq_stream_client::types::{
@@ -41,6 +43,57 @@ async fn producer_send_no_name_ok() {
     consumer.handle().close().await.unwrap();
 }
 
+#[tokio::test(flavor = "multi_thread")]
+async fn producer_send_name_deduplication_unique_ids() {
+    let env = TestEnvironment::create().await;
+
+    type Ids = Arc<Mutex<HashSet<u64>>>;
+
+    let ids = Ids::default();
+
+    let run = move |times, env: Environment, stream: String, ids: Ids| async move {
+        let (countdown, wait) = Countdown::new(times);
+        let mut producer = env
+            .producer()
+            .name("my_producer")
+            .build(&stream)
+            .await
+            .unwrap();
+
+        for _ in 0..times {
+            let cloned_ids = ids.clone();
+            let countdown = countdown.clone();
+            let _ = producer
+                .send(
+                    Message::builder().body(b"message".to_vec()).build(),
+                    move |result| {
+                        let value = cloned_ids.clone();
+
+                        let countdown = countdown.clone();
+                        async move {
+                            let _countdown = countdown;
+                            let id = result.unwrap().publishing_id();
+                            let mut guard = value.lock().await;
+                            guard.insert(id);
+                        }
+                    },
+                )
+                .await
+                .unwrap();
+        }
+
+        wait.await;
+
+        producer.close().await.unwrap();
+    };
+
+    run(10, env.env.clone(), env.stream.to_string(), ids.clone()).await;
+
+    run(10, env.env.clone(), env.stream.to_string(), ids.clone()).await;
+
+    assert_eq!(20, ids.lock().await.len());
+}
+
 #[tokio::test(flavor = "multi_thread")]
 async fn producer_send_name_with_deduplication_ok() {
     let env = TestEnvironment::create().await;