use cdrs_tokio::authenticators::NoneAuthenticatorProvider;
use cdrs_tokio::cluster::session::{Session, SessionBuilder, TcpSessionBuilder};
use cdrs_tokio::cluster::{NodeTcpConfigBuilder, TcpConnectionManager};
use cdrs_tokio::load_balancing::RoundRobinLoadBalancingStrategy;
use cdrs_tokio::query::*;
use cdrs_tokio::query_values;
use cdrs_tokio::transport::TransportTcp;
use cdrs_tokio::{IntoCdrsValue, TryFromRow};
use std::sync::Arc;
type CurrentSession = Session<
TransportTcp,
TcpConnectionManager,
RoundRobinLoadBalancingStrategy<TransportTcp, TcpConnectionManager>,
>;
#[derive(Clone, Debug, IntoCdrsValue, TryFromRow, PartialEq)]
struct RowStruct {
key: i32,
}
impl RowStruct {
fn into_query_values(self) -> QueryValues {
query_values!(self.key)
}
}
#[tokio::main]
async fn main() {
let cluster_config = NodeTcpConfigBuilder::new()
.with_contact_point("127.0.0.1:9042".into())
.with_authenticator_provider(Arc::new(NoneAuthenticatorProvider))
.build()
.await
.unwrap();
let lb = RoundRobinLoadBalancingStrategy::new();
let mut session = TcpSessionBuilder::new(lb, cluster_config)
.build()
.await
.unwrap();
create_keyspace(&mut session).await;
create_table(&mut session).await;
let insert_struct_cql = "INSERT INTO test_ks.my_test_table (key) VALUES (?)";
let prepared_query = session
.prepare(insert_struct_cql)
.await
.expect("Prepare query error");
for k in 100..110 {
let row = RowStruct { key: k };
insert_row(&mut session, row, &prepared_query).await;
}
batch_few_queries(&mut session, insert_struct_cql).await;
}
async fn create_keyspace(session: &mut CurrentSession) {
let create_ks: &'static str = "CREATE KEYSPACE IF NOT EXISTS test_ks WITH REPLICATION = { \
'class' : 'SimpleStrategy', 'replication_factor' : 1 };";
session
.query(create_ks)
.await
.expect("Keyspace creation error");
}
async fn create_table(session: &mut CurrentSession) {
let create_table_cql =
"CREATE TABLE IF NOT EXISTS test_ks.my_test_table (key int PRIMARY KEY);";
session
.query(create_table_cql)
.await
.expect("Table creation error");
}
async fn insert_row(session: &mut CurrentSession, row: RowStruct, prepared_query: &PreparedQuery) {
session
.exec_with_values(prepared_query, row.into_query_values())
.await
.expect("exec_with_values error");
}
async fn batch_few_queries(session: &mut CurrentSession, query: &str) {
let prepared_query = session.prepare(query).await.expect("Prepare query error");
let row_1 = RowStruct { key: 1001 };
let row_2 = RowStruct { key: 2001 };
let batch = BatchQueryBuilder::new()
.add_query_prepared(&prepared_query, row_1.into_query_values())
.add_query(query, row_2.into_query_values())
.build()
.expect("batch builder");
session.batch(batch).await.expect("batch query error");
}