test: switch to tokio-zookeeper

Switch to another clinet in order to improve stability

Author: DDtKey

Cargo.toml

@@ -125,7 +125,7 @@ tiberius = { version = "0.12.3", default-features = false, features = [
 ] }
 tokio = { version = "1", features = ["macros"] }
 tokio-util = { version = "0.7.10", features = ["compat"] }
-zookeeper-client = { version = "0.9.0" }
+tokio-zookeeper = "0.4.0"
 kube = { version = "0.98.0", default-features = false, features = [
     "client",
     "rustls-tls",

src/zookeeper/mod.rs

@@ -14,24 +14,28 @@ const TAG: &str = "3.9.0";
 /// # Example
 ///
 /// ```
-/// use testcontainers_modules::{testcontainers::runners::AsyncRunner, zookeeper};
-/// # tokio::runtime::Runtime::new().unwrap().block_on(async {
-/// let node = zookeeper::Zookeeper::default().start().await.unwrap();
-/// let zk_url = format!(
-///     "{}:{}",
-///     node.get_host().await.unwrap(),
-///     node.get_host_port_ipv4(2181).await.unwrap(),
-/// );
-/// let client = zookeeper_client::Client::connect(&zk_url)
-///     .await
-///     .expect("connect to Zookeeper");
+/// async {
+///     use testcontainers_modules::{testcontainers::runners::AsyncRunner, zookeeper};
 ///
-/// let path = "/test";
-/// let (_, stat_watcher) = client
-///     .check_and_watch_stat(path)
-///     .await
-///     .expect("stat watcher created");
-/// # })
+///     let node = zookeeper::Zookeeper::default().start().await.unwrap();
+///     let zk_url = format!(
+///         "{}:{}",
+///         node.get_host().await.unwrap(),
+///         node.get_host_port_ipv4(2181).await.unwrap(),
+///     );
+///     let zk_socket_addr = tokio::net::lookup_host(&zk_url)
+///         .await
+///         .unwrap()
+///         .next()
+///         .unwrap();
+///
+///     let (zk, default_watcher) = tokio_zookeeper::ZooKeeper::connect(&zk_socket_addr)
+///         .await
+///         .expect("connect to Zookeeper");
+///
+///     let path = "/test";
+///     let _stat = zk.watch().exists(path).await.expect("stat received");
+/// };
 /// ```
 ///
 ///
@@ -71,8 +75,10 @@ impl Image for Zookeeper {
 
 #[cfg(test)]
 mod tests {
+    use futures::StreamExt;
     use rustls::crypto::CryptoProvider;
-    use zookeeper_client::{Acls, Client, CreateMode, EventType};
+    use tokio::net::lookup_host;
+    use tokio_zookeeper::*;
 
     use crate::{testcontainers::runners::AsyncRunner, zookeeper::Zookeeper as ZookeeperImage};
 
@@ -88,27 +94,23 @@ mod tests {
 
         let node = ZookeeperImage::default().start().await?;
 
-        let host_ip = node.get_host().await?;
+        let host = node.get_host().await?;
         let host_port = node.get_host_port_ipv4(2181).await?;
-        let zk_url = format!("{host_ip}:{host_port}");
-        let client = Client::connect(&zk_url)
-            .await
-            .expect("connect to Zookeeper");
+        let zk_url = format!("{host}:{host_port}");
+        let zk_socket_addr = lookup_host(&zk_url).await?.next().unwrap();
+
+        let (zk, mut default_watcher) = ZooKeeper::connect(&zk_socket_addr).await.unwrap();
 
         let path = "/test";
-        let (_, stat_watcher) = client
-            .check_and_watch_stat(path)
-            .await
-            .expect("stat watcher created");
-
-        let create_options = CreateMode::Ephemeral.with_acls(Acls::anyone_all());
-        let (_, _) = client
-            .create(path, &[1, 2], &create_options)
-            .await
+        let _stat = zk.watch().exists(path).await.expect("stat requested");
+
+        let path = zk
+            .create(path, &[1, 2], Acl::open_unsafe(), CreateMode::Ephemeral)
+            .await?
             .expect("create a node");
 
-        let event = stat_watcher.changed().await;
-        assert_eq!(event.event_type, EventType::NodeCreated);
+        let event = default_watcher.next().await.expect("event received");
+        assert_eq!(event.event_type, WatchedEventType::NodeCreated);
         assert_eq!(event.path, path);
         Ok(())
     }