Repository: scylladb/scylla-rust-driver
Pull request #202 (2021-03-25 17:15:00)

Receive server events
## Description

- Reduced control connection count to 1
- Added `Register` and `Event` frames
- Modified connection to allow sending `REGISTER` message
- Added mechanism to receive events and propagate them to cluster module
- Made cluster react to `TOPOLOGY_CHANGE` and `STATUS_CHANGE` events (and trigger topology metadata refresh Fixes: #85) 

## Design

`TopologyReader` creates control connections using special `ConnectionConfig` in which a sender half of mpsc channel is embedded. Control connection sends `REGISTER` message to a server and pushes every received event into its mpsc sender. `ClusterWorker` holds other half of this channel and processes events pushed by connection (triggers topology info refresh or deletes down nodes from associated `ClusterData`).

## Pre-review checklist

- [x] I have split my patch into logically separate commits.
- [x] All commit messages clearly explain what they change and why.
- [ ] 
 I added relevant tests for new features and bug fixes. 

- [x] All commits compile, pass static checks and pass test.
- [x] PR description sums up the changes and reasons why they should be introduced.
- [x] I added appropriate `Fixes:` annotations to PR description.


havaker: I have set up a cluster using [these instructions](https://university.scylladb.com/setup-a-scylla-cluster/) and tested receiving events (by looking into driver's logs and wireshark). I was able to generate:

- `STATUS_CHANGE` up & down by starting and stopping containers (`docker-compose start/stop scylla-nodeX`)
- `TOPOLOGY_CHANGE` removed_node (by stopping container and later using `nodetool removenode`) and new_node (starting previously removed node with `docker-compose start`)

Driver behaved correctly & emitted appropriate logs.
When a node that provided events went down (node to which a control connection was made), connection to another one was established during periodic topology refresh.


Patch:
diff --git a/.github/workflows/book.yml b/.github/workflows/book.yml
index 57b2451d..441a4bf8 100644
--- a/.github/workflows/book.yml
+++ b/.github/workflows/book.yml
@@ -12,7 +12,7 @@ env:
 
 jobs:
   build:
-    container: scylladb/rust-mdbook
+    container: cvybhu/rust-mdbook
     runs-on: ubuntu-latest
     services:
       scylladb:
diff --git a/.github/workflows/tls.yml b/.github/workflows/tls.yml
index 3d99593f..cc49b0a2 100644
--- a/.github/workflows/tls.yml
+++ b/.github/workflows/tls.yml
@@ -14,7 +14,7 @@ jobs:
     runs-on: ubuntu-latest
     services:
       scylladb:
-        image: scylladb/scylla-tls
+        image: kejmer/scylla-tls
         ports:
           - 9042:9042
           - 9142:9142
diff --git a/README.md b/README.md
index e9a4073b..d1be51c7 100644
--- a/README.md
+++ b/README.md
@@ -5,7 +5,7 @@
 This is a client-side driver for [ScyllaDB] written in pure Rust with a fully async API using [Tokio].
 Although optimized for ScyllaDB, the driver is also compatible with [Apache Cassandra®].
 
-**Note: this driver is officially supported but currently available in beta. Bug reports and pull requests are welcome!**
+**Note: this driver is not officially supported.**
 
 ## Getting Started
 The [documentation book](https://cvybhu.github.io/scyllabook/index.html) is a good place to get started
diff --git a/book/src/connecting/authentication.md b/book/src/connecting/authentication.md
index 35f87761..0aa9f586 100644
--- a/book/src/connecting/authentication.md
+++ b/book/src/connecting/authentication.md
@@ -1,22 +1,3 @@
 # Authentication
 
-Driver supports authentication by username and password.
-
-To specify credentials use the `user` method in `SessionBuilder`:
-
-```rust
-# extern crate scylla;
-# extern crate tokio;
-# use std::error::Error;
-# async fn check_only_compiles() -> Result<(), Box<dyn Error>> {
-use scylla::{Session, SessionBuilder};
-
-let session: Session = SessionBuilder::new()
-    .known_node("127.0.0.1:9042")
-    .user("myusername", "mypassword")
-    .build()
-    .await?;
-
-# Ok(())
-# }
-```
\ No newline at end of file
+Authentication is still in development, see pull request [#177](https://github.com/scylladb/scylla-rust-driver/pull/177)
\ No newline at end of file
diff --git a/book/src/connecting/tls.md b/book/src/connecting/tls.md
index 320b436e..863349eb 100644
--- a/book/src/connecting/tls.md
+++ b/book/src/connecting/tls.md
@@ -10,7 +10,7 @@ It was chosen because [`rustls`](https://github.com/ctz/rustls) doesn't support
 
 To enable the `tls` feature add in `Cargo.toml`:
 ```toml
-scylla = { version = "0.1.0", features = ["ssl"] }
+scylla = { git = "https://github.com/...", branch = "main", features = ["ssl"] }
 openssl = "0.10.32"
 ```
 
diff --git a/book/src/load-balancing/load-balancing.md b/book/src/load-balancing/load-balancing.md
index 9dbeff4c..391132a9 100644
--- a/book/src/load-balancing/load-balancing.md
+++ b/book/src/load-balancing/load-balancing.md
@@ -12,9 +12,6 @@ Each of these basic load balancing strategies can be wrapped in `TokenAwarePolic
 > **Note**  
 > Only [prepared queries](../queries/prepared.md) use token aware load balancing
 
-All queries are shard aware, there is no way to turn off shard awareness.  
-If a token is available the query is sent to the correct shard, otherwise to a random one.
-
 So, the available load balancing policies are:
 * [Round robin](robin.md)
 * [DC Aware Round robin](dc-robin.md)
diff --git a/book/src/queries/batch.md b/book/src/queries/batch.md
index d9c873d8..1071e358 100644
--- a/book/src/queries/batch.md
+++ b/book/src/queries/batch.md
@@ -64,9 +64,6 @@ session.batch(&batch, ((), )).await?;
 # }
 ```
 
-See [Batch API documentation](https://docs.rs/scylla/0.1.0/scylla/statement/batch/struct.Batch.html)
-for more options
-
 ### Batch values
 Batch takes a tuple of values specified just like in [simple](simple.md) or [prepared](prepared.md) queries.
 
diff --git a/book/src/queries/prepared.md b/book/src/queries/prepared.md
index 66a2998b..d25145c4 100644
--- a/book/src/queries/prepared.md
+++ b/book/src/queries/prepared.md
@@ -66,9 +66,6 @@ session.execute(&prepared, (to_insert,)).await?;
 # }
 ```
 
-See [PreparedStatement API documentation](https://docs.rs/scylla/0.1.0/scylla/statement/prepared_statement/struct.PreparedStatement.html) 
-for more options
-
 ### Performance
 Prepared queries have good performance, much better than simple queries.  
 By default they use shard/token aware load balancing.
diff --git a/book/src/queries/simple.md b/book/src/queries/simple.md
index 5503951a..c90324d6 100644
--- a/book/src/queries/simple.md
+++ b/book/src/queries/simple.md
@@ -40,7 +40,7 @@ session.query(my_query, (to_insert,)).await?;
 # Ok(())
 # }
 ```
-See [Query API documentation](https://docs.rs/scylla/0.1.0/scylla/statement/query/struct.Query.html) for more options
+See [Query API documentation (TODO)]() for more options
 
 ### Second argument - the values
 Query text is constant, but the values might change.
diff --git a/book/src/quickstart/create-project.md b/book/src/quickstart/create-project.md
index fa50f26c..540488b6 100644
--- a/book/src/quickstart/create-project.md
+++ b/book/src/quickstart/create-project.md
@@ -8,7 +8,7 @@ cargo new myproject
 In `Cargo.toml` add useful dependencies:
 ```toml
 [dependencies]
-scylla = "0.1.0"
+scylla = { git = "https://github.com/scylladb/scylla-rust-driver", branch = "main" }
 tokio = { version = "1.1.0", features = ["full"] }
 futures = "0.3.6"
 uuid = "0.8.1"
@@ -17,6 +17,8 @@ num-bigint = "0.3"
 tracing = "0.1.25"
 tracing-subscriber = "0.2.16"
 ```
+> Note that when specifying a dependency as a git link, updates will not be automatically pulled.
+> Running `cargo update` will update the git dependency manually.
 
 In `main.rs` put:
 ```rust
diff --git a/book/src/title-page.md b/book/src/title-page.md
index e4ab2a11..de5a8d93 100644
--- a/book/src/title-page.md
+++ b/book/src/title-page.md
@@ -5,7 +5,7 @@ Although optimized for Scylla, the driver is also compatible with [Apache Cassan
 
 ### Other documentation
 * [Examples](https://github.com/scylladb/scylla-rust-driver/tree/main/examples)
-* [API documentation](https://docs.rs/scylla)
+*  *API documentation* (Coming soon, for now download the [repo](https://github.com/scylladb/scylla-rust-driver) and run `cargo doc`)
 * [Scylla documentation](https://docs.scylladb.com)
 * [Cassandra® documentation](https://cassandra.apache.org/doc/latest/)
 
diff --git a/scylla-macros/Cargo.toml b/scylla-macros/Cargo.toml
index 72c7cf18..09f28663 100644
--- a/scylla-macros/Cargo.toml
+++ b/scylla-macros/Cargo.toml
@@ -6,7 +6,6 @@ description = "proc macros for scylla async CQL driver"
 repository = "https://github.com/scylladb/scylla-rust-driver"
 readme = "../README.md"
 categories = ["database"]
-license = "MIT OR Apache-2.0"
 
 [lib]
 proc-macro = true
diff --git a/scylla/Cargo.toml b/scylla/Cargo.toml
index d1a79953..5afc886c 100644
--- a/scylla/Cargo.toml
+++ b/scylla/Cargo.toml
@@ -1,13 +1,12 @@
 [package]
 name = "scylla"
-version = "0.1.1"
+version = "0.1.0"
 edition = "2018"
 description = "Async CQL driver for Rust, optimized for Scylla, fully compatible with Apache Cassandra™"
 repository = "https://github.com/scylladb/scylla-rust-driver"
 readme = "../README.md"
 keywords = ["database", "scylla", "cql", "cassandra"]
 categories = ["database"]
-license = "MIT OR Apache-2.0"
 
 [features]
 defaults = []
diff --git a/scylla/src/frame/mod.rs b/scylla/src/frame/mod.rs
index 9ea6373c..bc3b5e6e 100644
--- a/scylla/src/frame/mod.rs
+++ b/scylla/src/frame/mod.rs
@@ -1,6 +1,7 @@
 pub mod frame_errors;
 pub mod request;
 pub mod response;
+pub mod server_event_type;
 pub mod types;
 pub mod value;
 
diff --git a/scylla/src/frame/request/mod.rs b/scylla/src/frame/request/mod.rs
index 745c7d87..f09dd3c8 100644
--- a/scylla/src/frame/request/mod.rs
+++ b/scylla/src/frame/request/mod.rs
@@ -4,6 +4,7 @@ pub mod execute;
 pub mod options;
 pub mod prepare;
 pub mod query;
+pub mod register;
 pub mod startup;
 
 use crate::frame::frame_errors::ParseError;
diff --git a/scylla/src/frame/request/register.rs b/scylla/src/frame/request/register.rs
new file mode 100644
index 00000000..2c008b05
--- /dev/null
+++ b/scylla/src/frame/request/register.rs
@@ -0,0 +1,27 @@
+use bytes::BufMut;
+
+use crate::frame::{
+    frame_errors::ParseError,
+    request::{Request, RequestOpcode},
+    server_event_type::EventType,
+    types,
+};
+
+pub struct Register {
+    pub event_types_to_register_for: Vec<EventType>,
+}
+
+impl Request for Register {
+    const OPCODE: RequestOpcode = RequestOpcode::Register;
+
+    fn serialize(&self, buf: &mut impl BufMut) -> Result<(), ParseError> {
+        let event_types_list = self
+            .event_types_to_register_for
+            .iter()
+            .map(|event| event.to_string())
+            .collect::<Vec<_>>();
+
+        types::write_string_list(&event_types_list, buf)?;
+        Ok(())
+    }
+}
diff --git a/scylla/src/frame/response/event.rs b/scylla/src/frame/response/event.rs
new file mode 100644
index 00000000..9678a8b7
--- /dev/null
+++ b/scylla/src/frame/response/event.rs
@@ -0,0 +1,71 @@
+use crate::frame::frame_errors::ParseError;
+use crate::frame::server_event_type::EventType;
+use crate::frame::types;
+use std::net::SocketAddr;
+
+#[derive(Debug)]
+pub enum Event {
+    TopologyChange(TopologyChangeEvent),
+    StatusChange(StatusChangeEvent),
+    SchemaChange, // TODO specify
+}
+
+#[derive(Debug)]
+pub enum TopologyChangeEvent {
+    NewNode(SocketAddr),
+    RemovedNode(SocketAddr),
+}
+
+#[derive(Debug)]
+pub enum StatusChangeEvent {
+    Up(SocketAddr),
+    Down(SocketAddr),
+}
+
+impl Event {
+    pub fn deserialize(buf: &mut &[u8]) -> Result<Self, ParseError> {
+        let event_type: EventType = types::read_string(buf)?.parse()?;
+        match event_type {
+            EventType::TopologyChange => {
+                Ok(Self::TopologyChange(TopologyChangeEvent::deserialize(buf)?))
+            }
+            EventType::StatusChange => Ok(Self::StatusChange(StatusChangeEvent::deserialize(buf)?)),
+            EventType::SchemaChange => {
+                // TODO implement deserialization of SchemaChange
+                Ok(Self::SchemaChange)
+            }
+        }
+    }
+}
+
+impl TopologyChangeEvent {
+    pub fn deserialize(buf: &mut &[u8]) -> Result<Self, ParseError> {
+        let type_of_change = types::read_string(buf)?;
+        let addr = types::read_inet(buf)?;
+
+        match type_of_change {
+            "NEW_NODE" => Ok(Self::NewNode(addr)),
+            "REMOVED_NODE" => Ok(Self::RemovedNode(addr)),
+            _ => Err(ParseError::BadData(format!(
+                "Invalid type of change ({}) in TopologyChangeEvent",
+                type_of_change
+            ))),
+        }
+    }
+}
+
+impl StatusChangeEvent {
+    pub fn deserialize(buf: &mut &[u8]) -> Result<Self, ParseError> {
+        let type_of_change = types::read_string(buf)?;
+        let addr = types::read_inet(buf)?;
+
+        match type_of_change {
+            "UP" => Ok(Self::Up(addr)),
+            "DOWN" => Ok(Self::Down(addr)),
+            _ => Err(ParseError::BadData(format!(
+                "Invalid type of status change ({}) in StatusChangeEvent",
+                type_of_change
+            ))),
+        }
+    }
+}
diff --git a/scylla/src/frame/response/mod.rs b/scylla/src/frame/response/mod.rs
index a701e862..e9fee538 100644
--- a/scylla/src/frame/response/mod.rs
+++ b/scylla/src/frame/response/mod.rs
@@ -1,6 +1,7 @@
 pub mod authenticate;
 pub mod cql_to_rust;
 pub mod error;
+pub mod event;
 pub mod result;
 pub mod supported;
 
@@ -32,6 +33,7 @@ pub enum Response {
     AuthSuccess(authenticate::AuthSuccess),
     AuthChallenge(authenticate::AuthChallenge),
     Supported(Supported),
+    Event(event::Event),
 }
 
 impl Response {
@@ -44,7 +46,7 @@ impl Response {
             }
             ResponseOpcode::Supported => Response::Supported(Supported::deserialize(buf)?),
             ResponseOpcode::Result => Response::Result(result::deserialize(buf)?),
-            ResponseOpcode::Event => unimplemented!(),
+            ResponseOpcode::Event => Response::Event(event::Event::deserialize(buf)?),
             ResponseOpcode::AuthChallenge => {
                 Response::AuthChallenge(authenticate::AuthChallenge::deserialize(buf)?)
             }
diff --git a/scylla/src/frame/server_event_type.rs b/scylla/src/frame/server_event_type.rs
new file mode 100644
index 00000000..2bfd64b4
--- /dev/null
+++ b/scylla/src/frame/server_event_type.rs
@@ -0,0 +1,37 @@
+use crate::frame::frame_errors::ParseError;
+use std::fmt;
+use std::str::FromStr;
+
+pub enum EventType {
+    TopologyChange,
+    StatusChange,
+    SchemaChange,
+}
+
+impl fmt::Display for EventType {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        let s = match &self {
+            Self::TopologyChange => "TOPOLOGY_CHANGE",
+            Self::StatusChange => "STATUS_CHANGE",
+            Self::SchemaChange => "SCHEMA_CHANGE",
+        };
+
+        write!(f, "{}", s)
+    }
+}
+
+impl FromStr for EventType {
+    type Err = ParseError;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        match s {
+            "TOPOLOGY_CHANGE" => Ok(Self::TopologyChange),
+            "STATUS_CHANGE" => Ok(Self::StatusChange),
+            "SCHEMA_CHANGE" => Ok(Self::SchemaChange),
+            _ => Err(ParseError::BadData(format!(
+                "Invalid type event type: {}",
+                s
+            ))),
+        }
+    }
+}
diff --git a/scylla/src/frame/types.rs b/scylla/src/frame/types.rs
index f1873acc..18abed6a 100644
--- a/scylla/src/frame/types.rs
+++ b/scylla/src/frame/types.rs
@@ -2,11 +2,13 @@
 
 use super::frame_errors::ParseError;
 use byteorder::{BigEndian, ReadBytesExt};
-use bytes::BufMut;
+use bytes::{Buf, BufMut};
 use num_enum::TryFromPrimitive;
 use std::collections::HashMap;
 use std::convert::TryFrom;
 use std::convert::TryInto;
+use std::net::IpAddr;
+use std::net::SocketAddr;
 use std::str;
 use uuid::Uuid;
 
@@ -420,3 +422,61 @@ fn type_consistency() {
     let err_str = format!("{}", c_result.unwrap_err());
     assert!(err_str.contains(&format!("{}", c)));
 }
+
+pub fn read_inet(buf: &mut &[u8]) -> Result<SocketAddr, ParseError> {
+    let len = buf.read_u8()?;
+    let ip_addr = match len {
+        4 => {
+            let ret = IpAddr::from(<[u8; 4]>::try_from(&buf[0..4])?);
+            buf.advance(4);
+            ret
+        }
+        16 => {
+            let ret = IpAddr::from(<[u8; 16]>::try_from(&buf[0..16])?);
+            buf.advance(16);
+            ret
+        }
+        v => {
+            return Err(ParseError::BadData(format!(
+                "Invalid inet bytes length: {}",
+                v
+            )))
+        }
+    };
+    let port = read_int(buf)?;
+
+    Ok(SocketAddr::new(ip_addr, port as u16))
+}
+
+pub fn write_inet(addr: SocketAddr, buf: &mut impl BufMut) {
+    match addr.ip() {
+        IpAddr::V4(v4) => {
+            buf.put_u8(4);
+            buf.put_slice(&v4.octets());
+        }
+        IpAddr::V6(v6) => {
+            buf.put_u8(16);
+            buf.put_slice(&v6.octets());
+        }
+    }
+
+    write_int(addr.port() as i32, buf)
+}
+
+#[test]
+fn type_inet() {
+    use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
+
+    let iv4 = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 1234);
+    let iv6 = SocketAddr::new(IpAddr::V6(Ipv6Addr::new(0, 0, 0, 0, 0, 0, 0, 1)), 2345);
+    let mut buf = Vec::new();
+
+    write_inet(iv4, &mut buf);
+    let read_iv4 = read_inet(&mut &*buf).unwrap();
+    assert_eq!(iv4, read_iv4);
+    buf.clear();
+
+    write_inet(iv6, &mut buf);
+    let read_iv6 = read_inet(&mut &*buf).unwrap();
+    assert_eq!(iv6, read_iv6);
+}
diff --git a/scylla/src/frame/value.rs b/scylla/src/frame/value.rs
index 3004fbaf..26aaa9f8 100644
--- a/scylla/src/frame/value.rs
+++ b/scylla/src/frame/value.rs
@@ -105,13 +105,6 @@ impl SerializedValues {
         }
     }
 
-    pub fn with_capacity(capacity: usize) -> Self {
-        SerializedValues {
-            serialized_values: Vec::with_capacity(capacity),
-            values_num: 0,
-        }
-    }
-
     /// A const empty instance, useful for taking references
     pub const EMPTY: &'static SerializedValues = &SerializedValues::new();
 
@@ -529,7 +522,7 @@ impl ValueList for [u8; 0] {
 // Implement ValueList for slices of Value types
 impl<T: Value> ValueList for &[T] {
     fn serialized(&self) -> SerializedResult<'_> {
-        let mut result = SerializedValues::with_capacity(self.len());
+        let mut result = SerializedValues::new();
         for val in *self {
             result.add_value(val)?;
         }
@@ -541,7 +534,7 @@ impl<T: Value> ValueList for &[T] {
 // Implement ValueList for Vec<Value>
 impl<T: Value> ValueList for Vec<T> {
     fn serialized(&self) -> SerializedResult<'_> {
-        let mut result = SerializedValues::with_capacity(self.len());
+        let mut result = SerializedValues::new();
         for val in self {
             result.add_value(val)?;
         }
@@ -556,20 +549,20 @@ impl<T: Value> ValueList for Vec<T> {
 // Further variants are done using a macro
 impl<T0: Value> ValueList for (T0,) {
     fn serialized(&self) -> SerializedResult<'_> {
-        let mut result = SerializedValues::with_capacity(1);
+        let mut result = SerializedValues::new();
         result.add_value(&self.0)?;
         Ok(Cow::Owned(result))
     }
 }
 
 macro_rules! impl_value_list_for_tuple {
-    ( $($Ti:ident),* ; $($FieldI:tt),* ; $size: expr) => {
+    ( $($Ti:ident),* ; $($FieldI:tt),* ) => {
         impl<$($Ti),+> ValueList for ($($Ti,)+)
         where
             $($Ti: Value),+
         {
             fn serialized(&self) -> SerializedResult<'_> {
-                let mut result = SerializedValues::with_capacity($size);
+                let mut result = SerializedValues::new();
                 $(
                     result.add_value(&self.$FieldI) ?;
                 )*
@@ -579,28 +572,28 @@ macro_rules! impl_value_list_for_tuple {
     }
 }
 
-impl_value_list_for_tuple!(T0, T1; 0, 1; 2);
-impl_value_list_for_tuple!(T0, T1, T2; 0, 1, 2; 3);
-impl_value_list_for_tuple!(T0, T1, T2, T3; 0, 1, 2, 3; 4);
-impl_value_list_for_tuple!(T0, T1, T2, T3, T4; 0, 1, 2, 3, 4; 5);
-impl_value_list_for_tuple!(T0, T1, T2, T3, T4, T5; 0, 1, 2, 3, 4, 5; 6);
-impl_value_list_for_tuple!(T0, T1, T2, T3, T4, T5, T6; 0, 1, 2, 3, 4, 5, 6; 7);
-impl_value_list_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7; 0, 1, 2, 3, 4, 5, 6, 7; 8);
-impl_value_list_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8; 0, 1, 2, 3, 4, 5, 6, 7, 8; 9);
+impl_value_list_for_tuple!(T0, T1; 0, 1);
+impl_value_list_for_tuple!(T0, T1, T2; 0, 1, 2);
+impl_value_list_for_tuple!(T0, T1, T2, T3; 0, 1, 2, 3);
+impl_value_list_for_tuple!(T0, T1, T2, T3, T4; 0, 1, 2, 3, 4);
+impl_value_list_for_tuple!(T0, T1, T2, T3, T4, T5; 0, 1, 2, 3, 4, 5);
+impl_value_list_for_tuple!(T0, T1, T2, T3, T4, T5, T6; 0, 1, 2, 3, 4, 5, 6);
+impl_value_list_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7; 0, 1, 2, 3, 4, 5, 6, 7);
+impl_value_list_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8; 0, 1, 2, 3, 4, 5, 6, 7, 8);
 impl_value_list_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8, T9;
-                           0, 1, 2, 3, 4, 5, 6, 7, 8, 9; 10);
+                           0, 1, 2, 3, 4, 5, 6, 7, 8, 9);
 impl_value_list_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10;
-                           0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10; 11);
+                           0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
 impl_value_list_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11;
-                           0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11; 12);
+                           0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11);
 impl_value_list_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12;
-                           0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12; 13);
+                           0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12);
 impl_value_list_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13;
-                           0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13; 14);
+                           0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13);
 impl_value_list_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14;
-                           0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14; 15);
+                           0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14);
 impl_value_list_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15;
-                           0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15; 16);
+                           0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15);
 
 // Every &impl ValueList should also implement ValueList
 impl<T: ValueList> ValueList for &T {
diff --git a/scylla/src/transport/cluster.rs b/scylla/src/transport/cluster.rs
index c769bdcc..bae1f8b7 100644
--- a/scylla/src/transport/cluster.rs
+++ b/scylla/src/transport/cluster.rs
@@ -1,3 +1,4 @@
+use crate::frame::response::event::{Event, StatusChangeEvent};
 /// Cluster manages up to date information and connections to database nodes
 use crate::routing::Token;
 use crate::transport::connection::{Connection, ConnectionConfig, VerifiedKeyspaceName};
@@ -11,6 +12,7 @@ use itertools::Itertools;
 use std::collections::{BTreeMap, HashMap};
 use std::net::SocketAddr;
 use std::sync::{Arc, RwLock};
+use tracing::{debug, warn};
 
 /// Cluster manages up to date information and connections to database nodes.
 /// All data can be accessed by cloning Arc<ClusterData> in the `data` field
@@ -23,11 +25,13 @@ pub struct Cluster {
     _worker_handle: RemoteHandle<()>,
 }
 
+#[derive(Clone)]
 pub struct Datacenter {
     pub nodes: Vec<Arc<Node>>,
     pub rack_count: usize,
 }
 
+#[derive(Clone)]
 pub struct ClusterData {
     pub known_peers: HashMap<SocketAddr, Arc<Node>>, // Invariant: nonempty after Cluster::new()
     pub ring: BTreeMap<Token, Arc<Node>>,            // Invariant: nonempty after Cluster::new()
@@ -51,6 +55,9 @@ struct ClusterWorker {
     // Channel used to receive use keyspace requests
     use_keyspace_channel: tokio::sync::mpsc::Receiver<UseKeyspaceRequest>,
 
+    // Channel used to receive server events
+    server_events_channel: tokio::sync::mpsc::Receiver<Event>,
+
     // Keyspace send in "USE <keyspace name>" when opening each connection
     used_keyspace: Option<VerifiedKeyspaceName>,
 }
@@ -80,16 +87,21 @@ impl Cluster {
         })));
 
         let (refresh_sender, refresh_receiver) = tokio::sync::mpsc::channel(32);
-
         let (use_keyspace_sender, use_keyspace_receiver) = tokio::sync::mpsc::channel(32);
+        let (server_events_sender, server_events_receiver) = tokio::sync::mpsc::channel(32);
 
         let worker = ClusterWorker {
             cluster_data: cluster_data.clone(),
 
-            topology_reader: TopologyReader::new(initial_peers, connection_config.clone()),
+            topology_reader: TopologyReader::new(
+                initial_peers,
+                connection_config.clone(),
+                server_events_sender,
+            ),
             connection_config,
 
             refresh_channel: refresh_receiver,
+            server_events_channel: server_events_receiver,
 
             use_keyspace_channel: use_keyspace_receiver,
             used_keyspace: None,
@@ -303,6 +315,29 @@ impl ClusterWorker {
                         None => return, // If refresh_channel was closed then cluster was dropped, we can stop working
                     }
                 }
+                recv_res = self.server_events_channel.recv() => {
+                    if let Some(event) = recv_res {
+                        debug!("Received server event: {:?}", event);
+                        match event {
+                            Event::TopologyChange(_) => (), // Refresh immediately
+                            Event::StatusChange(status) => {
+                                // If some node went down/up, update it's marker and refresh
+                                // later as planned.
+
+                                match status {
+                                    StatusChangeEvent::Down(addr) => self.change_node_down_marker(addr, true),
+                                    StatusChangeEvent::Up(addr) => self.change_node_down_marker(addr, false),
+                                }
+                                continue;
+                            },
+                            _ => continue, // Don't go to refreshing
+                        }
+                    } else {
+                        // If server_events_channel was closed, than TopologyReader was dropped,
+                        // so we can probably stop working too
+                        return;
+                    }
+                }
                 recv_res = self.use_keyspace_channel.recv() => {
                     match recv_res {
                         Some(request) => {
@@ -320,6 +355,7 @@ impl ClusterWorker {
             }
 
             // Perform the refresh
+            debug!("Requesting topology refresh");
             last_refresh_time = Instant::now();
             let refresh_res = self.perform_refresh().await;
 
@@ -331,6 +367,20 @@ impl ClusterWorker {
         }
     }
 
+    fn change_node_down_marker(&mut self, addr: SocketAddr, is_down: bool) {
+        let cluster_data = self.cluster_data.read().unwrap().clone();
+
+        let node = match cluster_data.known_peers.get(&addr) {
+            Some(node) => node,
+            None => {
+                warn!("Unknown node address {}", addr);
+                return;
+            }
+        };
+
+        node.change_down_marker(is_down);
+    }
+
     async fn handle_use_keyspace_request(
         cluster_data: Arc<ClusterData>,
         request: UseKeyspaceRequest,
@@ -387,19 +437,22 @@ impl ClusterWorker {
         let topo_info = self.topology_reader.read_topology_info().await?;
         let cluster_data: Arc<ClusterData> = self.cluster_data.read().unwrap().clone();
 
-        let mut new_cluster_data = Arc::new(ClusterData::new(
+        let new_cluster_data = Arc::new(ClusterData::new(
             topo_info,
             &self.connection_config,
             &cluster_data.known_peers,
             &self.used_keyspace,
         ));
 
-        // Update current cluster_data
-        // Use std::mem::swap to avoid dropping large structures with active write lock
+        self.update_cluster_data(new_cluster_data);
+
+        Ok(())
+    }
+
+    fn update_cluster_data(&mut self, mut new_cluster_data: Arc<ClusterData>) {
         let mut cluster_data_lock = self.cluster_data.write().unwrap();
+        // Use std::mem::swap to avoid dropping large structures with active write lock
         std::mem::swap(&mut *cluster_data_lock, &mut new_cluster_data);
         drop(cluster_data_lock);
-
-        Ok(())
     }
 }
diff --git a/scylla/src/transport/connection.rs b/scylla/src/transport/connection.rs
index fafa1d7a..a5690f5f 100644
--- a/scylla/src/transport/connection.rs
+++ b/scylla/src/transport/connection.rs
@@ -29,8 +29,9 @@ use super::errors::{BadKeyspaceName, BadQuery, DbError, QueryError};
 use crate::batch::{Batch, BatchStatement};
 use crate::frame::{
     self,
-    request::{self, batch, execute, query, Request, RequestOpcode},
-    response::{result, Response, ResponseOpcode},
+    request::{self, batch, execute, query, register, Request, RequestOpcode},
+    response::{event::Event, result, Response, ResponseOpcode},
+    server_event_type::EventType,
     value::{BatchValues, ValueList},
     FrameParams, RequestBodyWithExtensions,
 };
@@ -124,6 +125,9 @@ pub struct ConnectionConfig {
     pub ssl_context: Option<SslContext>,
     pub auth_username: Option<String>,
     pub auth_password: Option<String>,
+
+    // should be Some only in control connections,
+    pub event_sender: Option<mpsc::Sender<Event>>,
     /*
     These configuration options will be added in the future:
 
@@ -141,6 +145,7 @@ impl Default for ConnectionConfig {
         Self {
             compression: None,
             tcp_nodelay: false,
+            event_sender: None,
             #[cfg(feature = "ssl")]
             ssl_context: None,
             auth_username: None,
@@ -158,7 +163,7 @@ impl Connection {
         addr: SocketAddr,
         source_port: Option<u16>,
         config: ConnectionConfig,
-    ) -> Result<(Self, ErrorReceiver), std::io::Error> {
+    ) -> Result<(Self, ErrorReceiver), QueryError> {
         let stream = match source_port {
             Some(p) => connect_with_source_port(addr, p).await?,
             None => TcpStream::connect(addr).await?,
@@ -171,7 +176,8 @@ impl Connection {
 
         let (error_sender, error_receiver) = tokio::sync::oneshot::channel();
 
-        let _worker_handle = Self::run_router(&config, stream, receiver, error_sender).await?;
+        let _worker_handle =
+            Self::run_router(config.clone(), stream, receiver, error_sender).await?;
 
         let connection = Connection {
             submit_channel: sender,
@@ -418,6 +424,27 @@ impl Connection {
         }
     }
 
+    async fn register(
+        &self,
+        event_types_to_register_for: Vec<EventType>,
+    ) -> Result<(), QueryError> {
+        let register_frame = register::Register {
+            event_types_to_register_for,
+        };
+
+        match self
+            .send_request(&register_frame, true, false)
+            .await?
+            .response
+        {
+            Response::Ready => Ok(()),
+            Response::Error(err) => Err(err.into()),
+            _ => Err(QueryError::ProtocolError(
+                "Unexpected response to REGISTER message",
+            )),
+        }
+    }
+
     async fn send_request<R: Request>(
         &self,
         request: &R,
@@ -458,9 +485,17 @@ impl Connection {
                 "Connection broken",
             )))
         })??;
+
+        Self::parse_response(task_response, self.config.compression)
+    }
+
+    fn parse_response(
+        task_response: TaskResponse,
+        compression: Option<Compression>,
+    ) -> Result<QueryResponse, QueryError> {
         let body_with_ext = frame::parse_response_body_extensions(
             task_response.params.flags,
-            self.config.compression,
+            compression,
             task_response.body,
         )?;
 
@@ -479,7 +514,7 @@ impl Connection {
 
     #[cfg(feature = "ssl")]
     async fn run_router(
-        config: &ConnectionConfig,
+        config: ConnectionConfig,
         stream: TcpStream,
         receiver: mpsc::Receiver<Task>,
         error_sender: tokio::sync::oneshot::Sender<QueryError>,
@@ -489,30 +524,36 @@ impl Connection {
                 let ssl = Ssl::new(context)?;
                 let mut stream = SslStream::new(ssl, stream)?;
                 let _pin = Pin::new(&mut stream).connect().await;
-                Self::run_router_spawner(stream, receiver, error_sender)
+                Self::run_router_spawner(stream, receiver, error_sender, config)
             }
-            None => Self::run_router_spawner(stream, receiver, error_sender),
+            None => Self::run_router_spawner(stream, receiver, error_sender, config),
         };
         Ok(res)
     }
 
     #[cfg(not(feature = "ssl"))]
     async fn run_router(
-        config: &ConnectionConfig,
+        config: ConnectionConfig,
         stream: TcpStream,
         receiver: mpsc::Receiver<Task>,
         error_sender: tokio::sync::oneshot::Sender<QueryError>,
     ) -> Result<RemoteHandle<()>, std::io::Error> {
         let _config = config;
-        Ok(Self::run_router_spawner(stream, receiver, error_sender))
+        Ok(Self::run_router_spawner(
+            stream,
+            receiver,
+            error_sender,
+            config,
+        ))
     }
 
     fn run_router_spawner(
         stream: (impl AsyncRead + AsyncWrite + Send + 'static),
         receiver: mpsc::Receiver<Task>,
         error_sender: tokio::sync::oneshot::Sender<QueryError>,
+        config: ConnectionConfig,
     ) -> RemoteHandle<()> {
-        let (task, handle) = Self::router(stream, receiver, error_sender).remote_handle();
+        let (task, handle) = Self::router(stream, receiver, error_sender, config).remote_handle();
         tokio::task::spawn(task);
         handle
     }
@@ -521,6 +562,7 @@ impl Connection {
         stream: (impl AsyncRead + AsyncWrite),
         receiver: mpsc::Receiver<Task>,
         error_sender: tokio::sync::oneshot::Sender<QueryError>,
+        config: ConnectionConfig,
     ) {
         let (read_half, write_half) = split(stream);
         // Why are using a mutex here?
@@ -537,7 +579,7 @@ impl Connection {
         // across .await points. Therefore, it should not be too expensive.
         let handler_map = StdMutex::new(ResponseHandlerMap::new());
 
-        let r = Self::reader(read_half, &handler_map);
+        let r = Self::reader(read_half, &handler_map, config);
         let w = Self::writer(write_half, &handler_map, receiver);
 
         let result = futures::try_join!(r, w);
@@ -563,9 +605,15 @@ impl Connection {
     async fn reader(
         mut read_half: (impl AsyncRead + Unpin),
         handler_map: &StdMutex<ResponseHandlerMap>,
+        config: ConnectionConfig,
     ) -> Result<(), QueryError> {
         loop {
             let (params, opcode, body) = frame::read_response_frame(&mut read_half).await?;
+            let response = TaskResponse {
+                params,
+                opcode,
+                body,
+            };
 
             match params.stream.cmp(&-1) {
                 Ordering::Less => {
@@ -575,7 +623,9 @@ impl Connection {
                     continue;
                 }
                 Ordering::Equal => {
-                    // TODO: Server events
+                    if let Some(event_sender) = config.event_sender.as_ref() {
+                        Self::handle_event(response, config.compression, event_sender).await?;
+                    }
                     continue;
                 }
                 _ => {}
@@ -592,11 +642,7 @@ impl Connection {
                 // Don't care if sending of the response fails. This must
                 // mean that the receiver side was impatient and is not
                 // waiting for the result anymore.
-                let _ = handler.send(Ok(TaskResponse {
-                    params,
-                    opcode,
-                    body,
-                }));
+                let _ = handler.send(Ok(response));
             } else {
                 // Unsolicited frame. This should not happen and indicates
                 // a bug either in the driver, or in the database
@@ -649,6 +695,28 @@ impl Connection {
         Ok(())
     }
 
+    async fn handle_event(
+        task_response: TaskResponse,
+        compression: Option<Compression>,
+        event_sender: &mpsc::Sender<Event>,
+    ) -> Result<(), QueryError> {
+        let response = Self::parse_response(task_response, compression)?.response;
+        let event = match response {
+            Response::Event(e) => e,
+            _ => {
+                warn!("Expected to receive Event response, got {:?}", response);
+                return Ok(());
+            }
+        };
+
+        event_sender.send(event).await.map_err(|_| {
+            QueryError::IoError(Arc::new(std::io::Error::new(
+                ErrorKind::Other,
+                "Connection broken",
+            )))
+        })
+    }
+
     pub fn get_shard_info(&self) -> &Option<ShardInfo> {
         &self.shard_info
     }
@@ -783,6 +851,15 @@ pub async fn open_named_connection(
         }
     }
 
+    if connection.config.event_sender.is_some() {
+        let all_event_types = vec![
+            EventType::TopologyChange,
+            EventType::StatusChange,
+            EventType::SchemaChange,
+        ];
+        connection.register(all_event_types).await?;
+    }
+
     Ok((connection, error_receiver))
 }
 
diff --git a/scylla/src/transport/connection_keeper.rs b/scylla/src/transport/connection_keeper.rs
index 60051f56..d694dc8e 100644
--- a/scylla/src/transport/connection_keeper.rs
+++ b/scylla/src/transport/connection_keeper.rs
@@ -309,8 +309,7 @@ mod tests {
             tcp_nodelay: false,
             #[cfg(feature = "ssl")]
             ssl_context: None,
-            auth_password: None,
-            auth_username: None,
+            ..Default::default()
         };
 
         // Get shard info from a single connection, all connections will open to this shard
diff --git a/scylla/src/transport/node.rs b/scylla/src/transport/node.rs
index a3e4d328..ec100183 100644
--- a/scylla/src/transport/node.rs
+++ b/scylla/src/transport/node.rs
@@ -8,10 +8,15 @@ use futures::future::join_all;
 
 use futures::{future::RemoteHandle, FutureExt};
 use rand::Rng;
-use std::convert::TryInto;
-use std::hash::{Hash, Hasher};
-use std::net::SocketAddr;
-use std::sync::{Arc, RwLock};
+use std::{
+    convert::TryInto,
+    hash::{Hash, Hasher},
+    net::SocketAddr,
+    sync::{
+        atomic::{AtomicBool, Ordering},
+        Arc, RwLock,
+    },
+};
 
 /// Node represents a cluster node along with it's data and connections
 pub struct Node {
@@ -21,6 +26,8 @@ pub struct Node {
 
     pub connections: Arc<RwLock<Arc<NodeConnections>>>,
 
+    down_marker: AtomicBool,
+
     use_keyspace_channel: tokio::sync::mpsc::Sender<UseKeyspaceRequest>,
 
     _worker_handle: RemoteHandle<()>,
@@ -108,6 +115,7 @@ impl Node {
             datacenter,
             rack,
             connections,
+            down_marker: false.into(),
             use_keyspace_channel: use_keyspace_sender,
             _worker_handle: worker_handle,
         }
@@ -149,6 +157,14 @@ impl Node {
         }
     }
 
+    pub fn is_down(&self) -> bool {
+        self.down_marker.load(Ordering::Relaxed)
+    }
+
+    pub fn change_down_marker(&self, is_down: bool) {
+        self.down_marker.store(is_down, Ordering::Relaxed);
+    }
+
     // Tries to get a connection to given shard, if it's broken returns any working connection
     async fn connection_for_shard(
         shard: u16,
diff --git a/scylla/src/transport/session.rs b/scylla/src/transport/session.rs
index 43980708..c0f60e3e 100644
--- a/scylla/src/transport/session.rs
+++ b/scylla/src/transport/session.rs
@@ -176,6 +176,7 @@ impl SessionConfig {
             ssl_context: self.ssl_context.clone(),
             auth_username: self.auth_username.to_owned(),
             auth_password: self.auth_password.to_owned(),
+            ..Default::default()
         }
     }
 }
diff --git a/scylla/src/transport/topology.rs b/scylla/src/transport/topology.rs
index dbaaa88c..23495869 100644
--- a/scylla/src/transport/topology.rs
+++ b/scylla/src/transport/topology.rs
@@ -1,18 +1,26 @@
+use crate::frame::response::event::Event;
 use crate::routing::Token;
 use crate::transport::connection::{Connection, ConnectionConfig};
 use crate::transport::connection_keeper::ConnectionKeeper;
 use crate::transport::errors::QueryError;
 use crate::transport::session::IntoTypedRows;
 
+use rand::seq::SliceRandom;
+use rand::thread_rng;
 use std::collections::HashMap;
 use std::net::{IpAddr, SocketAddr};
 use std::str::FromStr;
+use tokio::sync::mpsc;
+use tracing::{debug, error, warn};
 
 /// Allows to read current topology info from the cluster
 pub struct TopologyReader {
-    control_connections: HashMap<SocketAddr, ConnectionKeeper>,
-    connect_port: u16,
     connection_config: ConnectionConfig,
+    control_connection_address: SocketAddr,
+    control_connection: ConnectionKeeper,
+
+    // when control connection fails, TopologyReader tries to connect to one of known_peers
+    known_peers: Vec<SocketAddr>,
 }
 
 /// Describes all topology information retrieved from the cluster
@@ -51,79 +59,108 @@ pub enum Strategy {
 
 impl TopologyReader {
     /// Creates new TopologyReader, which connects to known_peers in the background
-    pub fn new(known_peers: &[SocketAddr], connection_config: ConnectionConfig) -> Self {
-        let connect_port: u16 = known_peers
-            .first()
-            .expect("Tried to initialize TopologyReader with empty known_peers list!")
-            .port();
-
-        let mut control_connections: HashMap<SocketAddr, ConnectionKeeper> =
-            HashMap::with_capacity(known_peers.len());
-
-        for address in known_peers {
-            control_connections.insert(
-                *address,
-                ConnectionKeeper::new(*address, connection_config.clone(), None, None, None),
-            );
-        }
+    pub fn new(
+        known_peers: &[SocketAddr],
+        mut connection_config: ConnectionConfig,
+        server_event_sender: mpsc::Sender<Event>,
+    ) -> Self {
+        let control_connection_address = *known_peers
+            .choose(&mut thread_rng())
+            .expect("Tried to initialize TopologyReader with empty known_peers list!");
+
+        // setting event_sender field in connection config will cause control connection to
+        // - send REGISTER message to receive server events
+        // - send received events via server_event_sender
+        connection_config.event_sender = Some(server_event_sender);
+
+        let control_connection = ConnectionKeeper::new(
+            control_connection_address,
+            connection_config.clone(),
+            None,
+            None,
+            None,
+        );
 
         TopologyReader {
-            control_connections,
-            connect_port,
+            control_connection_address,
+            control_connection,
             connection_config,
+            known_peers: known_peers.into(),
         }
     }
 
     /// Fetches current topology info from the cluster
     pub async fn read_topology_info(&mut self) -> Result<TopologyInfo, QueryError> {
-        let result = self.fetch_topology_info().await;
-
-        if let Ok(topology_info) = &result {
-            self.update_control_connections(topology_info);
+        let mut result = self.fetch_topology_info().await;
+        if result.is_ok() {
+            return result;
         }
 
-        return result;
-    }
+        // shuffle known_peers to iterate through them in random order later
+        self.known_peers.shuffle(&mut thread_rng());
 
-    async fn fetch_topology_info(&self) -> Result<TopologyInfo, QueryError> {
-        // TODO: Timeouts? New attempts in parallel?
+        let address_of_failed_control_connection = self.control_connection_address;
+        let filtered_known_peers = self
+            .known_peers
+            .iter()
+            .filter(|&peer| peer != &address_of_failed_control_connection);
+
+        // if fetching topology info on current control connection failed,
+        // try to fetch topology info from other known peer
+        for peer in filtered_known_peers {
+            let err = match result {
+                Ok(_) => break,
+                Err(err) => err,
+            };
 
-        let mut last_error: QueryError = QueryError::ProtocolError(
-            "Invariant broken - TopologyReader control_connections not empty",
-        );
+            warn!(
+                control_connection_address = self.control_connection_address.to_string().as_str(),
+                error = err.to_string().as_str(),
+                "Falied to fetch topology info using current control connection"
+            );
 
-        for conn in self.control_connections.values() {
-            match query_topology_info(conn, self.connect_port).await {
-                Ok(info) => return Ok(info),
-                Err(e) => last_error = e,
-            };
+            self.control_connection_address = *peer;
+            self.control_connection = ConnectionKeeper::new(
+                self.control_connection_address,
+                self.connection_config.clone(),
+                None,
+                None,
+                None,
+            );
+
+            result = self.fetch_topology_info().await;
+        }
+
+        match &result {
+            Ok(topology_info) => {
+                self.update_known_peers(topology_info);
+                debug!("Fetched new topology info");
+            }
+            Err(error) => error!(
+                error = error.to_string().as_str(),
+                "Could not fetch topology info"
+            ),
         }
 
-        Err(last_error)
+        result
     }
 
-    fn update_control_connections(&mut self, topology_info: &TopologyInfo) {
-        let mut new_control_connections: HashMap<SocketAddr, ConnectionKeeper> =
-            HashMap::with_capacity(topology_info.peers.len());
-
-        for peer in &topology_info.peers {
-            let cur_connection = self
-                .control_connections
-                .remove(&peer.address)
-                .unwrap_or_else(|| {
-                    ConnectionKeeper::new(
-                        peer.address,
-                        self.connection_config.clone(),
-                        None,
-                        None,
-                        None,
-                    )
-                });
-
-            new_control_connections.insert(peer.address, cur_connection);
-        }
+    async fn fetch_topology_info(&self) -> Result<TopologyInfo, QueryError> {
+        // TODO: Timeouts?
+
+        query_topology_info(
+            &self.control_connection,
+            self.control_connection_address.port(),
+        )
+        .await
+    }
 
-        self.control_connections = new_control_connections;
+    fn update_known_peers(&mut self, topology_info: &TopologyInfo) {
+        self.known_peers = topology_info
+            .peers
+            .iter()
+            .map(|peer| peer.address)
+            .collect();
     }
 }
 


Test patch:
diff --git a/.github/workflows/authenticate_test.yml b/.github/workflows/authenticate_test.yml
index 620f7e9e..7727dc47 100644
--- a/.github/workflows/authenticate_test.yml
+++ b/.github/workflows/authenticate_test.yml
@@ -15,7 +15,7 @@ jobs:
     runs-on: ubuntu-latest
     services:
       scylladb:
-        image: scylladb/scylla-passauth
+        image: cvybhu/scylla-passauth
         ports:
           - 9042:9042
         options: --health-cmd "cqlsh --username cassandra --password cassandra --debug" --health-interval 5s --health-retries 30
@@ -24,4 +24,4 @@ jobs:
     steps:
     - uses: actions/checkout@v2
     - name: Run tests
-      run: cargo test --verbose --  --ignored
+      run: cargo test --verbose --  --ignored
\ No newline at end of file