CDRS

CDRS is a native driver for Apache Cassandra written in Rust. The motivation to write it in Rust is a lack of native one. Existing ones are bindings to C clients.

Documentation

CDRS completely implements 4-th version of Cassandra protocol. Also it provides tools for mapping results into Rust structures.

Content

Creating new connection

use cdrs::client::CDRS;
use cdrs::authenticators::NoneAuthenticator;
use cdrs::transport::TransportTcp;

After that you can create a new instance of CDRS and establish new connection:

let authenticator = NoneAuthenticator;
let addr = "127.0.0.1:9042";
let tcp_transport = TransportTcp::new(addr).unwrap();

// pass authenticator and transport into CDRS' constructor
let client = CDRS::new(tcp_transport, authenticator);
use cdrs::compression;
// start session without compression
let mut session = try!(client.start(compression::None));

Creating new connection with authentication

To use password authenticator, just include the one implemented in cdrs::authenticators.

use cdrs::client::CDRS;
use cdrs::authenticators::PasswordAuthenticator;
use cdrs::transport::TransportTcp;

After that you can create a new instance of CDRS and establish new connection:

let authenticator = PasswordAuthenticator::new("user", "pass");
let addr = "127.0.0.1:9042";
let tcp_transport = TransportTcp::new(addr).unwrap();

// pass authenticator and transport into CDRS' constructor
let client = CDRS::new(tcp_transport, authenticator);
use cdrs::compression;
// start session without compression
let mut session = try!(client.start(compression::None));

Creating new encrypted connection

To be able to create SSL-encrypted connection CDRS should be used with ssl feature enabled. Apart of CDRS itself openssl must also be imported.

[dependencies]
openssl = "0.9.6"

[dependencies.cdrs]
version = "*"
features = ["ssl"]

use cdrs::client::CDRS;
use cdrs::authenticators::PasswordAuthenticator;
use cdrs::transport::TransportTls;
use openssl::ssl::{SslConnectorBuilder, SslMethod};
use std::path::Path;

After that you can create a new instance of CDRS and establish new connection:

let authenticator = PasswordAuthenticator::new("user", "pass");
let addr = "127.0.0.1:9042";

// here needs to be a path of 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();

let ssl_transport = TransportTls::new(addr, &connector).unwrap();

// pass authenticator and SSL transport into CDRS' constructor
let client = CDRS::new(ssl_transport, authenticator);

Connecting via r2d2 connection pool

There is an option to create r2d2 connection pool of CDRS connections both plain and SSL-encrypted:

use cdrs::connection_manager::ConnectionManager;

let transport = TransportTcp::new(ADDR).unwrap();
let authenticator = PasswordAuthenticator::new(USER, PASS);
let manager = ConnectionManager::new(transport, authenticator, Compression::None);
let pool = r2d2::Pool::builder()
    .max_size(15)
    .build(manager).unwrap();

for _ in 0..20 {
    let pool = pool.clone();
    thread::spawn(move || {
        let conn = pool.get().unwrap();
        // use the connection
        // it will be returned to the pool when it falls out of scope.
    });
}

There is a related example.

Getting supported options

Before session established an application may want to know which options are supported by a server (for instance to figure out which compression to use). That's why CDRS instance has a method get_options which could be called before session get started. Options are presented as HashMap<String, Vec<String>>.

let options = try!(client.get_options());

This should be called before session started to let you know which compression to choose and because session object borrows CDRS instance.

Using compression

Two types of compression are supported - snappy and lz4. To use compression just start connection with desired type:

use cdrs::compression::Compression;
// session without compression
let mut session_res = client.start(Compression::None);
// session  lz4 compression
let mut session_res = client.start(Compression::Lz4);
// v with snappy compression
let mut session_res = client.start(Compression::Snappy);

Query execution

Query execution is provided in scope of Session. So to start executing queries you need to start Session first.

Use Query:

let create_query: Query = QueryBuilder::new("USE my_namespace;").finalize();
let with_tracing = false;
let with_warnings = false;

match session.query(create_query, with_tracing, with_warnings) {
    Ok(set_keyspace) => {
        // use_keyspace is a result frame of type SetKeyspace
    },
    Err(err) => log!(err)
}

Create Query:

Creating new table could be performed via session.query. In case of success method returns Schema Change frame that contains Change Type, Target and options that contain namespace and a name of newly created table.

use std::default::Default;
use cdrs::query::{Query, QueryBuilder};
use cdrs::consistency::Consistency;

let mut create_query: Query = QueryBuilder::new("CREATE TABLE keyspace.emp (
    empID int,
    deptID int,
    first_name varchar,
    last_name varchar,
    PRIMARY KEY (empID, deptID)
    );")
    .consistency(Consistency::One)
    .finalize();
let with_tracing = false;
let with_warnings = false;

let table_created = session.query(create_query, with_tracing, with_warnings).is_ok();

Insert/Update Query (Structures serialization)

