fixed infinite set keyspace notification loop

fixes #100

cdrs-tokio/src/cluster/config_tcp.rs

@@ -1,10 +1,10 @@
+use cassandra_protocol::authenticators::{NoneAuthenticatorProvider, SaslAuthenticatorProvider};
+use cassandra_protocol::error::Result;
+use cassandra_protocol::frame::Version;
 use std::net::SocketAddr;
 use std::sync::Arc;
 
 use crate::cluster::NodeAddress;
-use cassandra_protocol::authenticators::{NoneAuthenticatorProvider, SaslAuthenticatorProvider};
-use cassandra_protocol::error::Result;
-use cassandra_protocol::frame::Version;
 
 /// Single node TCP connection config.
 #[derive(Clone)]

cdrs-tokio/src/cluster/keyspace_holder.rs

@@ -25,12 +25,25 @@ impl KeyspaceHolder {
 
     #[inline]
     pub fn update_current_keyspace(&self, keyspace: String) {
-        self.update_current_keyspace_without_notification(keyspace.clone());
-        let _ = self.keyspace_sender.send(Some(keyspace));
+        let old_keyspace = self.current_keyspace.swap(Some(Arc::new(keyspace.clone())));
+        match &old_keyspace {
+            None => {
+                self.send_notification(keyspace);
+            }
+            Some(old_keyspace) if **old_keyspace != keyspace => {
+                self.send_notification(keyspace);
+            }
+            _ => {}
+        }
     }
 
     #[inline]
     pub fn update_current_keyspace_without_notification(&self, keyspace: String) {
         self.current_keyspace.store(Some(Arc::new(keyspace)));
     }
+
+    #[inline]
+    fn send_notification(&self, keyspace: String) {
+        let _ = self.keyspace_sender.send(Some(keyspace));
+    }
 }

changelog.md

@@ -5,6 +5,7 @@
 * `ExponentialReconnectionSchedule` duration overflow.
 * Forgetting real error type in certain transport error situations.
 * Not sending re-preparation statements to correct nodes.
+* Infinite set keyspace notification loop.
 
 ### New