feat: option to fail eagerly with Error::NoHosts

This reverses the default behavior and give us chance to not fail with
Error::NoHosts.

This commit also refactors test_readonly a bit with help from above.

Closes #34.

Author: kezhuw

src/client/mod.rs

@@ -1541,6 +1541,7 @@ pub struct Connector {
     session: Option<SessionInfo>,
     readonly: bool,
     detached: bool,
+    fail_eagerly: bool,
     server_version: Version,
     session_timeout: Duration,
     connection_timeout: Duration,
@@ -1555,6 +1556,7 @@ impl Connector {
             session: None,
             readonly: false,
             detached: false,
+            fail_eagerly: false,
             server_version: Version(u32::MAX, u32::MAX, u32::MAX),
             session_timeout: Duration::ZERO,
             connection_timeout: Duration::ZERO,
@@ -1621,6 +1623,15 @@ impl Connector {
         self
     }
 
+    /// Fail session establishment eagerly with [Error::NoHosts] when all hosts has been tried.
+    ///
+    /// This permits fail-fast without wait up to [Self::session_timeout] in [Self::connect]. This
+    /// is not suitable for situations where ZooKeeper cluster is accessible via a single virtual IP.
+    pub fn fail_eagerly(&mut self) -> &mut Self {
+        self.fail_eagerly = true;
+        self
+    }
+
     async fn connect_internally(&mut self, secure: bool, cluster: &str) -> Result<Client> {
         let (endpoints, chroot) = endpoint::parse_connect_string(cluster, secure)?;
         if let Some(session) = self.session.as_ref() {
@@ -1647,6 +1658,9 @@ impl Connector {
             self.connection_timeout,
         );
         let mut endpoints = IterableEndpoints::from(endpoints.as_slice());
+        if !self.fail_eagerly {
+            endpoints.cycle();
+        }
         let mut buf = Vec::with_capacity(4096);
         let mut connecting_depot = Depot::for_connecting();
         let conn = session.start(&mut endpoints, &mut buf, &mut connecting_depot).await?;
@@ -1677,8 +1691,9 @@ impl Connector {
     /// plaintext protocol, while `server3` uses tls encrypted protocol.
     ///
     /// # Notable errors
-    /// * [Error::NoHosts] if no host is available
+    /// * [Error::NoHosts] if no host is available and [Self::fail_eagerly] is turn on
     /// * [Error::SessionExpired] if specified session expired
+    /// * [Error::Timeout] if no session established with in approximate [Self::session_timeout]
     ///
     /// # Notable behaviors
     /// The state of this connector is undefined after connection attempt no matter whether it is

tests/zookeeper.rs

@@ -156,7 +156,23 @@ async fn connect(cluster: &Cluster, chroot: &str) -> zk::Client {
 
 #[test_log::test(tokio::test)]
 async fn test_connect_nohosts() {
-    assert_that!(zk::Client::connect("127.0.0.1:100,127.0.0.1:101").await.unwrap_err()).is_equal_to(zk::Error::NoHosts);
+    assert_that!(zk::Client::connector()
+        .session_timeout(Duration::from_secs(24 * 3600))
+        .fail_eagerly()
+        .connect("127.0.0.1:100,127.0.0.1:101")
+        .await
+        .unwrap_err())
+    .is_equal_to(zk::Error::NoHosts);
+}
+
+#[test_log::test(tokio::test)]
+async fn test_connect_timeout() {
+    assert_that!(zk::Client::connector()
+        .session_timeout(Duration::from_secs(1))
+        .connect("127.0.0.1:100,127.0.0.1:101")
+        .await
+        .unwrap_err())
+    .is_equal_to(zk::Error::Timeout);
 }
 
 #[test_log::test(tokio::test)]
@@ -1807,18 +1823,12 @@ async fn test_readonly(tls: bool) {
 
     logs.wait_for_message("Read-only server started").unwrap();
 
-    // It takes time for the last server to serve request, so let's spin on connecting.
-    let mut timeout = tokio::time::sleep(Duration::from_secs(60));
-    let client = loop {
-        let mut connector = zk::Client::connector();
-        connector.session_timeout(Duration::from_secs(60)).readonly(true);
-        select! {
-            _ = unsafe { Pin::new_unchecked(&mut timeout) } => panic!("expect ConnectedReadOnly, but got no session"),
-            result = connector.connect("localhost:4001,localhost:4002,localhost:4003") => if let Ok(client) = result {
-                break client
-            },
-        }
-    };
+    let client = zk::Client::connector()
+        .readonly(true)
+        .session_timeout(Duration::from_secs(60))
+        .connect("localhost:4001,localhost:4002,localhost:4003")
+        .await
+        .unwrap();
     assert_that!(client.create("/y", b"", PERSISTENT_OPEN).await.unwrap_err()).is_equal_to(zk::Error::NotReadOnly);
 
     let session = client.session().clone();