In order to perform insertion/updating of a Rust structure you'd need to perform its serialization first. CDRS provides Into<Value> implementation (and since CDRS version 1.2.0 derivable IntoCDRSValue) trait.

After that it's possible to create a query with values:

let ints = IntsV3 {
      bigint: 123,
      int: 234,
  };
let values: Vec<Value> = vec![ints.bigint.into(), ints.int.into()];

let insert_query = QueryBuilder::new(INSERT_QUERY_STRING)
    .values(values)
    .finalize();

Conversion of nested structures could be performed if we'd consider inner structures as Cassandra User Defined Type:

#[derive(IntoCDRSValues)]
struct SomeUdt {
  value: i32
}

struct RowObj {
  id: i64,
  udt: SomeUdt
}

let obj = {
  id: 1,
  udt: SomeUdt { value: 128 }
};

let values: Vec<Value> = vec![obj.id.into_cdrs_value(), obj.udt.into_cdrs_value()];

let insert_query = QueryBuilder::new(INSERT_QUERY_STRING)
    .values(values)
    .finalize();

Select Query:

As a response to select query CDRS returns a result frame of type Rows with data items (columns) encoded in Cassandra's way.

use std::default::Default;
use cdrs::client::Query;
use cdrs::consistency::Consistency;

let select_query: Query = QueryBuilder::new("SELECT * FROM keyspace.table;").finalize();
let with_tracing = false;
let with_warnings = false;

match session.query(select_query, with_tracing, with_warnings) {
    Ok(res) => println!("Result frame: {:?},\nparsed body: {:?}", res, res.get_body());,
    Err(err) => log!(err)
}

Select Query (mapping results):

Once CDRS got response to SELECT query you can map rows encapsulated within Result frame into Rust values or into List, Map or UDT helper structures which provide a way to convert wrapped values into plain ones.

As an example let's consider a case when application gets a collection of messages of following format:

struct Message {
    pub author: String,
    pub text: String,
    pub optional_field: Option<String>
}

To get a collection of messages Vec<Message> let's convert a result of query into collection of rows Vec<cdrs::types::row::Row> and then convert each column into appropriate Rust type:

use cdrs::error::{Result as CResult};

let res_body = parsed.get_body().unwrap();
let rows = res_body.into_rows().unwrap();
let messages: Vec<CResult<Message>> = rows
    .iter()
    .map(|row| Message {
        author: row.r_by_name("author").unwrap(),
        text: row.r_by_name("text").unwrap(),
        optional_field: row.get_by_name("optional_field")
    })
    .collect();

or by column position:

let messages: Vec<CResult<Message>> = rows
    .iter()
    .map(|row| Message {
        author: row.r_by_index(0).unwrap(),
        text: row.r_by_index(1).unwrap(),
        optional_field: row.get_by_index(2)
    })
    .collect();

There is no difference between Cassandra's List and Sets in terms of Rust. They could be represented as Vec<T>. To convert a frame into a structure that contains a collection of elements do as follows:

struct Author {
    pub name: String,
    pub messages: Vec<String>
}

//...
use cdrs::error::{Result as CResult};
let res_body = parsed.get_body().unwrap();
let rows = res_body.into_rows().unwrap();
let messages: Vec<CAuthor> = rows
    .iter()
    .map(|row| {
        let name: String = row.r_by_name("name").unwrap();
        let messages: Vec<String> = row
            // unwrap Option<CResult<T>>, where T implements AsRust
            .r_by_name("messages").unwrap()
            .as_r_rust().unwrap();
        return Author {
            author: name,
            text: messages
        };
    })
    .collect();

Prepare and execute a query:

Prepare-execute query is also supported:

  // NOTE: keyspace "keyspace" should already exist
  let create_table_cql = "USE keyspace;".to_string();
  let with_tracing = false;
  let with_warnings = false;

  // prepare a query
  let prepared = session.prepare(create_table_cql, with_tracing, with_warnings)
    .unwrap()
    .get_body()
    .into_prepared()
    .unwrap();

  // execute prepared query
  let execution_params = QueryParamsBuilder::new(Consistency::One).finalize();
  let query_id = prepared.id;
  let executed = session.execute(query_id, execution_params, false, false)
    .unwrap()
    .get_body()
    .unwrap()
    .into_set_keyspace()
    .unwrap();

It's also makes sense to use prepare query in pair with batching few queries.

Listen to Server events

CDRS provides functionality which allows listening to server events. Events inform user about following changes:

Topology change - events related to change in the cluster topology. Currently, events are sent when new nodes are added to the cluster, and when nodes are removed.
Status change - events related to change of node status. Currently, up/down events are sent.
Schema_change - events related to schema change.

Current implementation allows to move listener and stream handler into separate threads so then (as we believe) developers could leverage whatever async IO library they want.

To find an examples please refer to examples.

Cassandra clusters and load balancing

