Repository: AlexPikalov/cdrs (language: rust)

Issue:
[v2.0.0] Session externally non-mutable
Currently when using Session it cannot be shared among types. I've historically organized my database functions into DAO classes that provide additional functionality as needed.

As an example...

```rust
fn get_user_group(pool: &r2d2::Pool...) {
   let sess = pool.get();
   let users_dao = UsersDAO::new(&sess);
   let groups_dao = GroupsDAO::new(&sess);
   let user = users_dao.get(...);
   let group = groups_dao.get(...);
}
```

Notably passing &sess isn't possible in the above example in 1.x as sess needs to be mutable. By instead changing sess to only be internally mutable using Cell/RefCell would allow external users to share a non-mutable reference to Session in such a manner.

To be clear I could wrap the entire Session in an ```Rc<RefCell<Session>>``` to mimic what I'd like to do here, but I think it would be nice if the cdrs crate supported the above pattern without additional work for me, the user.


Discussion / hints:
@bfrog I see what you mean. I'll try to play around session in v2 and will let you know about my results.
I'd gladly create a PR for this, and its not exactly a breaking change though it will probably give a bunch of warnings about mut not being needed anymore
Sure, feel free to submit a PR.

Fix patch:
diff --git a/examples/crud_operations.rs b/examples/crud_operations.rs
index 8744fd4..5f27229 100644
--- a/examples/crud_operations.rs
+++ b/examples/crud_operations.rs
@@ -5,33 +5,34 @@ extern crate cdrs_helpers_derive;
 #[macro_use]
 extern crate maplit;
 
+use std::cell::RefCell;
 use std::collections::HashMap;
 
 use cdrs::authenticators::NoneAuthenticator;
 use cdrs::cluster::{Cluster, Session};
+use cdrs::load_balancing::RoundRobin;
 use cdrs::query::*;
-use cdrs::load_balancing::{RoundRobin};
 use cdrs::transport::TransportTcp;
 
-use cdrs::types::prelude::*;
-use cdrs::types::from_cdrs::FromCDRSByName;
 use cdrs::frame::IntoBytes;
+use cdrs::types::from_cdrs::FromCDRSByName;
+use cdrs::types::prelude::*;
 
-type CurrentSession = Session<RoundRobin<TransportTcp>, NoneAuthenticator>;
+type CurrentSession = Session<RoundRobin<RefCell<TransportTcp>>, NoneAuthenticator>;
 
 fn main() {
   let cluster = Cluster::new(vec!["127.0.0.1:9042"], NoneAuthenticator {});
-  let mut no_compression = cluster
+  let no_compression = cluster
     .connect(RoundRobin::new())
     .expect("No compression connection error");
 
-  create_keyspace(&mut no_compression);
-  create_udt(&mut no_compression);
-  create_table(&mut no_compression);
-  insert_struct(&mut no_compression);
-  select_struct(&mut no_compression);
-  update_struct(&mut no_compression);
-  delete_struct(&mut no_compression);
+  create_keyspace(&no_compression);
+  create_udt(&no_compression);
+  create_table(&no_compression);
+  insert_struct(&no_compression);
+  select_struct(&no_compression);
+  update_struct(&no_compression);
+  delete_struct(&no_compression);
 }
 
 #[derive(Clone, Debug, IntoCDRSValue, TryFromRow, PartialEq)]
@@ -53,39 +54,38 @@ struct User {
   username: String,
 }
 
