Repository: AlexPikalov/cdrs (language: rust)

Issue:
Fix SSL session initialisation for cluster
Currently [`Session::new_ssl`](https://github.com/AlexPikalov/cdrs/blob/master/src/cluster/session.rs#L263) takes single `SslConnector` while each node needs to have its own connector. Instead all SSL-encrypted must take a vector of `SslConnectors` one per node address


Discussion / hints:

Fix patch:
diff --git a/README.md b/README.md
index e76e6e7..6089b82 100644
--- a/README.md
+++ b/README.md
@@ -1,272 +1,68 @@
 # CDRS [![crates.io version](https://img.shields.io/crates/v/cdrs.svg)](https://crates.io/crates/cdrs) [![Build Status](https://travis-ci.org/AlexPikalov/cdrs.svg?branch=master)](https://travis-ci.org/AlexPikalov/cdrs) [![Build status](https://ci.appveyor.com/api/projects/status/sirj4flws6o0dvb7/branch/master?svg=true)](https://ci.appveyor.com/project/harrydevnull/cdrs/branch/master)
 
-CDRS is Apache **C**assandra **d**river written in pure **R**u**s**t. The driver implements all the features described in Cassandra binary protocol specification (versions 3 and 4).
+<p align="center">
+  <img src="./cdrs-logo.png" alt="CDRS - Apache Cassandra driver"/>
+</p>
 
-### Describing Cassandra Cluster and starting new Session
+CDRS is Apache **C**assandra **d**river written in pure **R**u**s**t.
 
-In order to start any communication with Cassandra cluster that requires authentication there should be provided a list of Cassandra nodes (IP addresses of machines where Cassandra is installed and included into a cluster). To get more details how to configure multinode cluster refer, for instance, to [DataStax documentation](https://docs.datastax.com/en/cassandra/3.0/cassandra/initialize/initTOC.html).
+## Features
 
-```rust
-use cdrs::cluster::Cluster;
-let cluster = Cluster::new(vec!["youraddress_1:9042", "youraddress_2:9042"], authenticator);
-```
-
-First agrument is a Rust `Vec` of node addresses, the second argument could be any structure that implements `cdrs::authenticators::Authenticator` trait. This allows to use custom authentication strategies, but in this case developers should implement authenticators by themselves. Out of the box CDRS provides two types of authenticators:
-
-* `cdrs::authenticators::NoneAuthenticator` that should be used if authentication is disabled ([Cassandra authenticator](http://cassandra.apache.org/doc/latest/configuration/cassandra_config_file.html#authenticator) is set to `AllowAllAuthenticator`) on server.
-
-* `cdrs::authenticators::PasswordAuthenticator` that should be used if authentication is enabled on the server and [authenticator](http://cassandra.apache.org/doc/latest/configuration/cassandra_config_file.html#authenticator) is `PasswordAuthenticator`:
-
-```rust
-use cdrs::authenticators::PasswordAuthenticator;
-let authenticator = PasswordAuthenticator::new("user", "pass");
-```
+- TCP/SSL connection;
+- Load balancing;
+- Connection pooling;
+- LZ4, Snappy compression;
+- Cassandra-to-Rust data deserialization;
+- Pluggable authentication strategies;
+- [ScyllaDB](https://www.scylladb.com/) support;
+- Server events listening;
+- Multiple CQL version support (3, 4), full spec implementation;
+- Query tracing information.
 
-When cluster nodes are described new `Session` could be established. Each new `Session` has its own load balancing strategy as well as data compression. (CDRS supports both [Snappy](<https://en.wikipedia.org/wiki/Snappy_(compression)>) and [LZ4](<https://en.wikipedia.org/wiki/LZ4_(compression_algorithm)>) compresssions)
+## Documentation and examples
 
-```rust
-let mut no_compression = cluster.connect(RoundRobin::new())
-                                  .expect("No compression connection error");
-  let mut lz4_compression = cluster.connect_lz4(RoundRobin::new())
-                                   .expect("LZ4 compression connection error");
-  let mut snappy_compression = cluster.connect_snappy(RoundRobin::new())
-                                      .expect("Snappy compression connection error");
-```
+- [User guide](./documentation).
+- [Examples](./examples).
+- API docs (release).
 
-where the first argument of each connect methods is a load balancer. Each structure that implements `cdrs::load_balancing::LoadBalancingStrategy` could be used as a load balancer during establishing new `Session`. CDRS provides two strategies out of the box: `cdrs::load_balancing::{RoundRobin, Random}`. Having been set once at the start load balancing strategy cannot be changed during the session.
+## Getting started
 
-Unlike to load balancing compression method could be changed without session restart:
-
-```rust
-use compression::Compression;
-let mut session = cluster.connect(RoundRobin::new())
-                                  .expect("No compression connection error");
-session.compression = Compression::LZ4;
-```
-
-### Starting new SSL-encrypted Session
-
-SSL-encrypted connection is also awailable with CDRS however to get this working CDRS itself should be imported with `ssl` feature enabled:
+Add CDRS to your `Cargo.toml` file as a dependency:
 
 ```toml
-[dependencies]
-openssl = "0.9.6"
-
-[dependencies.cdrs]
-version = "*"
-features = ["ssl"]
-```
-
-Another difference comparing to non-encrypted connection is necessity to create [`SSLConnector`](https://docs.rs/openssl/0.10.2/openssl/ssl/struct.SslConnector.html)
-
-```rust
-use std::path::Path;
-use openssl::ssl::{SslConnectorBuilder, SslMethod};
-use cdrs::client::CDRS;
-use cdrs::authenticators::PasswordAuthenticator;
-use cdrs::transport::TransportTls;
-
-// here needs to be a path to your SSL certificate
-let path = Path::new("./node0.cer.pem");
-let mut ssl_connector_builder = SslConnectorBuilder::new(SslMethod::tls()).unwrap();
-ssl_connector_builder.builder_mut().set_ca_file(path).unwrap();
-let connector = ssl_connector_builder.build();
-```
-
-When these preparation are done we're good to start SSL-encrypted session.
-
-```rust
-let mut no_compression = cluster.connect_ssl(RoundRobin::new(), authenticator, connector)
-                                  .expect("No compression connection error");
-let mut lz4_compression = cluster.connect_lz4_ssl(RoundRobin::new(), authenticator, connector)
-                                  .expect("LZ4 compression connection error");
-let mut snappy_compression = cluster.connect_snappy_ssl(RoundRobin::new(), authenticator, connector)
-                                  .expect("Snappy compression connection error");
-```
-
-More details regarding configuration Cassandra server for SSL-encrypted Client-Node communication could be found, for instance, on [DataStax website](https://docs.datastax.com/en/cassandra/3.0/cassandra/configuration/secureSSLClientToNode.html).
-
-### Executing queries
-
-CDRS `Session` implements `cdrs::query::QueryExecutor` trait that provides few options for immediate query execution:
-
-```rust
-// simple query
-session.query("SELECT * from my.store").unwrap();
-
-// simple query with tracing and warnings
-let with_tracing = true;
-let with_warnings = true;
-session.query_tw("SELECT * FROM my.store", with_tracing, with_warnings).unwrap();
-
-// query with query values
-let values = query_values!(1 as i32, 1 as i64);
-session.query_tw("INSERT INTO my.numbers (my_int, my_bigint) VALUES (?, ?)", values).unwrap();
-
-// query with query values, tracing and warnings
-let with_tracing = true;
-let with_warnings = true;
-let values = query_values!(1 as i32, 1 as i64);
-session.query_tw("INSERT INTO my.numbers (my_int, my_bigint) VALUES (?, ?)", values, with_tracing, with_warnings).unwrap();
-
-// query with query params
-use cdrs::query::QueryParamsBuilder;
-use cdrs::consistency::Consistency;
-
-let mut params = QueryParamsBuilder::new();
-params = params.consistency(Consistency::Any);
-session.query_with_params("SELECT * FROM my.store", params.finalize()).unwrap();
-
-// query with query params and tracing, warnings
-use cdrs::query::QueryParamsBuilder;
-use cdrs::consistency::Consistency;
-
-let with_tracing = true;
-let with_warnings = true;
-
-let mut params = QueryParamsBuilder::new();
-params = params.consistency(Consistency::Any);
-
-session.query_with_params_tw("SELECT * FROM my.store", params.finalize(), with_tracing, with_warnings).unwrap();
-```
-
-### Preparing queries
-
-During preparing a query a server parses the query, saves parsing result into cache and returns back to a client an ID that could be further used for executing prepared statement with different parameters (such as values, consistency etc.). When a server executes prepared query it doesn't need to parse it so parsing step will be skipped.
-
-CDRS `Session` implements `cdrs::query::PrepareExecutor` trait that provides few option for preparing query:
-
-```rust
-let prepared_query = session.prepare("INSERT INTO my.store (my_int, my_bigint) VALUES (?, ?)").unwrap();
-
-// or with tracing and warnings
-let with_tracing = true;
-let with_warnings = true;
-
-let prepred_query = session.prepare_tw("INSERT INTO my.store (my_int, my_bigint) VALUES (?, ?)", with_tracing, with_warnings).unwrap();
+cdrs = { version = "2" }
 ```
 
-### Executing prepared queries
-
-When query is prepared on the server client gets prepared query id of type `cdrs::query::PreparedQuery`. Having such id it's possible to execute prepared query using session methods from `cdrs::query::ExecExecutor`:
+Then add it as an external crate to your `main.rs`:
 
 ```rust
-// execute prepared query without specifying any extra parameters or values
-session.exec(&preparedQuery).unwrap();
-
-// execute prepared query with tracing and warning information
-let with_tracing = true;
-let with_warnings = true;
-
-session.exec_tw(&preparedQuery, with_tracing, with_warnings).unwrap();
-
-// execute prepared query with values
-let values_with_names = query_values!{"my_bigint" => bigint, "my_int" => int};
-
-session.exec_with_values(&preparedQuery, values_with_names).unwrap();
-
-// execute prepared query with values with warnings and tracing information
-let with_tracing = true;
-let with_warnings = true;
-
-let values_with_names = query_values!{"my_bigint" => 1 as i64, "my_int" => 2 as i32};
-
-session.exec_with_values_tw(&preparedQuery, values_with_names, with_tracing, with_warnings).unwrap();
-
-// execute prepared query with parameters
-use cdrs::query::QueryParamsBuilder;
-use cdrs::consistency::Consistency;
-
-let mut params = QueryParamsBuilder::new();
-params = params.consistency(Consistency::Any);
-session.exec_with_parameters(&preparedQuery, params.finalize()).unwrap();
-
-// execute prepared query with parameters, tracing and warning information
-use cdrs::query::QueryParamsBuilder;
-use cdrs::consistency::Consistency;
-
-let with_tracing = true;
-let with_warnings = true;
-let mut params = QueryParamsBuilder::new();
-params = params.consistency(Consistency::Any);
-session.exec_with_parameters_tw(&preparedQuery, params.finalize(), with_tracing, with_warnings).unwrap();
-```
-
-### Batch queries
-
-CDRS `Session` supports batching few queries in a single request to Apache Cassandra via implementing `cdrs::query::BatchExecutor` trait:
-
-```rust
-// batch two queries
-use cdrs::query::{BatchQueryBuilder, QueryBatch};
-
-let mut queries = BatchQueryBuilder::new();
-queries = queries.add_query_prepared(&prepared_query);
-queries = queries.add_query("INSERT INTO my.store (my_int) VALUES (?)", query_values!(1 as i32));
-session.batch_with_params(queries.finalyze());
-
-// batch queries with tracing and warning information
-use cdrs::query::{BatchQueryBuilder, QueryBatch};
-
-let with_tracing = true;
-let with_warnings = true;
-let mut queries = BatchQueryBuilder::new();
-queries = queries.add_query_prepared(&prepared_query);
-queries = queries.add_query("INSERT INTO my.store (my_int) VALUES (?)", query_values!(1 as i32));
-session.batch_with_params_tw(queries.finalyze(), with_tracing, with_warnings);
-```
-
-### Query values types
-
-Accordingly to specification along with queries there could be provided something that is called values. Apache Cassandra server will use values instead of `?` symbols from a query string.
-
-There are two types of queries defined in the spec and supported by CDRS driver. Each of these two types could be easily constructed via provided `query_values!` macros.
-
-* simple values - could be imagine as a list of values. The order of simple values matters because server will put them in the same number as columns were provided in query string.
-
-```rust
-let simple_values = query_values!(1 as i32, 2 as i32);
-```
-
-* named values are similar to hash maps, where keys represent column names which the a value has to be assigned to.
-
-```rust
-let values_with_names = query_values!{"my_bigint" => 1 as i64, "my_int" => 2 as i32};
-```
-
-Each type that implements `Into<cdrs::types::value::Value>` could be used as a value in `query_values!` macros. For primitive types please refer to following [wrapper CDRS types](https://github.com/AlexPikalov/cdrs/blob/master/type-mapping.md) that could be easily converted to `Value`. For custom types (in Cassandra terminology User Defined Types) `IntoCDRSValue` derive could be used:
-
-```rust
-#[derive(Debug, IntoCDRSValue)]
-struct Udt {
-    pub number: i32,
-    pub number_16: i16,
-    pub number_8: N,
-}
-
-// for nested structures it works as well
-#[derive(Debug, IntoCDRSValue)]
-struct N {
-    pub n: i16,
+extern crate cdrs;
+
+use cdrs::authenticators::NoneAuthenticator;
+use cdrs::cluster::session::{new as new_session};
+use cdrs::cluster::{ClusterTcpConfig, NodeTcpConfigBuilder};
+use cdrs::load_balancing::RoundRobin;
+use cdrs::query::*;
+
+fn main() {
+  let node = NodeTcpConfigBuilder::new("127.0.0.1:9042", NoneAuthenticator {}).build();
+  let cluster_config = ClusterTcpConfig(vec![node]);
+  let no_compression =
+    new_session(&cluster_config, RoundRobin::new()).expect("session should be created");
+
+  let create_ks: &'static str = "CREATE KEYSPACE IF NOT EXISTS test_ks WITH REPLICATION = { \
+                                 'class' : 'SimpleStrategy', 'replication_factor' : 1 };";
+  no_compression.query(create_ks).expect("Keyspace create error");
 }
 ```
 
-Look into this [link](https://github.com/AlexPikalov/into-cdrs-value-derive/tree/master/example) to find a full example how to use CDRS + [_into-cdrs-value-derive_](https://github.com/AlexPikalov/into-cdrs-value-derive) crate.
-
-### Mapping results into Rust structures
-
-In ordert to query information from Cassandra DB and transform results to Rust types an structures each row in a query result should be transformed leveraging one of following traits provided by CDRS `cdrs::types::{AsRustType, AsRust, IntoRustByName, ByName, IntoRustByIndex, ByIndex}`.
-
-* `AsRustType` may be used in order to transform such complex structures as Cassandra lists, sets, tuples. The Cassandra value in this case could non-set and null values.
-
-* `AsRust` trait may be used for similar purposes as `AsRustType` but it assumes that Cassandra value is neither non-set nor null value. Otherwise it panics.
-
-* `IntoRustByName` trait may be used to access a value as a Rust structure/type by name. Such as in case of rows where each column has its own name, and maps. These values may be as well non-set and null.
+This example configures a cluster consisting of a single node, and uses round robin load balancing and default `r2d2` values for connection pool.
 
-* `ByName` trait is the same as `IntoRustByName` but value should be neither non-set nor null. Otherwise it panics.
+## License
 
-* `IntoRustByIndex` is the same as `IntoRustByName` but values could be accessed via column index basing on their order provided in query. These values may be as well non-set and null.
+This project is licensed under either of
 
-* `ByIndex` is the same as `IntoRustByIndex` but value can be neither non-set nor null. Otherwise it panics.
+- Apache License, Version 2.0, ([LICENSE-APACHE](LICENSE-APACHE) or [http://www.apache.org/licenses/LICENSE-2.0](http://www.apache.org/licenses/LICENSE-2.0))
+- MIT license ([LICENSE-MIT](LICENSE-MIT) or [http://opensource.org/licenses/MIT](http://opensource.org/licenses/MIT))
 
-Relations between Cassandra and Rust types are described in [type-mapping.md](https://github.com/AlexPikalov/cdrs/blob/master/type-mapping.md). For details see examples.
+at your option.
diff --git a/cdrs-logo.png b/cdrs-logo.png
new file mode 100644
index 0000000..e72f700
Binary files /dev/null and b/cdrs-logo.png differ
diff --git a/documentation/README.md b/documentation/README.md
new file mode 100644
index 0000000..6a39043
--- /dev/null
+++ b/documentation/README.md
@@ -0,0 +1,10 @@
+# Definitive guide
+
+- [Cluster configuration](./cluster-configuration.md).
+- [CDRS session](./cdrs-session.md):
+- [SSL session](./cdrs-session.md)
+- [Making queries](./making-query.md).
+- [Query values](./query-values.md)
+- [Cassandra-to-Rust deserialization](./deserialization.md).
+- [Preparing and executing queries](./preparing-and-executing-queries.md).
+- [Batching multiple queries](./batching-multiple-queries.md).
diff --git a/documentation/batching-multiple-queries.md b/documentation/batching-multiple-queries.md
new file mode 100644
index 0000000..67a264b
--- /dev/null
+++ b/documentation/batching-multiple-queries.md
@@ -0,0 +1,25 @@
+### Batch queries
+
+_TODO: rework this section_
+
+CDRS `Session` supports batching few queries in a single request to Apache Cassandra via implementing `cdrs::query::BatchExecutor` trait:
+
+```rust
+// batch two queries
+use cdrs::query::{BatchQueryBuilder, QueryBatch};
+
+let mut queries = BatchQueryBuilder::new();
+queries = queries.add_query_prepared(&prepared_query);
+queries = queries.add_query("INSERT INTO my.store (my_int) VALUES (?)", query_values!(1 as i32));
+session.batch_with_params(queries.finalyze());
+
+// batch queries with tracing and warning information
+use cdrs::query::{BatchQueryBuilder, QueryBatch};
+
+let with_tracing = true;
+let with_warnings = true;
+let mut queries = BatchQueryBuilder::new();
+queries = queries.add_query_prepared(&prepared_query);
+queries = queries.add_query("INSERT INTO my.store (my_int) VALUES (?)", query_values!(1 as i32));
+session.batch_with_params_tw(queries.finalyze(), with_tracing, with_warnings);
+```
diff --git a/documentation/cdrs-session.md b/documentation/cdrs-session.md
new file mode 100644
index 0000000..9d5e669
--- /dev/null
+++ b/documentation/cdrs-session.md
@@ -0,0 +1,77 @@
+# CDRS Session
+
+`Session` is a structure that holds as set pools of connections authorised by a Cluster. As well it provides data decompressing and load balancing mechanisms used for Cassandra frame exchange, or querying in other words.
+
+```rust
+use cdrs::load_balancing::RoundRobin;
+use cdrs::cluster::session::{new as new_session};
+
+
+let load_balancing = RoundRobin::new();
+let session = new_session(&cluster_config, RoundRobin::new())
+  .expect("session should be created");
+```
+
+Here, in order to create new session a [cluster config](./cluster-configuration.md) and a load balancing strategy must be provided. Load balancing strategy is used when some query should be performed by driver. At that moment load balancer returns a pool of connections for a node that was picked up in accordance to a strategy. After that CDRS gets from r2d2 pool one of available connections, and then this connection will be used for frames exchange. Such logic guarantees that nodes' loads are balanced and there is no need to establish new connection if there is a one that is released after previous query.
+
+This is how the architecture looks like:
+
+<p align="center">
+  <img src="./schemes/cdrs-load-balancing.png" alt="CDRS load balancing architecture"/>
+</p>
+
+## Load balancing
+
+Any structure that implements `LoadBalancingStrategy` trait can be used in `Session` as a load balancer.
+
+CDRS provides few strategies out of the box so no additional development may not be needed:
+
+- `cdrs::load_balancing::Random` randomly picks up a node from a cluster.
+
+- `cdrs::load_balancing::RoundRobinSync` thread safe round robin balancing strategy.
+
+- `cdrs::load_balancing::RoundRobin` light weight round robin strategy that is not thread safe though. So it should be used in mono thread apps only.
+
+Along with that any custom load balancing strategy may be implemented and used with CDRS. The only requirement is the structure must implement `LoadBalancingStrategy` trait.
+
+## Data compression
+
+CQL binary protocol allows using LZ4 and Snappy data compression in order to reduce trafic between Node and Client.
+
+CDRS provides methods for creating `Session` with different compression contexts:
+
+- `cdrs::session::new(&cluster_config, load_balancer)` creates new `Session` that will be exchanging non-compressed frames with a Cluster.
+
+- `cdrs::session::new_snappy(&cluster_config, load_balancer)` creates new `Session` that will be exchanging Snappy-compressed frames with a Cluster.
+
+- `cdrs::session::new_lz4(&cluster_config, load_balancer)` creates new `Session` that will be exchanging LZ4-compressed frames with a Cluster.
+
+Once `Session` is successfully created it can be used for communication with Cluster.
+
+## Making queries
+
+By default `Session` structure doesn't provide an API for making queries. Query functionality becomes enabled after importing one or few of following traits:
+
+```rust
+use cdrs::query::QueryExecutor;
+```
+
+```rust
+use cdrs::query::PrepareExecutor;
+```
+
+```rust
+use cdrs::query::ExecExecutor;
+```
+
+```rust
+use cdrs::query::BatchExecutor;
+```
+
+Detailed Query API and those traits overview please find in [making query](./making-query.md) section.
+
+### Reference
+
+1. LZ4 compression algorythm https://en.wikipedia.org/wiki/LZ4_(compression_algorithm).
+
+2. Snappy compression alhorythm https://en.wikipedia.org/wiki/Snappy_(compression).
diff --git a/documentation/cluster-configuration.md b/documentation/cluster-configuration.md
new file mode 100644
index 0000000..c47f94a
--- /dev/null
+++ b/documentation/cluster-configuration.md
@@ -0,0 +1,55 @@
+### Cluster configuration
+
+Apache Cassandra is designed to be a scalable and higly available database. So most often developers work with multi node Cassandra clusters. For instance Apple's setup includes 75k nodes, Netflix 2.5k nodes, Ebay >100 nodes.
+
+That's why CDRS driver was designed with multinode support in mind. In order to connect to Cassandra cluster via CDRS connection configuration should be provided:
+
+```rust
+use cdrs::authenticators::NoneAuthenticator;
+use cdrs::cluster::{ClusterTcpConfig, NodeTcpConfigBuilder, TcpConnectionPool};
+
+fn main() {
+  let node = NodeTcpConfigBuilder::new("127.0.0.1:9042", NoneAuthenticator {}).build();
+  let cluster_config = ClusterTcpConfig(vec![node]);
+}
+```
+
+`ClusterTcpConfig` receives a vector of Cassandra nodes configurations. `NodeTcpConfigBuilder` is a builder that provides methods for configuring r2d2 pool of connections to a given node:
+
+```rust
+let node_address = "127.0.0.1:9042";
+let authenticator = NoneAuthenticator {};
+let node = NodeTcpConfigBuilder::new(node_address, authenticator)
+  .max_size(5)
+  .min_idle(4)
+  .max_lifetime(Some(Duration::from_secs(60)))
+  .idle_timeout(Duration::from_secs(60))
+  .build();
+let no_compression: CurrentSession =
+  new_session(&cluster_config, RoundRobin::new()).expect("session should be created");
+```
+
+All existing `NodeTcpConfigBuilder` methods have the same behaviour as ones from `r2d2::Builder`, so for more details please refer to [r2d2](https://docs.rs/r2d2/0.8.2/r2d2/struct.Builder.html) official documentation.
+
+For each node configuration, `Authenticator` should be provided. `Authenticator` is a trat a structure should implement so it can be used by CDRS session. Out of the box CDRS provides two types of authenticators:
+
+- `cdrs::authenticators::NoneAuthenticator` that should be used if authentication is disabled by a node ([Cassandra authenticator](http://cassandra.apache.org/doc/latest/configuration/cassandra_config_file.html#authenticator) is set to `AllowAllAuthenticator`) on server.
+
+- `cdrs::authenticators::PasswordAuthenticator` that should be used if authentication is enabled on the server and [authenticator](http://cassandra.apache.org/doc/latest/configuration/cassandra_config_file.html#authenticator) is `PasswordAuthenticator`.
+
+```rust
+use cdrs::authenticators::PasswordAuthenticator;
+let authenticator = PasswordAuthenticator::new("user", "pass");
+```
+
+If a node has a custom authentication strategy, corresponded `Authenticator` should be implemented by a developer and further used in `NodeTcpConfigBuilder`.
+
+To figure out how a custom `Authenticator` should be implemented refer to [./src/authenticators.rs](./src/authenticators.rs).
+
+### Reference
+
+1. Cassandra cluster configuration https://docs.datastax.com/en/cassandra/3.0/cassandra/initialize/initTOC.html.
+
+2. ScyllaDB cluster configuration https://docs.scylladb.com/operating-scylla/ (see Cluster Management section).
+
+3. `r2d2::Builder` documentation https://docs.rs/r2d2/0.8.2/r2d2/struct.Builder.html.
diff --git a/documentation/deserialization.md b/documentation/deserialization.md
new file mode 100644
index 0000000..1dd251b
--- /dev/null
+++ b/documentation/deserialization.md
@@ -0,0 +1,19 @@
+### Mapping results into Rust structures
+
+_TODO: provide more details and link to examples_
+
+In ordert to query information from Cassandra DB and transform results to Rust types an structures each row in a query result should be transformed leveraging one of following traits provided by CDRS `cdrs::types::{AsRustType, AsRust, IntoRustByName, ByName, IntoRustByIndex, ByIndex}`.
+
+- `AsRustType` may be used in order to transform such complex structures as Cassandra lists, sets, tuples. The Cassandra value in this case could non-set and null values.
+
+- `AsRust` trait may be used for similar purposes as `AsRustType` but it assumes that Cassandra value is neither non-set nor null value. Otherwise it panics.
+
+- `IntoRustByName` trait may be used to access a value as a Rust structure/type by name. Such as in case of rows where each column has its own name, and maps. These values may be as well non-set and null.
+
+- `ByName` trait is the same as `IntoRustByName` but value should be neither non-set nor null. Otherwise it panics.
+
+- `IntoRustByIndex` is the same as `IntoRustByName` but values could be accessed via column index basing on their order provided in query. These values may be as well non-set and null.
+
+- `ByIndex` is the same as `IntoRustByIndex` but value can be neither non-set nor null. Otherwise it panics.
+
+Relations between Cassandra and Rust types are described in [type-mapping.md](https://github.com/AlexPikalov/cdrs/blob/master/type-mapping.md). For details see examples.
diff --git a/documentation/making-query.md b/documentation/making-query.md
new file mode 100644
index 0000000..23d3945
--- /dev/null
+++ b/documentation/making-query.md
@@ -0,0 +1,100 @@
+# Making queries
+
+By default `Session` structure doesn't provide an API for making queries. Query functionality bacomes enabled after importing one or few of following traits:
+
+```rust
+use cdrs::query::QueryExecutor;
+```
+
+This trait provides an API for making plain queries and immediately receiving responses.
+
+```rust
+use cdrs::query::PrepareExecutor;
+```
+
+This trait enables query preparation on the server. After query preparation it's enough to exectute query via `cdrs::query::ExecExecutor` API prviding a query ID returned by `cdrs::query::PrepareExecutor`
+
+```rust
+use cdrs::query::ExecExecutor;
+```
+
+This trait provides an API for query exectution. `PrepareExecutor` and `ExecExecutor` APIs are considered in [next sections](./preparing-and-executing-queries.md).
+
+```rust
+use cdrs::query::BatchExecutor;
+```
+
+`BatchExecutor` provides functionality for executing multiple queries in a single request to Cluster. For more details refer to [Batch Queries](./batching-multiple-queries.md) section.
+
+### `CDRSSession` trait
+
+Each of traits enumered beyond provides just a piece of full query API. They can be used independently one from another though. However if the whole query functionality is needed in a programm `use cdrs::cluster::CDRSSession` should be considered instead.
+
+`CDRSSession` source code looks following
+
+```rust
+pub trait CDRSSession<
+  'a,
+  T: CDRSTransport + 'static,
+  M: r2d2::ManageConnection<Connection = cell::RefCell<T>, Error = error::Error>,
+>:
+  GetCompressor<'static>
+  + GetConnection<T, M>
+  + QueryExecutor<T, M>
+  + PrepareExecutor<T, M>
+  + ExecExecutor<T, M>
+  + BatchExecutor<T, M>
+{
+}
+```
+
+It includes all the functionality related to making queries.
+
+### `QueryExecutor` API
+
+`QueryExecutor` trait provides various methods for immediate query execution.
+
+```rust
+session.query("INSERT INTO my.numbers (my_int, my_bigint) VALUES (1, 2)").unwrap();
+```
+
+`query` method receives a single argument which is a CQL query string. It returns `cdrs::error::Result` that in case of `SELECT` query can be mapped on corresponded Rust structure. See [CRUD example](../examples/crud_operations.rs) for details.
+
+The same query could be made leveraging something that is called Values. It allows to have generic query strings independent from actuall values.
+
+```rust
+#[macro_use]
+extern crate cdrs;
+//...
+
+const insert_numbers_query: &'static str = "INSERT INTO my.numbers (my_int, my_bigint) VALUES (?, ?)";
+let values = query_values!(1 as i32, 1 as i64);
+
+session.query_with_values(insert_numbers_query, values).unwrap();
+```
+
+Here we've provided a generic query string for inserting numbers into `my.numbers` table. This query string doesn't have actual values hardcoded so exactly the same query can be used for multiple insert operations. Such sort of query strings can be used for Prepare-and-Execute operations when a query string is sent just once during Prepare step and then Execution operation is performed each time new values should be inserted. For more detailes see [Preparing and Executing](./preparing-and-executing-queries.md) section.
+
+However the full controll over the query can be achieved via `cdrs::query::QueryParamsBuilder`:
+
+```rust
+use cdrs::query::QueryParamsBuilder;
+use cdrs::consistency::Consistency;
+
+let query_params = QueryParamsBuilder::new()
+  .consistency(Consistency::Any)
+  .finalize();
+session.query_with_params("SELECT * FROM my.store", query_params).unwrap();
+```
+
+`QueryParamsBuilder` allows to precise all possible parameters of a query: consistency, values, paging properties and others. To get all parameters please refer to CDRS API [docs](https://docs.rs/cdrs/2.0.0-beta.1/cdrs/query/struct.QueryParamsBuilder.html).
+
+Usually developers don't need to use `query_with_params` as almost all functionality is provided by such ergonomic methods as `query_with_values`, `pager` etc.
+
+### Reference
+
+1. `QueryParamsBuilder` API docs https://docs.rs/cdrs/2.0.0-beta.1/cdrs/query/struct.QueryParamsBuilder.html.
+
+2. The Cassandra Query Language (CQL) http://cassandra.apache.org/doc/4.0/cql/.
+
+3. [CDRS: Preparing and executing queries](./preparing-and-executing-queries.md)
diff --git a/documentation/preparing-and-executing-queries.md b/documentation/preparing-and-executing-queries.md
new file mode 100644
index 0000000..f5e4451
--- /dev/null
+++ b/documentation/preparing-and-executing-queries.md
@@ -0,0 +1,63 @@
+### Preparing queries
+
+_TODO: provide more details_
+
+During preparing a query a server parses the query, saves parsing result into cache and returns back to a client an ID that could be further used for executing prepared statement with different parameters (such as values, consistency etc.). When a server executes prepared query it doesn't need to parse it so parsing step will be skipped.
+
+CDRS `Session` implements `cdrs::query::PrepareExecutor` trait that provides few option for preparing query:
+
+```rust
+let prepared_query = session.prepare("INSERT INTO my.store (my_int, my_bigint) VALUES (?, ?)").unwrap();
+
+// or with tracing and warnings
+let with_tracing = true;
+let with_warnings = true;
+
+let prepred_query = session.prepare_tw("INSERT INTO my.store (my_int, my_bigint) VALUES (?, ?)", with_tracing, with_warnings).unwrap();
+```
+
+### Executing prepared queries
+
+When query is prepared on the server client gets prepared query id of type `cdrs::query::PreparedQuery`. Having such id it's possible to execute prepared query using session methods from `cdrs::query::ExecExecutor`:
+
+```rust
+// execute prepared query without specifying any extra parameters or values
+session.exec(&preparedQuery).unwrap();
+
+// execute prepared query with tracing and warning information
+let with_tracing = true;
+let with_warnings = true;
+
+session.exec_tw(&preparedQuery, with_tracing, with_warnings).unwrap();
+
+// execute prepared query with values
+let values_with_names = query_values!{"my_bigint" => bigint, "my_int" => int};
+
+session.exec_with_values(&preparedQuery, values_with_names).unwrap();
+
+// execute prepared query with values with warnings and tracing information
+let with_tracing = true;
+let with_warnings = true;
+
+let values_with_names = query_values!{"my_bigint" => 1 as i64, "my_int" => 2 as i32};
+
+session.exec_with_values_tw(&preparedQuery, values_with_names, with_tracing, with_warnings).unwrap();
+
+// execute prepared query with parameters
+use cdrs::query::QueryParamsBuilder;
+use cdrs::consistency::Consistency;
+
+let mut params = QueryParamsBuilder::new();
+params = params.consistency(Consistency::Any);
+session.exec_with_parameters(&preparedQuery, params.finalize()).unwrap();
+
+// execute prepared query with parameters, tracing and warning information
+use cdrs::query::QueryParamsBuilder;
+use cdrs::consistency::Consistency;
+
+let with_tracing = true;
+let with_warnings = true;
+let mut params = QueryParamsBuilder::new();
+params = params.consistency(Consistency::Any);
+session.exec_with_parameters_tw(&preparedQuery, params.finalize(), with_tracing, with_warnings).unwrap();
+```
diff --git a/documentation/query-values.md b/documentation/query-values.md
new file mode 100644
index 0000000..204b23a
--- /dev/null
+++ b/documentation/query-values.md
@@ -0,0 +1,61 @@
+# Query `Value`
+
+Query `Value`-s can be used along with query string templates. Query string templates is a special sort of query string that contains `?` sign. `?` will be subsituted by CDRS driver with query `Values`.
+
+For instance:
+
+```rust
+const insert_numbers_query: &'static str = "INSERT INTO my.numbers (my_int, my_bigint) VALUES (?, ?)";
+let values = query_values!(1 as i32, 1 as i64);
+
+session.query_with_values(insert_numbers_query, values).unwrap();
+```
+
+`insert_numbers_query` is a typical query template. `session::query_with_values` method provides an API for using such query strings along with query values.
+
+There is full list of `CDRSSession` methods that allow using values and query templates:
+
+- `exec_with_values` - executes previously preared query with provided values (see [example](../examples/prepare_batch_execute.rs) and/or [Preparing and Executing](./preparing-and-executing-queries.md) section);
+
+- `query_with_params_tw` - immediately executes a query using provided values (see [example](../examples/crud_operations.rs))
+
+## Simple `Value` and `Value` with names
+
+There are two type of query values supported by CDRS:
+
+- simple `Value`-s may be imagined as a tuple of actual values. This values will be inserted instead of a `?` that has the same index number as a `Value` within a tuple. To easy create `Value`-s CDRS provides `query_values!` macro:
+
+```rust
+#[macro_use]
+extern crate cdrs;
+let values = query_values!(1 as i32, 1 as i64);
+```
+
+- `Value`-s with names may be imagined as a `Map` that links a table column name with a value that should be inserted in a column. It means that `Value`-s with maps should not necesarily have the same order as a corresponded `?` in a query template:
+
+```rust
+#[macro_use]
+extern crate cdrs;
+//...
+
+const insert_numbers_query: &'static str = "INSERT INTO my.numbers (my_int, my_bigint) VALUES (?, ?)";
+let values = query_values!(my_bigint => 1 as i64, my_int => 1 as i64);
+
+session.query_with_values(insert_numbers_query, values).unwrap();
+```
+
+What kind of values can be used as `query_values!` arguments? All types that have implementations of [`Into<Bytes>` trait](https://docs.rs/cdrs/2.0.0-beta.1/cdrs/types/value/struct.Bytes.html).
+
+For Rust structs represented by [Cassandra User Defined types](http://cassandra.apache.org/doc/4.0/cql/types.html#grammar-token-user_defined_type) `#[derive(IntoCDRSValue)]` can be used for recurcive implementation. See [CRUD example](../examples/crud_operations.rs).
+
+### Reference
+
+1. Cassandra official docs - User Defined Types http://cassandra.apache.org/doc/4.0/cql/types.html#grammar-token-user_defined_type.
+
+2. Datastax - User Defined Types https://docs.datastax.com/en/cql/3.3/cql/cql_using/useCreateUDT.html.
+
+3. ScyllaDB - User Defined Types https://docs.scylladb.com/getting-started/types/
+
+4. [CDRS CRUD Example](../examples/crud_operations.rs)
+
+5. https://github.com/AlexPikalov/cdrs-helpers-derive
diff --git a/documentation/schemes/cdrs-load-balancing.png b/documentation/schemes/cdrs-load-balancing.png
new file mode 100644
index 0000000..1ceb020
Binary files /dev/null and b/documentation/schemes/cdrs-load-balancing.png differ
diff --git a/examples/README.md b/examples/README.md
new file mode 100644
index 0000000..3fcb8e7
--- /dev/null
+++ b/examples/README.md
@@ -0,0 +1,7 @@
+# CDRS examples
+
+- [`crud_operations.rs`](./crud_operations.rs) demonstrates how to create keyspace, table and user defined type. As well basic CRUD (Create, Read, Update, Delete) operations are shown;
+- [`multiple_thread.rs`](./multiple_thread.rs) shows how to use CDRS in multi thread applications;
+- [`paged_query.rs`](./multiple_thread.rs) uncovers query paging;
+- [`prepare_batch_execute.rs`](./prepare_batch_execute.rs) provides an example of query preparation and batching;
+- [`server_events.rs`](./server_events.rs) illustrates a process of server events (create table, schema change etc.) listening.
diff --git a/examples/crud_operations.rs b/examples/crud_operations.rs
index 5f27229..3b16280 100644
--- a/examples/crud_operations.rs
+++ b/examples/crud_operations.rs
@@ -5,26 +5,25 @@ 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::cluster::session::{new as new_session, Session};
+use cdrs::cluster::{ClusterTcpConfig, NodeTcpConfigBuilder, TcpConnectionPool};
 use cdrs::load_balancing::RoundRobin;
 use cdrs::query::*;
-use cdrs::transport::TransportTcp;
 
 use cdrs::frame::IntoBytes;
 use cdrs::types::from_cdrs::FromCDRSByName;
 use cdrs::types::prelude::*;
 
-type CurrentSession = Session<RoundRobin<RefCell<TransportTcp>>, NoneAuthenticator>;
+type CurrentSession = Session<RoundRobin<TcpConnectionPool<NoneAuthenticator>>>;
 
 fn main() {
-  let cluster = Cluster::new(vec!["127.0.0.1:9042"], NoneAuthenticator {});
-  let no_compression = cluster
-    .connect(RoundRobin::new())
-    .expect("No compression connection error");
+  let node = NodeTcpConfigBuilder::new("127.0.0.1:9042", NoneAuthenticator {}).build();
+  let cluster_config = ClusterTcpConfig(vec![node]);
+  let no_compression: CurrentSession =
+    new_session(&cluster_config, RoundRobin::new()).expect("session should be created");
 
   create_keyspace(&no_compression);
   create_udt(&no_compression);
diff --git a/examples/multiple_thread.rs b/examples/multiple_thread.rs
new file mode 100644
index 0000000..ba9d703
--- /dev/null
+++ b/examples/multiple_thread.rs
@@ -0,0 +1,95 @@
+#[macro_use]
+extern crate cdrs;
+#[macro_use]
+extern crate cdrs_helpers_derive;
+
+use std::sync::Arc;
+use std::thread;
+
+use cdrs::authenticators::NoneAuthenticator;
+use cdrs::cluster::session::{new as new_session, Session};
+use cdrs::cluster::{ClusterTcpConfig, NodeTcpConfigBuilder, TcpConnectionPool};
+use cdrs::load_balancing::RoundRobinSync;
+use cdrs::query::*;
+
+use cdrs::frame::IntoBytes;
+use cdrs::types::from_cdrs::FromCDRSByName;
+use cdrs::types::prelude::*;
+
+type CurrentSession = Session<RoundRobinSync<TcpConnectionPool<NoneAuthenticator>>>;
+
+fn main() {
+  let node = NodeTcpConfigBuilder::new("127.0.0.1:9042", NoneAuthenticator {}).build();
+  let cluster_config = ClusterTcpConfig(vec![node]);
+  let lb = RoundRobinSync::new();
+  let no_compression: Arc<CurrentSession> =
+    Arc::new(new_session(&cluster_config, lb).expect("session should be created"));
+
+  create_keyspace(&no_compression.clone());
+  create_table(&no_compression.clone());
+
+  for i in 0..20 {
+    let thread_session = no_compression.clone();
+    thread::spawn(move || {
+      insert_struct(&thread_session, i);
+    }).join()
+      .expect("thread error");
+  }
+
+  select_struct(&no_compression);
+}
+
+#[derive(Clone, Debug, IntoCDRSValue, TryFromRow, PartialEq)]
+struct RowStruct {
+  key: i32,
+}
+
+impl RowStruct {
+  fn into_query_values(self) -> QueryValues {
+    query_values!("key" => self.key)
+  }
+}
+
+#[derive(Debug, Clone, PartialEq, IntoCDRSValue, TryFromUDT)]
+struct User {
+  username: String,
+}
+
+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: &CurrentSession) {
+  let create_table_cql =
+    "CREATE TABLE IF NOT EXISTS test_ks.multi_thread_table (key int PRIMARY KEY);";
+  session
+    .query(create_table_cql)
+    .expect("Table creation error");
+}
+
+fn insert_struct(session: &CurrentSession, key: i32) {
+  let row = RowStruct { key };
+
+  let insert_struct_cql = "INSERT INTO test_ks.multi_thread_table (key) VALUES (?)";
+  session
+    .query_with_values(insert_struct_cql, row.into_query_values())
+    .expect("insert");
+}
+
+fn select_struct(session: &CurrentSession) {
+  let select_struct_cql = "SELECT * FROM test_ks.multi_thread_table";
+  let rows = session
+    .query(select_struct_cql)
+    .expect("query")
+    .get_body()
+    .expect("get body")
+    .into_rows()
+    .expect("into rows");
+
+  for row in rows {
+    let my_row: RowStruct = RowStruct::try_from_row(row).expect("into RowStruct");
+    println!("struct got: {:?}", my_row);
+  }
+}
diff --git a/examples/paged_query.rs b/examples/paged_query.rs
index 3e909ee..61464b3 100644
--- a/examples/paged_query.rs
+++ b/examples/paged_query.rs
@@ -3,19 +3,17 @@ 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::cluster::session::{new as new_session, Session};
+use cdrs::cluster::{ClusterTcpConfig, NodeTcpConfigBuilder, TcpConnectionPool};
 use cdrs::load_balancing::RoundRobin;
 use cdrs::query::*;
-use cdrs::transport::TransportTcp;
 
 use cdrs::frame::IntoBytes;
 use cdrs::types::from_cdrs::FromCDRSByName;
 use cdrs::types::prelude::*;
 
-type CurrentSession = Session<RoundRobin<RefCell<TransportTcp>>, NoneAuthenticator>;
+type CurrentSession = Session<RoundRobin<TcpConnectionPool<NoneAuthenticator>>>;
 
 #[derive(Clone, Debug, IntoCDRSValue, TryFromRow, PartialEq)]
 struct RowStruct {
@@ -29,10 +27,10 @@ impl RowStruct {
 }
 
 fn main() {
-  let cluster = Cluster::new(vec!["127.0.0.1:9042"], NoneAuthenticator {});
-  let no_compression = cluster
-    .connect(RoundRobin::new())
-    .expect("No compression connection error");
+  let node = NodeTcpConfigBuilder::new("127.0.0.1:9042", NoneAuthenticator {}).build();
+  let cluster_config = ClusterTcpConfig(vec![node]);
+  let lb = RoundRobin::new();
+  let no_compression = new_session(&cluster_config, lb).expect("session should be created");
 
   create_keyspace(&no_compression);
   create_table(&no_compression);
diff --git a/examples/prepare_batch_execute.rs b/examples/prepare_batch_execute.rs
index 79fdda0..a6261e2 100644
--- a/examples/prepare_batch_execute.rs
+++ b/examples/prepare_batch_execute.rs
@@ -2,22 +2,18 @@
 extern crate cdrs;
 #[macro_use]
 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::cluster::session::{new as new_session, Session};
+use cdrs::cluster::{ClusterTcpConfig, NodeTcpConfigBuilder, TcpConnectionPool};
 use cdrs::load_balancing::RoundRobin;
 use cdrs::query::*;
-use cdrs::transport::TransportTcp;
 
 use cdrs::frame::IntoBytes;
 use cdrs::types::from_cdrs::FromCDRSByName;
 use cdrs::types::prelude::*;
 
-type CurrentSession = Session<RoundRobin<RefCell<TransportTcp>>, NoneAuthenticator>;
+type CurrentSession = Session<RoundRobin<TcpConnectionPool<NoneAuthenticator>>>;
 
 #[derive(Clone, Debug, IntoCDRSValue, TryFromRow, PartialEq)]
 struct RowStruct {
@@ -33,10 +29,10 @@ impl RowStruct {
 }
 
 fn main() {
-  let cluster = Cluster::new(vec!["127.0.0.1:9042"], NoneAuthenticator {});
-  let no_compression = cluster
-    .connect(RoundRobin::new())
-    .expect("No compression connection error");
+  let node = NodeTcpConfigBuilder::new("127.0.0.1:9042", NoneAuthenticator {}).build();
+  let cluster_config = ClusterTcpConfig(vec![node]);
+  let lb = RoundRobin::new();
+  let no_compression = new_session(&cluster_config, lb).expect("session should be created");
 
   create_keyspace(&no_compression);
   create_table(&no_compression);
diff --git a/examples/server_events.rs b/examples/server_events.rs
index 5a5d854..0d2c11a 100644
--- a/examples/server_events.rs
+++ b/examples/server_events.rs
@@ -1,28 +1,29 @@
 extern crate cdrs;
 
-use std::cell::RefCell;
 use std::iter::Iterator;
 use std::thread;
 
 use cdrs::authenticators::NoneAuthenticator;
-use cdrs::cluster::{Cluster, Session};
+use cdrs::cluster::session::new as new_session;
+use cdrs::cluster::{ClusterTcpConfig, NodeTcpConfigBuilder};
 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<RefCell<TransportTcp>>, NoneAuthenticator>;
-
 fn main() {
-  let cluster = Cluster::new(vec!["127.0.0.1:9042"], NoneAuthenticator {});
-  let no_compression: CurrentSession = cluster
-    .connect(RoundRobin::new())
-    .expect("No compression connection error");
+  let node = NodeTcpConfigBuilder::new("127.0.0.1:9042", NoneAuthenticator {}).build();
+  let cluster_config = ClusterTcpConfig(vec![node]);
+  let lb = RoundRobin::new();
+  let no_compression = new_session(&cluster_config, lb).expect("session should be created");
 
   let (listener, stream) = no_compression
-    .listen("127.0.0.1:9042", vec![SimpleServerEvent::SchemaChange])
+    .listen(
+      "127.0.0.1:9042",
+      NoneAuthenticator {},
+      vec![SimpleServerEvent::SchemaChange],
+    )
     .expect("listen error");
 
   thread::spawn(move || listener.start(&Compression::None).unwrap());
diff --git a/src/authenticators.rs b/src/authenticators.rs
index 019e4ff..75a6cbc 100644
--- a/src/authenticators.rs
+++ b/src/authenticators.rs
@@ -1,6 +1,6 @@
 use types::CBytes;
 
-pub trait Authenticator: Clone {
+pub trait Authenticator: Clone + Send + Sync {
     fn get_auth_token(&self) -> CBytes;
     fn get_cassandra_name(&self) -> Option<&str>;
 }
@@ -13,8 +13,10 @@ pub struct PasswordAuthenticator<'a> {
 
 impl<'a> PasswordAuthenticator<'a> {
     pub fn new<'b>(username: &'b str, password: &'b str) -> PasswordAuthenticator<'b> {
-        PasswordAuthenticator { username: username,
-                                password: password, }
+        PasswordAuthenticator {
+            username: username,
+            password: password,
+        }
     }
 }
 
@@ -64,8 +66,10 @@ mod tests {
     #[test]
     fn test_password_authenticator_get_cassandra_name() {
         let auth = PasswordAuthenticator::new("foo", "bar");
-        assert_eq!(auth.get_cassandra_name(),
-                   Some("org.apache.cassandra.auth.PasswordAuthenticator"));
+        assert_eq!(
+            auth.get_cassandra_name(),
+            Some("org.apache.cassandra.auth.PasswordAuthenticator")
+        );
     }
 
     #[test]
diff --git a/src/cluster/cluster.rs b/src/cluster/cluster.rs
deleted file mode 100644
index 1b6c681..0000000
--- a/src/cluster/cluster.rs
+++ /dev/null
@@ -1,87 +0,0 @@
-use std::cell::RefCell;
-
-use authenticators::Authenticator;
-use cluster::session::Session;
-use error;
-use load_balancing::LoadBalancingStrategy;
-use transport::TransportTcp;
-
-#[cfg(feature = "ssl")]
-use openssl::ssl::SslConnector;
-#[cfg(feature = "ssl")]
-use transport::TransportTls;
-
-pub struct Cluster<A> {
-  nodes_addrs: Vec<&'static str>,
-  authenticator: A,
-}
-
-impl<'a, A: Authenticator + Sized> Cluster<A> {
-  pub fn new(nodes_addrs: Vec<&'static str>, authenticator: A) -> Cluster<A> {
-    Cluster {
-      nodes_addrs,
-      authenticator,
-    }
-  }
-
-  pub fn connect<LB>(&self, lb: LB) -> error::Result<Session<LB, A>>
-  where
-    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<RefCell<TransportTcp>> + Sized,
-    A: Authenticator + 'a + Sized,
-  {
-    Session::new_snappy(&self.nodes_addrs, lb, self.authenticator.clone())
-  }
-
-  pub fn connect_lz4<LB>(&self, lb: LB) -> error::Result<Session<LB, A>>
-  where
-    LB: LoadBalancingStrategy<RefCell<TransportTcp>> + Sized,
-  {
-    Session::new_lz4(&self.nodes_addrs, lb, self.authenticator.clone())
-  }
-
-  #[cfg(feature = "ssl")]
-  pub fn connect_ssl<LB>(
-    &self,
-    lb: LB,
-    authenticator: A,
-    ssl_connector: &SslConnector,
-  ) -> error::Result<Session<LB, A>>
-  where
-    LB: LoadBalancingStrategy<RefCell<TransportTls>> + Sized,
-  {
-    Session::new_ssl(&self.nodes_addrs, lb, authenticator, ssl_connector)
-  }
-
-  #[cfg(feature = "ssl")]
-  pub fn connect_snappy_ssl<LB>(
-    &self,
-    lb: LB,
-    authenticator: A,
-    ssl_connector: &SslConnector,
-  ) -> error::Result<Session<LB, A>>
-  where
-    LB: LoadBalancingStrategy<RefCell<TransportTls>> + Sized,
-  {
-    Session::new_snappy_ssl(&self.nodes_addrs, lb, authenticator, ssl_connector)
-  }
-
-  #[cfg(feature = "ssl")]
-  pub fn connect_lz4_ssl<LB>(
-    &self,
-    lb: LB,
-    authenticator: A,
-    ssl_connector: &SslConnector,
-  ) -> error::Result<Session<LB, A>>
-  where
-    LB: LoadBalancingStrategy<RefCell<TransportTls>> + Sized,
-  {
-    Session::new_lz4_ssl(&self.nodes_addrs, lb, authenticator, ssl_connector)
-  }
-}
diff --git a/src/cluster/config_ssl.rs b/src/cluster/config_ssl.rs
new file mode 100644
index 0000000..eee2401
--- /dev/null
+++ b/src/cluster/config_ssl.rs
@@ -0,0 +1,123 @@
+#[cfg(feature = "ssl")]
+use openssl::ssl::SslConnector;
+use std::time::Duration;
+
+use authenticators::Authenticator;
+
+/// Cluster configuration that holds per node SSL configs
+pub struct ClusterSslConfig<'a, A: Authenticator + Sized>(pub Vec<NodeSslConfig<'a, A>>);
+
+/// Single node SSL connection config.
+#[derive(Clone)]
+pub struct NodeSslConfig<'a, A> {
+  pub addr: &'a str,
+  pub authenticator: A,
+  pub ssl_connector: SslConnector,
+  pub max_size: u32,
+  pub min_idle: Option<u32>,
+  pub max_lifetime: Option<Duration>,
+  pub idle_timeout: Option<Duration>,
+  pub connection_timeout: Duration,
+}
+
+/// Builder structure that helps to configure SSL connection for node.
+pub struct NodeSslConfigBuilder<'a, A> {
+  addr: &'a str,
+  authenticator: A,
+  ssl_connector: SslConnector,
+  max_size: Option<u32>,
+  min_idle: Option<u32>,
+  max_lifetime: Option<Duration>,
+  idle_timeout: Option<Duration>,
+  connection_timeout: Option<Duration>,
+}
+
+impl<'a, A: Authenticator + Sized> NodeSslConfigBuilder<'a, A> {
+  const DEFAULT_MAX_SIZE: u32 = 10;
+  const DEFAULT_CONNECTION_TIMEOUT: Duration = Duration::from_secs(30);
+
+  /// `NodeSslConfigBuilder` constructor function. It receives
+  /// * node socket address as a string
+  /// * authenticator
+  /// * SSL connector structure (for more details see [openssl docs](https://docs.rs/openssl/0.10.12/openssl/ssl/struct.SslConnector.html))
+  }); 
+  pub fn new<'b>(
+    addr: &'b str,
+    authenticator: A,
+    ssl_connector: SslConnector,
+  ) -> NodeSslConfigBuilder<'b, A> {
+    NodeSslConfigBuilder {
+      addr,
+      authenticator,
+      ssl_connector,
+      max_size: None,
+      min_idle: None,
+      max_lifetime: None,
+      idle_timeout: None,
+      connection_timeout: None,
+    }
+  }
+
+  /// Sets the maximum number of connections managed by the pool.
+  /// Defaults to 10.
+  pub fn max_size(mut self, size: u32) -> Self {
+    self.max_size = Some(size);
+    self
+  }
+
+  /// Sets the minimum idle connection count maintained by the pool.
+  /// If set, the pool will try to maintain at least this many idle
+  /// connections at all times, while respecting the value of `max_size`.
+  /// Defaults to None (equivalent to the value of `max_size`).
+  pub fn min_idle(mut self, min_idle: Option<u32>) -> Self {
+    self.max_size = min_idle;
+    self
+  }
+
+  /// Sets the maximum lifetime of connections in the pool.
+  /// If set, connections will be closed after existing for at most 30 seconds beyond this duration.
+  /// If a connection reaches its maximum lifetime while checked out it will be closed when it is returned to the pool.
+  /// Defaults to 30 minutes.
+  pub fn max_lifetime(mut self, max_lifetime: Option<Duration>) -> Self {
+    self.max_lifetime = max_lifetime;
+    self
+  }
+
+  /// Sets the idle timeout used by the pool.
+  /// If set, connections will be closed after sitting idle for at most 30 seconds beyond this duration.
+  /// Defaults to 10 minutes.
+  pub fn idle_timeout(mut self, idle_timeout: Option<Duration>) -> Self {
+    self.idle_timeout = idle_timeout;
+    self
+  }
+
+  /// Sets the connection timeout used by the pool.
+  /// Defaults to 30 seconds.
+  pub fn connection_timeout(mut self, connection_timeout: Duration) -> Self {
+    self.connection_timeout = Some(connection_timeout);
+    self
+  }
+
+  /// Sets new authenticator.
+  pub fn authenticator(mut self, authenticator: A) -> Self {
+    self.authenticator = authenticator;
+    self
+  }
+
+  /// Finalizes building process and returns `NodeSslConfig`
+  pub fn build(self) -> NodeSslConfig<'a, A> {
+    NodeSslConfig {
+      addr: self.addr,
+      authenticator: self.authenticator,
+      ssl_connector: self.ssl_connector,
+
+      max_size: self.max_size.unwrap_or(Self::DEFAULT_MAX_SIZE),
+      min_idle: self.min_idle,
+      max_lifetime: self.max_lifetime,
+      idle_timeout: self.idle_timeout,
+      connection_timeout: self
+        .connection_timeout
+        .unwrap_or(Self::DEFAULT_CONNECTION_TIMEOUT),
+    }
+  }
+}
diff --git a/src/cluster/config_tcp.rs b/src/cluster/config_tcp.rs
new file mode 100644
index 0000000..40442ab
--- /dev/null
+++ b/src/cluster/config_tcp.rs
@@ -0,0 +1,111 @@
+use std::time::Duration;
+
+use authenticators::Authenticator;
+
+/// Cluster configuration that holds per node TCP configs
+pub struct ClusterTcpConfig<'a, A: Authenticator + Sized>(pub Vec<NodeTcpConfig<'a, A>>);
+
+/// Single node TCP connection config.
+#[derive(Clone)]
+pub struct NodeTcpConfig<'a, A> {
+  pub addr: &'a str,
+  pub authenticator: A,
+  pub max_size: u32,
+  pub min_idle: Option<u32>,
+  pub max_lifetime: Option<Duration>,
+  pub idle_timeout: Option<Duration>,
+  pub connection_timeout: Duration,
+}
+
+/// Builder structure that helps to configure TCP connection for node.
+pub struct NodeTcpConfigBuilder<'a, A> {
+  addr: &'a str,
+  authenticator: A,
+  max_size: Option<u32>,
+  min_idle: Option<u32>,
+  max_lifetime: Option<Duration>,
+  idle_timeout: Option<Duration>,
+  connection_timeout: Option<Duration>,
+}
+
+impl<'a, A: Authenticator + Sized> NodeTcpConfigBuilder<'a, A> {
+  const DEFAULT_MAX_SIZE: u32 = 10;
+  const DEFAULT_CONNECTION_TIMEOUT: Duration = Duration::from_secs(30);
+
+  /// `NodeTcpConfigBuilder` constructor function. It receivesthread::spawn(move || {
+  /// * node socket address as a string
+  /// * authenticator
+  pub fn new<'b>(addr: &'b str, authenticator: A) -> NodeTcpConfigBuilder<'b, A> {
+    NodeTcpConfigBuilder {
+      addr,
+      authenticator,
+      max_size: None,
+      min_idle: None,
+      max_lifetime: None,
+      idle_timeout: None,
+      connection_timeout: None,
+    }
+  }
+
+  /// Sets the maximum number of connections managed by the pool.
+  /// Defaults to 10.
+  pub fn max_size(mut self, size: u32) -> Self {
+    self.max_size = Some(size);
+    self
+  }
+
+  /// Sets the minimum idle connection count maintained by the pool.
+  /// If set, the pool will try to maintain at least this many idle
+  /// connections at all times, while respecting the value of `max_size`.
+  /// Defaults to None (equivalent to the value of `max_size`).
+  pub fn min_idle(mut self, min_idle: Option<u32>) -> Self {
+    self.max_size = min_idle;
+    self
+  }
+
+  /// Sets the maximum lifetime of connections in the pool.
+  /// If set, connections will be closed after existing for at most 30 seconds beyond this duration.
+  /// If a connection reaches its maximum lifetime while checked out it will be closed when it is returned to the pool.
+  /// Defaults to 30 minutes.
+  pub fn max_lifetime(mut self, max_lifetime: Option<Duration>) -> Self {
+    self.max_lifetime = max_lifetime;
+    self
+  }
+
+  /// Sets the idle timeout used by the pool.
+  /// If set, connections will be closed after sitting idle for at most 30 seconds beyond this duration.
+  /// Defaults to 10 minutes.
+  pub fn idle_timeout(mut self, idle_timeout: Option<Duration>) -> Self {
+    self.idle_timeout = idle_timeout;
+    self
+  }
+
+  /// Sets the connection timeout used by the pool.
+  /// Defaults to 30 seconds.
+  pub fn connection_timeout(mut self, connection_timeout: Duration) -> Self {
+    self.connection_timeout = Some(connection_timeout);
+    self
+  }
+
+  /// Sets new authenticator.
+  pub fn authenticator(mut self, authenticator: A) -> Self {
+    self.authenticator = authenticator;
+    self
+  }
+
+  /// Finalizes building process and returns `NodeSslConfig`
+  pub fn build(self) -> NodeTcpConfig<'a, A> {
+    NodeTcpConfig {
+      addr: self.addr,
+      authenticator: self.authenticator,
+
+      max_size: self.max_size.unwrap_or(Self::DEFAULT_MAX_SIZE),
+      min_idle: self.min_idle,
+      max_lifetime: self.max_lifetime,
+      idle_timeout: self.idle_timeout,
+      connection_timeout: self
+        .connection_timeout
+        .unwrap_or(Self::DEFAULT_CONNECTION_TIMEOUT),
+    }
+  }
+}
diff --git a/src/cluster/mod.rs b/src/cluster/mod.rs
index f205834..cdff7a8 100644
--- a/src/cluster/mod.rs
+++ b/src/cluster/mod.rs
@@ -1,31 +1,60 @@
+use r2d2;
 use std::cell;
 
-mod cluster;
+#[cfg(feature = "ssl")]
+mod config_ssl;
+mod config_tcp;
 mod pager;
-mod session;
+pub mod session;
+#[cfg(feature = "ssl")]
+mod ssl_connection_pool;
+mod tcp_connection_pool;
 
-pub use cluster::cluster::Cluster;
+#[cfg(feature = "ssl")]
+pub use cluster::config_ssl::{ClusterSslConfig, NodeSslConfig, NodeSslConfigBuilder};
+pub use cluster::config_tcp::{ClusterTcpConfig, NodeTcpConfig, NodeTcpConfigBuilder};
 pub use cluster::pager::{QueryPager, SessionPager};
-pub use cluster::session::Session;
+#[cfg(feature = "ssl")]
+pub use cluster::ssl_connection_pool::{new_ssl_pool, SslConnectionPool, SslConnectionsManager};
+pub use cluster::tcp_connection_pool::{
+  new_tcp_pool, startup, TcpConnectionPool, TcpConnectionsManager,
+};
 
 use compression::Compression;
+use error;
 use query::{BatchExecutor, ExecExecutor, PrepareExecutor, QueryExecutor};
 use transport::CDRSTransport;
 
-pub trait GetTransport<'a, T: CDRSTransport + 'a> {
-  fn get_transport(&self) -> Option<&cell::RefCell<T>>;
+/// `GetConnection` trait provides a unified interface for Session to get a connection
+/// from a load balancer
+pub trait GetConnection<
+  T: CDRSTransport + Send + Sync + 'static,
+  M: r2d2::ManageConnection<Connection = cell::RefCell<T>, Error = error::Error>,
+>
+{
+  /// Returns connection from a load balancer.
+  fn get_connection(&self) -> Option<r2d2::PooledConnection<M>>;
 }
 
+/// `GetCompressor` trait provides a unified interface for Session to get a compressor
+/// for further decompressing received data.
 pub trait GetCompressor<'a> {
+  /// Returns actual compressor.
   fn get_compressor(&self) -> Compression;
 }
 
-pub trait CDRSSession<'a, T: CDRSTransport + 'static>:
+/// `CDRSSession` trait wrap ups whole query functionality. Use it only if whole query
+/// machinery is needed and direct sub traits otherwise.
+pub trait CDRSSession<
+  'a,
+  T: CDRSTransport + 'static,
+  M: r2d2::ManageConnection<Connection = cell::RefCell<T>, Error = error::Error>,
+>:
   GetCompressor<'static>
-  + GetTransport<'static, T>
-  + QueryExecutor<T>
-  + PrepareExecutor<T>
-  + ExecExecutor<T>
-  + BatchExecutor<T>
+  + GetConnection<T, M>
+  + QueryExecutor<T, M>
+  + PrepareExecutor<T, M>
+  + ExecExecutor<T, M>
+  + BatchExecutor<T, M>
 {
 }
diff --git a/src/cluster/pager.rs b/src/cluster/pager.rs
index bc4881e..020ba69 100644
--- a/src/cluster/pager.rs
+++ b/src/cluster/pager.rs
@@ -1,3 +1,5 @@
+use r2d2;
+use std::cell::RefCell;
 use std::marker::PhantomData;
 
 use cluster::CDRSSession;
@@ -8,22 +10,36 @@ use transport::CDRSTransport;
 use types::rows::Row;
 use types::CBytes;
 
-pub struct SessionPager<'a, S: CDRSSession<'static, T> + 'a, T: CDRSTransport + 'static> {
+pub struct SessionPager<
+  'a,
+  M: r2d2::ManageConnection<Connection = RefCell<T>, Error = error::Error> + Sized,
+  S: CDRSSession<'static, T, M> + 'a,
+  T: CDRSTransport + 'static,
+> {
   page_size: i32,
   session: &'a S,
   transport_type: PhantomData<&'a T>,
+  connection_type: PhantomData<&'a M>,
 }
 
-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> {
+impl<
+    'a,
+    'b: 'a,
+    M: r2d2::ManageConnection<Connection = RefCell<T>, Error = error::Error> + Sized,
+    S: CDRSSession<'static, T, M>,
+    T: CDRSTransport + 'static,
+  > SessionPager<'a, M, S, T>
+{
+  pub fn new(session: &'b S, page_size: i32) -> SessionPager<'a, M, S, T> {
     SessionPager {
       session,
       page_size,
       transport_type: PhantomData,
+      connection_type: PhantomData,
     }
   }
 
-  pub fn query<Q>(&'a mut self, query: Q) -> QueryPager<'a, Q, SessionPager<'a, S, T>>
+  pub fn query<Q>(&'a mut self, query: Q) -> QueryPager<'a, Q, SessionPager<'a, M, S, T>>
   where
     Q: ToString,
   {
@@ -35,7 +51,7 @@ impl<'a, 'b: 'a, S: CDRSSession<'static, T>, T: CDRSTransport + 'static> Session
     }
   }
 
-  pub fn exec(&'a mut self, query: &'a PreparedQuery) -> ExecPager<'a, SessionPager<'a, S, T>> {
+  pub fn exec(&'a mut self, query: &'a PreparedQuery) -> ExecPager<'a, SessionPager<'a, M, S, T>> {
     ExecPager {
       pager: self,
       paging_state: None,
@@ -52,8 +68,13 @@ pub struct QueryPager<'a, Q: ToString, P: 'a> {
   query: Q,
 }
 
-impl<'a, Q: ToString, T: CDRSTransport + 'static, S: CDRSSession<'static, T>>
-  QueryPager<'a, Q, SessionPager<'a, S, T>>
+impl<
+    'a,
+    Q: ToString,
+    T: CDRSTransport + 'static,
+    M: r2d2::ManageConnection<Connection = RefCell<T>, Error = error::Error> + Sized,
+    S: CDRSSession<'static, T, M>,
+  > QueryPager<'a, Q, SessionPager<'a, M, S, T>>
 {
   pub fn next(&mut self) -> error::Result<Vec<Row>> {
     let mut params = QueryParamsBuilder::new().page_size(self.pager.page_size);
@@ -91,8 +112,12 @@ pub struct ExecPager<'a, P: 'a> {
   query: &'a PreparedQuery,
 }
 
-impl<'a, T: CDRSTransport + 'static, S: CDRSSession<'static, T>>
-  ExecPager<'a, SessionPager<'a, S, T>>
+impl<
+    'a,
+    T: CDRSTransport + 'static,
+    M: r2d2::ManageConnection<Connection = RefCell<T>, Error = error::Error> + Sized,
+    S: CDRSSession<'static, T, M>,
+  > ExecPager<'a, SessionPager<'a, M, S, T>>
 {
   pub fn next(&mut self) -> error::Result<Vec<Row>> {
     let mut params = QueryParamsBuilder::new().page_size(self.pager.page_size);
diff --git a/src/cluster/session.rs b/src/cluster/session.rs
index 259747f..7568040 100644
--- a/src/cluster/session.rs
+++ b/src/cluster/session.rs
@@ -1,18 +1,24 @@
+use r2d2;
 use std::cell::RefCell;
-use std::io;
 use std::io::Write;
 
-use cluster::{CDRSSession, GetCompressor, GetTransport, SessionPager};
+#[cfg(feature = "ssl")]
+use cluster::{new_ssl_pool, ClusterSslConfig, SslConnectionPool};
+use cluster::{
+  new_tcp_pool, startup, CDRSSession, ClusterTcpConfig, GetCompressor, GetConnection,
+  TcpConnectionPool,
+};
 use error;
 use load_balancing::LoadBalancingStrategy;
 use transport::{CDRSTransport, TransportTcp};
 
 use authenticators::Authenticator;
+use cluster::SessionPager;
 use compression::Compression;
 use events::{new_listener, EventStream, Listener};
 use frame::events::SimpleServerEvent;
 use frame::parser::parse_frame;
-use frame::{Frame, IntoBytes, Opcode};
+use frame::{Frame, IntoBytes};
 use query::{BatchExecutor, ExecExecutor, PrepareExecutor, QueryExecutor};
 
 #[cfg(feature = "ssl")]
@@ -20,236 +26,186 @@ use openssl::ssl::SslConnector;
 #[cfg(feature = "ssl")]
 use transport::TransportTls;
 
-pub struct Session<LB, A> {
+/// CDRS session that holds one pool of authorized connecitons per node.
+/// `compression` field contains data compressor that will be used
+/// for decompressing data received from Cassandra server.
+pub struct Session<LB> {
   load_balancing: LB,
   #[allow(dead_code)]
-  authenticator: A,
   pub compression: Compression,
 }
 
-impl<'a, LB, A> GetCompressor<'a> for Session<LB, A> {
+impl<'a, LB> GetCompressor<'a> for Session<LB> {
+  /// Returns compression that current session has.
   fn get_compressor(&self) -> Compression {
     self.compression.clone()
   }
 }
 
-impl<'a, LB: Sized, A: Authenticator + 'a + Sized> Session<LB, A> {
-  pub fn paged<T: CDRSTransport + 'static>(
+impl<'a, LB: Sized> Session<LB> {
+  /// Basing on current session returns new `SessionPager` that can be used
+  /// for performing paged queries.
+  pub fn paged<
+    T: CDRSTransport + 'static,
+    M: r2d2::ManageConnection<Connection = RefCell<T>, Error = error::Error>,
+  >(
     &'a self,
     page_size: i32,
-  ) -> SessionPager<'a, Session<LB, A>, T>
+  ) -> SessionPager<'a, M, Session<LB>, T>
   where
-    Session<LB, A>: CDRSSession<'static, T>,
+    Session<LB>: CDRSSession<'static, T, M>,
   {
     return SessionPager::new(self, page_size);
   }
-
-  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();
-
-    transport.borrow_mut().write(startup_frame.as_slice())?;
-
-    let start_response = try!(parse_frame(transport, compression));
-
-    if start_response.opcode == Opcode::Ready {
-      return Ok(());
-    }
-
-    if start_response.opcode == Opcode::Authenticate {
-      let body = start_response.get_body()?;
-      let authenticator = body.get_authenticator().expect(
-        "Cassandra Server did communicate that it needed password
-                authentication but the auth schema was missing in the body response",
-      );
-
-      // This creates a new scope; avoiding a clone
-      // and we check whether
-      // 1. any authenticators has been passed in by client and if not send error back
-      // 2. authenticator is provided by the client and `auth_scheme` presented by
-      //      the server and client are same if not send error back
-      // 3. if it falls through it means the preliminary conditions are true
-
-      let auth_check = session_authenticator
-        .get_cassandra_name()
-        .ok_or(error::Error::General(
-          "No authenticator was provided".to_string(),
-        ))
-        .map(|auth| {
-          if authenticator != auth {
-            let io_err = io::Error::new(
-              io::ErrorKind::NotFound,
-              format!(
-                "Unsupported type of authenticator. {:?} got,
-                             but {} is supported.",
-                authenticator, auth
-              ),
-            );
-            return Err(error::Error::Io(io_err));
-          }
-          Ok(())
-        });
-
-      if let Err(err) = auth_check {
-        return Err(err);
-      }
-
-      let auth_token_bytes = session_authenticator.get_auth_token().into_cbytes();
-      try!(
-        transport.borrow_mut().write(
-          Frame::new_req_auth_response(auth_token_bytes)
-            .into_cbytes()
-            .as_slice()
-        )
-      );
-      try!(parse_frame(transport, compression));
-
-      return Ok(());
-    }
-
-    unimplemented!();
-  }
 }
 
 impl<
-    'a,
-    T: CDRSTransport + 'a,
-    LB: LoadBalancingStrategy<RefCell<T>> + Sized,
-    A: Authenticator + Sized,
-  > GetTransport<'a, T> for Session<LB, A>
+    T: CDRSTransport + Send + Sync + 'static,
+    M: r2d2::ManageConnection<Connection = RefCell<T>, Error = error::Error> + Sized,
+    LB: LoadBalancingStrategy<r2d2::Pool<M>>,
+  > GetConnection<T, M> for Session<LB>
 {
-  fn get_transport(&self) -> Option<&RefCell<T>> {
-    self.load_balancing.next()
+  fn get_connection(&self) -> Option<r2d2::PooledConnection<M>> {
+    self.load_balancing.next().and_then(|pool| pool.get().ok())
   }
 }
 
 impl<
     'a,
     T: CDRSTransport + 'static,
-    LB: LoadBalancingStrategy<RefCell<T>> + Sized,
-    A: Authenticator + Sized,
-  > QueryExecutor<T> for Session<LB, A>
-{
-}
+    M: r2d2::ManageConnection<Connection = RefCell<T>, Error = error::Error> + Sized,
+    LB: LoadBalancingStrategy<r2d2::Pool<M>> + Sized,
+  > QueryExecutor<T, M> for Session<LB>
+{}
 
 impl<
     'a,
     T: CDRSTransport + 'static,
-    LB: LoadBalancingStrategy<RefCell<T>> + Sized,
-    A: Authenticator + Sized,
-  > PrepareExecutor<T> for Session<LB, A>
-{
-}
+    LB: LoadBalancingStrategy<r2d2::Pool<M>> + Sized,
+    M: r2d2::ManageConnection<Connection = RefCell<T>, Error = error::Error> + Sized,
+  > PrepareExecutor<T, M> for Session<LB>
+{}
 
 impl<
     'a,
     T: CDRSTransport + 'static,
-    LB: LoadBalancingStrategy<RefCell<T>> + Sized,
-    A: Authenticator + Sized,
-  > ExecExecutor<T> for Session<LB, A>
-{
-}
+    LB: LoadBalancingStrategy<r2d2::Pool<M>> + Sized,
+    M: r2d2::ManageConnection<Connection = RefCell<T>, Error = error::Error> + Sized,
+  > ExecExecutor<T, M> for Session<LB>
+{}
 
 impl<
     'a,
     T: CDRSTransport + 'static,
-    LB: LoadBalancingStrategy<RefCell<T>> + Sized,
-    A: Authenticator + Sized,
-  > BatchExecutor<T> for Session<LB, A>
-{
-}
+    LB: LoadBalancingStrategy<r2d2::Pool<M>> + Sized,
+    M: r2d2::ManageConnection<Connection = RefCell<T>, Error = error::Error> + Sized,
+  > BatchExecutor<T, M> for Session<LB>
+{}
 
 impl<
     'a,
     T: CDRSTransport + 'static,
-    LB: LoadBalancingStrategy<RefCell<T>> + Sized,
-    A: Authenticator + Sized,
-  > CDRSSession<'a, T> for Session<LB, A>
+    M: r2d2::ManageConnection<Connection = RefCell<T>, Error = error::Error> + Sized,
+    LB: LoadBalancingStrategy<r2d2::Pool<M>> + Sized,
+  > CDRSSession<'a, T, M> for Session<LB>
+{}
+
+/// Creates new session that will perform queries without any compression. `Compression` type
+/// can be changed at any time.
+/// As a parameter it takes:
+/// * cluster config
+/// * load balancing strategy (cannot be changed during `Session` life time).
+pub fn new<A, LB>(
+  node_configs: &ClusterTcpConfig<'static, A>,
+  mut load_balancing: LB,
+) -> error::Result<Session<LB>>
+where
+  A: Authenticator + 'static + Sized,
+  LB: LoadBalancingStrategy<TcpConnectionPool<A>> + Sized,
 {
+  let mut nodes: Vec<TcpConnectionPool<A>> = Vec::with_capacity(node_configs.0.len());
+
+  for node_config in &node_configs.0 {
+    let node_connection_pool = new_tcp_pool(node_config.clone())?;
+    nodes.push(node_connection_pool);
+  }
+
+  load_balancing.init(nodes);
+
+  Ok(Session {
+    load_balancing,
+    compression: Compression::None,
+  })
 }
 
-impl<'a, LB: LoadBalancingStrategy<RefCell<TransportTcp>> + Sized, A: Authenticator + 'a + Sized>
-  Session<LB, A>
+/// Creates new session that will perform queries with Snappy compression. `Compression` type
+/// can be changed at any time.
+/// As a parameter it takes:
+/// * cluster config
+/// * load balancing strategy (cannot be changed during `Session` life time).
+pub fn new_snappy<A, LB>(
+  node_configs: &ClusterTcpConfig<'static, A>,
+  mut load_balancing: LB,
+) -> error::Result<Session<LB>>
+where
+  A: Authenticator + 'static + Sized,
+  LB: LoadBalancingStrategy<TcpConnectionPool<A>> + Sized,
 {
-  pub fn new(
-    addrs: &Vec<&str>,
-    mut load_balancing: LB,
-    authenticator: A,
-  ) -> error::Result<Session<LB, A>> {
-    let mut nodes: Vec<RefCell<TransportTcp>> = Vec::with_capacity(addrs.len());
-
-    for addr in addrs {
-      let transport = RefCell::new(TransportTcp::new(&addr)?);
-      Self::startup(&transport, &authenticator)?;
-      nodes.push(transport);
-    }
-
-    load_balancing.init(nodes);
-
-    Ok(Session {
-      load_balancing,
-      authenticator,
-      compression: Compression::None,
-    })
-  }
+  let mut nodes: Vec<TcpConnectionPool<A>> = Vec::with_capacity(node_configs.0.len());
 
-  pub fn new_snappy(
-    addrs: &Vec<&str>,
-    mut load_balancing: LB,
-    authenticator: A,
-  ) -> error::Result<Session<LB, A>> {
-    let mut nodes: Vec<RefCell<TransportTcp>> = Vec::with_capacity(addrs.len());
-
-    for addr in addrs {
-      let transport = RefCell::new(TransportTcp::new(&addr)?);
-      Self::startup(&transport, &authenticator)?;
-      nodes.push(transport);
-    }
-
-    load_balancing.init(nodes);
-
-    Ok(Session {
-      load_balancing,
-      authenticator,
-      compression: Compression::Snappy,
-    })
+  for node_config in &node_configs.0 {
+    let node_connection_pool = new_tcp_pool(node_config.clone())?;
+    nodes.push(node_connection_pool);
   }
 
-  pub fn new_lz4(
-    addrs: &Vec<&str>,
-    mut load_balancing: LB,
-    authenticator: A,
-  ) -> error::Result<Session<LB, A>> {
-    let mut nodes: Vec<RefCell<TransportTcp>> = Vec::with_capacity(addrs.len());
-
-    for addr in addrs {
-      let transport = RefCell::new(TransportTcp::new(&addr)?);
-      Self::startup(&transport, &authenticator)?;
-      nodes.push(transport);
-    }
-
-    load_balancing.init(nodes);
-
-    Ok(Session {
-      load_balancing,
-      authenticator,
-      compression: Compression::Lz4,
-    })
+  load_balancing.init(nodes);
+
+  Ok(Session {
+    load_balancing,
+    compression: Compression::Snappy,
+  })
+}
+
+/// Creates new session that will perform queries with LZ4 compression. `Compression` type
+/// can be changed at any time.
+/// As a parameter it takes:
+/// * cluster config
+/// * load balancing strategy (cannot be changed during `Session` life time).
+pub fn new_lz4<A, LB>(
+  node_configs: &ClusterTcpConfig<'static, A>,
+  mut load_balancing: LB,
+) -> error::Result<Session<LB>>
+where
+  A: Authenticator + 'static + Sized,
+  LB: LoadBalancingStrategy<TcpConnectionPool<A>> + Sized,
+{
+  let mut nodes: Vec<TcpConnectionPool<A>> = Vec::with_capacity(node_configs.0.len());
+
+  for node_config in &node_configs.0 {
+    let node_connection_pool = new_tcp_pool(node_config.clone())?;
+    nodes.push(node_connection_pool);
   }
 
-  pub fn listen(
+  load_balancing.init(nodes);
+
+  Ok(Session {
+    load_balancing,
+    compression: Compression::Lz4,
+  })
+}
+
+impl<'a, L> Session<L> {
+  /// Returns new event listener.
+  pub fn listen<A: Authenticator + 'static + Sized>(
     &self,
     node: &str,
+    authenticator: A,
     events: Vec<SimpleServerEvent>,
   ) -> error::Result<(Listener<RefCell<TransportTcp>>, EventStream)> {
-    let authenticator = self.authenticator.clone();
     let compression = self.get_compressor();
     let transport = TransportTcp::new(&node).map(RefCell::new)?;
 
-    Self::startup(&transport, &authenticator)?;
+    startup(&transport, &authenticator)?;
 
     let query_frame = Frame::new_req_register(events).into_cbytes();
     transport.borrow_mut().write(query_frame.as_slice())?;
@@ -259,97 +215,112 @@ impl<'a, LB: LoadBalancingStrategy<RefCell<TransportTcp>> + Sized, A: Authentica
   }
 }
 
+/// Creates new SSL-based session that will perform queries without any compression. `Compression` type
+/// can be changed at any time.
+/// As a parameter it takes:
+/// * SSL cluster config
+/// * load balancing strategy (cannot be changed during `Session` life time).
 #[cfg(feature = "ssl")]
-impl<'a, LB: LoadBalancingStrategy<RefCell<TransportTls>> + Sized, A: Authenticator + 'a + Sized>
-  Session<LB, A>
+pub fn new_ssl<A, LB>(
+  node_configs: &ClusterSslConfig<'static, A>,
+  mut load_balancing: LB,
+) -> error::Result<Session<LB>>
+where
+  A: Authenticator + 'static + Sized,
+  LB: LoadBalancingStrategy<SslConnectionPool<A>> + Sized,
 {
-  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<RefCell<TransportTls>> = Vec::with_capacity(addrs.len());
-
-    for addr in addrs {
-      let transport = RefCell::new(TransportTls::new(&addr, ssl_connector)?);
-      Self::startup(&transport, &authenticator)?;
-      nodes.push(transport);
-    }
-
-    load_balancing.init(nodes);
-
-    Ok(Session {
-      load_balancing,
-      authenticator,
-      compression: Compression::None,
-    })
+  let mut nodes: Vec<SslConnectionPool<A>> = Vec::with_capacity(node_configs.0.len());
+
+  for node_config in &node_configs.0 {
+    let node_connection_pool = new_ssl_pool(node_config.clone())?;
+    nodes.push(node_connection_pool);
   }
 
-  pub fn new_snappy_ssl(
-    addrs: &Vec<&str>,
-    mut load_balancing: LB,
-    authenticator: A,
-    ssl_connector: &SslConnector,
-  ) -> error::Result<Session<LB, A>> {
-    let mut nodes: Vec<RefCell<TransportTls>> = Vec::with_capacity(addrs.len());
-
-    for addr in addrs {
-      let transport = RefCell::new(TransportTls::new(&addr, ssl_connector)?);
-      Self::startup(&transport, &authenticator)?;
-      nodes.push(transport);
-    }
-
-    load_balancing.init(nodes);
-
-    Ok(Session {
-      load_balancing,
-      authenticator,
-      compression: Compression::Snappy,
-    })
+  load_balancing.init(nodes);
+
+  Ok(Session {
+    load_balancing,
+    compression: Compression::None,
+  })
+}
+
+/// Creates new SSL-based session that will perform queries with Snappy compression. `Compression` type
+/// can be changed at any time.
+/// As a parameter it takes:
+/// * SSL cluster config
+/// * load balancing strategy (cannot be changed during `Session` life time).
+#[cfg(feature = "ssl")]
+pub fn new_snappy_ssl<A, LB>(
+  node_configs: &ClusterSslConfig<'static, A>,
+  mut load_balancing: LB,
+) -> error::Result<Session<LB>>
+where
+  A: Authenticator + 'static + Sized,
+  LB: LoadBalancingStrategy<SslConnectionPool<A>> + Sized,
+{
+  let mut nodes: Vec<SslConnectionPool<A>> = Vec::with_capacity(node_configs.0.len());
+
+  for node_config in &node_configs.0 {
+    let node_connection_pool = new_ssl_pool(node_config.clone())?;
+    nodes.push(node_connection_pool);
   }
 
-  pub fn new_lz4_ssl(
-    addrs: &Vec<&str>,
-    mut load_balancing: LB,
-    authenticator: A,
-    ssl_connector: &SslConnector,
-  ) -> error::Result<Session<LB, A>> {
-    let mut nodes: Vec<RefCell<TransportTls>> = Vec::with_capacity(addrs.len());
-
-    for addr in addrs {
-      let transport = RefCell::new(TransportTls::new(&addr, ssl_connector)?);
-      Self::startup(&transport, &authenticator)?;
-      nodes.push(transport);
-    }
-
-    load_balancing.init(nodes);
-
-    Ok(Session {
-      load_balancing,
-      authenticator,
-      compression: Compression::Lz4,
-    })
+  load_balancing.init(nodes);
+
+  Ok(Session {
+    load_balancing,
+    compression: Compression::Snappy,
+  })
+}
+
+/// Creates new SSL-based session that will perform queries with LZ4 compression. `Compression` type
+/// can be changed at any time.
+/// As a parameter it takes:
+/// * SSL cluster config
+/// * load balancing strategy (cannot be changed during `Session` life time).
+#[cfg(feature = "ssl")]
+pub fn new_lz4_ssl<A, LB>(
+  node_configs: &ClusterSslConfig<'static, A>,
+  mut load_balancing: LB,
+) -> error::Result<Session<LB>>
+where
+  A: Authenticator + 'static + Sized,
+  LB: LoadBalancingStrategy<SslConnectionPool<A>> + Sized,
+{
+  let mut nodes: Vec<SslConnectionPool<A>> = Vec::with_capacity(node_configs.0.len());
+
+  for node_config in &node_configs.0 {
+    let node_connection_pool = new_ssl_pool(node_config.clone())?;
+    nodes.push(node_connection_pool);
   }
 
-  pub fn listen_ssl(
+  load_balancing.init(nodes);
+
+  Ok(Session {
+    load_balancing,
+    compression: Compression::Lz4,
+  })
+}
+
+/// Returns new SSL-based event listener.
+#[cfg(feature = "ssl")]
+impl<'a, L> Session<L> {
+  pub fn listen_ssl<A: Authenticator + 'static + Sized>(
     &self,
     node: (&str, &SslConnector),
+    authenticator: A,
     events: Vec<SimpleServerEvent>,
   ) -> error::Result<(Listener<RefCell<TransportTls>>, EventStream)> {
-    let (addr, ssl_connector) = node;
-    let authenticator = self.authenticator.clone();
+    let (addr_ref, ssl_connector_ref) = node;
     let compression = self.get_compressor();
-    let transport = self
-      .get_transport()
-      .ok_or("Cannot connect to a cluster - no nodes provided")?;
-    let transport_cell = RefCell::new(TransportTls::new(&addr, ssl_connector)?);
-    Self::startup(&transport, &authenticator)?;
+    let transport = TransportTls::new(addr_ref, ssl_connector_ref).map(RefCell::new)?;
+
+    startup(&transport, &authenticator)?;
 
     let query_frame = Frame::new_req_register(events).into_cbytes();
-    transport_cell.borrow_mut().write(query_frame.as_slice())?;
-    parse_frame(&transport_cell, &compression)?;
+    transport.borrow_mut().write(query_frame.as_slice())?;
+    parse_frame(&transport, &compression)?;
 
-    Ok(new_listener(transport_cell))
+    Ok(new_listener(transport))
   }
 }
diff --git a/src/cluster/ssl_connection_pool.rs b/src/cluster/ssl_connection_pool.rs
new file mode 100644
index 0000000..8881c90
--- /dev/null
+++ b/src/cluster/ssl_connection_pool.rs
@@ -0,0 +1,79 @@
+use openssl::ssl::SslConnector;
+use r2d2::{Builder, ManageConnection, Pool};
+use std::cell::RefCell;
+use std::error::Error;
+use std::io::Write;
+
+use authenticators::Authenticator;
+use cluster::{startup, NodeSslConfig};
+use compression::Compression;
+use error;
+use frame::parser::parse_frame;
+use frame::{Frame, IntoBytes};
+use transport::CDRSTransport;
+use transport::TransportTls;
+
+/// Shortcut for `r2d2::Pool` type of SSL-based CDRS connections.
+pub type SslConnectionPool<A> = Pool<SslConnectionsManager<A>>;
+
+/// `r2d2::Pool` of SSL-based CDRS connections.
+///
+/// Used internally for SSL Session for holding connections to a specific Cassandra node.
+pub fn new_ssl_pool<A: Authenticator + Send + Sync + 'static>(
+  node_config: NodeSslConfig<'static, A>,
+) -> error::Result<SslConnectionPool<A>> {
+  let manager = SslConnectionsManager::new(
+    node_config.addr,
+    node_config.authenticator,
+    node_config.ssl_connector,
+  );
+
+  Builder::new()
+    .max_size(node_config.max_size)
+    .min_idle(node_config.min_idle)
+    .max_lifetime(node_config.max_lifetime)
+    .idle_timeout(node_config.idle_timeout)
+    .connection_timeout(node_config.connection_timeout)
+    .build(manager)
+    .map_err(|err| error::Error::from(err.description()))
+}
+
+/// `r2d2` connection manager.
+pub struct SslConnectionsManager<A> {
+  addr: &'static str,
+  ssl_connector: SslConnector,
+  auth: A,
+}
+
+impl<A> SslConnectionsManager<A> {
+  pub fn new(addr: &'static str, auth: A, ssl_connector: SslConnector) -> Self {
+    SslConnectionsManager {
+      addr,
+      auth,
+      ssl_connector,
+    }
+  }
+}
+
+impl<A: Authenticator + 'static + Send + Sync> ManageConnection for SslConnectionsManager<A> {
+  type Connection = RefCell<TransportTls>;
+  type Error = error::Error;
+
+  fn connect(&self) -> Result<Self::Connection, Self::Error> {
+    let transport = RefCell::new(TransportTls::new(&self.addr, &self.ssl_connector)?);
+    startup(&transport, &self.auth)?;
+
+    Ok(transport)
+  }
+
+  fn is_valid(&self, conn: &mut Self::Connection) -> Result<(), Self::Error> {
+    let options_frame = Frame::new_req_options().into_cbytes();
+    try!(conn.borrow_mut().write(options_frame.as_slice()));
+
+    parse_frame(conn, &Compression::None {}).map(|_| ())
+  }
+
+  fn has_broken(&self, conn: &mut Self::Connection) -> bool {
+    conn.borrow().is_alive()
+  }
+}
diff --git a/src/cluster/tcp_connection_pool.rs b/src/cluster/tcp_connection_pool.rs
new file mode 100644
index 0000000..8680338
--- /dev/null
+++ b/src/cluster/tcp_connection_pool.rs
@@ -0,0 +1,138 @@
+use r2d2::{Builder, ManageConnection, Pool};
+use std::cell::RefCell;
+use std::error::Error;
+use std::io;
+use std::io::Write;
+
+use authenticators::Authenticator;
+use cluster::NodeTcpConfig;
+use compression::Compression;
+use error;
+use frame::parser::parse_frame;
+use frame::{Frame, IntoBytes, Opcode};
+use transport::{CDRSTransport, TransportTcp};
+
+/// Shortcut for `r2d2::Pool` type of TCP-based CDRS connections.
+pub type TcpConnectionPool<A> = Pool<TcpConnectionsManager<A>>;
+
+/// `r2d2::Pool` of TCP-based CDRS connections.
+///
+/// Used internally for TCP Session for holding connections to a specific Cassandra node.
+pub fn new_tcp_pool<A: Authenticator + Send + Sync + 'static>(
+  node_config: NodeTcpConfig<'static, A>,
+) -> error::Result<TcpConnectionPool<A>> {
+  let manager = TcpConnectionsManager::new(node_config.addr, node_config.authenticator);
+
+  Builder::new()
+    .max_size(node_config.max_size)
+    .min_idle(node_config.min_idle)
+    .max_lifetime(node_config.max_lifetime)
+    .idle_timeout(node_config.idle_timeout)
+    .connection_timeout(node_config.connection_timeout)
+    .build(manager)
+    .map_err(|err| error::Error::from(err.description()))
+}
+
+/// `r2d2` connection manager.
+pub struct TcpConnectionsManager<A> {
+  addr: &'static str,
+  auth: A,
+}
+
+impl<A> TcpConnectionsManager<A> {
+  pub fn new(addr: &'static str, auth: A) -> Self {
+    TcpConnectionsManager { addr, auth }
+  }
+}
+
+impl<A: Authenticator + 'static + Send + Sync> ManageConnection for TcpConnectionsManager<A> {
+  type Connection = RefCell<TransportTcp>;
+  type Error = error::Error;
+
+  fn connect(&self) -> Result<Self::Connection, Self::Error> {
+    let transport = RefCell::new(TransportTcp::new(&self.addr)?);
+    startup(&transport, &self.auth)?;
+
+    Ok(transport)
+  }
+
+  fn is_valid(&self, conn: &mut Self::Connection) -> Result<(), Self::Error> {
+    let options_frame = Frame::new_req_options().into_cbytes();
+    try!(conn.borrow_mut().write(options_frame.as_slice()));
+
+    parse_frame(conn, &Compression::None {}).map(|_| ())
+  }
+
+  fn has_broken(&self, conn: &mut Self::Connection) -> bool {
+    conn.borrow().is_alive()
+  }
+}
+
+pub fn startup<'b, T: CDRSTransport + 'static, A: Authenticator + 'static + Sized>(
+  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();
+
+  transport.borrow_mut().write(startup_frame.as_slice())?;
+
+  let start_response = try!(parse_frame(transport, compression));
+
+  if start_response.opcode == Opcode::Ready {
+    return Ok(());
+  }
+
+  if start_response.opcode == Opcode::Authenticate {
+    let body = start_response.get_body()?;
+    let authenticator = body.get_authenticator().expect(
+      "Cassandra Server did communicate that it neededs
+                authentication but the auth schema was missing in the body response",
+    );
+
+    // This creates a new scope; avoiding a clone
+    // and we check whether
+    // 1. any authenticators has been passed in by client and if not send error back
+    // 2. authenticator is provided by the client and `auth_scheme` presented by
+    //      the server and client are same if not send error back
+    // 3. if it falls through it means the preliminary conditions are true
+
+    let auth_check = session_authenticator
+      .get_cassandra_name()
+      .ok_or(error::Error::General(
+        "No authenticator was provided".to_string(),
+      ))
+      .map(|auth| {
+        if authenticator != auth {
+          let io_err = io::Error::new(
+            io::ErrorKind::NotFound,
+            format!(
+              "Unsupported type of authenticator. {:?} got,
+                             but {} is supported.",
+              authenticator, auth
+            ),
+          );
+          return Err(error::Error::Io(io_err));
+        }
+        Ok(())
+      });
+
+    if let Err(err) = auth_check {
+      return Err(err);
+    }
+
+    let auth_token_bytes = session_authenticator.get_auth_token().into_cbytes();
+    try!(
+      transport.borrow_mut().write(
+        Frame::new_req_auth_response(auth_token_bytes)
+          .into_cbytes()
+          .as_slice()
+      )
+    );
+    try!(parse_frame(transport, compression));
+
+    return Ok(());
+  }
+
+  unreachable!();
+}
diff --git a/src/compression.rs b/src/compression.rs
index 296705a..612a8d0 100644
--- a/src/compression.rs
+++ b/src/compression.rs
@@ -7,30 +7,15 @@
 //!must never be compressed.  However, once the STARTUP frame has been received
 //!by the server, messages can be compressed (including the response to the STARTUP
 //!request).
-//!
-//!```no_run
-//!use cdrs::cluster::Cluster;
-//!use cdrs::authenticators::NoneAuthenticator;
-//!use cdrs::load_balancing::RoundRobin;
-//!
-//!const _ADDR: &'static str = "127.0.0.1:9042";
-//!let cluster = Cluster::new(vec![_ADDR], NoneAuthenticator {});
-//!let mut no_compression = cluster.connect(RoundRobin::new())
-//!                                .expect("No compression connection error");
-//!let mut lz4_compression = cluster.connect_lz4(RoundRobin::new())
-//!                                 .expect("LZ4 compression connection error");
-//!let mut snappy_compression = cluster.connect_snappy(RoundRobin::new())
-//!                                  .expect("Snappy compression connection error");
-//!```
 
 use std::convert::From;
 use std::error::Error;
-use std::result;
 use std::fmt;
 use std::io;
+use std::result;
 
-use snap;
 use lz4_compress as lz4;
+use snap;
 
 type Result<T> = result::Result<T, CompressionError>;
 
@@ -147,14 +132,16 @@ impl Compression {
 
     fn encode_snappy(bytes: Vec<u8>) -> Result<Vec<u8>> {
         let mut encoder = snap::Encoder::new();
-        encoder.compress_vec(bytes.as_slice())
-               .map_err(CompressionError::Snappy)
+        encoder
+            .compress_vec(bytes.as_slice())
+            .map_err(CompressionError::Snappy)
     }
 
     fn decode_snappy(bytes: Vec<u8>) -> Result<Vec<u8>> {
         let mut decoder = snap::Decoder::new();
-        decoder.decompress_vec(bytes.as_slice())
-               .map_err(CompressionError::Snappy)
+        decoder
+            .decompress_vec(bytes.as_slice())
+            .map_err(CompressionError::Snappy)
     }
 
     fn encode_lz4(bytes: Vec<u8>) -> Result<Vec<u8>> {
@@ -215,8 +202,9 @@ mod tests {
     fn test_compression_encode_snappy() {
         let snappy_compression = Compression::Snappy;
         let bytes = String::from("Hello World").into_bytes().to_vec();
-        snappy_compression.encode(bytes.clone())
-                          .expect("Should work without exceptions");
+        snappy_compression
+            .encode(bytes.clone())
+            .expect("Should work without exceptions");
     }
 
     #[test]
@@ -231,8 +219,9 @@ mod tests {
     fn test_compression_encode_lz4() {
         let snappy_compression = Compression::Lz4;
         let bytes = String::from("Hello World").into_bytes().to_vec();
-        snappy_compression.encode(bytes.clone())
-                          .expect("Should work without exceptions");
+        snappy_compression
+            .encode(bytes.clone())
+            .expect("Should work without exceptions");
     }
 
     #[test]
@@ -250,8 +239,9 @@ mod tests {
     fn test_compression_encode_none() {
         let none_compression = Compression::None;
         let bytes = String::from("Hello World").into_bytes().to_vec();
-        none_compression.encode(bytes.clone())
-                        .expect("Should work without exceptions");
+        none_compression
+            .encode(bytes.clone())
+            .expect("Should work without exceptions");
     }
 
     #[test]
@@ -275,8 +265,9 @@ mod tests {
     fn test_compression_encode_snappy_with_non_utf8() {
         let snappy_compression = Compression::Snappy;
         let v = vec![0xff, 0xff];
-        let encoded = snappy_compression.encode(v.clone())
-                                        .expect("Should work without exceptions");
+        let encoded = snappy_compression
+            .encode(v.clone())
+            .expect("Should work without exceptions");
         assert_eq!(snappy_compression.decode(encoded).unwrap(), v);
     }
 
diff --git a/src/consistency.rs b/src/consistency.rs
index e142578..74dbf59 100644
--- a/src/consistency.rs
+++ b/src/consistency.rs
@@ -1,11 +1,11 @@
 #![warn(missing_docs)]
 //! The module contains Rust representation of Cassandra consistency levels.
-use std::io;
 use std::convert::From;
 use std::default::Default;
+use std::io;
 
-use frame::{FromBytes, FromCursor, IntoBytes};
 use error;
+use frame::{FromBytes, FromCursor, IntoBytes};
 use types::*;
 
 /// `Consistency` is an enum which represents Cassandra's consistency levels.
@@ -23,7 +23,6 @@ pub enum Consistency {
     /// A write must be written to the commit log and memtable of at least two replica nodes.
     /// Similar to ONE.
     Two,
-
     /// A write must be written to the commit log and memtable of at least three replica nodes.
     /// Similar to TWO.
     Three,
@@ -120,19 +119,19 @@ impl From<i32> for Consistency {
 impl FromBytes for Consistency {
     fn from_bytes(bytes: &[u8]) -> error::Result<Consistency> {
         try_from_bytes(bytes).map_err(Into::into).map(|b| match b {
-                                                          0x0000 => Consistency::Any,
-                                                          0x0001 => Consistency::One,
-                                                          0x0002 => Consistency::Two,
-                                                          0x0003 => Consistency::Three,
-                                                          0x0004 => Consistency::Quorum,
-                                                          0x0005 => Consistency::All,
-                                                          0x0006 => Consistency::LocalQuorum,
-                                                          0x0007 => Consistency::EachQuorum,
-                                                          0x0008 => Consistency::Serial,
-                                                          0x0009 => Consistency::LocalSerial,
-                                                          0x000A => Consistency::LocalOne,
-                                                          _ => Consistency::Unknown,
-                                                      })
+            0x0000 => Consistency::Any,
+            0x0001 => Consistency::One,
+            0x0002 => Consistency::Two,
+            0x0003 => Consistency::Three,
+            0x0004 => Consistency::Quorum,
+            0x0005 => Consistency::All,
+            0x0006 => Consistency::LocalQuorum,
+            0x0007 => Consistency::EachQuorum,
+            0x0008 => Consistency::Serial,
+            0x0009 => Consistency::LocalSerial,
+            0x000A => Consistency::LocalOne,
+            _ => Consistency::Unknown,
+        })
     }
 }
 
@@ -145,9 +144,9 @@ impl FromCursor for Consistency {
 
 #[cfg(test)]
 mod tests {
-    use std::io::Cursor;
-    use frame::traits::{FromBytes, FromCursor, IntoBytes};
     use super::*;
+    use frame::traits::{FromBytes, FromCursor, IntoBytes};
+    use std::io::Cursor;
 
     #[test]
     fn test_consistency_into_cbytes() {
@@ -186,49 +185,87 @@ mod tests {
         assert_eq!(Consistency::from_bytes(&[0, 0]).unwrap(), Consistency::Any);
         assert_eq!(Consistency::from_bytes(&[0, 1]).unwrap(), Consistency::One);
         assert_eq!(Consistency::from_bytes(&[0, 2]).unwrap(), Consistency::Two);
-        assert_eq!(Consistency::from_bytes(&[0, 3]).unwrap(),
-                   Consistency::Three);
-        assert_eq!(Consistency::from_bytes(&[0, 4]).unwrap(),
-                   Consistency::Quorum);
+        assert_eq!(
+            Consistency::from_bytes(&[0, 3]).unwrap(),
+            Consistency::Three
+        );
+        assert_eq!(
+            Consistency::from_bytes(&[0, 4]).unwrap(),
+            Consistency::Quorum
+        );
         assert_eq!(Consistency::from_bytes(&[0, 5]).unwrap(), Consistency::All);
-        assert_eq!(Consistency::from_bytes(&[0, 6]).unwrap(),
-                   Consistency::LocalQuorum);
-        assert_eq!(Consistency::from_bytes(&[0, 7]).unwrap(),
-                   Consistency::EachQuorum);
-        assert_eq!(Consistency::from_bytes(&[0, 8]).unwrap(),
-                   Consistency::Serial);
-        assert_eq!(Consistency::from_bytes(&[0, 9]).unwrap(),
-                   Consistency::LocalSerial);
-        assert_eq!(Consistency::from_bytes(&[0, 10]).unwrap(),
-                   Consistency::LocalOne);
-        assert_eq!(Consistency::from_bytes(&[0, 11]).unwrap(),
-                   Consistency::Unknown);
+        assert_eq!(
+            Consistency::from_bytes(&[0, 6]).unwrap(),
+            Consistency::LocalQuorum
+        );
+        assert_eq!(
+            Consistency::from_bytes(&[0, 7]).unwrap(),
+            Consistency::EachQuorum
+        );
+        assert_eq!(
+            Consistency::from_bytes(&[0, 8]).unwrap(),
+            Consistency::Serial
+        );
+        assert_eq!(
+            Consistency::from_bytes(&[0, 9]).unwrap(),
+            Consistency::LocalSerial
+        );
+        assert_eq!(
+            Consistency::from_bytes(&[0, 10]).unwrap(),
+            Consistency::LocalOne
+        );
+        assert_eq!(
+            Consistency::from_bytes(&[0, 11]).unwrap(),
+            Consistency::Unknown
+        );
     }
 
     #[test]
     fn test_consistency_from_cursor() {
-        assert_eq!(Consistency::from_cursor(&mut Cursor::new(&[0, 0])).unwrap(),
-                   Consistency::Any);
-        assert_eq!(Consistency::from_cursor(&mut Cursor::new(&[0, 1])).unwrap(),
-                   Consistency::One);
-        assert_eq!(Consistency::from_cursor(&mut Cursor::new(&[0, 2])).unwrap(),
-                   Consistency::Two);
-        assert_eq!(Consistency::from_cursor(&mut Cursor::new(&[0, 3])).unwrap(),
-                   Consistency::Three);
-        assert_eq!(Consistency::from_cursor(&mut Cursor::new(&[0, 4])).unwrap(),
-                   Consistency::Quorum);
-        assert_eq!(Consistency::from_cursor(&mut Cursor::new(&[0, 5])).unwrap(),
-                   Consistency::All);
-        assert_eq!(Consistency::from_cursor(&mut Cursor::new(&[0, 6])).unwrap(),
-                   Consistency::LocalQuorum);
-        assert_eq!(Consistency::from_cursor(&mut Cursor::new(&[0, 7])).unwrap(),
-                   Consistency::EachQuorum);
-        assert_eq!(Consistency::from_cursor(&mut Cursor::new(&[0, 8])).unwrap(),
-                   Consistency::Serial);
-        assert_eq!(Consistency::from_cursor(&mut Cursor::new(&[0, 9])).unwrap(),
-                   Consistency::LocalSerial);
-        assert_eq!(Consistency::from_cursor(&mut Cursor::new(&[0, 10])).unwrap(),
-                   Consistency::LocalOne);
+        assert_eq!(
+            Consistency::from_cursor(&mut Cursor::new(&[0, 0])).unwrap(),
+            Consistency::Any
+        );
+        assert_eq!(
+            Consistency::from_cursor(&mut Cursor::new(&[0, 1])).unwrap(),
+            Consistency::One
+        );
+        assert_eq!(
+            Consistency::from_cursor(&mut Cursor::new(&[0, 2])).unwrap(),
+            Consistency::Two
+        );
+        assert_eq!(
+            Consistency::from_cursor(&mut Cursor::new(&[0, 3])).unwrap(),
+            Consistency::Three
+        );
+        assert_eq!(
+            Consistency::from_cursor(&mut Cursor::new(&[0, 4])).unwrap(),
+            Consistency::Quorum
+        );
+        assert_eq!(
+            Consistency::from_cursor(&mut Cursor::new(&[0, 5])).unwrap(),
+            Consistency::All
+        );
+        assert_eq!(
+            Consistency::from_cursor(&mut Cursor::new(&[0, 6])).unwrap(),
+            Consistency::LocalQuorum
+        );
+        assert_eq!(
+            Consistency::from_cursor(&mut Cursor::new(&[0, 7])).unwrap(),
+            Consistency::EachQuorum
+        );
+        assert_eq!(
+            Consistency::from_cursor(&mut Cursor::new(&[0, 8])).unwrap(),
+            Consistency::Serial
+        );
+        assert_eq!(
+            Consistency::from_cursor(&mut Cursor::new(&[0, 9])).unwrap(),
+            Consistency::LocalSerial
+        );
+        assert_eq!(
+            Consistency::from_cursor(&mut Cursor::new(&[0, 10])).unwrap(),
+            Consistency::LocalOne
+        );
     }
 
 }
diff --git a/src/frame/parser.rs b/src/frame/parser.rs
index 9d2b701..7e496a3 100644
--- a/src/frame/parser.rs
+++ b/src/frame/parser.rs
@@ -1,14 +1,28 @@
+use r2d2;
 use std::cell::RefCell;
 use std::io::{Cursor, Read};
+use std::ops::Deref;
 
 use super::*;
 use compression::Compression;
 use error;
 use frame::frame_response::ResponseBody;
 use frame::FromCursor;
+use transport::CDRSTransport;
 use types::data_serialization_types::decode_timeuuid;
 use types::{from_bytes, from_u16_bytes, CStringList, UUID_LEN};
 
+pub fn from_connection<M, T>(
+    conn: &r2d2::PooledConnection<M>,
+    compressor: &Compression,
+) -> error::Result<Frame>
+where
+    T: CDRSTransport + 'static,
+    M: r2d2::ManageConnection<Connection = RefCell<T>, Error = error::Error> + Sized,
+{
+    parse_frame(conn.deref(), compressor)
+}
+
 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];
diff --git a/src/lib.rs b/src/lib.rs
index 1759444..db1f3c2 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -10,6 +10,7 @@ extern crate log;
 extern crate lz4_compress;
 #[cfg(feature = "ssl")]
 extern crate openssl;
+extern crate r2d2;
 extern crate rand;
 extern crate time;
 extern crate uuid;
diff --git a/src/load_balancing/mod.rs b/src/load_balancing/mod.rs
index a0539de..3429926 100644
--- a/src/load_balancing/mod.rs
+++ b/src/load_balancing/mod.rs
@@ -1,8 +1,10 @@
 mod random;
 mod round_robin;
+mod round_robin_sync;
 
 pub use load_balancing::random::Random;
 pub use load_balancing::round_robin::RoundRobin;
+pub use load_balancing::round_robin_sync::RoundRobinSync;
 
 pub trait LoadBalancingStrategy<N>: Sized {
   fn init(&mut self, cluster: Vec<N>);
diff --git a/src/load_balancing/round_robin_sync.rs b/src/load_balancing/round_robin_sync.rs
new file mode 100644
index 0000000..cd543bd
--- /dev/null
+++ b/src/load_balancing/round_robin_sync.rs
@@ -0,0 +1,59 @@
+use std::sync::Mutex;
+
+use super::LoadBalancingStrategy;
+
+pub struct RoundRobinSync<N> {
+  cluster: Vec<N>,
+  prev_idx: Mutex<usize>,
+}
+
+impl<N> RoundRobinSync<N> {
+  pub fn new() -> Self {
+    RoundRobinSync {
+      prev_idx: Mutex::new(0),
+      cluster: vec![],
+    }
+  }
+}
+
+impl<N> From<Vec<N>> for RoundRobinSync<N> {
+  fn from(cluster: Vec<N>) -> RoundRobinSync<N> {
+    RoundRobinSync {
+      prev_idx: Mutex::new(0),
+      cluster: cluster,
+    }
+  }
+}
+
+impl<N> LoadBalancingStrategy<N> for RoundRobinSync<N> {
+  fn init(&mut self, cluster: Vec<N>) {
+    self.cluster = cluster;
+  }
+
+  /// Returns next node from a cluster
+  fn next(&self) -> Option<&N> {
+    let mut prev_idx = self.prev_idx.try_lock();
+    if let Ok(ref mut mtx) = prev_idx {
+      let next_idx = (**mtx + 1) % self.cluster.len();
+      **mtx = next_idx;
+      self.cluster.get(next_idx)
+    } else {
+      return None;
+    }
+  }
+}
+
+#[cfg(test)]
+mod tests {
+  use super::*;
+
+  #[test]
+  fn round_robin() {
+    let nodes = vec!["a", "b", "c"];
+    let nodes_c = nodes.clone();
+    let load_balancer = RoundRobinSync::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 d53a712..dcb62cb 100644
--- a/src/query/batch_executor.rs
+++ b/src/query/batch_executor.rs
@@ -1,13 +1,18 @@
-use cluster::{GetCompressor, GetTransport};
+use r2d2;
+use std::cell::RefCell;
+
+use cluster::{GetCompressor, GetConnection};
 use error;
-use frame::parser::parse_frame;
+use frame::parser::from_connection;
 use frame::traits::IntoBytes;
 use frame::{Flag, Frame};
 use query::batch_query_builder::QueryBatch;
 use transport::CDRSTransport;
 
-pub trait BatchExecutor<T: CDRSTransport + 'static>:
-  GetTransport<'static, T> + GetCompressor<'static>
+pub trait BatchExecutor<
+  T: CDRSTransport + 'static,
+  M: r2d2::ManageConnection<Connection = RefCell<T>, Error = error::Error> + Sized,
+>: GetConnection<T, M> + GetCompressor<'static>
 {
   fn batch_with_params_tw(
     &self,
@@ -29,7 +34,7 @@ pub trait BatchExecutor<T: CDRSTransport + 'static>:
     let ref compression = self.get_compressor();
 
     self
-      .get_transport()
+      .get_connection()
       .ok_or(error::Error::from("Unable to get transport"))
       .and_then(|transport_cell| {
         let write_res = transport_cell
@@ -38,7 +43,7 @@ pub trait BatchExecutor<T: CDRSTransport + 'static>:
           .map_err(error::Error::from);
         write_res.map(|_| transport_cell)
       })
-      .and_then(|transport_cell| parse_frame(transport_cell, compression))
+      .and_then(|transport_cell| from_connection(&transport_cell, compression))
   }
 
   fn batch_with_params(&self, batch: QueryBatch) -> error::Result<Frame> {
diff --git a/src/query/exec_executor.rs b/src/query/exec_executor.rs
index 88e585a..0d19003 100644
--- a/src/query/exec_executor.rs
+++ b/src/query/exec_executor.rs
@@ -1,6 +1,9 @@
-use cluster::{GetCompressor, GetTransport};
+use r2d2;
+use std::cell::RefCell;
+
+use cluster::{GetCompressor, GetConnection};
 use error;
-use frame::parser::parse_frame;
+use frame::parser::from_connection;
 use frame::{Flag, Frame, IntoBytes};
 use query::{QueryParams, QueryParamsBuilder, QueryValues};
 use transport::CDRSTransport;
@@ -8,8 +11,10 @@ use types::CBytesShort;
 
 pub type PreparedQuery = CBytesShort;
 
-pub trait ExecExecutor<T: CDRSTransport + 'static>:
-  GetTransport<'static, T> + GetCompressor<'static>
+pub trait ExecExecutor<
+  T: CDRSTransport + 'static,
+  M: r2d2::ManageConnection<Connection = RefCell<T>, Error = error::Error> + Sized,
+>: GetConnection<T, M> + GetCompressor<'static>
 {
   fn exec_with_params_tw(
     &self,
@@ -30,7 +35,7 @@ pub trait ExecExecutor<T: CDRSTransport + 'static>:
     let ref compression = self.get_compressor();
 
     self
-      .get_transport()
+      .get_connection()
       .ok_or(error::Error::from("Unable to get transport"))
       .and_then(|transport_cell| {
         let write_res = transport_cell
@@ -39,7 +44,7 @@ pub trait ExecExecutor<T: CDRSTransport + 'static>:
           .map_err(error::Error::from);
         write_res.map(|_| transport_cell)
       })
-      .and_then(|transport_cell| parse_frame(transport_cell, compression))
+      .and_then(|transport_cell| from_connection(&transport_cell, compression))
   }
 
   fn exec_with_params(
diff --git a/src/query/mod.rs b/src/query/mod.rs
index 1a9ca59..c32744b 100644
--- a/src/query/mod.rs
+++ b/src/query/mod.rs
@@ -2,20 +2,20 @@ mod batch_executor;
 mod batch_query_builder;
 mod exec_executor;
 mod prepare_executor;
+mod query;
 mod query_executor;
 mod query_flags;
-mod query;
+mod query_params;
 mod query_params_builder;
 mod query_values;
-mod query_params;
 
 pub use query::batch_executor::BatchExecutor;
 pub use query::batch_query_builder::{BatchQueryBuilder, QueryBatch};
 pub use query::exec_executor::ExecExecutor;
 pub use query::prepare_executor::{PrepareExecutor, PreparedQuery};
+pub use query::query::Query;
 pub use query::query_executor::QueryExecutor;
 pub use query::query_flags::QueryFlags;
-pub use query::query::Query;
-pub use query::query_params_builder::QueryParamsBuilder;
 pub use query::query_params::QueryParams;
+pub use query::query_params_builder::QueryParamsBuilder;
 pub use query::query_values::QueryValues;
diff --git a/src/query/prepare_executor.rs b/src/query/prepare_executor.rs
index 5886c64..60da28b 100644
--- a/src/query/prepare_executor.rs
+++ b/src/query/prepare_executor.rs
@@ -1,14 +1,19 @@
-use cluster::{GetCompressor, GetTransport};
+use r2d2;
+use std::cell::RefCell;
+
+use cluster::{GetCompressor, GetConnection};
 use error;
-use frame::parser::parse_frame;
+use frame::parser::from_connection;
 use frame::{Flag, Frame, IntoBytes};
 use transport::CDRSTransport;
 use types::CBytesShort;
 
 pub type PreparedQuery = CBytesShort;
 
-pub trait PrepareExecutor<T: CDRSTransport + 'static>:
-  GetTransport<'static, T> + GetCompressor<'static>
+pub trait PrepareExecutor<
+  T: CDRSTransport + 'static,
+  M: r2d2::ManageConnection<Connection = RefCell<T>, Error = error::Error> + Sized,
+>: GetConnection<T, M> + GetCompressor<'static>
 {
   /// It prepares a query for execution, along with query itself
   /// the method takes `with_tracing` and `with_warnings` flags
@@ -31,7 +36,7 @@ pub trait PrepareExecutor<T: CDRSTransport + 'static>:
     let ref compression = self.get_compressor();
 
     self
-      .get_transport()
+      .get_connection()
       .ok_or(error::Error::from("Unable to get transport"))
       .and_then(|transport_cell| {
         let write_res = transport_cell
@@ -40,7 +45,7 @@ pub trait PrepareExecutor<T: CDRSTransport + 'static>:
           .map_err(error::Error::from);
         write_res.map(|_| transport_cell)
       })
-      .and_then(|transport_cell| parse_frame(transport_cell, compression))
+      .and_then(|transport_cell| from_connection(&transport_cell, compression))
       .and_then(|response| response.get_body())
       .and_then(|body| {
         Ok(
diff --git a/src/query/query_executor.rs b/src/query/query_executor.rs
index bbaca24..6f11967 100644
--- a/src/query/query_executor.rs
+++ b/src/query/query_executor.rs
@@ -1,12 +1,17 @@
-use cluster::{GetCompressor, GetTransport};
+use r2d2;
+use std::cell::RefCell;
+
+use cluster::{GetCompressor, GetConnection};
 use error;
-use frame::parser::parse_frame;
+use frame::parser::from_connection;
 use frame::{Flag, Frame, IntoBytes};
 use query::{Query, QueryParams, QueryParamsBuilder, QueryValues};
 use transport::CDRSTransport;
 
-pub trait QueryExecutor<T: CDRSTransport + 'static>:
-  GetTransport<'static, T> + GetCompressor<'static>
+pub trait QueryExecutor<
+  T: CDRSTransport + 'static,
+  M: r2d2::ManageConnection<Connection = RefCell<T>, Error = error::Error> + Sized,
+>: GetConnection<T, M> + GetCompressor<'static>
 {
   fn query_with_params_tw<Q: ToString>(
     &self,
@@ -34,7 +39,7 @@ pub trait QueryExecutor<T: CDRSTransport + 'static>:
     let ref compression = self.get_compressor();
 
     self
-      .get_transport()
+      .get_connection()
       .ok_or(error::Error::from("Unable to get transport"))
       .and_then(|transport_cell| {
         let write_res = transport_cell
@@ -43,7 +48,7 @@ pub trait QueryExecutor<T: CDRSTransport + 'static>:
           .map_err(error::Error::from);
         write_res.map(|_| transport_cell)
       })
-      .and_then(|transport_cell| parse_frame(transport_cell, compression))
+      .and_then(|transport_cell| from_connection(&transport_cell, compression))
   }
 
   /// Executes a query with default parameters:


Test patch:
diff --git a/tests/common/mod.rs b/tests/common/mod.rs
index 4b8d851..46601a6 100644
--- a/tests/common/mod.rs
+++ b/tests/common/mod.rs
@@ -1,29 +1,24 @@
-use std::cell::RefCell;
-
 use cdrs::authenticators::NoneAuthenticator;
-use cdrs::cluster::{Cluster, Session};
-// use cdrs::compression::Compression;
+use cdrs::cluster::session::{new as new_session, Session};
+use cdrs::cluster::{ClusterTcpConfig, NodeTcpConfigBuilder, TcpConnectionPool};
 use cdrs::error::Result;
 use cdrs::load_balancing::RoundRobin;
 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<RefCell<TransportTcp>>, NoneAuthenticator>;
+type CurrentSession = Session<RoundRobin<TcpConnectionPool<NoneAuthenticator>>>;
 
-pub fn setup(create_table_cql: &'static str) -> Result<CSession> {
+pub fn setup(create_table_cql: &'static str) -> Result<CurrentSession> {
   setup_multiple(&[create_table_cql])
 }
 
-pub fn setup_multiple(create_cqls: &[&'static str]) -> Result<CSession> {
-  let authenticator = NoneAuthenticator {};
-  let nodes = vec![ADDR];
-  let cluster = Cluster::new(nodes, authenticator);
-  let session = cluster
-    .connect(RoundRobin::new())
-    .expect("No compression connection error");
+pub fn setup_multiple(create_cqls: &[&'static str]) -> Result<CurrentSession> {
+  let node = NodeTcpConfigBuilder::new(ADDR, NoneAuthenticator {}).build();
+  let cluster_config = ClusterTcpConfig(vec![node]);
+  let lb = RoundRobin::new();
+  let session = new_session(&cluster_config, lb).expect("session should be created");
   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 \
diff --git a/tests/keyspace.rs b/tests/keyspace.rs
index eceaa56..d3b1ba5 100644
--- a/tests/keyspace.rs
+++ b/tests/keyspace.rs
@@ -3,7 +3,8 @@ extern crate cdrs;
 use std::collections::HashMap;
 
 use cdrs::authenticators::NoneAuthenticator;
-use cdrs::cluster::Cluster;
+use cdrs::cluster::session::{new as new_session, Session};
+use cdrs::cluster::{ClusterTcpConfig, NodeTcpConfigBuilder, TcpConnectionPool};
 use cdrs::load_balancing::RoundRobin;
 use cdrs::query::QueryExecutor;
 use cdrs::types::map::Map;
@@ -14,9 +15,10 @@ const ADDR: &'static str = "127.0.0.1:9042";
 #[test]
 #[ignore]
 fn create_keyspace() {
-        let cluster = Cluster::new(vec!["127.0.0.1:9042"], NoneAuthenticator {});
-        let session = cluster.connect(RoundRobin::new())
-                .expect("No compression connection error");
+        let node = NodeTcpConfigBuilder::new("127.0.0.1:9042", NoneAuthenticator {}).build();
+        let cluster_config = ClusterTcpConfig(vec![node]);
+        let lb = RoundRobin::new();
+        let session = new_session(&cluster_config, lb).expect("session should be created");
 
         let drop_query = "DROP KEYSPACE IF EXISTS create_ks_test";
         let keyspace_droped = session.query(drop_query).is_ok();
@@ -33,7 +35,8 @@ fn create_keyspace() {
 
         let select_query =
                 "SELECT * FROM system_schema.keyspaces WHERE keyspace_name = 'create_ks_test'";
-        let keyspace_selected = session.query(select_query)
+        let keyspace_selected = session
+                .query(select_query)
                 .expect("select keyspace query")
                 .get_body()
                 .expect("get select keyspace query body")
@@ -43,7 +46,8 @@ fn create_keyspace() {
         assert_eq!(keyspace_selected.len(), 1);
         let keyspace = &keyspace_selected[0];
 
-        let keyspace_name: String = keyspace.get_r_by_name("keyspace_name")
+        let keyspace_name: String = keyspace
+                .get_r_by_name("keyspace_name")
                 .expect("keyspace name into rust error");
         assert_eq!(
                 keyspace_name,
@@ -51,7 +55,8 @@ fn create_keyspace() {
                 "wrong keyspace name"
         );
 
-        let durable_writes: bool = keyspace.get_r_by_name("durable_writes")
+        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");
 
@@ -61,7 +66,8 @@ fn create_keyspace() {
                 "class".to_string(),
                 "org.apache.cassandra.locator.SimpleStrategy".to_string(),
         );
-        let strategy_options: HashMap<String, String> = keyspace.r_by_name::<Map>("replication")
+        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");
@@ -74,9 +80,10 @@ fn create_keyspace() {
 #[test]
 #[ignore]
 fn alter_keyspace() {
-        let cluster = Cluster::new(vec!["127.0.0.1:9042"], NoneAuthenticator {});
-        let session = cluster.connect(RoundRobin::new())
-                .expect("No compression connection error");
+        let node = NodeTcpConfigBuilder::new("127.0.0.1:9042", NoneAuthenticator {}).build();
+        let cluster_config = ClusterTcpConfig(vec![node]);
+        let lb = RoundRobin::new();
+        let session = new_session(&cluster_config, lb).expect("session should be created");
 
         let drop_query = "DROP KEYSPACE IF EXISTS alter_ks_test";
         let keyspace_droped = session.query(drop_query).is_ok();
@@ -101,7 +108,8 @@ fn alter_keyspace() {
 
         let select_query =
                 "SELECT * FROM system_schema.keyspaces WHERE keyspace_name = 'alter_ks_test'";
-        let keyspace_selected = session.query(select_query)
+        let keyspace_selected = session
+                .query(select_query)
                 .expect("select keyspace query")
                 .get_body()
                 .expect("get select keyspace query body")
@@ -111,7 +119,8 @@ fn alter_keyspace() {
         assert_eq!(keyspace_selected.len(), 1);
         let keyspace = &keyspace_selected[0];
 
-        let strategy_options: HashMap<String, String> = keyspace.r_by_name::<Map>("replication")
+        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");
@@ -127,9 +136,10 @@ fn alter_keyspace() {
 #[test]
 #[ignore]
 fn use_keyspace() {
-        let cluster = Cluster::new(vec!["127.0.0.1:9042"], NoneAuthenticator {});
-        let session = cluster.connect(RoundRobin::new())
-                .expect("No compression connection error");
+        let node = NodeTcpConfigBuilder::new("127.0.0.1:9042", NoneAuthenticator {}).build();
+        let cluster_config = ClusterTcpConfig(vec![node]);
+        let lb = RoundRobin::new();
+        let session = new_session(&cluster_config, lb).expect("session should be created");
 
         let create_query = "CREATE KEYSPACE IF NOT EXISTS use_ks_test WITH \
                             replication = {'class': 'SimpleStrategy', 'replication_factor': 1} \
@@ -141,7 +151,8 @@ fn use_keyspace() {
         );
 
         let use_query = "USE use_ks_test";
-        let keyspace_used = session.query(use_query)
+        let keyspace_used = session
+                .query(use_query)
                 .expect("should use selected")
                 .get_body()
                 .expect("should get body")
@@ -154,9 +165,10 @@ fn use_keyspace() {
 #[test]
 #[ignore]
 fn drop_keyspace() {
-        let cluster = Cluster::new(vec!["127.0.0.1:9042"], NoneAuthenticator {});
-        let session = cluster.connect(RoundRobin::new())
-                .expect("No compression connection error");
+        let node = NodeTcpConfigBuilder::new("127.0.0.1:9042", NoneAuthenticator {}).build();
+        let cluster_config = ClusterTcpConfig(vec![node]);
+        let lb = RoundRobin::new();
+        let session = new_session(&cluster_config, lb).expect("session should be created");
 
         let create_query = "CREATE KEYSPACE IF NOT EXISTS drop_ks_test WITH \
                             replication = {'class': 'SimpleStrategy', 'replication_factor': 1} \
diff --git a/tests/native_types.rs b/tests/native_types.rs
index 1597925..decee9a 100644
--- a/tests/native_types.rs
+++ b/tests/native_types.rs
@@ -11,7 +11,7 @@ use common::*;
 use cdrs::query::QueryExecutor;
 use cdrs::types::blob::Blob;
 use cdrs::types::map::Map;
-use cdrs::types::value::{Bytes, Value};
+use cdrs::types::value::Bytes;
 use cdrs::types::{AsRust, ByName, IntoRustByName};
 use uuid::Uuid;
 
@@ -254,7 +254,8 @@ fn blob() {
     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")
+        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");