Introduction to CDRS, Apache Cassandra driver completely written in Rust
CDRS ( Apache C assandra d river written in R u s t) is my own open source project, which I decided to develop after I discovered that there was a shortage of drivers for Cassandra in the Rust ecosystem.
Of course, I will not say that they are not at all. They are, but one part is the packages abandoned in the infancy of Hello World, and the second part is probably the only binding to the driver from DataStax, written in C ++ .
As for the CDRS, with Rust, he fully implements the specification of the 4th protocol version .
cargo.toml
To include the driver in your project, as usual, you need the following.
First, add CDRS to dependenciesyour cargo.tomlfile section :
[dependencies]
cdrs = "1.0.0-beta.1"This will allow you to use a TCP connection without encryption.
If you intend to create an SSL-encrypted connection with your database, then CDRS must be enabled with the "ssl" feature:
[dependencies]
openssl = "0.9.6"
[dependencies.cdrs]
version = "1.0.0-beta.1"
features = ["ssl"]Second, add it to lib.rs
extern crate CDRSEstablish a connection
TCP connection
To establish a non-encrypted connection, you will need the following modules
use cdrs::client::CDRS;
use cdrs::authenticators::{NoneAuthenticator, PasswordAuthenticator};
use cdrs::transport::TransportPlain;If it happens that your cluster does not require password authentication, then the connection can be established as follows:
let authenticator = NoneAuthenticator;
let addr = "127.0.0.1:9042";
let tcp_transport = TransportPlain::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));To establish a connection that requires password authentication, instead of NoneAuthenticatorusing PasswordAuthenticator:
let authenticator = PasswordAuthenticator::new("user", "pass");TLS connection
Establishing a TLS connection is much like the process described in the previous step, except that you need a PEM certificate to create an SSL transport.
use cdrs::client::CDRS;
use cdrs::authenticators::PasswordAuthenticator;
use cdrs::transport::TransportTls;
use openssl::ssl::{SslConnectorBuilder, SslMethod};
use std::path::Path;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);Connection pool
For easier management of existing connections, CDRS contains ConnectionManager, which is essentially an adapter for r2d2 .
use cdrs::connection_manager::ConnectionManager;
//...
let config = r2d2::Config::builder()
.pool_size(3)
.build();
let transport = TransportPlain::new(ADDR).unwrap();
let authenticator = PasswordAuthenticator::new(USER, PASS);
let manager = ConnectionManager::new(transport, authenticator, Compression::None);
let pool = r2d2::Pool::new(config, 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.
});
}Compression - lz4 and snappy
To use lz4and snappycompression, transfer enough constructor session desired decoder:
// session without compression
let mut session_res = client.start(compression::None);
// session with lz4 compression
let mut session_res = client.start(compression::Lz4);
// session with snappy compression
let mut session_res = client.start(compression::Snappy);Further, the CDRS will independently inform the cluster that it is ready to receive information in compressed form with the selected decoder. Further unpacking will take place automatically and does not require any further action from the developer.
Query execution
The execution of requests to the Cassandra server takes place exclusively within the framework of the existing session, after choosing authorization methods, compression, and also the type of transport.
To execute a particular request, it is necessary to create an objectQuery that at first glance may seem somewhat redundant for simple queries, since it contains many parameters that are probably not used as often.
For this reason, a was created builderthat simplifies the process of configuring the request. For example, for a simple ' USE my_namespace;' simply
let create_query: Query = QueryBuilder::new("USE my_namespace;").finalize();
let with_tracing = false;
let with_warnings = false;
let switched = session.query(create_query, with_tracing, with_warnings).is_ok();Create a new table
To create a new table in the Cassandra cluster, as before, you must first configure Queryand then run the query:
use std::default::Default;
use cdrs::query::{Query, QueryBuilder};
use cdrs::consistency::Consistency;
let mut create_query: Query = QueryBuilder::new("CREATE TABLE keyspace.authors (
id int,
name text,
messages list,
PRIMARY KEY (id)
);")
.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(); As for the CQL query itself, creating a new table, it is better to turn to specialized resources, for example, DataStax , for more complete information .
SELECT query and results mapping
Suppose that in our database there is a table of authors, with each author having a list of their posts. Let these messages be stored inside the list column. In Rust terms, the author should look like this:
struct Author {
pub name: String,
pub messages: Vec
} The query itself can be executed through a Session::querymethod, as was done in the case of creating a table. Naturally, CQL should be something like in this case SELECT * FROM keyspace.authors;. ' ' If the table contains data about some authors, we can try to map the received data to a collection of Rust structures, such as ' 'Vec
//...
use cdrs::error::{Result as CResult};
let res_body = parsed.get_body();
let rows = res_body.into_rows().unwrap();
let messages: Vec = rows
.iter()
.map(|row| {
let name: String = row.get_by_name("name").unwrap();
let messages: Vec = row
// unwrap Option>, where T implements AsRust
.get_by_name("messages").unwrap().unwrap()
.as_rust().unwrap();
return Author {
author: name,
text: messages
};
})
.collect(); When displaying the results, you should pay attention to the following traits:
IntoRustByName. Говоря простым языком, этот трейт применяется по отношению к сложным типам Cassandra таким, как row (которая, строго говоря не является отдельным типом, определенным в спецификации, но по своему внутреннему устройству может рассматриваться, как что-то близкое к User Defined Type) и UDT. Грубо говоря,
get_by_nameпытается отыскать "свойство" по его имени, и если находит, то возвращает результат преобразования этого свойства к Rust типу или к CDRS типам, таким какList, 'Map',UDT. Сами же эти типы есть отображение соответствующих типов данных определенных в спецификации.- AsRust. Этот трейт предназначен для конечного отображения в Rust типы. Полный список имплиментаторов можно увидеть в приведенной ссылке.
Prepare & Execute
It is sometimes convenient at first to prepare a complex query once, and then execute it several times with different data at different times. Prepare & Execute is perfect for this.
// prepare just once
let insert_table_cql = " insert into user_keyspace.users (user_name, password, gender, session_token, state) values (?, ?, ?, ?, ?)";
let prepared = session.prepare(insert_table_cql.to_string(), true, true)
.unwrap()
.get_body()
.into_prepared()
.unwrap();
// execute later and possible few times with different values
let v: Vec = vec![Value::new_normal(String::from("john").into_bytes()),
Value::new_normal(String::from("pwd").into_bytes()),
Value::new_normal(String::from("male").into_bytes()),
Value::new_normal(String::from("09000").into_bytes()),
Value::new_normal(String::from("FL").into_bytes())];
let execution_params = QueryParamsBuilder::new(Consistency::One).values(v).finalize();
// without tracing and warnings
let executed = session.execute(prepared.id, execution_params, false, false); It also makes sense to combine Prepare & Batch to execute several prepared queries at once. The simplest Batch example can also be found in the examples .
Cassandra events
In addition to all of the above, CDRS provides the ability to subscribe and follow the events that the server publishes.
let (mut listener, stream) = session.listen_for(vec![SimpleServerEvent::SchemaChange]).unwrap();
thread::spawn(move || listener.start(&Compression::None).unwrap());
let topology_changes = stream
// inspects all events in a stream
.inspect(|event| println!("inspect event {:?}", event))
// filter by event's type: topology changes
.filter(|event| event == &SimpleServerEvent::TopologyChange)
// filter by event's specific information: new node was added
.filter(|event| {
match event {
&ServerEvent::TopologyChange(ref event) => {
event.change_type == TopologyChangeType::NewNode
},
_ => false
}
});
println!("Start listen for server events");
for change in topology_changes {
println!("server event {:?}", change);
}To find a complete list of events, it is best to consult the specification itself , as well as the driver documentation .
In the future, there are plans to use events for smart load balancing.
useful links
- CDRS repository , examples , documentation
- Protocol version 4 specification
- DataStax CQL , cluster configuration .