-fn create_keyspace(session: &mut CurrentSession) {
+fn create_keyspace(session: &CurrentSession) {
   let create_ks: &'static str = "CREATE KEYSPACE IF NOT EXISTS test_ks WITH REPLICATION = { \
                                  'class' : 'SimpleStrategy', 'replication_factor' : 1 };";
   session.query(create_ks).expect("Keyspace creation error");
 }
 
-fn create_udt(session: &mut CurrentSession) {
+fn create_udt(session: &CurrentSession) {
   let create_type_cql = "CREATE TYPE IF NOT EXISTS test_ks.user (username text)";
   session
     .query(create_type_cql)
     .expect("Keyspace creation error");
 }
 
-fn create_table(session: &mut CurrentSession) {
-  let create_table_cql = "CREATE TABLE IF NOT EXISTS test_ks.my_test_table (key int PRIMARY KEY, \
-                          user test_ks.user, map map<text, frozen<test_ks.user>>, list list<frozen<test_ks.user>>);";
+fn create_table(session: &CurrentSession) {
+  let create_table_cql =
+    "CREATE TABLE IF NOT EXISTS test_ks.my_test_table (key int PRIMARY KEY, \
+     user test_ks.user, map map<text, frozen<test_ks.user>>, list list<frozen<test_ks.user>>);";
   session
     .query(create_table_cql)
     .expect("Table creation error");
 }
 
-fn insert_struct(session: &mut CurrentSession) {
+fn insert_struct(session: &CurrentSession) {
   let row = RowStruct {
     key: 3i32,
     user: User {
       username: "John".to_string(),
     },
     map: hashmap! { "John".to_string() => User { username: "John".to_string() } },
-    list: vec![
-      User {
-        username: "John".to_string(),
-      },
-    ],
+    list: vec![User {
+      username: "John".to_string(),
+    }],
   };
 
   let insert_struct_cql = "INSERT INTO test_ks.my_test_table \
@@ -95,7 +95,7 @@ fn insert_struct(session: &mut CurrentSession) {
     .expect("insert");
 }
 
-fn select_struct(session: &mut CurrentSession) {
+fn select_struct(session: &CurrentSession) {
   let select_struct_cql = "SELECT * FROM test_ks.my_test_table";
   let rows = session
     .query(select_struct_cql)
@@ -111,7 +111,7 @@ fn select_struct(session: &mut CurrentSession) {
   }
 }
 
-fn update_struct(session: &mut CurrentSession) {
+fn update_struct(session: &CurrentSession) {
   let update_struct_cql = "UPDATE test_ks.my_test_table SET user = ? WHERE key = ?";
   let upd_user = User {
     username: "Marry".to_string(),
@@ -122,7 +122,7 @@ fn update_struct(session: &mut CurrentSession) {
     .expect("update");
 }
 
-fn delete_struct(session: &mut CurrentSession) {
+fn delete_struct(session: &CurrentSession) {
   let delete_struct_cql = "DELETE FROM test_ks.my_test_table WHERE key = ?";
   let user_key = 1i32;
   session
diff --git a/examples/paged_query.rs b/examples/paged_query.rs
index ca3a927..3e909ee 100644
--- a/examples/paged_query.rs
+++ b/examples/paged_query.rs
@@ -3,17 +3,19 @@ extern crate cdrs;
 #[macro_use]
 extern crate cdrs_helpers_derive;
 
+use std::cell::RefCell;
+
 use cdrs::authenticators::NoneAuthenticator;
 use cdrs::cluster::{Cluster, Session};
-use cdrs::query::*;
 use cdrs::load_balancing::RoundRobin;
+use cdrs::query::*;
 use cdrs::transport::TransportTcp;
 
-use cdrs::types::prelude::*;
-use cdrs::types::from_cdrs::FromCDRSByName;
 use cdrs::frame::IntoBytes;
+use cdrs::types::from_cdrs::FromCDRSByName;
+use cdrs::types::prelude::*;
 
-type CurrentSession = Session<RoundRobin<TransportTcp>, NoneAuthenticator>;
+type CurrentSession = Session<RoundRobin<RefCell<TransportTcp>>, NoneAuthenticator>;
 
 #[derive(Clone, Debug, IntoCDRSValue, TryFromRow, PartialEq)]
 struct RowStruct {
@@ -28,35 +30,36 @@ impl RowStruct {
 
 fn main() {
   let cluster = Cluster::new(vec!["127.0.0.1:9042"], NoneAuthenticator {});
-  let mut no_compression = cluster
+  let no_compression = cluster
     .connect(RoundRobin::new())
     .expect("No compression connection error");
 
-  create_keyspace(&mut no_compression);
-  create_table(&mut no_compression);
-  fill_table(&mut no_compression);
-  paged_selection_query(&mut no_compression);
+  create_keyspace(&no_compression);
+  create_table(&no_compression);
+  fill_table(&no_compression);
+  paged_selection_query(&no_compression);
 }
 
-fn create_keyspace(session: &mut CurrentSession) {
+fn create_keyspace(session: &CurrentSession) {
   let create_ks: &'static str = "CREATE KEYSPACE IF NOT EXISTS test_ks WITH REPLICATION = { \
                                  'class' : 'SimpleStrategy', 'replication_factor' : 1 };";
   session.query(create_ks).expect("Keyspace creation error");
 }
 
-fn create_table(session: &mut CurrentSession) {
-  let create_table_cql = "CREATE TABLE IF NOT EXISTS test_ks.my_test_table (key int PRIMARY KEY, \
-                          user test_ks.user, map map<text, frozen<test_ks.user>>, list list<frozen<test_ks.user>>);";
+fn create_table(session: &CurrentSession) {
+  let create_table_cql =
+    "CREATE TABLE IF NOT EXISTS test_ks.my_test_table (key int PRIMARY KEY, \
+     user test_ks.user, map map<text, frozen<test_ks.user>>, list list<frozen<test_ks.user>>);";
   session
     .query(create_table_cql)
     .expect("Table creation error");
 }
 
-fn fill_table(session: &mut CurrentSession) {
+fn fill_table(session: &CurrentSession) {
   let insert_struct_cql = "INSERT INTO test_ks.my_test_table (key) VALUES (?)";
 
   for k in 100..110 {
-    let row = RowStruct {key: k as i32};
+    let row = RowStruct { key: k as i32 };
 
     session
       .query_with_values(insert_struct_cql, row.into_query_values())
@@ -64,7 +67,7 @@ fn fill_table(session: &mut CurrentSession) {
   }
 }
 
-fn paged_selection_query(session: &mut CurrentSession) {
+fn paged_selection_query(session: &CurrentSession) {
   let q = "SELECT * FROM test_ks.my_test_table;";
   let mut pager = session.paged(2);
   let mut query_pager = pager.query(q);
diff --git a/examples/prepare_batch_execute.rs b/examples/prepare_batch_execute.rs
index 910a3e9..79fdda0 100644
--- a/examples/prepare_batch_execute.rs
+++ b/examples/prepare_batch_execute.rs
@@ -5,17 +5,19 @@ extern crate cdrs_helpers_derive;
 #[macro_use]
 extern crate maplit;
 
+use std::cell::RefCell;
+
 use cdrs::authenticators::NoneAuthenticator;
 use cdrs::cluster::{Cluster, Session};
-use cdrs::query::*;
 use cdrs::load_balancing::RoundRobin;
+use cdrs::query::*;
 use cdrs::transport::TransportTcp;
 
-use cdrs::types::prelude::*;
-use cdrs::types::from_cdrs::FromCDRSByName;
 use cdrs::frame::IntoBytes;
+use cdrs::types::from_cdrs::FromCDRSByName;
+use cdrs::types::prelude::*;
 
-type CurrentSession = Session<RoundRobin<TransportTcp>, NoneAuthenticator>;
+type CurrentSession = Session<RoundRobin<RefCell<TransportTcp>>, NoneAuthenticator>;
 
 #[derive(Clone, Debug, IntoCDRSValue, TryFromRow, PartialEq)]
 struct RowStruct {
@@ -32,12 +34,12 @@ impl RowStruct {
 
 fn main() {
   let cluster = Cluster::new(vec!["127.0.0.1:9042"], NoneAuthenticator {});
-  let mut no_compression = cluster
+  let no_compression = cluster
     .connect(RoundRobin::new())
     .expect("No compression connection error");
 
-  create_keyspace(&mut no_compression);
-  create_table(&mut no_compression);
+  create_keyspace(&no_compression);
+  create_table(&no_compression);
 
   let insert_struct_cql = "INSERT INTO test_ks.my_test_table (key) VALUES (?)";
   let prepared_query = no_compression
@@ -47,32 +49,32 @@ fn main() {
   for k in 100..110 {
     let row = RowStruct { key: k as i32 };
 
-    insert_row(&mut no_compression, row, &prepared_query);
+    insert_row(&no_compression, row, &prepared_query);
   }
 
-  batch_few_queries(&mut no_compression, &insert_struct_cql);
+  batch_few_queries(&no_compression, &insert_struct_cql);
 }
 
-fn create_keyspace(session: &mut CurrentSession) {
+fn create_keyspace(session: &CurrentSession) {
   let create_ks: &'static str = "CREATE KEYSPACE IF NOT EXISTS test_ks WITH REPLICATION = { \
                                  'class' : 'SimpleStrategy', 'replication_factor' : 1 };";
   session.query(create_ks).expect("Keyspace creation error");
 }
 
-fn create_table(session: &mut CurrentSession) {
+fn create_table(session: &CurrentSession) {
   let create_table_cql = "CREATE TABLE IF NOT EXISTS test_ks.my_test_table (key int PRIMARY KEY);";
   session
     .query(create_table_cql)
     .expect("Table creation error");
 }
 
-fn insert_row(session: &mut CurrentSession, row: RowStruct, prepared_query: &PreparedQuery) {
+fn insert_row(session: &CurrentSession, row: RowStruct, prepared_query: &PreparedQuery) {
   session
     .exec_with_values(prepared_query, row.into_query_values())
     .expect("exec_with_values error");
 }
 
-fn batch_few_queries(session: &mut CurrentSession, query: &str) {
+fn batch_few_queries(session: &CurrentSession, query: &str) {
   let prepared_query = session.prepare(query).expect("Prepare query error");
   let row_1 = RowStruct { key: 1001 as i32 };
   let row_2 = RowStruct { key: 2001 as i32 };
diff --git a/examples/server_events.rs b/examples/server_events.rs
index 96473be..5a5d854 100644
--- a/examples/server_events.rs
+++ b/examples/server_events.rs
@@ -1,27 +1,28 @@
 extern crate cdrs;
 
-use std::thread;
+use std::cell::RefCell;
 use std::iter::Iterator;
+use std::thread;
 
-use cdrs::authenticators::{Authenticator, NoneAuthenticator};
+use cdrs::authenticators::NoneAuthenticator;
 use cdrs::cluster::{Cluster, Session};
-use cdrs::load_balancing::RoundRobin;
-use cdrs::transport::TransportTcp;
 use cdrs::compression::Compression;
 use cdrs::frame::events::{ChangeType, ServerEvent, SimpleServerEvent, Target};
+use cdrs::load_balancing::RoundRobin;
+use cdrs::transport::TransportTcp;
 
 const _ADDR: &'static str = "127.0.0.1:9042";
 
-type CurrentSession = Session<RoundRobin<TransportTcp>, NoneAuthenticator>;
+type CurrentSession = Session<RoundRobin<RefCell<TransportTcp>>, NoneAuthenticator>;
 
 fn main() {
   let cluster = Cluster::new(vec!["127.0.0.1:9042"], NoneAuthenticator {});
-  let mut no_compression = cluster
+  let no_compression: CurrentSession = cluster
     .connect(RoundRobin::new())
     .expect("No compression connection error");
 
-  let (mut listener, stream) = no_compression
-    .listen(vec![SimpleServerEvent::SchemaChange])
+  let (listener, stream) = no_compression
+    .listen("127.0.0.1:9042", vec![SimpleServerEvent::SchemaChange])
     .expect("listen error");
 
   thread::spawn(move || listener.start(&Compression::None).unwrap());
diff --git a/src/cluster/cluster.rs b/src/cluster/cluster.rs
index faba7ae..1b6c681 100644
--- a/src/cluster/cluster.rs
+++ b/src/cluster/cluster.rs
@@ -1,14 +1,15 @@
-use error;
+use std::cell::RefCell;
 
 use authenticators::Authenticator;
-use transport::TransportTcp;
 use cluster::session::Session;
+use error;
 use load_balancing::LoadBalancingStrategy;
+use transport::TransportTcp;
 
-#[cfg(feature = "ssl")]
-use transport::TransportTls;
 #[cfg(feature = "ssl")]
 use openssl::ssl::SslConnector;
+#[cfg(feature = "ssl")]
+use transport::TransportTls;
 
 pub struct Cluster<A> {
   nodes_addrs: Vec<&'static str>,
@@ -25,14 +26,14 @@ impl<'a, A: Authenticator + Sized> Cluster<A> {
 
   pub fn connect<LB>(&self, lb: LB) -> error::Result<Session<LB, A>>
   where
-    LB: LoadBalancingStrategy<TransportTcp> + Sized,
+    LB: LoadBalancingStrategy<RefCell<TransportTcp>> + Sized,
   {
     Session::new(&self.nodes_addrs, lb, self.authenticator.clone())
   }
 
   pub fn connect_snappy<LB>(&self, lb: LB) -> error::Result<Session<LB, A>>
   where
-    LB: LoadBalancingStrategy<TransportTcp> + Sized,
+    LB: LoadBalancingStrategy<RefCell<TransportTcp>> + Sized,
     A: Authenticator + 'a + Sized,
   {
     Session::new_snappy(&self.nodes_addrs, lb, self.authenticator.clone())
@@ -40,7 +41,7 @@ impl<'a, A: Authenticator + Sized> Cluster<A> {
 
   pub fn connect_lz4<LB>(&self, lb: LB) -> error::Result<Session<LB, A>>
   where
-    LB: LoadBalancingStrategy<TransportTcp> + Sized,
+    LB: LoadBalancingStrategy<RefCell<TransportTcp>> + Sized,
   {
     Session::new_lz4(&self.nodes_addrs, lb, self.authenticator.clone())
   }
@@ -53,7 +54,7 @@ impl<'a, A: Authenticator + Sized> Cluster<A> {
     ssl_connector: &SslConnector,
   ) -> error::Result<Session<LB, A>>
   where
-    LB: LoadBalancingStrategy<TransportTls> + Sized,
+    LB: LoadBalancingStrategy<RefCell<TransportTls>> + Sized,
   {
     Session::new_ssl(&self.nodes_addrs, lb, authenticator, ssl_connector)
   }
@@ -66,7 +67,7 @@ impl<'a, A: Authenticator + Sized> Cluster<A> {
     ssl_connector: &SslConnector,
   ) -> error::Result<Session<LB, A>>
   where
-    LB: LoadBalancingStrategy<TransportTls> + Sized,
+    LB: LoadBalancingStrategy<RefCell<TransportTls>> + Sized,
   {
     Session::new_snappy_ssl(&self.nodes_addrs, lb, authenticator, ssl_connector)
   }
@@ -79,7 +80,7 @@ impl<'a, A: Authenticator + Sized> Cluster<A> {
     ssl_connector: &SslConnector,
   ) -> error::Result<Session<LB, A>>
   where
-    LB: LoadBalancingStrategy<TransportTls> + Sized,
+    LB: LoadBalancingStrategy<RefCell<TransportTls>> + Sized,
   {
     Session::new_lz4_ssl(&self.nodes_addrs, lb, authenticator, ssl_connector)
   }
diff --git a/src/cluster/mod.rs b/src/cluster/mod.rs
index 64f5f48..f205834 100644
--- a/src/cluster/mod.rs
+++ b/src/cluster/mod.rs
@@ -1,28 +1,31 @@
+use std::cell;
+
 mod cluster;
 mod pager;
 mod session;
 
 pub use cluster::cluster::Cluster;
-pub use cluster::session::Session;
 pub use cluster::pager::{QueryPager, SessionPager};
+pub use cluster::session::Session;
 
-use transport::CDRSTransport;
 use compression::Compression;
 use query::{BatchExecutor, ExecExecutor, PrepareExecutor, QueryExecutor};
+use transport::CDRSTransport;
 
 pub trait GetTransport<'a, T: CDRSTransport + 'a> {
-  fn get_transport(&mut self) -> Option<&mut T>;
+  fn get_transport(&self) -> Option<&cell::RefCell<T>>;
 }
 
 pub trait GetCompressor<'a> {
   fn get_compressor(&self) -> Compression;
 }
 
-pub trait CDRSSession<'a, T: CDRSTransport + 'a>
-  : GetCompressor<'a>
-  + GetTransport<'a, T>
-  + QueryExecutor<'a, T>
-  + PrepareExecutor<'a, T>
-  + ExecExecutor<'a, T>
-  + BatchExecutor<'a, T> {
+pub trait CDRSSession<'a, T: CDRSTransport + 'static>:
+  GetCompressor<'static>
+  + GetTransport<'static, T>
+  + QueryExecutor<T>
+  + PrepareExecutor<T>
+  + ExecExecutor<T>
+  + BatchExecutor<T>
+{
 }
diff --git a/src/cluster/pager.rs b/src/cluster/pager.rs
index b9999a9..bc4881e 100644
--- a/src/cluster/pager.rs
+++ b/src/cluster/pager.rs
@@ -1,40 +1,47 @@
 use std::marker::PhantomData;
 
+use cluster::CDRSSession;
 use error;
 use frame::frame_result::{RowsMetadata, RowsMetadataFlag};
-use cluster::CDRSSession;
 use query::{PreparedQuery, QueryParamsBuilder};
+use transport::CDRSTransport;
 use types::rows::Row;
 use types::CBytes;
-use transport::CDRSTransport;
 
-pub struct SessionPager<'a, S: CDRSSession<'a, T> + 'a, T: CDRSTransport + 'a> {
+pub struct SessionPager<'a, S: CDRSSession<'static, T> + 'a, T: CDRSTransport + 'static> {
   page_size: i32,
-  session: &'a mut S,
+  session: &'a S,
   transport_type: PhantomData<&'a T>,
 }
 
-impl<'a, 'b: 'a, S: CDRSSession<'a, T>, T: CDRSTransport + 'a> SessionPager<'a, S, T> {
-  pub fn new(session: &'b mut S, page_size: i32) -> SessionPager<'a, S, T> {
-    SessionPager { session,
-                   page_size,
-                   transport_type: PhantomData, }
+impl<'a, 'b: 'a, S: CDRSSession<'static, T>, T: CDRSTransport + 'static> SessionPager<'a, S, T> {
+  pub fn new(session: &'b S, page_size: i32) -> SessionPager<'a, S, T> {
+    SessionPager {
+      session,
+      page_size,
+      transport_type: PhantomData,
+    }
   }
 
   pub fn query<Q>(&'a mut self, query: Q) -> QueryPager<'a, Q, SessionPager<'a, S, T>>
-    where Q: ToString
+  where
+    Q: ToString,
   {
-    QueryPager { pager: self,
-                 paging_state: None,
-                 has_more_pages: None,
-                 query, }
+    QueryPager {
+      pager: self,
+      paging_state: None,
+      has_more_pages: None,
+      query,
+    }
   }
 
   pub fn exec(&'a mut self, query: &'a PreparedQuery) -> ExecPager<'a, SessionPager<'a, S, T>> {
-    ExecPager { pager: self,
-                paging_state: None,
-                has_more_pages: None,
-                query, }
+    ExecPager {
+      pager: self,
+      paging_state: None,
+      has_more_pages: None,
+      query,
+    }
   }
 }
 
@@ -45,28 +52,31 @@ pub struct QueryPager<'a, Q: ToString, P: 'a> {
   query: Q,
 }
 
-impl<'a, Q: ToString, T: CDRSTransport + 'a, S: CDRSSession<'a, T>>
-  QueryPager<'a, Q, SessionPager<'a, S, T>> {
+impl<'a, Q: ToString, T: CDRSTransport + 'static, S: CDRSSession<'static, T>>
+  QueryPager<'a, Q, SessionPager<'a, S, T>>
+{
   pub fn next(&mut self) -> error::Result<Vec<Row>> {
     let mut params = QueryParamsBuilder::new().page_size(self.pager.page_size);
     if self.paging_state.is_some() {
       params = params.paging_state(self.paging_state.clone().unwrap());
     }
 
-    let body = self.pager
-                   .session
-                   .query_with_params(self.query.to_string(), params.finalize())
-                   .and_then(|frame| frame.get_body())?;
+    let body = self
+      .pager
+      .session
+      .query_with_params(self.query.to_string(), params.finalize())
+      .and_then(|frame| frame.get_body())?;
 
-    let metadata_res: error::Result<RowsMetadata> =
-      body.as_rows_metadata()
-          .ok_or("Pager query should yield a vector of rows".into());
+    let metadata_res: error::Result<RowsMetadata> = body
+      .as_rows_metadata()
+      .ok_or("Pager query should yield a vector of rows".into());
     let metadata = metadata_res?;
 
     self.has_more_pages = Some(RowsMetadataFlag::has_has_more_pages(metadata.flags.clone()));
     self.paging_state = metadata.paging_state.clone();
-    body.into_rows()
-        .ok_or("Pager query should yield a vector of rows".into())
+    body
+      .into_rows()
+      .ok_or("Pager query should yield a vector of rows".into())
   }
 
   pub fn has_more(&self) -> bool {
@@ -81,27 +91,31 @@ pub struct ExecPager<'a, P: 'a> {
   query: &'a PreparedQuery,
 }
 
-impl<'a, T: CDRSTransport + 'a, S: CDRSSession<'a, T>> ExecPager<'a, SessionPager<'a, S, T>> {
+impl<'a, T: CDRSTransport + 'static, S: CDRSSession<'static, T>>
+  ExecPager<'a, SessionPager<'a, S, T>>
+{
   pub fn next(&mut self) -> error::Result<Vec<Row>> {
     let mut params = QueryParamsBuilder::new().page_size(self.pager.page_size);
     if self.paging_state.is_some() {
       params = params.paging_state(self.paging_state.clone().unwrap());
     }
 
-    let body = self.pager
-                   .session
-                   .exec_with_params(self.query, params.finalize())
-                   .and_then(|frame| frame.get_body())?;
+    let body = self
+      .pager
+      .session
+      .exec_with_params(self.query, params.finalize())
+      .and_then(|frame| frame.get_body())?;
 
-    let metadata_res: error::Result<RowsMetadata> =
-      body.as_rows_metadata()
-          .ok_or("Pager query should yield a vector of rows".into());
+    let metadata_res: error::Result<RowsMetadata> = body
+      .as_rows_metadata()
+      .ok_or("Pager query should yield a vector of rows".into());
     let metadata = metadata_res?;
 
     self.has_more_pages = Some(RowsMetadataFlag::has_has_more_pages(metadata.flags.clone()));
     self.paging_state = metadata.paging_state.clone();
-    body.into_rows()
-        .ok_or("Pager query should yield a vector of rows".into())
+    body
+      .into_rows()
+      .ok_or("Pager query should yield a vector of rows".into())
   }
 
   pub fn has_more(&self) -> bool {
diff --git a/src/cluster/session.rs b/src/cluster/session.rs
index dd0cc01..259747f 100644
--- a/src/cluster/session.rs
+++ b/src/cluster/session.rs
@@ -1,28 +1,29 @@
+use std::cell::RefCell;
 use std::io;
 use std::io::Write;
 
+use cluster::{CDRSSession, GetCompressor, GetTransport, SessionPager};
 use error;
-use transport::{CDRSTransport, TransportTcp};
 use load_balancing::LoadBalancingStrategy;
-use cluster::{CDRSSession, GetCompressor, GetTransport, SessionPager};
+use transport::{CDRSTransport, TransportTcp};
 
 use authenticators::Authenticator;
 use compression::Compression;
-use frame::{Frame, IntoBytes, Opcode};
-use frame::parser::parse_frame;
-use query::{BatchExecutor, ExecExecutor, PrepareExecutor, QueryExecutor};
 use events::{new_listener, EventStream, Listener};
 use frame::events::SimpleServerEvent;
+use frame::parser::parse_frame;
+use frame::{Frame, IntoBytes, Opcode};
+use query::{BatchExecutor, ExecExecutor, PrepareExecutor, QueryExecutor};
 
-#[cfg(feature = "ssl")]
-use transport::TransportTls;
 #[cfg(feature = "ssl")]
 use openssl::ssl::SslConnector;
-
+#[cfg(feature = "ssl")]
+use transport::TransportTls;
 
 pub struct Session<LB, A> {
   load_balancing: LB,
-  #[allow(dead_code)] authenticator: A,
+  #[allow(dead_code)]
+  authenticator: A,
   pub compression: Compression,
 }
 
@@ -33,24 +34,25 @@ impl<'a, LB, A> GetCompressor<'a> for Session<LB, A> {
 }
 
 impl<'a, LB: Sized, A: Authenticator + 'a + Sized> Session<LB, A> {
-  pub fn paged<T: CDRSTransport>(
-    &'a mut self,
+  pub fn paged<T: CDRSTransport + 'static>(
+    &'a self,
     page_size: i32,
   ) -> SessionPager<'a, Session<LB, A>, T>
   where
-    Session<LB, A>: CDRSSession<'a, T>,
+    Session<LB, A>: CDRSSession<'static, T>,
   {
     return SessionPager::new(self, page_size);
   }
 
-  fn startup<'b, T: CDRSTransport>(
-    transport: &'b mut T,
+  fn startup<'b, T: CDRSTransport + 'static>(
+    transport: &RefCell<T>,
     session_authenticator: &'b A,
   ) -> error::Result<()> {
     let ref mut compression = Compression::None;
     let startup_frame = Frame::new_req_startup(compression.as_str()).into_cbytes();
 
-    try!(transport.write(startup_frame.as_slice()));
+    transport.borrow_mut().write(startup_frame.as_slice())?;
+
     let start_response = try!(parse_frame(transport, compression));
 
     if start_response.opcode == Opcode::Ready {
@@ -83,8 +85,7 @@ impl<'a, LB: Sized, A: Authenticator + 'a + Sized> Session<LB, A> {
               format!(
                 "Unsupported type of authenticator. {:?} got,
                              but {} is supported.",
-                authenticator,
-                auth
+                authenticator, auth
               ),
             );
             return Err(error::Error::Io(io_err));
@@ -98,7 +99,7 @@ impl<'a, LB: Sized, A: Authenticator + 'a + Sized> Session<LB, A> {
 
       let auth_token_bytes = session_authenticator.get_auth_token().into_cbytes();
       try!(
-        transport.write(
+        transport.borrow_mut().write(
           Frame::new_req_auth_response(auth_token_bytes)
             .into_cbytes()
             .as_slice()
@@ -114,68 +115,75 @@ impl<'a, LB: Sized, A: Authenticator + 'a + Sized> Session<LB, A> {
 }
 
 impl<
-  'a,
-  T: CDRSTransport + 'a,
-  LB: LoadBalancingStrategy<T> + Sized,
-  A: Authenticator + Sized,
-> GetTransport<'a, T> for Session<LB, A> {
-  fn get_transport(&mut self) -> Option<&mut T> {
+    'a,
+    T: CDRSTransport + 'a,
+    LB: LoadBalancingStrategy<RefCell<T>> + Sized,
+    A: Authenticator + Sized,
+  > GetTransport<'a, T> for Session<LB, A>
+{
+  fn get_transport(&self) -> Option<&RefCell<T>> {
     self.load_balancing.next()
   }
 }
 
 impl<
-  'a,
-  T: CDRSTransport + 'a,
-  LB: LoadBalancingStrategy<T> + Sized,
-  A: Authenticator + Sized,
-> QueryExecutor<'a, T> for Session<LB, A> {
+    'a,
+    T: CDRSTransport + 'static,
+    LB: LoadBalancingStrategy<RefCell<T>> + Sized,
+    A: Authenticator + Sized,
+  > QueryExecutor<T> for Session<LB, A>
+{
 }
 
 impl<
-  'a,
-  T: CDRSTransport + 'a,
-  LB: LoadBalancingStrategy<T> + Sized,
-  A: Authenticator + Sized,
-> PrepareExecutor<'a, T> for Session<LB, A> {
+    'a,
+    T: CDRSTransport + 'static,
+    LB: LoadBalancingStrategy<RefCell<T>> + Sized,
+    A: Authenticator + Sized,
+  > PrepareExecutor<T> for Session<LB, A>
+{
 }
 
 impl<
-  'a,
-  T: CDRSTransport + 'a,
-  LB: LoadBalancingStrategy<T> + Sized,
-  A: Authenticator + Sized,
-> ExecExecutor<'a, T> for Session<LB, A> {
+    'a,
+    T: CDRSTransport + 'static,
+    LB: LoadBalancingStrategy<RefCell<T>> + Sized,
+    A: Authenticator + Sized,
+  > ExecExecutor<T> for Session<LB, A>
+{
 }
 
 impl<
-  'a,
-  T: CDRSTransport + 'a,
-  LB: LoadBalancingStrategy<T> + Sized,
-  A: Authenticator + Sized,
-> BatchExecutor<'a, T> for Session<LB, A> {
+    'a,
+    T: CDRSTransport + 'static,
+    LB: LoadBalancingStrategy<RefCell<T>> + Sized,
+    A: Authenticator + Sized,
+  > BatchExecutor<T> for Session<LB, A>
+{
 }
 
 impl<
-  'a,
-  T: CDRSTransport + 'a,
-  LB: LoadBalancingStrategy<T> + Sized,
-  A: Authenticator + Sized,
-> CDRSSession<'a, T> for Session<LB, A> {
+    'a,
+    T: CDRSTransport + 'static,
+    LB: LoadBalancingStrategy<RefCell<T>> + Sized,
+    A: Authenticator + Sized,
+  > CDRSSession<'a, T> for Session<LB, A>
+{
 }
 
-impl<'a, LB: LoadBalancingStrategy<TransportTcp> + Sized, A: Authenticator + 'a + Sized>
-  Session<LB, A> {
+impl<'a, LB: LoadBalancingStrategy<RefCell<TransportTcp>> + Sized, A: Authenticator + 'a + Sized>
+  Session<LB, A>
+{
   pub fn new(
     addrs: &Vec<&str>,
     mut load_balancing: LB,
     authenticator: A,
   ) -> error::Result<Session<LB, A>> {
-    let mut nodes: Vec<TransportTcp> = Vec::with_capacity(addrs.len());
+    let mut nodes: Vec<RefCell<TransportTcp>> = Vec::with_capacity(addrs.len());
 
     for addr in addrs {
-      let mut transport = TransportTcp::new(&addr)?;
-      Self::startup(&mut transport, &authenticator)?;
+      let transport = RefCell::new(TransportTcp::new(&addr)?);
+      Self::startup(&transport, &authenticator)?;
       nodes.push(transport);
     }
 
@@ -193,11 +201,11 @@ impl<'a, LB: LoadBalancingStrategy<TransportTcp> + Sized, A: Authenticator + 'a
     mut load_balancing: LB,
     authenticator: A,
   ) -> error::Result<Session<LB, A>> {
-    let mut nodes: Vec<TransportTcp> = Vec::with_capacity(addrs.len());
+    let mut nodes: Vec<RefCell<TransportTcp>> = Vec::with_capacity(addrs.len());
 
     for addr in addrs {
-      let mut transport = TransportTcp::new(&addr)?;
-      Self::startup(&mut transport, &authenticator)?;
+      let transport = RefCell::new(TransportTcp::new(&addr)?);
+      Self::startup(&transport, &authenticator)?;
       nodes.push(transport);
     }
 
@@ -215,11 +223,11 @@ impl<'a, LB: LoadBalancingStrategy<TransportTcp> + Sized, A: Authenticator + 'a
     mut load_balancing: LB,
     authenticator: A,
   ) -> error::Result<Session<LB, A>> {
-    let mut nodes: Vec<TransportTcp> = Vec::with_capacity(addrs.len());
+    let mut nodes: Vec<RefCell<TransportTcp>> = Vec::with_capacity(addrs.len());
 
     for addr in addrs {
-      let mut transport = TransportTcp::new(&addr)?;
-      Self::startup(&mut transport, &authenticator)?;
+      let transport = RefCell::new(TransportTcp::new(&addr)?);
+      Self::startup(&transport, &authenticator)?;
       nodes.push(transport);
     }
 
@@ -233,40 +241,39 @@ impl<'a, LB: LoadBalancingStrategy<TransportTcp> + Sized, A: Authenticator + 'a
   }
 
   pub fn listen(
-    &mut self,
+    &self,
+    node: &str,
     events: Vec<SimpleServerEvent>,
-  ) -> error::Result<(Listener<TransportTcp>, EventStream)> {
+  ) -> error::Result<(Listener<RefCell<TransportTcp>>, EventStream)> {
     let authenticator = self.authenticator.clone();
     let compression = self.get_compressor();
-    let transport = self
-      .get_transport()
-      .ok_or("Cannot connect to a cluster - no nodes provided")?;
-    let mut transport = transport.try_clone()?;
-    Self::startup(&mut transport, &authenticator)?;
+    let transport = TransportTcp::new(&node).map(RefCell::new)?;
+
+    Self::startup(&transport, &authenticator)?;
 
     let query_frame = Frame::new_req_register(events).into_cbytes();
-    try!(transport.write(query_frame.as_slice()));
-    println!("send register");
-    try!(parse_frame(&mut transport, &compression));
-    println!("receive response");
+    transport.borrow_mut().write(query_frame.as_slice())?;
+    parse_frame(&transport, &compression)?;
+
     Ok(new_listener(transport))
   }
 }
 
 #[cfg(feature = "ssl")]
-impl<'a, LB: LoadBalancingStrategy<TransportTls> + Sized, A: Authenticator + 'a + Sized>
-  Session<LB, A> {
+impl<'a, LB: LoadBalancingStrategy<RefCell<TransportTls>> + Sized, A: Authenticator + 'a + Sized>
+  Session<LB, A>
+{
   pub fn new_ssl(
     addrs: &Vec<&str>,
     mut load_balancing: LB,
     authenticator: A,
     ssl_connector: &SslConnector,
   ) -> error::Result<Session<LB, A>> {
-    let mut nodes: Vec<TransportTls> = Vec::with_capacity(addrs.len());
+    let mut nodes: Vec<RefCell<TransportTls>> = Vec::with_capacity(addrs.len());
 
     for addr in addrs {
-      let mut transport = TransportTls::new(&addr, ssl_connector)?;
-      Self::startup(&mut transport, &authenticator)?;
+      let transport = RefCell::new(TransportTls::new(&addr, ssl_connector)?);
+      Self::startup(&transport, &authenticator)?;
       nodes.push(transport);
     }
 
@@ -285,11 +292,11 @@ impl<'a, LB: LoadBalancingStrategy<TransportTls> + Sized, A: Authenticator + 'a
     authenticator: A,
     ssl_connector: &SslConnector,
   ) -> error::Result<Session<LB, A>> {
-    let mut nodes: Vec<TransportTls> = Vec::with_capacity(addrs.len());
+    let mut nodes: Vec<RefCell<TransportTls>> = Vec::with_capacity(addrs.len());
 
     for addr in addrs {
-      let mut transport = TransportTls::new(&addr, ssl_connector)?;
-      Self::startup(&mut transport, &authenticator)?;
+      let transport = RefCell::new(TransportTls::new(&addr, ssl_connector)?);
+      Self::startup(&transport, &authenticator)?;
       nodes.push(transport);
     }
 
@@ -308,11 +315,11 @@ impl<'a, LB: LoadBalancingStrategy<TransportTls> + Sized, A: Authenticator + 'a
     authenticator: A,
     ssl_connector: &SslConnector,
   ) -> error::Result<Session<LB, A>> {
-    let mut nodes: Vec<TransportTls> = Vec::with_capacity(addrs.len());
+    let mut nodes: Vec<RefCell<TransportTls>> = Vec::with_capacity(addrs.len());
 
     for addr in addrs {
-      let mut transport = TransportTls::new(&addr, ssl_connector)?;
-      Self::startup(&mut transport, &authenticator)?;
+      let transport = RefCell::new(TransportTls::new(&addr, ssl_connector)?);
+      Self::startup(&transport, &authenticator)?;
       nodes.push(transport);
     }
 
@@ -326,20 +333,23 @@ impl<'a, LB: LoadBalancingStrategy<TransportTls> + Sized, A: Authenticator + 'a
   }
 
   pub fn listen_ssl(
-    &mut self,
+    &self,
+    node: (&str, &SslConnector),
     events: Vec<SimpleServerEvent>,
-  ) -> error::Result<(Listener<TransportTls>, EventStream)> {
+  ) -> error::Result<(Listener<RefCell<TransportTls>>, EventStream)> {
+    let (addr, ssl_connector) = node;
     let authenticator = self.authenticator.clone();
     let compression = self.get_compressor();
     let transport = self
       .get_transport()
       .ok_or("Cannot connect to a cluster - no nodes provided")?;
-    let mut transport = transport.try_clone()?;
-    Self::startup(&mut transport, &authenticator)?;
+    let transport_cell = RefCell::new(TransportTls::new(&addr, ssl_connector)?);
+    Self::startup(&transport, &authenticator)?;
 
     let query_frame = Frame::new_req_register(events).into_cbytes();
-    try!(transport.write(query_frame.as_slice()));
-    try!(parse_frame(&mut transport, &compression));
-    Ok(new_listener(transport))
+    transport_cell.borrow_mut().write(query_frame.as_slice())?;
+    parse_frame(&transport_cell, &compression)?;
+
+    Ok(new_listener(transport_cell))
   }
 }
diff --git a/src/events.rs b/src/events.rs
index 06d1011..1e923d0 100644
--- a/src/events.rs
+++ b/src/events.rs
@@ -1,12 +1,15 @@
-use std::sync::mpsc::{channel, Receiver, Sender};
+use std::cell::RefCell;
 use std::iter::Iterator;
+use std::sync::mpsc::{channel, Receiver, Sender};
 
-use std::error::Error;
+use compression::Compression;
 use error;
-use frame::events::{SchemaChange as FrameSchemaChange, ServerEvent as FrameServerEvent,
-                    SimpleServerEvent as FrameSimpleServerEvent};
+use frame::events::{
+    SchemaChange as FrameSchemaChange, ServerEvent as FrameServerEvent,
+    SimpleServerEvent as FrameSimpleServerEvent,
+};
 use frame::parser::parse_frame;
-use compression::Compression;
+use std::error::Error;
 use transport::CDRSTransport;
 
 /// Full Server Event which includes all details about occured change.
@@ -29,8 +32,10 @@ pub type SchemaChange = FrameSchemaChange;
 /// It is similar to `Receiver::iter`.
 pub fn new_listener<X>(transport: X) -> (Listener<X>, EventStream) {
     let (tx, rx) = channel();
-    let listener = Listener { transport: transport,
-                              tx: tx, };
+    let listener = Listener {
+        transport: transport,
+        tx: tx,
+    };
     let stream = EventStream { rx: rx };
     (listener, stream)
 }
@@ -44,15 +49,16 @@ pub struct Listener<X> {
     tx: Sender<ServerEvent>,
 }
 
-impl<X: CDRSTransport> Listener<X> {
+impl<X: CDRSTransport + 'static> Listener<RefCell<X>> {
     /// It starts a process of listening to new events. Locks a frame.
-    pub fn start(&mut self, compressor: &Compression) -> error::Result<()> {
+    pub fn start(self, compressor: &Compression) -> error::Result<()> {
         loop {
-            let event_opt = try!(parse_frame(&mut self.transport, compressor)).get_body()?
-                                                                              .into_server_event();
+            let event_opt = try!(parse_frame(&self.transport, compressor))
+                .get_body()?
+                .into_server_event();
 
             let event = if event_opt.is_some() {
-                // unwrap is safe is we've checked that event_opt.is_some()
+                // unwrap is safe as we've checked that event_opt.is_some()
                 event_opt.unwrap().event as ServerEvent
             } else {
                 continue;
diff --git a/src/frame/parser.rs b/src/frame/parser.rs
index e2dfcc6..9d2b701 100644
--- a/src/frame/parser.rs
+++ b/src/frame/parser.rs
@@ -1,19 +1,21 @@
+use std::cell::RefCell;
 use std::io::{Cursor, Read};
 
-use frame::FromCursor;
+use super::*;
 use compression::Compression;
+use error;
 use frame::frame_response::ResponseBody;
-use super::*;
-use types::{from_bytes, CStringList, from_u16_bytes, UUID_LEN};
+use frame::FromCursor;
 use types::data_serialization_types::decode_timeuuid;
-use error;
+use types::{from_bytes, from_u16_bytes, CStringList, UUID_LEN};
 
-pub fn parse_frame(cursor: &mut Read, compressor: &Compression) -> error::Result<Frame> {
+pub fn parse_frame(cursor_cell: &RefCell<Read>, compressor: &Compression) -> error::Result<Frame> {
     let mut version_bytes = [0; Version::BYTE_LENGTH];
     let mut flag_bytes = [0; Flag::BYTE_LENGTH];
     let mut opcode_bytes = [0; Opcode::BYTE_LENGTH];
     let mut stream_bytes = [0; STREAM_LEN];
     let mut length_bytes = [0; LENGTH_LEN];
+    let mut cursor = cursor_cell.borrow_mut();
 
     // NOTE: order of reads matters
     try!(cursor.read_exact(&mut version_bytes));
@@ -66,27 +68,25 @@ pub fn parse_frame(cursor: &mut Read, compressor: &Compression) -> error::Result
 
     try!(body_cursor.read_to_end(&mut body));
 
-    let frame = Frame { version: version,
-                        flags: flags,
-                        opcode: opcode,
-                        stream: stream,
-                        body: body,
-                        tracing_id: tracing_id,
-                        warnings: warnings, };
+    let frame = Frame {
+        version: version,
+        flags: flags,
+        opcode: opcode,
+        stream: stream,
+        body: body,
+        tracing_id: tracing_id,
+        warnings: warnings,
+    };
 
     convert_frame_into_result(frame)
 }
 
 fn convert_frame_into_result(frame: Frame) -> error::Result<Frame> {
     match frame.opcode {
-        Opcode::Error => {
-            frame.get_body().and_then(|err| match err {
-                                          ResponseBody::Error(err) => {
-                                              Err(error::Error::Server(err))
-                                          }
-                                          _ => unreachable!(),
-                                      })
-        }
+        Opcode::Error => frame.get_body().and_then(|err| match err {
+            ResponseBody::Error(err) => Err(error::Error::Server(err)),
+            _ => unreachable!(),
+        }),
         _ => Ok(frame),
     }
 }
diff --git a/src/load_balancing/mod.rs b/src/load_balancing/mod.rs
index f99cb2e..a0539de 100644
--- a/src/load_balancing/mod.rs
+++ b/src/load_balancing/mod.rs
@@ -6,5 +6,5 @@ pub use load_balancing::round_robin::RoundRobin;
 
 pub trait LoadBalancingStrategy<N>: Sized {
   fn init(&mut self, cluster: Vec<N>);
-  fn next(&mut self) -> Option<&mut N>;
+  fn next(&self) -> Option<&N>;
 }
diff --git a/src/load_balancing/random.rs b/src/load_balancing/random.rs
index a707861..68b40a1 100644
--- a/src/load_balancing/random.rs
+++ b/src/load_balancing/random.rs
@@ -3,12 +3,12 @@ use rand;
 use super::LoadBalancingStrategy;
 
 pub struct Random<N> {
-  cluster: Vec<N>,
+  pub cluster: Vec<N>,
 }
 
 impl<N> Random<N> {
-  pub fn new() -> Self {
-    Random { cluster: vec![] }
+  pub fn new(cluster: Vec<N>) -> Self {
+    Random { cluster }
   }
 
   /// Returns random number from a range
@@ -31,9 +31,9 @@ impl<N> LoadBalancingStrategy<N> for Random<N> {
     self.cluster = cluster;
   }
   /// Returns next random node from a cluster
-  fn next(&mut self) -> Option<&mut N> {
+  fn next(&self) -> Option<&N> {
     let len = self.cluster.len();
-    self.cluster.get_mut(Self::rnd_idx((0, len)))
+    self.cluster.get(Self::rnd_idx((0, len)))
   }
 }
 
@@ -44,7 +44,7 @@ mod tests {
   #[test]
   fn random() {
     let nodes = vec!["a", "b", "c", "d", "e", "f", "g"];
-    let mut load_balancer = Random::from(nodes);
+    let load_balancer = Random::from(nodes);
     for _ in 0..100 {
       let s = load_balancer.next();
       assert!(s.is_some());
diff --git a/src/load_balancing/round_robin.rs b/src/load_balancing/round_robin.rs
index 08e810e..eec34c7 100644
--- a/src/load_balancing/round_robin.rs
+++ b/src/load_balancing/round_robin.rs
@@ -1,21 +1,27 @@
+use std::cell::RefCell;
+
 use super::LoadBalancingStrategy;
 
 pub struct RoundRobin<N> {
   cluster: Vec<N>,
-  prev_idx: usize,
+  prev_idx: RefCell<usize>,
 }
 
 impl<N> RoundRobin<N> {
   pub fn new() -> Self {
-    RoundRobin { prev_idx: 0,
-                 cluster: vec![], }
+    RoundRobin {
+      prev_idx: RefCell::new(0),
+      cluster: vec![],
+    }
   }
 }
 
 impl<N> From<Vec<N>> for RoundRobin<N> {
   fn from(cluster: Vec<N>) -> RoundRobin<N> {
-    RoundRobin { prev_idx: 0,
-                 cluster, }
+    RoundRobin {
+      prev_idx: RefCell::new(0),
+      cluster: cluster,
+    }
   }
 }
 
@@ -25,9 +31,11 @@ impl<N> LoadBalancingStrategy<N> for RoundRobin<N> {
   }
 
   /// Returns next node from a cluster
-  fn next(&mut self) -> Option<&mut N> {
-    self.prev_idx = (self.prev_idx + 1) % self.cluster.len();
-    self.cluster.get_mut(self.prev_idx)
+  fn next(&self) -> Option<&N> {
+    let prev_idx = *self.prev_idx.borrow();
+    let next_idx = (prev_idx + 1) % self.cluster.len();
+    self.prev_idx.replace(next_idx);
+    self.cluster.get(next_idx)
   }
 }
 
@@ -39,7 +47,7 @@ mod tests {
   fn round_robin() {
     let nodes = vec!["a", "b", "c"];
     let nodes_c = nodes.clone();
-    let mut load_balancer = RoundRobin::from(nodes);
+    let load_balancer = RoundRobin::from(nodes);
     for i in 0..10 {
       assert_eq!(&nodes_c[(i + 1) % 3], load_balancer.next().unwrap());
     }
diff --git a/src/query/batch_executor.rs b/src/query/batch_executor.rs
index 96c19bd..d53a712 100644
--- a/src/query/batch_executor.rs
+++ b/src/query/batch_executor.rs
@@ -1,18 +1,20 @@
-use error;
-use transport::CDRSTransport;
 use cluster::{GetCompressor, GetTransport};
-use frame::{Flag, Frame};
+use error;
 use frame::parser::parse_frame;
 use frame::traits::IntoBytes;
+use frame::{Flag, Frame};
 use query::batch_query_builder::QueryBatch;
+use transport::CDRSTransport;
 
-pub trait BatchExecutor<'a, T: CDRSTransport + 'a>
-  : GetTransport<'a, T> + GetCompressor<'a> {
-  fn batch_with_params_tw(&mut self,
-                          batch: QueryBatch,
-                          with_tracing: bool,
-                          with_warnings: bool)
-                          -> error::Result<Frame> {
+pub trait BatchExecutor<T: CDRSTransport + 'static>:
+  GetTransport<'static, T> + GetCompressor<'static>
+{
+  fn batch_with_params_tw(
+    &self,
+    batch: QueryBatch,
+    with_tracing: bool,
+    with_warnings: bool,
+  ) -> error::Result<Frame> {
     let mut flags = vec![];
 
     if with_tracing {
@@ -25,13 +27,21 @@ pub trait BatchExecutor<'a, T: CDRSTransport + 'a>
 
     let query_frame = Frame::new_req_batch(batch, flags).into_cbytes();
     let ref compression = self.get_compressor();
-    let transport = self.get_transport().ok_or("Unable to get transport")?;
 
-    try!(transport.write(query_frame.as_slice()));
-    parse_frame(transport, compression)
+    self
+      .get_transport()
+      .ok_or(error::Error::from("Unable to get transport"))
+      .and_then(|transport_cell| {
+        let write_res = transport_cell
+          .borrow_mut()
+          .write(query_frame.as_slice())
+          .map_err(error::Error::from);
+        write_res.map(|_| transport_cell)
+      })
+      .and_then(|transport_cell| parse_frame(transport_cell, compression))
   }
 
-  fn batch_with_params(&mut self, batch: QueryBatch) -> error::Result<Frame> {
+  fn batch_with_params(&self, batch: QueryBatch) -> error::Result<Frame> {
     self.batch_with_params_tw(batch, false, false)
   }
 }
diff --git a/src/query/exec_executor.rs b/src/query/exec_executor.rs
index c2cc3ac..88e585a 100644
--- a/src/query/exec_executor.rs
+++ b/src/query/exec_executor.rs
@@ -1,21 +1,23 @@
+use cluster::{GetCompressor, GetTransport};
 use error;
-use query::{QueryParams, QueryParamsBuilder, QueryValues};
-use frame::{Flag, Frame, IntoBytes};
 use frame::parser::parse_frame;
-use types::CBytesShort;
-use cluster::{GetCompressor, GetTransport};
+use frame::{Flag, Frame, IntoBytes};
+use query::{QueryParams, QueryParamsBuilder, QueryValues};
 use transport::CDRSTransport;
+use types::CBytesShort;
 
 pub type PreparedQuery = CBytesShort;
 
-pub trait ExecExecutor<'a, T: CDRSTransport + 'a>
-  : GetTransport<'a, T> + GetCompressor<'a> {
-  fn exec_with_params_tw(&mut self,
-                         prepared: &PreparedQuery,
-                         query_parameters: QueryParams,
-                         with_tracing: bool,
-                         with_warnings: bool)
-                         -> error::Result<Frame> {
+pub trait ExecExecutor<T: CDRSTransport + 'static>:
+  GetTransport<'static, T> + GetCompressor<'static>
+{
+  fn exec_with_params_tw(
+    &self,
+    prepared: &PreparedQuery,
+    query_parameters: QueryParams,
+    with_tracing: bool,
+    with_warnings: bool,
+  ) -> error::Result<Frame> {
     let mut flags = vec![];
     if with_tracing {
       flags.push(Flag::Tracing);
@@ -26,47 +28,59 @@ pub trait ExecExecutor<'a, T: CDRSTransport + 'a>
 
     let options_frame = Frame::new_req_execute(prepared, query_parameters, flags).into_cbytes();
     let ref compression = self.get_compressor();
-    let transport = self.get_transport().ok_or("Unable to get transport")?;
 
-    (transport.write(options_frame.as_slice()))?;
-    parse_frame(transport, compression)
+    self
+      .get_transport()
+      .ok_or(error::Error::from("Unable to get transport"))
+      .and_then(|transport_cell| {
+        let write_res = transport_cell
+          .borrow_mut()
+          .write(options_frame.as_slice())
+          .map_err(error::Error::from);
+        write_res.map(|_| transport_cell)
+      })
+      .and_then(|transport_cell| parse_frame(transport_cell, compression))
   }
 
-  fn exec_with_params(&mut self,
-                      prepared: &PreparedQuery,
-                      query_parameters: QueryParams)
-                      -> error::Result<Frame> {
+  fn exec_with_params(
+    &self,
+    prepared: &PreparedQuery,
+    query_parameters: QueryParams,
+  ) -> error::Result<Frame> {
     self.exec_with_params_tw(prepared, query_parameters, false, false)
   }
 
-  fn exec_with_values_tw<V: Into<QueryValues>>(&mut self,
-                                               prepared: &PreparedQuery,
-                                               values: V,
-                                               with_tracing: bool,
-                                               with_warnings: bool)
-                                               -> error::Result<Frame> {
+  fn exec_with_values_tw<V: Into<QueryValues>>(
+    &self,
+    prepared: &PreparedQuery,
+    values: V,
+    with_tracing: bool,
+    with_warnings: bool,
+  ) -> error::Result<Frame> {
     let query_params_builder = QueryParamsBuilder::new();
     let query_params = query_params_builder.values(values.into()).finalize();
     self.exec_with_params_tw(prepared, query_params, with_tracing, with_warnings)
   }
 
-  fn exec_with_values<V: Into<QueryValues>>(&mut self,
-                                            prepared: &PreparedQuery,
-                                            values: V)
-                                            -> error::Result<Frame> {
+  fn exec_with_values<V: Into<QueryValues>>(
+    &self,
+    prepared: &PreparedQuery,
+    values: V,
+  ) -> error::Result<Frame> {
     self.exec_with_values_tw(prepared, values, false, false)
   }
 
-  fn exec_tw(&mut self,
-             prepared: &PreparedQuery,
-             with_tracing: bool,
-             with_warnings: bool)
-             -> error::Result<Frame> {
+  fn exec_tw(
+    &self,
+    prepared: &PreparedQuery,
+    with_tracing: bool,
+    with_warnings: bool,
+  ) -> error::Result<Frame> {
     let query_params = QueryParamsBuilder::new().finalize();
     self.exec_with_params_tw(prepared, query_params, with_tracing, with_warnings)
   }
 
-  fn exec(&'a mut self, prepared: &PreparedQuery) -> error::Result<Frame> {
+  fn exec(&mut self, prepared: &PreparedQuery) -> error::Result<Frame> {
     self.exec_tw(prepared, false, false)
   }
 }
diff --git a/src/query/prepare_executor.rs b/src/query/prepare_executor.rs
index a1a994a..5886c64 100644
--- a/src/query/prepare_executor.rs
+++ b/src/query/prepare_executor.rs
@@ -1,22 +1,24 @@
+use cluster::{GetCompressor, GetTransport};
 use error;
+use frame::parser::parse_frame;
 use frame::{Flag, Frame, IntoBytes};
-use types::CBytesShort;
-use cluster::{GetCompressor, GetTransport};
 use transport::CDRSTransport;
-use frame::parser::parse_frame;
+use types::CBytesShort;
 
 pub type PreparedQuery = CBytesShort;
 
-pub trait PrepareExecutor<'a, T: CDRSTransport + 'a>
-  : GetTransport<'a, T> + GetCompressor<'a> {
+pub trait PrepareExecutor<T: CDRSTransport + 'static>:
+  GetTransport<'static, T> + GetCompressor<'static>
+{
   /// It prepares a query for execution, along with query itself
   /// the method takes `with_tracing` and `with_warnings` flags
   /// to get tracing information and warnings.
-  fn prepare_tw<Q: ToString>(&'a mut self,
-                             query: Q,
-                             with_tracing: bool,
-                             with_warnings: bool)
-                             -> error::Result<PreparedQuery> {
+  fn prepare_tw<Q: ToString>(
+    &self,
+    query: Q,
+    with_tracing: bool,
+    with_warnings: bool,
+  ) -> error::Result<PreparedQuery> {
     let mut flags = vec![];
     if with_tracing {
       flags.push(Flag::Tracing);
@@ -25,17 +27,33 @@ pub trait PrepareExecutor<'a, T: CDRSTransport + 'a>
       flags.push(Flag::Warning);
     }
 
-    let options_frame = Frame::new_req_prepare(query.to_string(), flags).into_cbytes();
+    let query_frame = Frame::new_req_prepare(query.to_string(), flags).into_cbytes();
     let ref compression = self.get_compressor();
-    let transport = self.get_transport().ok_or("Unable to get transport")?;
 
-    try!(transport.write(options_frame.as_slice()));
-    parse_frame(transport, compression).and_then(|response| response.get_body())
-                                       .and_then(|body| Ok(body.into_prepared().expect("").id))
+    self
+      .get_transport()
+      .ok_or(error::Error::from("Unable to get transport"))
+      .and_then(|transport_cell| {
+        let write_res = transport_cell
+          .borrow_mut()
+          .write(query_frame.as_slice())
+          .map_err(error::Error::from);
+        write_res.map(|_| transport_cell)
+      })
+      .and_then(|transport_cell| parse_frame(transport_cell, compression))
+      .and_then(|response| response.get_body())
+      .and_then(|body| {
+        Ok(
+          body
+            .into_prepared()
+            .expect("CDRS BUG: cannot convert frame into prepared")
+            .id,
+        )
+      })
   }
 
   /// It prepares query without additional tracing information and warnings.
-  fn prepare<Q: ToString>(&'a mut self, query: Q) -> error::Result<PreparedQuery> {
+  fn prepare<Q: ToString>(&self, query: Q) -> error::Result<PreparedQuery> {
     self.prepare_tw(query, false, false)
   }
 }
diff --git a/src/query/query_executor.rs b/src/query/query_executor.rs
index 1e5cd87..bbaca24 100644
--- a/src/query/query_executor.rs
+++ b/src/query/query_executor.rs
@@ -1,20 +1,24 @@
+use cluster::{GetCompressor, GetTransport};
 use error;
+use frame::parser::parse_frame;
 use frame::{Flag, Frame, IntoBytes};
 use query::{Query, QueryParams, QueryParamsBuilder, QueryValues};
-use cluster::{GetCompressor, GetTransport};
 use transport::CDRSTransport;
-use frame::parser::parse_frame;
 
-pub trait QueryExecutor<'a, T: CDRSTransport + 'a>
-  : GetTransport<'a, T> + GetCompressor<'a> {
-  fn query_with_params_tw<Q: ToString>(&mut self,
-                                       query: Q,
-                                       query_params: QueryParams,
-                                       with_tracing: bool,
-                                       with_warnings: bool)
-                                       -> error::Result<Frame> {
-    let query = Query { query: query.to_string(),
-                        params: query_params, };
+pub trait QueryExecutor<T: CDRSTransport + 'static>:
+  GetTransport<'static, T> + GetCompressor<'static>
+{
+  fn query_with_params_tw<Q: ToString>(
+    &self,
+    query: Q,
+    query_params: QueryParams,
+    with_tracing: bool,
+    with_warnings: bool,
+  ) -> error::Result<Frame> {
+    let query = Query {
+      query: query.to_string(),
+      params: query_params,
+    };
 
     let mut flags = vec![];
 
@@ -28,54 +32,67 @@ pub trait QueryExecutor<'a, T: CDRSTransport + 'a>
 
     let query_frame = Frame::new_query(query, flags).into_cbytes();
     let ref compression = self.get_compressor();
-    let transport = self.get_transport().ok_or("Unable to get transport")?;
-    try!(transport.write(query_frame.as_slice()));
-    parse_frame(transport, compression)
+
+    self
+      .get_transport()
+      .ok_or(error::Error::from("Unable to get transport"))
+      .and_then(|transport_cell| {
+        let write_res = transport_cell
+          .borrow_mut()
+          .write(query_frame.as_slice())
+          .map_err(error::Error::from);
+        write_res.map(|_| transport_cell)
+      })
+      .and_then(|transport_cell| parse_frame(transport_cell, compression))
   }
 
   /// Executes a query with default parameters:
   /// * TDB
-  fn query<Q: ToString>(&mut self, query: Q) -> error::Result<Frame> {
+  fn query<Q: ToString>(&self, query: Q) -> error::Result<Frame> {
     self.query_tw(query, false, false)
   }
 
   /// Executes a query with ability to trace it and see warnings, and default parameters:
   /// * TBD
-  fn query_tw<Q: ToString>(&mut self,
-                           query: Q,
-                           with_tracing: bool,
-                           with_warnings: bool)
-                           -> error::Result<Frame> {
+  fn query_tw<Q: ToString>(
+    &self,
+    query: Q,
+    with_tracing: bool,
+    with_warnings: bool,
+  ) -> error::Result<Frame> {
     let query_params = QueryParamsBuilder::new().finalize();
     self.query_with_params_tw(query, query_params, with_tracing, with_warnings)
   }
 
   /// Executes a query with bounded values (either with or without names).
-  fn query_with_values<Q: ToString, V: Into<QueryValues>>(&mut self,
-                                                          query: Q,
-                                                          values: V)
-                                                          -> error::Result<Frame> {
+  fn query_with_values<Q: ToString, V: Into<QueryValues>>(
+    &self,
+    query: Q,
+    values: V,
+  ) -> error::Result<Frame> {
     self.query_with_values_tw(query, values, false, false)
   }
 
   /// Executes a query with bounded values (either with or without names)
   /// and ability to see warnings, trace a request and default parameters.
-  fn query_with_values_tw<Q: ToString, V: Into<QueryValues>>(&mut self,
-                                                             query: Q,
-                                                             values: V,
-                                                             with_tracing: bool,
-                                                             with_warnings: bool)
-                                                             -> error::Result<Frame> {
+  fn query_with_values_tw<Q: ToString, V: Into<QueryValues>>(
+    &self,
+    query: Q,
+    values: V,
+    with_tracing: bool,
+    with_warnings: bool,
+  ) -> error::Result<Frame> {
     let query_params_builder = QueryParamsBuilder::new();
     let query_params = query_params_builder.values(values.into()).finalize();
     self.query_with_params_tw(query, query_params, with_tracing, with_warnings)
   }
 
   /// Executes a query with query params without warnings and tracing.
-  fn query_with_params<Q: ToString>(&mut self,
-                                    query: Q,
-                                    query_params: QueryParams)
-                                    -> error::Result<Frame> {
+  fn query_with_params<Q: ToString>(
+    &self,
+    query: Q,
+    query_params: QueryParams,
+  ) -> error::Result<Frame> {
     self.query_with_params_tw(query, query_params, false, false)
   }
 }


Test patch:
diff --git a/tests/collection_types.rs b/tests/collection_types.rs
index 098631c..38a19d2 100644
--- a/tests/collection_types.rs
+++ b/tests/collection_types.rs
@@ -10,16 +10,16 @@ mod common;
 
 use common::*;
 
-use uuid::Uuid;
 use cdrs::query::QueryExecutor;
-use cdrs::types::ByName;
-use cdrs::types::{AsRust, IntoRustByName};
+use cdrs::types::blob::Blob;
 use cdrs::types::list::List;
 use cdrs::types::map::Map;
-use cdrs::types::blob::Blob;
+use cdrs::types::AsRust;
+use cdrs::types::ByName;
+use uuid::Uuid;
 
-use std::str::FromStr;
 use std::collections::HashMap;
+use std::str::FromStr;
 
 #[test]
 #[ignore]
@@ -27,7 +27,7 @@ fn list() {
     let cql = "CREATE TABLE IF NOT EXISTS cdrs_test.test_lists \
                (my_text_list frozen<list<text>> PRIMARY KEY, \
                my_nested_list list<frozen<list<int>>>)";
-    let mut session = setup(cql).expect("setup");
+    let session = setup(cql).expect("setup");
 
     let my_text_list = vec!["text1", "text2", "text3"];
     let my_nested_list: Vec<Vec<i32>> =
@@ -36,32 +36,34 @@ fn list() {
 
     let cql = "INSERT INTO cdrs_test.test_lists \
                (my_text_list, my_nested_list) VALUES (?, ?)";
-    session.query_with_values(cql, values)
-           .expect("insert lists error");
+    session
+        .query_with_values(cql, values)
+        .expect("insert lists error");
 
     let cql = "SELECT * FROM cdrs_test.test_lists";
-    let rows = session.query(cql)
-                      .expect("query lists error")
-                      .get_body()
-                      .expect("get body with lists error")
-                      .into_rows()
-                      .expect("converting body with lists into rows error");
+    let rows = session
+        .query(cql)
+        .expect("query lists error")
+        .get_body()
+        .expect("get body with lists error")
+        .into_rows()
+        .expect("converting body with lists into rows error");
 
     assert_eq!(rows.len(), 1);
     for row in rows {
-        let my_text_list_row: Vec<String> =
-            row.r_by_name::<List>("my_text_list").expect("my_text_list")
-               .as_r_rust()
-               .expect("my_text_list as rust");
-        let my_nested_list_outer_row: Vec<List> =
-            row.r_by_name::<List>("my_nested_list").expect("my_nested_list")
-               .as_r_rust()
-               .expect("my_nested_list (outer) as rust");
+        let my_text_list_row: Vec<String> = row.r_by_name::<List>("my_text_list")
+            .expect("my_text_list")
+            .as_r_rust()
+            .expect("my_text_list as rust");
+        let my_nested_list_outer_row: Vec<List> = row.r_by_name::<List>("my_nested_list")
+            .expect("my_nested_list")
+            .as_r_rust()
+            .expect("my_nested_list (outer) as rust");
         let mut my_nested_list_row = Vec::with_capacity(my_nested_list_outer_row.len());
         for my_nested_list_inner_row in my_nested_list_outer_row {
-            let my_nested_list_inner_row: Vec<i32> =
-                my_nested_list_inner_row.as_r_rust()
-                                        .expect("my_nested_list (inner) as rust");
+            let my_nested_list_inner_row: Vec<i32> = my_nested_list_inner_row
+                .as_r_rust()
+                .expect("my_nested_list (inner) as rust");
             my_nested_list_row.push(my_nested_list_inner_row);
         }
         assert_eq!(my_text_list_row, vec!["text1", "text2", "text3"]);
@@ -72,16 +74,17 @@ fn list() {
 #[test]
 #[ignore]
 #[cfg(feature = "v4")]
-#[ignore]
 fn list_v4() {
     let cql = "CREATE TABLE IF NOT EXISTS cdrs_test.test_lists_v4 \
                (my_text_list frozen<list<text>> PRIMARY KEY, \
                my_nested_list list<frozen<list<smallint>>>)";
-    let mut session = setup(cql).expect("setup");
+    let session = setup(cql).expect("setup");
 
-    let my_text_list = vec!["text1".to_string(),
-                            "text2".to_string(),
-                            "text3".to_string()];
+    let my_text_list = vec![
+        "text1".to_string(),
+        "text2".to_string(),
+        "text3".to_string(),
+    ];
     let my_nested_list: Vec<Vec<i16>> =
         vec![vec![1, 2, 3], vec![999, 888, 777, 666, 555], vec![-1, -2]];
     let values = query_values!(my_text_list.clone(), my_nested_list.clone());
@@ -91,28 +94,29 @@ fn list_v4() {
     session.query_with_values(cql, values).expect("insert");
 
     let cql = "SELECT * FROM cdrs_test.test_lists_v4";
-    let rows = session.query(cql)
-                      .expect("query")
-                      .get_body()
-                      .expect("get body")
-                      .into_rows()
-                      .expect("into rows");
+    let rows = session
+        .query(cql)
+        .expect("query")
+        .get_body()
+        .expect("get body")
+        .into_rows()
+        .expect("into rows");
 
     assert_eq!(rows.len(), 1);
     for row in rows {
-        let my_text_list_row: Vec<String> =
-            row.r_by_name::<List>("my_text_list").expect("my_text_list")
-               .as_r_rust()
-               .expect("my_text_list as rust");
-        let my_nested_list_outer_row: Vec<List> =
-            row.r_by_name::<List>("my_nested_list").expect("my_nested_list")
-               .as_r_rust()
-               .expect("my_nested_list (outer) as rust");
+        let my_text_list_row: Vec<String> = row.r_by_name::<List>("my_text_list")
+            .expect("my_text_list")
+            .as_r_rust()
+            .expect("my_text_list as rust");
+        let my_nested_list_outer_row: Vec<List> = row.r_by_name::<List>("my_nested_list")
+            .expect("my_nested_list")
+            .as_r_rust()
+            .expect("my_nested_list (outer) as rust");
         let mut my_nested_list_row = Vec::with_capacity(my_nested_list_outer_row.len());
         for my_nested_list_inner_row in my_nested_list_outer_row {
-            let my_nested_list_inner_row: Vec<i16> =
-                my_nested_list_inner_row.as_r_rust()
-                                        .expect("my_nested_list (inner) as rust");
+            let my_nested_list_inner_row: Vec<i16> = my_nested_list_inner_row
+                .as_r_rust()
+                .expect("my_nested_list (inner) as rust");
             my_nested_list_row.push(my_nested_list_inner_row);
         }
         assert_eq!(my_text_list_row, my_text_list);
@@ -126,11 +130,13 @@ fn set() {
     let cql = "CREATE TABLE IF NOT EXISTS cdrs_test.test_sets \
                (my_text_set frozen<set<text>> PRIMARY KEY, \
                my_nested_set set<frozen<set<int>>>)";
-    let mut session = setup(cql).expect("setup");
+    let session = setup(cql).expect("setup");
 
-    let my_text_set = vec!["text1".to_string(),
-                           "text2".to_string(),
-                           "text3".to_string()];
+    let my_text_set = vec![
+        "text1".to_string(),
+        "text2".to_string(),
+        "text3".to_string(),
+    ];
     let my_nested_set: Vec<Vec<i32>> =
         vec![vec![-2, -1], vec![1, 2, 3], vec![555, 666, 777, 888, 999]];
     let values = query_values!(my_text_set.clone(), my_nested_set.clone());
@@ -140,28 +146,29 @@ fn set() {
     session.query_with_values(cql, values).expect("insert");
 
     let cql = "SELECT * FROM cdrs_test.test_sets";
-    let rows = session.query(cql)
-                      .expect("query")
-                      .get_body()
-                      .expect("get body")
-                      .into_rows()
-                      .expect("into rows");
+    let rows = session
+        .query(cql)
+        .expect("query")
+        .get_body()
+        .expect("get body")
+        .into_rows()
+        .expect("into rows");
 
     assert_eq!(rows.len(), 1);
     for row in rows {
-        let my_text_set_row: Vec<String> =
-            row.r_by_name::<List>("my_text_set").expect("my_text_set")
-               .as_r_rust()
-               .expect("my_text_set as rust");
-        let my_nested_set_outer_row: Vec<List> =
-            row.r_by_name::<List>("my_nested_set").expect("my_nested_set")
-               .as_r_rust()
-               .expect("my_nested_set (outer) as rust");
+        let my_text_set_row: Vec<String> = row.r_by_name::<List>("my_text_set")
+            .expect("my_text_set")
+            .as_r_rust()
+            .expect("my_text_set as rust");
+        let my_nested_set_outer_row: Vec<List> = row.r_by_name::<List>("my_nested_set")
+            .expect("my_nested_set")
+            .as_r_rust()
+            .expect("my_nested_set (outer) as rust");
         let mut my_nested_set_row = Vec::with_capacity(my_nested_set_outer_row.len());
         for my_nested_set_inner_row in my_nested_set_outer_row {
-            let my_nested_set_inner_row: Vec<i32> =
-                my_nested_set_inner_row.as_r_rust()
-                                       .expect("my_nested_set (inner) as rust");
+            let my_nested_set_inner_row: Vec<i32> = my_nested_set_inner_row
+                .as_r_rust()
+                .expect("my_nested_set (inner) as rust");
             my_nested_set_row.push(my_nested_set_inner_row);
         }
         assert_eq!(my_text_set_row, my_text_set);
@@ -172,16 +179,17 @@ fn set() {
 #[test]
 #[ignore]
 #[cfg(feature = "v4")]
-#[ignore]
 fn set_v4() {
     let cql = "CREATE TABLE IF NOT EXISTS cdrs_test.test_sets_v4 \
                (my_text_set frozen<set<text>> PRIMARY KEY, \
                my_nested_set set<frozen<set<smallint>>>)";
-    let mut session = setup(cql).expect("setup");
+    let session = setup(cql).expect("setup");
 
-    let my_text_set = vec!["text1".to_string(),
-                           "text2".to_string(),
-                           "text3".to_string()];
+    let my_text_set = vec![
+        "text1".to_string(),
+        "text2".to_string(),
+        "text3".to_string(),
+    ];
     let my_nested_set: Vec<Vec<i16>> =
         vec![vec![-2, -1], vec![1, 2, 3], vec![555, 666, 777, 888, 999]];
     let values = query_values!(my_text_set.clone(), my_nested_set.clone());
@@ -191,28 +199,29 @@ fn set_v4() {
     session.query_with_values(cql, values).expect("insert");
 
     let cql = "SELECT * FROM cdrs_test.test_sets_v4";
-    let rows = session.query(cql)
-                      .expect("query")
-                      .get_body()
-                      .expect("get body")
-                      .into_rows()
-                      .expect("into rows");
+    let rows = session
+        .query(cql)
+        .expect("query")
+        .get_body()
+        .expect("get body")
+        .into_rows()
+        .expect("into rows");
 
     assert_eq!(rows.len(), 1);
     for row in rows {
-        let my_text_set_row: Vec<String> =
-            row.r_by_name::<List>("my_text_set").expect("my_text_set")
-               .as_r_rust()
-               .expect("my_text_set as rust");
-        let my_nested_set_outer_row: Vec<List> =
-            row.r_by_name::<List>("my_nested_set").expect("my_nested_set")
-               .as_r_rust()
-               .expect("my_nested_set (outer) as rust");
+        let my_text_set_row: Vec<String> = row.r_by_name::<List>("my_text_set")
+            .expect("my_text_set")
+            .as_r_rust()
+            .expect("my_text_set as rust");
+        let my_nested_set_outer_row: Vec<List> = row.r_by_name::<List>("my_nested_set")
+            .expect("my_nested_set")
+            .as_r_rust()
+            .expect("my_nested_set (outer) as rust");
         let mut my_nested_set_row = Vec::with_capacity(my_nested_set_outer_row.len());
         for my_nested_set_inner_row in my_nested_set_outer_row {
-            let my_nested_set_inner_row: Vec<i16> =
-                my_nested_set_inner_row.as_r_rust()
-                                       .expect("my_nested_set (inner) as rust");
+            let my_nested_set_inner_row: Vec<i16> = my_nested_set_inner_row
+                .as_r_rust()
+                .expect("my_nested_set (inner) as rust");
             my_nested_set_row.push(my_nested_set_inner_row);
         }
         assert_eq!(my_text_set_row, my_text_set);
@@ -227,7 +236,7 @@ fn map_without_blob() {
                (my_key int PRIMARY KEY, \
                my_text_map map<text, text>, \
                my_nested_map map<uuid, frozen<map<bigint, int>>>)";
-    let mut session = setup(cql).expect("setup");
+    let session = setup(cql).expect("setup");
 
     let my_text_map = hashmap!{
         "key1".to_string() => "value1".to_string(),
@@ -255,28 +264,29 @@ fn map_without_blob() {
     session.query_with_values(cql, values).expect("insert");
 
     let cql = "SELECT * FROM cdrs_test.test_maps_without_blob";
-    let rows = session.query(cql)
-                      .expect("query")
-                      .get_body()
-                      .expect("get body")
-                      .into_rows()
-                      .expect("into rows");
+    let rows = session
+        .query(cql)
+        .expect("query")
+        .get_body()
+        .expect("get body")
+        .into_rows()
+        .expect("into rows");
 
     assert_eq!(rows.len(), 1);
     for row in rows {
-        let my_text_map_row: HashMap<String, String> =
-            row.r_by_name::<Map>("my_text_map").expect("my_text_map")
-               .as_r_rust()
-               .expect("my_text_map as rust");
-        let my_nested_map_outer_row: HashMap<Uuid, Map> =
-            row.r_by_name::<Map>("my_nested_map").expect("my_nested_map")
-               .as_r_rust()
-               .expect("my_nested_map (outer) as rust");
+        let my_text_map_row: HashMap<String, String> = row.r_by_name::<Map>("my_text_map")
+            .expect("my_text_map")
+            .as_r_rust()
+            .expect("my_text_map as rust");
+        let my_nested_map_outer_row: HashMap<Uuid, Map> = row.r_by_name::<Map>("my_nested_map")
+            .expect("my_nested_map")
+            .as_r_rust()
+            .expect("my_nested_map (outer) as rust");
         let mut my_nested_map_row = HashMap::with_capacity(my_nested_map_outer_row.len());
         for (index, my_nested_map_inner_row) in my_nested_map_outer_row {
-            let my_nested_map_inner_row: HashMap<i64, i32> =
-                my_nested_map_inner_row.as_r_rust()
-                                       .expect("my_nested_map (inner) as rust");
+            let my_nested_map_inner_row: HashMap<i64, i32> = my_nested_map_inner_row
+                .as_r_rust()
+                .expect("my_nested_map (inner) as rust");
             my_nested_map_row.insert(index, my_nested_map_inner_row);
         }
         assert_eq!(my_text_map_row, my_text_map);
@@ -287,12 +297,11 @@ fn map_without_blob() {
 #[test]
 #[ignore]
 #[cfg(feature = "v4")]
-#[ignore]
 fn map_without_blob_v4() {
     let cql = "CREATE TABLE IF NOT EXISTS cdrs_test.test_maps_without_blob_v4 \
                (my_text_map frozen<map<text, text>> PRIMARY KEY, \
                my_nested_map map<uuid, frozen<map<bigint, tinyint>>>)";
-    let mut session = setup(cql).expect("setup");
+    let session = setup(cql).expect("setup");
 
     let my_text_map = hashmap!{
         "key1".to_string() => "value1".to_string(),
@@ -320,28 +329,29 @@ fn map_without_blob_v4() {
     session.query_with_values(cql, values).expect("insert");
 
     let cql = "SELECT * FROM cdrs_test.test_maps_without_blob_v4";
-    let rows = session.query(cql)
-                      .expect("query")
-                      .get_body()
-                      .expect("get body")
-                      .into_rows()
-                      .expect("into rows");
+    let rows = session
+        .query(cql)
+        .expect("query")
+        .get_body()
+        .expect("get body")
+        .into_rows()
+        .expect("into rows");
 
     assert_eq!(rows.len(), 1);
     for row in rows {
-        let my_text_map_row: HashMap<String, String> =
-            row.r_by_name::<Map>("my_text_map").expect("my_text_map")
-               .as_r_rust()
-               .expect("my_text_map as rust");
-        let my_nested_map_outer_row: HashMap<Uuid, Map> =
-            row.r_by_name::<Map>("my_nested_map").expect("my_nested_map")
-               .as_r_rust()
-               .expect("my_nested_map (outer) as rust");
+        let my_text_map_row: HashMap<String, String> = row.r_by_name::<Map>("my_text_map")
+            .expect("my_text_map")
+            .as_r_rust()
+            .expect("my_text_map as rust");
+        let my_nested_map_outer_row: HashMap<Uuid, Map> = row.r_by_name::<Map>("my_nested_map")
+            .expect("my_nested_map")
+            .as_r_rust()
+            .expect("my_nested_map (outer) as rust");
         let mut my_nested_map_row = HashMap::with_capacity(my_nested_map_outer_row.len());
         for (index, my_nested_map_inner_row) in my_nested_map_outer_row {
-            let my_nested_map_inner_row: HashMap<i64, i8> =
-                my_nested_map_inner_row.as_r_rust()
-                                       .expect("my_nested_map (inner) as rust");
+            let my_nested_map_inner_row: HashMap<i64, i8> = my_nested_map_inner_row
+                .as_r_rust()
+                .expect("my_nested_map (inner) as rust");
             my_nested_map_row.insert(index, my_nested_map_inner_row);
         }
         assert_eq!(my_text_map_row, my_text_map);
@@ -355,7 +365,7 @@ fn map() {
     let cql = "CREATE TABLE IF NOT EXISTS cdrs_test.test_maps \
                (my_text_map frozen<map<text, text>> PRIMARY KEY, \
                my_nested_map map<uuid, frozen<map<bigint, blob>>>)";
-    let mut session = setup(cql).expect("setup");
+    let session = setup(cql).expect("setup");
 
     let my_text_map = hashmap!{
         "key1".to_string() => "value1".to_string(),
@@ -383,28 +393,29 @@ fn map() {
     session.query_with_values(cql, values).expect("insert");
 
     let cql = "SELECT * FROM cdrs_test.test_maps";
-    let rows = session.query(cql)
-                      .expect("query")
-                      .get_body()
-                      .expect("get body")
-                      .into_rows()
-                      .expect("into rows");
+    let rows = session
+        .query(cql)
+        .expect("query")
+        .get_body()
+        .expect("get body")
+        .into_rows()
+        .expect("into rows");
 
     assert_eq!(rows.len(), 1);
     for row in rows {
-        let my_text_map_row: HashMap<String, String> =
-            row.r_by_name::<Map>("my_text_map").expect("my_text_map")
-               .as_r_rust()
-               .expect("my_text_map as rust");
-        let my_nested_map_outer_row: HashMap<Uuid, Map> =
-            row.r_by_name::<Map>("my_nested_map").expect("my_nested_map")
-               .as_r_rust()
-               .expect("my_nested_map (outer) as rust");
+        let my_text_map_row: HashMap<String, String> = row.r_by_name::<Map>("my_text_map")
+            .expect("my_text_map")
+            .as_r_rust()
+            .expect("my_text_map as rust");
+        let my_nested_map_outer_row: HashMap<Uuid, Map> = row.r_by_name::<Map>("my_nested_map")
+            .expect("my_nested_map")
+            .as_r_rust()
+            .expect("my_nested_map (outer) as rust");
         let mut my_nested_map_row = HashMap::with_capacity(my_nested_map_outer_row.len());
         for (index, my_nested_map_inner_row) in my_nested_map_outer_row {
-            let my_nested_map_inner_row: HashMap<i64, Blob> =
-                my_nested_map_inner_row.as_r_rust()
-                                       .expect("my_nested_map (inner) as rust");
+            let my_nested_map_inner_row: HashMap<i64, Blob> = my_nested_map_inner_row
+                .as_r_rust()
+                .expect("my_nested_map (inner) as rust");
             my_nested_map_row.insert(index, my_nested_map_inner_row);
         }
         assert_eq!(my_text_map_row, my_text_map);
diff --git a/tests/common/mod.rs b/tests/common/mod.rs
index 3bc5e7c..4b8d851 100644
--- a/tests/common/mod.rs
+++ b/tests/common/mod.rs
@@ -1,15 +1,17 @@
-use regex::Regex;
+use std::cell::RefCell;
+
 use cdrs::authenticators::NoneAuthenticator;
-use cdrs::compression::Compression;
+use cdrs::cluster::{Cluster, Session};
+// use cdrs::compression::Compression;
 use cdrs::error::Result;
 use cdrs::load_balancing::RoundRobin;
-use cdrs::cluster::{Cluster, Session};
 use cdrs::query::QueryExecutor;
 use cdrs::transport::TransportTcp;
+use regex::Regex;
 
 const ADDR: &'static str = "127.0.0.1:9042";
 
-pub type CSession = Session<RoundRobin<TransportTcp>, NoneAuthenticator>;
+pub type CSession = Session<RoundRobin<RefCell<TransportTcp>>, NoneAuthenticator>;
 
 pub fn setup(create_table_cql: &'static str) -> Result<CSession> {
   setup_multiple(&[create_table_cql])
@@ -17,10 +19,11 @@ pub fn setup(create_table_cql: &'static str) -> Result<CSession> {
 
 pub fn setup_multiple(create_cqls: &[&'static str]) -> Result<CSession> {
   let authenticator = NoneAuthenticator {};
-  let nodes = vec!["127.0.0.1:9042"];
+  let nodes = vec![ADDR];
   let cluster = Cluster::new(nodes, authenticator);
-  let mut session = cluster.connect(RoundRobin::new())
-                           .expect("No compression connection error");
+  let session = cluster
+    .connect(RoundRobin::new())
+    .expect("No compression connection error");
   let re_table_name = Regex::new(r"CREATE TABLE IF NOT EXISTS (\w+\.\w+)").unwrap();
 
   let create_keyspace_query = "CREATE KEYSPACE IF NOT EXISTS cdrs_test WITH \
@@ -29,8 +32,9 @@ pub fn setup_multiple(create_cqls: &[&'static str]) -> Result<CSession> {
   session.query(create_keyspace_query)?;
 
   for create_cql in create_cqls.iter() {
-    let table_name = re_table_name.captures(create_cql)
-                                  .map(|cap| cap.get(1).unwrap().as_str());
+    let table_name = re_table_name
+      .captures(create_cql)
+      .map(|cap| cap.get(1).unwrap().as_str());
 
     // Re-using tables is a lot faster than creating/dropping them for every test.
     // But if table definitions change while editing tests
diff --git a/tests/derive_traits.rs b/tests/derive_traits.rs
index 21a311b..72190ce 100644
--- a/tests/derive_traits.rs
+++ b/tests/derive_traits.rs
@@ -12,18 +12,18 @@ mod common;
 
 use common::*;
 
-use cdrs::query::QueryValues;
-use cdrs::query::QueryExecutor;
-use cdrs::types::{AsRust, AsRustType, IntoRustByName};
-use cdrs::types::value::{Bytes, Value};
-use cdrs::types::rows::Row;
-use cdrs::types::udt::UDT;
-use cdrs::types::map::Map;
 use cdrs::error::Result as CDRSResult;
 use cdrs::frame::IntoBytes;
-use time::Timespec;
 use cdrs::frame::{TryFromRow, TryFromUDT};
+use cdrs::query::QueryExecutor;
+use cdrs::query::QueryValues;
 use cdrs::types::from_cdrs::FromCDRSByName;
+use cdrs::types::map::Map;
+use cdrs::types::rows::Row;
+use cdrs::types::udt::UDT;
+use cdrs::types::value::{Bytes, Value};
+use cdrs::types::{AsRust, AsRustType, IntoRustByName};
+use time::Timespec;
 
 use std::collections::HashMap;
 
@@ -33,7 +33,7 @@ fn simple_udt() {
   let create_type_cql = "CREATE TYPE IF NOT EXISTS cdrs_test.derive_udt (my_text text)";
   let create_table_cql = "CREATE TABLE IF NOT EXISTS cdrs_test.test_derived_udt \
                           (my_key int PRIMARY KEY, my_udt derive_udt)";
-  let mut session = setup_multiple(&[create_type_cql, create_table_cql]).expect("setup");
+  let session = setup_multiple(&[create_type_cql, create_table_cql]).expect("setup");
 
   #[derive(Clone, Debug, IntoCDRSValue, TryFromRow, PartialEq)]
   struct RowStruct {
@@ -89,7 +89,7 @@ fn nested_udt() {
                           (my_inner_udt frozen<nested_inner_udt>)";
   let create_table_cql = "CREATE TABLE IF NOT EXISTS cdrs_test.test_nested_udt \
                           (my_key int PRIMARY KEY, my_outer_udt nested_outer_udt)";
-  let mut session =
+  let session =
     setup_multiple(&[create_type1_cql, create_type2_cql, create_table_cql]).expect("setup");
 
   #[derive(Clone, Debug, IntoCDRSValue, TryFromRow, PartialEq)]
@@ -153,7 +153,7 @@ fn alter_udt_add() {
   let create_type_cql = "CREATE TYPE cdrs_test.alter_udt_add_udt (my_text text)";
   let create_table_cql = "CREATE TABLE IF NOT EXISTS cdrs_test.test_alter_udt_add \
                           (my_key int PRIMARY KEY, my_map frozen<map<text, alter_udt_add_udt>>)";
-  let mut session = setup_multiple(&[
+  let session = setup_multiple(&[
     drop_table_cql,
     drop_type_cql,
     create_type_cql,
diff --git a/tests/keyspace.rs b/tests/keyspace.rs
index ba1703c..eceaa56 100644
--- a/tests/keyspace.rs
+++ b/tests/keyspace.rs
@@ -6,8 +6,8 @@ use cdrs::authenticators::NoneAuthenticator;
 use cdrs::cluster::Cluster;
 use cdrs::load_balancing::RoundRobin;
 use cdrs::query::QueryExecutor;
-use cdrs::types::{AsRust, ByName, IntoRustByName};
 use cdrs::types::map::Map;
+use cdrs::types::{AsRust, ByName, IntoRustByName};
 
 const ADDR: &'static str = "127.0.0.1:9042";
 
@@ -15,8 +15,8 @@ const ADDR: &'static str = "127.0.0.1:9042";
 #[ignore]
 fn create_keyspace() {
         let cluster = Cluster::new(vec!["127.0.0.1:9042"], NoneAuthenticator {});
-        let mut session =
-                cluster.connect(RoundRobin::new()).expect("No compression connection error");
+        let session = cluster.connect(RoundRobin::new())
+                .expect("No compression connection error");
 
         let drop_query = "DROP KEYSPACE IF EXISTS create_ks_test";
         let keyspace_droped = session.query(drop_query).is_ok();
@@ -26,50 +26,57 @@ fn create_keyspace() {
                             replication = {'class': 'SimpleStrategy', 'replication_factor': 1} \
                             AND durable_writes = false";
         let keyspace_created = session.query(create_query).is_ok();
-        assert!(keyspace_created,
-                "Should create new keyspace without errors");
+        assert!(
+                keyspace_created,
+                "Should create new keyspace without errors"
+        );
 
         let select_query =
                 "SELECT * FROM system_schema.keyspaces WHERE keyspace_name = 'create_ks_test'";
-        let keyspace_selected = session.query(select_query).expect("select keyspace query")
-                                       .get_body()
-                                       .expect("get select keyspace query body")
-                                       .into_rows()
-                                       .expect("convert keyspaces results into rows");
+        let keyspace_selected = session.query(select_query)
+                .expect("select keyspace query")
+                .get_body()
+                .expect("get select keyspace query body")
+                .into_rows()
+                .expect("convert keyspaces results into rows");
 
         assert_eq!(keyspace_selected.len(), 1);
         let keyspace = &keyspace_selected[0];
 
-        let keyspace_name: String =
-                keyspace.get_r_by_name("keyspace_name").expect("keyspace name into rust error");
-        assert_eq!(keyspace_name,
-                   "create_ks_test".to_string(),
-                   "wrong keyspace name");
+        let keyspace_name: String = keyspace.get_r_by_name("keyspace_name")
+                .expect("keyspace name into rust error");
+        assert_eq!(
+                keyspace_name,
+                "create_ks_test".to_string(),
+                "wrong keyspace name"
+        );
 
-        let durable_writes: bool =
-                keyspace.get_r_by_name("durable_writes").expect("durable writes into rust error");
+        let durable_writes: bool = keyspace.get_r_by_name("durable_writes")
+                .expect("durable writes into rust error");
         assert_eq!(durable_writes, false, "wrong durable writes");
 
         let mut expected_strategy_options: HashMap<String, String> = HashMap::new();
         expected_strategy_options.insert("replication_factor".to_string(), "1".to_string());
-        expected_strategy_options.insert("class".to_string(),
-                                         "org.apache.cassandra.locator.SimpleStrategy".to_string());
-        let strategy_options: HashMap<String, String> =
-    keyspace.r_by_name::<Map>("replication")
-            .expect("strategy optioins into rust error")
-            .as_r_rust()
-            .expect("uuid_key_map");
-        assert_eq!(expected_strategy_options,
-                   strategy_options,
-                   "wrong strategy options");
+        expected_strategy_options.insert(
+                "class".to_string(),
+                "org.apache.cassandra.locator.SimpleStrategy".to_string(),
+        );
+        let strategy_options: HashMap<String, String> = keyspace.r_by_name::<Map>("replication")
+                .expect("strategy optioins into rust error")
+                .as_r_rust()
+                .expect("uuid_key_map");
+        assert_eq!(
+                expected_strategy_options, strategy_options,
+                "wrong strategy options"
+        );
 }
 
 #[test]
 #[ignore]
 fn alter_keyspace() {
         let cluster = Cluster::new(vec!["127.0.0.1:9042"], NoneAuthenticator {});
-        let mut session =
-                cluster.connect(RoundRobin::new()).expect("No compression connection error");
+        let session = cluster.connect(RoundRobin::new())
+                .expect("No compression connection error");
 
         let drop_query = "DROP KEYSPACE IF EXISTS alter_ks_test";
         let keyspace_droped = session.query(drop_query).is_ok();
@@ -79,58 +86,68 @@ fn alter_keyspace() {
                             replication = {'class': 'SimpleStrategy', 'replication_factor': 1} \
                             AND durable_writes = false";
         let keyspace_created = session.query(create_query).is_ok();
-        assert!(keyspace_created,
-                "Should create new keyspace without errors");
+        assert!(
+                keyspace_created,
+                "Should create new keyspace without errors"
+        );
 
         let alter_query = "ALTER KEYSPACE alter_ks_test WITH \
                            replication = {'class': 'SimpleStrategy', 'replication_factor': 3} \
                            AND durable_writes = false";
-        assert!(session.query(alter_query).is_ok(),
-                "alter should be without errors");
+        assert!(
+                session.query(alter_query).is_ok(),
+                "alter should be without errors"
+        );
 
         let select_query =
                 "SELECT * FROM system_schema.keyspaces WHERE keyspace_name = 'alter_ks_test'";
-        let keyspace_selected = session.query(select_query).expect("select keyspace query")
-                                       .get_body()
-                                       .expect("get select keyspace query body")
-                                       .into_rows()
-                                       .expect("convert keyspaces results into rows");
+        let keyspace_selected = session.query(select_query)
+                .expect("select keyspace query")
+                .get_body()
+                .expect("get select keyspace query body")
+                .into_rows()
+                .expect("convert keyspaces results into rows");
 
         assert_eq!(keyspace_selected.len(), 1);
         let keyspace = &keyspace_selected[0];
 
-        let strategy_options: HashMap<String, String> =
-    keyspace.r_by_name::<Map>("replication")
-            .expect("strategy optioins into rust error")
-            .as_r_rust()
-            .expect("uuid_key_map");
-
-        assert_eq!(strategy_options.get("replication_factor")
-                                   .expect("replication_factor unwrap"),
-                   &"3".to_string());
+        let strategy_options: HashMap<String, String> = keyspace.r_by_name::<Map>("replication")
+                .expect("strategy optioins into rust error")
+                .as_r_rust()
+                .expect("uuid_key_map");
+
+        assert_eq!(
+                strategy_options
+                        .get("replication_factor")
+                        .expect("replication_factor unwrap"),
+                &"3".to_string()
+        );
 }
 
 #[test]
 #[ignore]
 fn use_keyspace() {
         let cluster = Cluster::new(vec!["127.0.0.1:9042"], NoneAuthenticator {});
-        let mut session =
-                cluster.connect(RoundRobin::new()).expect("No compression connection error");
+        let session = cluster.connect(RoundRobin::new())
+                .expect("No compression connection error");
 
         let create_query = "CREATE KEYSPACE IF NOT EXISTS use_ks_test WITH \
                             replication = {'class': 'SimpleStrategy', 'replication_factor': 1} \
                             AND durable_writes = false";
         let keyspace_created = session.query(create_query).is_ok();
-        assert!(keyspace_created,
-                "Should create new keyspace without errors");
+        assert!(
+                keyspace_created,
+                "Should create new keyspace without errors"
+        );
 
         let use_query = "USE use_ks_test";
-        let keyspace_used = session.query(use_query).expect("should use selected")
-                                   .get_body()
-                                   .expect("should get body")
-                                   .into_set_keyspace()
-                                   .expect("set keyspace")
-                                   .body;
+        let keyspace_used = session.query(use_query)
+                .expect("should use selected")
+                .get_body()
+                .expect("should get body")
+                .into_set_keyspace()
+                .expect("set keyspace")
+                .body;
         assert_eq!(keyspace_used.as_str(), "use_ks_test", "wrong kespace used");
 }
 
@@ -138,15 +155,17 @@ fn use_keyspace() {
 #[ignore]
 fn drop_keyspace() {
         let cluster = Cluster::new(vec!["127.0.0.1:9042"], NoneAuthenticator {});
-        let mut session =
-                cluster.connect(RoundRobin::new()).expect("No compression connection error");
+        let session = cluster.connect(RoundRobin::new())
+                .expect("No compression connection error");
 
         let create_query = "CREATE KEYSPACE IF NOT EXISTS drop_ks_test WITH \
                             replication = {'class': 'SimpleStrategy', 'replication_factor': 1} \
                             AND durable_writes = false";
         let keyspace_created = session.query(create_query).is_ok();
-        assert!(keyspace_created,
-                "Should create new keyspace without errors");
+        assert!(
+                keyspace_created,
+                "Should create new keyspace without errors"
+        );
 
         let drop_query = "DROP KEYSPACE drop_ks_test";
         let keyspace_droped = session.query(drop_query).is_ok();
diff --git a/tests/native_types.rs b/tests/native_types.rs
index 81ca746..1597925 100644
--- a/tests/native_types.rs
+++ b/tests/native_types.rs
@@ -8,23 +8,23 @@ mod common;
 
 use common::*;
 
-use uuid::Uuid;
+use cdrs::query::QueryExecutor;
+use cdrs::types::blob::Blob;
 use cdrs::types::map::Map;
 use cdrs::types::value::{Bytes, Value};
 use cdrs::types::{AsRust, ByName, IntoRustByName};
-use cdrs::types::blob::Blob;
-use cdrs::query::QueryExecutor;
+use uuid::Uuid;
 
-use std::str::FromStr;
-use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
 use std::collections::HashMap;
+use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
+use std::str::FromStr;
 
 #[test]
 #[ignore]
 fn string() {
     let cql = "CREATE TABLE IF NOT EXISTS cdrs_test.test_string \
                (my_ascii ascii PRIMARY KEY, my_text text, my_varchar varchar)";
-    let mut session = setup(cql).expect("setup");
+    let session = setup(cql).expect("setup");
 
     let my_ascii = "my_ascii";
     let my_text = "my_text";
@@ -33,16 +33,18 @@ fn string() {
 
     let query = "INSERT INTO cdrs_test.test_string \
                  (my_ascii, my_text, my_varchar) VALUES (?, ?, ?)";
-    session.query_with_values(query, values)
-           .expect("insert stings error");
+    session
+        .query_with_values(query, values)
+        .expect("insert stings error");
 
     let cql = "SELECT * FROM cdrs_test.test_string";
-    let rows = session.query(cql)
-                      .expect("select strings query error")
-                      .get_body()
-                      .expect("get body error")
-                      .into_rows()
-                      .expect("converting into rows error");
+    let rows = session
+        .query(cql)
+        .expect("select strings query error")
+        .get_body()
+        .expect("get body error")
+        .into_rows()
+        .expect("converting into rows error");
 
     assert_eq!(rows.len(), 1);
     for row in rows {
@@ -60,7 +62,7 @@ fn string() {
 fn counter() {
     let cql = "CREATE TABLE IF NOT EXISTS cdrs_test.test_counter \
                (my_bigint bigint PRIMARY KEY, my_counter counter)";
-    let mut session = setup(cql).expect("setup");
+    let session = setup(cql).expect("setup");
 
     let my_bigint: i64 = 10_000_000_000_000_000;
     let my_counter: i64 = 100_000_000;
@@ -71,12 +73,13 @@ fn counter() {
     session.query_with_values(query, values).expect("insert");
 
     let cql = "SELECT * FROM cdrs_test.test_counter";
-    let rows = session.query(cql)
-                      .expect("select counter query error")
-                      .get_body()
-                      .expect("get counter body error")
-                      .into_rows()
-                      .expect("converting coutner body into rows error");
+    let rows = session
+        .query(cql)
+        .expect("select counter query error")
+        .get_body()
+        .expect("get counter body error")
+        .into_rows()
+        .expect("converting coutner body into rows error");
 
     assert_eq!(rows.len(), 1);
     for row in rows {
@@ -93,7 +96,7 @@ fn counter() {
 fn integer() {
     let cql = "CREATE TABLE IF NOT EXISTS cdrs_test.test_integer \
                (my_bigint bigint PRIMARY KEY, my_int int, my_boolean boolean)";
-    let mut session = setup(cql).expect("setup");
+    let session = setup(cql).expect("setup");
 
     let my_bigint: i64 = 10_000_000_000_000_000;
     let my_int: i32 = 100_000_000;
@@ -102,16 +105,18 @@ fn integer() {
 
     let query = "INSERT INTO cdrs_test.test_integer \
                  (my_bigint, my_int, my_boolean) VALUES (?, ?, ?)";
-    session.query_with_values(query, values)
-           .expect("insert integers error");
+    session
+        .query_with_values(query, values)
+        .expect("insert integers error");
 
     let cql = "SELECT * FROM cdrs_test.test_integer";
-    let rows = session.query(cql)
-                      .expect("select integers query error")
-                      .get_body()
-                      .expect("get body with integers error")
-                      .into_rows()
-                      .expect("converting body with integers into rows error");
+    let rows = session
+        .query(cql)
+        .expect("select integers query error")
+        .get_body()
+        .expect("get body with integers error")
+        .into_rows()
+        .expect("converting body with integers into rows error");
 
     assert_eq!(rows.len(), 1);
     for row in rows {
@@ -128,12 +133,11 @@ fn integer() {
 #[test]
 #[ignore]
 #[cfg(feature = "v4")]
-#[ignore]
 fn integer_v4() {
     let cql = "CREATE TABLE IF NOT EXISTS cdrs_test.test_integer_v4 \
                (my_bigint bigint PRIMARY KEY, my_int int, my_smallint smallint, \
                my_tinyint tinyint, my_boolean boolean)";
-    let mut session = setup(cql).expect("setup");
+    let session = setup(cql).expect("setup");
 
     let my_bigint: i64 = 10_000_000_000_000_000;
     let my_int: i32 = 100_000_000;
@@ -144,16 +148,18 @@ fn integer_v4() {
 
     let query = "INSERT INTO cdrs_test.test_integer_v4 \
                  (my_bigint, my_int, my_smallint, my_tinyint, my_boolean) VALUES (?, ?, ?, ?, ?)";
-    session.query_with_values(query, values)
-           .expect("insert integers error");
+    session
+        .query_with_values(query, values)
+        .expect("insert integers error");
 
     let cql = "SELECT * FROM cdrs_test.test_integer_v4";
-    let rows = session.query(cql)
-                      .expect("query integers error")
-                      .get_body()
-                      .expect("get body with integers error")
-                      .into_rows()
-                      .expect("converting body with integers into rows error");
+    let rows = session
+        .query(cql)
+        .expect("query integers error")
+        .get_body()
+        .expect("get body with integers error")
+        .into_rows()
+        .expect("converting body with integers into rows error");
 
     assert_eq!(rows.len(), 1);
     for row in rows {
@@ -176,23 +182,25 @@ fn integer_v4() {
 fn float() {
     let cql = "CREATE TABLE IF NOT EXISTS cdrs_test.test_float \
                (my_float float PRIMARY KEY, my_double double)";
-    let mut session = setup(cql).expect("setup");
+    let session = setup(cql).expect("setup");
 
     let my_float: f32 = 123.456;
     let my_double: f64 = 987.654;
     let values = query_values!(my_float, my_double);
 
     let query = "INSERT INTO cdrs_test.test_float (my_float, my_double) VALUES (?, ?)";
-    session.query_with_values(query, values)
-           .expect("insert floats error");
+    session
+        .query_with_values(query, values)
+        .expect("insert floats error");
 
     let cql = "SELECT * FROM cdrs_test.test_float";
-    let rows = session.query(cql)
-                      .expect("query floats error")
-                      .get_body()
-                      .expect("get body with floats error")
-                      .into_rows()
-                      .expect("converting body with floats into rows error");
+    let rows = session
+        .query(cql)
+        .expect("query floats error")
+        .get_body()
+        .expect("get body with floats error")
+        .into_rows()
+        .expect("converting body with floats into rows error");
 
     assert_eq!(rows.len(), 1);
     for row in rows {
@@ -208,7 +216,7 @@ fn float() {
 fn blob() {
     let cql = "CREATE TABLE IF NOT EXISTS cdrs_test.test_blob \
                (my_blob blob PRIMARY KEY, my_mapblob map<text, blob>)";
-    let mut session = setup(cql).expect("setup");
+    let session = setup(cql).expect("setup");
 
     let my_blob: Blob = vec![0, 1, 2, 4, 8, 16, 32, 64, 128, 255].into();
     let my_map: HashMap<String, Blob> = [
@@ -220,33 +228,36 @@ fn blob() {
         .map(|x| x.clone())
         .collect();
 
-    let val_map: HashMap<String, Bytes> = my_map.clone()
-                                                .into_iter()
-                                                .map(|(k, v)| (k, Bytes::new(v.into_vec())))
-                                                .collect();
+    let val_map: HashMap<String, Bytes> = my_map
+        .clone()
+        .into_iter()
+        .map(|(k, v)| (k, Bytes::new(v.into_vec())))
+        .collect();
 
     let values = query_values!(my_blob.clone(), val_map);
 
     let query = "INSERT INTO cdrs_test.test_blob (my_blob, my_mapblob) VALUES (?,?)";
-    session.query_with_values(query, values)
-           .expect("insert blob error");
+    session
+        .query_with_values(query, values)
+        .expect("insert blob error");
 
     let cql = "SELECT * FROM cdrs_test.test_blob";
-    let rows = session.query(cql)
-                      .expect("query blobs error")
-                      .get_body()
-                      .expect("get body with blobs error")
-                      .into_rows()
-                      .expect("converting body with blobs into rows error");
+    let rows = session
+        .query(cql)
+        .expect("query blobs error")
+        .get_body()
+        .expect("get body with blobs error")
+        .into_rows()
+        .expect("converting body with blobs into rows error");
 
     assert_eq!(rows.len(), 1);
     for row in rows {
         let my_blob_row: Blob = row.get_r_by_name("my_blob").expect("my_blob");
         assert_eq!(my_blob_row, my_blob);
-        let my_map_row: HashMap<String, Blob> =
-            row.r_by_name::<Map>("my_mapblob").expect("my_mapblob by name")
-               .as_r_rust()
-               .expect("my_mapblob as r rust");
+        let my_map_row: HashMap<String, Blob> = row.r_by_name::<Map>("my_mapblob")
+            .expect("my_mapblob by name")
+            .as_r_rust()
+            .expect("my_mapblob as r rust");
         assert_eq!(my_map_row, my_map);
     }
 }
@@ -257,22 +268,24 @@ fn blob() {
 fn uuid() {
     let cql = "CREATE TABLE IF NOT EXISTS cdrs_test.test_uuid \
                (my_uuid uuid PRIMARY KEY)";
-    let mut session = setup(cql).expect("setup");
+    let session = setup(cql).expect("setup");
 
     let my_uuid = Uuid::from_str("bb16106a-10bc-4a07-baa3-126ffe208c43").unwrap();
     let values = query_values!(my_uuid);
 
     let query = "INSERT INTO cdrs_test.test_uuid (my_uuid) VALUES (?)";
-    session.query_with_values(query, values)
-           .expect("insert UUID error");
+    session
+        .query_with_values(query, values)
+        .expect("insert UUID error");
 
     let cql = "SELECT * FROM cdrs_test.test_uuid";
-    let rows = session.query(cql)
-                      .expect("query UUID error")
-                      .get_body()
-                      .expect("get body with UUID error")
-                      .into_rows()
-                      .expect("convertion body with UUID into rows error");
+    let rows = session
+        .query(cql)
+        .expect("query UUID error")
+        .get_body()
+        .expect("get body with UUID error")
+        .into_rows()
+        .expect("convertion body with UUID into rows error");
 
     assert_eq!(rows.len(), 1);
     for row in rows {
@@ -287,30 +300,34 @@ fn uuid() {
 fn time() {
     let cql = "CREATE TABLE IF NOT EXISTS cdrs_test.test_time \
                (my_timestamp timestamp PRIMARY KEY)";
-    let mut session = setup(cql).expect("setup");
+    let session = setup(cql).expect("setup");
 
     let my_timestamp = time::get_time();
     let values = query_values!(my_timestamp);
 
     let query = "INSERT INTO cdrs_test.test_time (my_timestamp) VALUES (?)";
-    session.query_with_values(query, values)
-           .expect("insert timestamp error");
+    session
+        .query_with_values(query, values)
+        .expect("insert timestamp error");
 
     let cql = "SELECT * FROM cdrs_test.test_time";
-    let rows = session.query(cql)
-                      .expect("query with time error")
-                      .get_body()
-                      .expect("get body with time error")
-                      .into_rows()
-                      .expect("converting body with time into rows error");
+    let rows = session
+        .query(cql)
+        .expect("query with time error")
+        .get_body()
+        .expect("get body with time error")
+        .into_rows()
+        .expect("converting body with time into rows error");
 
     assert_eq!(rows.len(), 1);
     for row in rows {
         let my_timestamp_row: time::Timespec =
             row.get_r_by_name("my_timestamp").expect("my_timestamp");
         assert_eq!(my_timestamp_row.sec, my_timestamp.sec);
-        assert_eq!(my_timestamp_row.nsec / 1_000_000,
-                   my_timestamp.nsec / 1_000_000); // C* `timestamp` has millisecond precision
+        assert_eq!(
+            my_timestamp_row.nsec / 1_000_000,
+            my_timestamp.nsec / 1_000_000
+        ); // C* `timestamp` has millisecond precision
     }
 }
 
@@ -319,23 +336,25 @@ fn time() {
 fn inet() {
     let cql = "CREATE TABLE IF NOT EXISTS cdrs_test.test_inet \
                (my_inet_v4 inet PRIMARY KEY, my_inet_v6 inet)";
-    let mut session = setup(cql).expect("setup");
+    let session = setup(cql).expect("setup");
 
     let my_inet_v4 = IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1));
     let my_inet_v6 = IpAddr::V6(Ipv6Addr::new(0, 0, 0, 0, 0, 0, 0, 1));
     let values = query_values!(my_inet_v4.clone(), my_inet_v6.clone());
 
     let query = "INSERT INTO cdrs_test.test_inet (my_inet_v4, my_inet_v6) VALUES (?, ?)";
-    session.query_with_values(query, values)
-           .expect("insert inet error");
+    session
+        .query_with_values(query, values)
+        .expect("insert inet error");
 
     let query = "SELECT * FROM cdrs_test.test_inet";
-    let rows = session.query(query)
-                      .expect("query inet error")
-                      .get_body()
-                      .expect("get body with inet error")
-                      .into_rows()
-                      .expect("converting body with inet into rows error");
+    let rows = session
+        .query(query)
+        .expect("query inet error")
+        .get_body()
+        .expect("get body with inet error")
+        .into_rows()
+        .expect("converting body with inet into rows error");
 
     assert_eq!(rows.len(), 1);
     for row in rows {
diff --git a/tests/tuple_types.rs b/tests/tuple_types.rs
index 3c1b57f..37454c6 100644
--- a/tests/tuple_types.rs
+++ b/tests/tuple_types.rs
@@ -8,15 +8,15 @@ mod common;
 
 use common::*;
 
-use uuid::Uuid;
-use time::Timespec;
-use cdrs::query::QueryExecutor;
-use cdrs::types::{IntoRustByIndex, IntoRustByName};
-use cdrs::types::value::{Bytes, Value};
-use cdrs::types::tuple::Tuple;
-use cdrs::types::blob::Blob;
 use cdrs::error::Result;
 use cdrs::frame::IntoBytes;
+use cdrs::query::QueryExecutor;
+use cdrs::types::blob::Blob;
+use cdrs::types::tuple::Tuple;
+use cdrs::types::value::{Bytes, Value};
+use cdrs::types::{IntoRustByIndex, IntoRustByName};
+use time::Timespec;
+use uuid::Uuid;
 
 use std::str::FromStr;
 
@@ -25,7 +25,7 @@ use std::str::FromStr;
 fn simple_tuple() {
   let cql = "CREATE TABLE IF NOT EXISTS cdrs_test.simple_tuple \
              (my_tuple tuple<text, int> PRIMARY KEY)";
-  let mut session = setup(cql).expect("setup");
+  let session = setup(cql).expect("setup");
 
   #[derive(Debug, Clone, PartialEq)]
   struct MyTuple {
@@ -37,8 +37,10 @@ fn simple_tuple() {
     pub fn try_from(tuple: Tuple) -> Result<MyTuple> {
       let my_text: String = tuple.get_r_by_index(0)?;
       let my_int: i32 = tuple.get_r_by_index(1)?;
-      Ok(MyTuple { my_text: my_text,
-                   my_int: my_int, })
+      Ok(MyTuple {
+        my_text: my_text,
+        my_int: my_int,
+      })
     }
   }
 
@@ -53,8 +55,10 @@ fn simple_tuple() {
     }
   }
 
-  let my_tuple = MyTuple { my_text: "my_text".to_string(),
-                           my_int: 0, };
+  let my_tuple = MyTuple {
+    my_text: "my_text".to_string(),
+    my_int: 0,
+  };
   let values = query_values!(my_tuple.clone());
 
   let cql = "INSERT INTO cdrs_test.simple_tuple \
@@ -62,12 +66,13 @@ fn simple_tuple() {
   session.query_with_values(cql, values).expect("insert");
 
   let cql = "SELECT * FROM cdrs_test.simple_tuple";
-  let rows = session.query(cql)
-                    .expect("query")
-                    .get_body()
-                    .expect("get body")
-                    .into_rows()
-                    .expect("into rows");
+  let rows = session
+    .query(cql)
+    .expect("query")
+    .get_body()
+    .expect("get body")
+    .into_rows()
+    .expect("into rows");
 
   assert_eq!(rows.len(), 1);
   for row in rows {
@@ -83,7 +88,7 @@ fn nested_tuples() {
   let cql = "CREATE TABLE IF NOT EXISTS cdrs_test.test_nested_tuples \
              (my_key int PRIMARY KEY, \
              my_outer_tuple tuple<uuid, blob, tuple<text, int, timestamp>>)";
-  let mut session = setup(cql).expect("setup");
+  let session = setup(cql).expect("setup");
 
   #[derive(Debug, Clone, PartialEq)]
   struct MyInnerTuple {
@@ -97,9 +102,11 @@ fn nested_tuples() {
       let my_text: String = tuple.get_r_by_index(0)?;
       let my_int: i32 = tuple.get_r_by_index(1)?;
       let my_timestamp: Timespec = tuple.get_r_by_index(2)?;
-      Ok(MyInnerTuple { my_text: my_text,
-                        my_int: my_int,
-                        my_timestamp: my_timestamp, })
+      Ok(MyInnerTuple {
+        my_text: my_text,
+        my_int: my_int,
+        my_timestamp: my_timestamp,
+      })
     }
   }
 
@@ -129,9 +136,11 @@ fn nested_tuples() {
       let my_blob: Blob = tuple.get_r_by_index(1)?;
       let my_inner_tuple: Tuple = tuple.get_r_by_index(2)?;
       let my_inner_tuple = MyInnerTuple::try_from(my_inner_tuple).expect("from tuple");
-      Ok(MyOuterTuple { my_uuid: my_uuid,
-                        my_blob: my_blob.into_vec(),
-                        my_inner_tuple: my_inner_tuple, })
+      Ok(MyOuterTuple {
+        my_uuid: my_uuid,
+        my_blob: my_blob.into_vec(),
+        my_inner_tuple: my_inner_tuple,
+      })
     }
   }
 
@@ -150,13 +159,19 @@ fn nested_tuples() {
 
   let my_uuid = Uuid::from_str("bb16106a-10bc-4a07-baa3-126ffe208c43").unwrap();
   let my_blob: Vec<u8> = vec![0, 1, 2, 4, 8, 16, 32, 64, 128, 255];
-  let my_inner_tuple = MyInnerTuple { my_text: "my_text".to_string(),
-                                      my_int: 1_000,
-                                      my_timestamp: Timespec { sec: 1,
-                                                               nsec: 999_000_000, }, };
-  let my_outer_tuple = MyOuterTuple { my_uuid: my_uuid,
-                                      my_blob: my_blob,
-                                      my_inner_tuple: my_inner_tuple, };
+  let my_inner_tuple = MyInnerTuple {
+    my_text: "my_text".to_string(),
+    my_int: 1_000,
+    my_timestamp: Timespec {
+      sec: 1,
+      nsec: 999_000_000,
+    },
+  };
+  let my_outer_tuple = MyOuterTuple {
+    my_uuid: my_uuid,
+    my_blob: my_blob,
+    my_inner_tuple: my_inner_tuple,
+  };
   let values = query_values!(0i32, my_outer_tuple.clone());
 
   let cql = "INSERT INTO cdrs_test.test_nested_tuples \
@@ -164,12 +179,13 @@ fn nested_tuples() {
   session.query_with_values(cql, values).expect("insert");
 
   let cql = "SELECT * FROM cdrs_test.test_nested_tuples";
-  let rows = session.query(cql)
-                    .expect("query")
-                    .get_body()
-                    .expect("get body")
-                    .into_rows()
-                    .expect("into rows");
+  let rows = session
+    .query(cql)
+    .expect("query")
+    .get_body()
+    .expect("get body")
+    .into_rows()
+    .expect("into rows");
 
   assert_eq!(rows.len(), 1);
   for row in rows {
diff --git a/tests/user_defined_types.rs b/tests/user_defined_types.rs
index a5097ef..f7fc081 100644
--- a/tests/user_defined_types.rs
+++ b/tests/user_defined_types.rs
@@ -10,13 +10,13 @@ mod common;
 
 use common::*;
 
-use cdrs::query::QueryExecutor;
-use cdrs::types::{AsRust, IntoRustByName};
-use cdrs::types::value::{Bytes, Value};
-use cdrs::types::udt::UDT;
-use cdrs::types::map::Map;
 use cdrs::error::Result;
 use cdrs::frame::IntoBytes;
+use cdrs::query::QueryExecutor;
+use cdrs::types::map::Map;
+use cdrs::types::udt::UDT;
+use cdrs::types::value::{Bytes, Value};
+use cdrs::types::{AsRust, IntoRustByName};
 use time::Timespec;
 
 use std::collections::HashMap;
@@ -27,7 +27,7 @@ fn simple_udt() {
   let create_type_cql = "CREATE TYPE IF NOT EXISTS cdrs_test.simple_udt (my_text text)";
   let create_table_cql = "CREATE TABLE IF NOT EXISTS cdrs_test.test_simple_udt \
                           (my_key int PRIMARY KEY, my_udt simple_udt)";
-  let mut session = setup_multiple(&[create_type_cql, create_table_cql]).expect("setup");
+  let session = setup_multiple(&[create_type_cql, create_table_cql]).expect("setup");
 
   #[derive(Debug, Clone, PartialEq)]
   struct MyUdt {
@@ -50,7 +50,9 @@ fn simple_udt() {
     }
   }
 
-  let my_udt = MyUdt { my_text: "my_text".to_string(), };
+  let my_udt = MyUdt {
+    my_text: "my_text".to_string(),
+  };
   let values = query_values!(0i32, my_udt.clone());
 
   let cql = "INSERT INTO cdrs_test.test_simple_udt \
@@ -58,12 +60,13 @@ fn simple_udt() {
   session.query_with_values(cql, values).expect("insert");
 
   let cql = "SELECT * FROM cdrs_test.test_simple_udt";
-  let rows = session.query(cql)
-                    .expect("query")
-                    .get_body()
-                    .expect("get body")
-                    .into_rows()
-                    .expect("into rows");
+  let rows = session
+    .query(cql)
+    .expect("query")
+    .get_body()
+    .expect("get body")
+    .into_rows()
+    .expect("into rows");
 
   assert_eq!(rows.len(), 1);
   for row in rows {
@@ -81,7 +84,7 @@ fn nested_udt() {
                           (my_inner_udt frozen<nested_inner_udt>)";
   let create_table_cql = "CREATE TABLE IF NOT EXISTS cdrs_test.test_nested_udt \
                           (my_key int PRIMARY KEY, my_outer_udt nested_outer_udt)";
-  let mut session =
+  let session =
     setup_multiple(&[create_type1_cql, create_type2_cql, create_table_cql]).expect("setup");
 
   #[derive(Debug, Clone, PartialEq)]
@@ -114,7 +117,9 @@ fn nested_udt() {
     pub fn try_from(udt: UDT) -> Result<MyOuterUdt> {
       let my_inner_udt: UDT = udt.get_r_by_name("my_inner_udt")?;
       let my_inner_udt = MyInnerUdt::try_from(my_inner_udt).expect("from udt");
-      Ok(MyOuterUdt { my_inner_udt: my_inner_udt, })
+      Ok(MyOuterUdt {
+        my_inner_udt: my_inner_udt,
+      })
     }
   }
 
@@ -127,8 +132,12 @@ fn nested_udt() {
     }
   }
 
-  let my_inner_udt = MyInnerUdt { my_text: "my_text".to_string(), };
-  let my_outer_udt = MyOuterUdt { my_inner_udt: my_inner_udt, };
+  let my_inner_udt = MyInnerUdt {
+    my_text: "my_text".to_string(),
+  };
+  let my_outer_udt = MyOuterUdt {
+    my_inner_udt: my_inner_udt,
+  };
   let values = query_values!(0i32, my_outer_udt.clone());
 
   let cql = "INSERT INTO cdrs_test.test_nested_udt \
@@ -136,12 +145,13 @@ fn nested_udt() {
   session.query_with_values(cql, values).expect("insert");
 
   let cql = "SELECT * FROM cdrs_test.test_nested_udt";
-  let rows = session.query(cql)
-                    .expect("query")
-                    .get_body()
-                    .expect("get body")
-                    .into_rows()
-                    .expect("into rows");
+  let rows = session
+    .query(cql)
+    .expect("query")
+    .get_body()
+    .expect("get body")
+    .into_rows()
+    .expect("into rows");
 
   assert_eq!(rows.len(), 1);
   for row in rows {
@@ -159,7 +169,7 @@ fn alter_udt_add() {
   let create_type_cql = "CREATE TYPE cdrs_test.alter_udt_add_udt (my_text text)";
   let create_table_cql = "CREATE TABLE IF NOT EXISTS cdrs_test.test_alter_udt_add \
                           (my_key int PRIMARY KEY, my_map frozen<map<text, alter_udt_add_udt>>)";
-  let mut session = setup_multiple(&[
+  let session = setup_multiple(&[
     drop_table_cql,
     drop_type_cql,
     create_type_cql,
@@ -190,12 +200,16 @@ fn alter_udt_add() {
     pub fn try_from(udt: UDT) -> Result<MyUdtB> {
       let my_text: String = udt.get_r_by_name("my_text")?;
       let my_timestamp: Option<Timespec> = udt.get_by_name("my_timestamp")?;
-      Ok(MyUdtB { my_text: my_text,
-                  my_timestamp: my_timestamp, })
+      Ok(MyUdtB {
+        my_text: my_text,
+        my_timestamp: my_timestamp,
+      })
     }
   }
 
-  let my_udt_a = MyUdtA { my_text: "my_text".to_string(), };
+  let my_udt_a = MyUdtA {
+    my_text: "my_text".to_string(),
+  };
   let my_map_a = hashmap! { "1" => my_udt_a.clone() };
   let values = query_values!(0i32, my_map_a.clone());
 
@@ -206,16 +220,19 @@ fn alter_udt_add() {
   let cql = "ALTER TYPE cdrs_test.alter_udt_add_udt ADD my_timestamp timestamp";
   session.query(cql).expect("alter type");
 
-  let my_udt_b = MyUdtB { my_text: my_udt_a.my_text,
-                          my_timestamp: None, };
+  let my_udt_b = MyUdtB {
+    my_text: my_udt_a.my_text,
+    my_timestamp: None,
+  };
 
   let cql = "SELECT * FROM cdrs_test.test_alter_udt_add";
-  let rows = session.query(cql)
-                    .expect("query")
-                    .get_body()
-                    .expect("get body")
-                    .into_rows()
-                    .expect("into rows");
+  let rows = session
+    .query(cql)
+    .expect("query")
+    .get_body()
+    .expect("get body")
+    .into_rows()
+    .expect("into rows");
 
   assert_eq!(rows.len(), 1);
   for row in rows {