Begin implementing the sniffing of nodes

WIP
elastic · Sep 21, 2020 · 4de791d · 4de791d
1 parent 2705663
commit 4de791d
Showing 1 changed file with 18 additions and 3 deletions.
diff --git a/elasticsearch/src/http/transport.rs b/elasticsearch/src/http/transport.rs
@@ -34,6 +34,7 @@ use crate::{
         response::Response,
         Method,
     },
+    nodes::{NodesInfo, NodesInfoParts},
 };
 use base64::write::EncoderWriter as Base64Encoder;
 use bytes::BytesMut;
@@ -380,8 +381,11 @@ impl Transport {
         Q: Serialize + ?Sized,
     {
         if self.conn_pool.reseedable() {
-            // Reseed nodes
-            println!("Reseeding!");
+            self.conn_pool.reseeding();
+            // NodesInfo::new(&self, NodesInfoParts::None)
+            //     .send()
+            //     .await
+            //     .expect("Could not retrieve nodes for refresh");
         }
         let connection = self.conn_pool.next();
         let url = connection.url.join(path.trim_start_matches('/'))?;
@@ -471,6 +475,9 @@ pub trait ConnectionPool: Debug + dyn_clone::DynClone + Sync + Send {
         false
     }
 
+    // NOOP
+    fn reseeding(&self) {}
+
     // NOOP by default
     fn reseed(&self, _connection: Vec<Connection>) {}
 }
@@ -637,6 +644,7 @@ pub struct MultiNodeConnectionPool<TStrategy = RoundRobin> {
 
 #[derive(Debug, Clone)]
 pub struct MultiNodeConnectionPoolInner {
+    reseeding: bool,
     last_update: Option<Instant>,
     connections: Vec<Connection>,
 }
@@ -660,14 +668,20 @@ where
             .last_update
             .as_ref()
             .map(|last_update| last_update.elapsed() > wait);
-        last_update_is_stale.unwrap_or(true)
+        last_update_is_stale.unwrap_or(true) && !inner.reseeding
+    }
+
+    fn reseeding(&self) {
+        let mut inner = self.inner.write().expect("Lock Poisoned");
+        inner.reseeding = true
     }
 
     fn reseed(&self, mut connection: Vec<Connection>) {
         let mut inner = self.inner.write().expect("lock poisoned");
         inner.last_update = Some(Instant::now());
         inner.connections.clear();
         inner.connections.append(&mut connection);
+        inner.reseeding = false;
     }
 }
 
@@ -678,6 +692,7 @@ impl MultiNodeConnectionPool<RoundRobin> {
 
         let inner: Arc<RwLock<MultiNodeConnectionPoolInner>> =
             Arc::new(RwLock::new(MultiNodeConnectionPoolInner {
+                reseeding: false,
                 last_update: None,
                 connections,
             }));