CDRS supports Apache Cassandra clusters and load balancing. In order to connect to desired nodes you have to provide related transports (either TCP or TLS) and to configure r2d2 pool.

let cluster = vec![_ADDR1, _ADDR2]
    .iter()
    .map(|addr| TransportTcp::new(addr).unwrap())
    .collect();
let pool = r2d2::Pool::builder()
    .max_size(15)
    .build(manager).expect("Failed to initialize Cassandra connection pool");

After that you need to choose desired load balancing strategy and instantiate cluster collection manager. At current moment two static strategies were implemented: Random and RoundRobin.

let load_balancer = LoadBalancer::new(cluster, LoadBalancingStrategy::RoundRobin);
let manager = ClusterConnectionManager::new(load_balancer, authenticator, Compression::None);

After that you'll be able to communicate with cluster via r2d2 connection pool.

Performance

Folder ./benches contains benchmark tests. This is an attempt to measure the driver performance and to compare it with already existing solutions. Also periodically running benchmark tests might help to identify slow parts of CDRS and prevent performance degradation caused by changes.

To get current result you need either to have nightly Rust installed on your machine or install rustup.

In case you have nightly Rust, just run cargo bench. If you have rustup -- rustup run nightly cargo bench.

To find last results refer to benchmarks.md

Supported features

Frames

Request

Response

Issues

Feel free to submit issues and enhancement requests.

Contributing

Please refer to each project's style guidelines and guidelines for submitting patches and additions. In general, we follow the "fork-and-pull" Git workflow.

Fork the repo on GitHub
Clone the project to your own machine
Commit changes to your own branch
**Run cargo test --all-features && cargo fmt -- --write-mode=diff
Push your work back up to your fork
Submit a Pull request so that we can review your changes

NOTE: Be sure to merge the latest from "upstream" before making a pull request! while running the tests you might need a local cassandra server working. The easiest way was to run cassandra on docker on local machine

Running Cassandra on Local

If you have docker on the machine type the below command

docker run --name cassandra-1 -d -p 9042:9042 -p 9160:9160 cassandra:2.2.1

docker ps should show an output like below

   CONTAINER ID        IMAGE               COMMAND                  CREATED             STATUS              PORTS                                                                     NAMES
   a78c3a43bf1b        cassandra:2.2.1     "/docker-entrypoin..."   4 days ago          Up 4 days           7000-7001/tcp, 0.0.0.0:9042->9042/tcp, 7199/tcp, 0.0.0.0:9160->9160/tcp   cassandra-1

If docker is new to your tool set; it is never too late to know this awesome tool https://docs.docker.com/docker-for-mac/

Running Cassandra Cluster on local

To start Apache Cassandra cluster on local just run tests/build-cluster.sh. This script will create two nodes of Apache Cassandra 3.9 with following exposed ports: 9042 and 9043.

Name		Name	Last commit message	Last commit date
Latest commit History 678 Commits
benches		benches
examples		examples
fixtures/frame/events		fixtures/frame/events
src		src
tests		tests
.gitignore		.gitignore
.travis.yml		.travis.yml
CONTRIBUTORS.md		CONTRIBUTORS.md
Cargo.toml		Cargo.toml
LICENSE-APACHE		LICENSE-APACHE
LICENSE-MIT		LICENSE-MIT
README.md		README.md
appveyor.yml		appveyor.yml
benchmarks.md		benchmarks.md
cassandra-ports.txt		cassandra-ports.txt
changelog.md		changelog.md
rustfmt.toml		rustfmt.toml
type-mapping.md		type-mapping.md

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Licenses found

Repository files navigation

CDRS

Content

Creating new connection

Creating new connection with authentication

Creating new encrypted connection

Connecting via r2d2 connection pool

Getting supported options

Using compression

Query execution

Use Query:

Create Query:

Insert/Update Query (Structures serialization)

Select Query:

Select Query (mapping results):

Prepare and execute a query:

Listen to Server events

Cassandra clusters and load balancing

Performance

Supported features

Frames

Request

Response

Issues

Contributing

Running Cassandra on Local

Running Cassandra Cluster on local

License

About

Licenses found

Releases 31

Sponsor this project

Packages

Contributors 28

Languages

License

Licenses found

AlexPikalov/cdrs

Folders and files

Latest commit

History

Repository files navigation

CDRS

Content

Creating new connection

Creating new connection with authentication

Creating new encrypted connection

Connecting via r2d2 connection pool

Getting supported options

Using compression

Query execution

Use Query:

Create Query:

Insert/Update Query (Structures serialization)

Select Query:

Select Query (mapping results):

Prepare and execute a query:

Listen to Server events

Cassandra clusters and load balancing

Performance

Supported features

Frames

Request

Response

Issues

Contributing

Running Cassandra on Local

Running Cassandra Cluster on local

License

About

Topics

Resources

License

Licenses found

Stars

Watchers

Forks

Releases 31

Sponsor this project

Packages 0

Contributors 28

Languages

Packages