diff --git a/Cargo.toml b/Cargo.toml index 60b531c..b1be495 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -150,6 +150,7 @@ azure_storage_blobs = "0.21.0" azure_storage = "0.21.0" base64 = "0.22.1" fantoccini = "0.21" +cdrs-tokio = "9.0.2" [[example]] name = "anvil" diff --git a/src/cassandra/mod.rs b/src/cassandra/mod.rs index fac6528..182e50b 100644 --- a/src/cassandra/mod.rs +++ b/src/cassandra/mod.rs @@ -1,3 +1,5 @@ +use std::borrow::Cow; + use testcontainers::{core::WaitFor, Image}; const NAME: &str = "cassandra"; @@ -11,27 +13,16 @@ const TAG: &str = "5.0.6"; /// ``` /// use std::time::Duration; /// -/// use scylla::client::{session::Session, session_builder::SessionBuilder}; /// use testcontainers::{runners::AsyncRunner, ImageExt}; /// /// #[tokio::test] /// async fn default_cassandra() -> Result<(), Box> { -/// let image = ScyllaDB::default(); +/// let image = Cassandra::default().with_startup_timeout(Duration::from_secs(120)); /// let instance = image.start().await?; /// let host = instance.get_host().await?; /// let port = instance.get_host_port_ipv4(9042).await?; /// let hostname = format!("{host}:{port}"); -/// let session: Session = SessionBuilder::new().known_node(hostname).build().await?; -/// -/// let prepared_statement = session -/// .prepare("SELECT release_version FROM system.local") -/// .await?; -/// let rows = session -/// .execute_unpaged(&prepared_statement, &[]) -/// .await? -/// .into_rows_result()?; -/// let (version,) = rows.single_row::<(String,)>()?; -/// assert_eq!(version, "5.0.6"); +/// // do something using a driver /// Ok(()) /// } /// ``` @@ -50,6 +41,22 @@ impl Image for Cassandra { TAG } + fn env_vars( + &self, + ) -> impl IntoIterator>, impl Into>)> { + [ + ( + "JVM_EXTRA_OPTS", + "-Dcassandra.skip_wait_for_gossip_to_settle=0 -Dcassandra.initial_token=0", + ), + ("CASSANDRA_DC", "dc1"), + ("CASSANDRA_SNITCH", "GossipingPropertyFileSnitch"), + ("CASSANDRA_ENDPOINT_SNITCH", "GossipingPropertyFileSnitch"), + ("HEAP_NEWSIZE", "128M"), + ("MAX_HEAP_SIZE", "1024M"), + ] + } + fn ready_conditions(&self) -> Vec { vec![WaitFor::message_on_either_std("Startup complete")] } @@ -59,29 +66,50 @@ impl Image for Cassandra { mod tests { use std::time::Duration; - use scylla::client::{session::Session, session_builder::SessionBuilder}; + use cdrs_tokio::{ + cluster::{ + session::{SessionBuilder, TcpSessionBuilder}, + NodeTcpConfigBuilder, + }, + load_balancing::RoundRobinLoadBalancingStrategy, + types::ByName, + }; use testcontainers::{runners::AsyncRunner, ImageExt}; use super::*; #[tokio::test] async fn cassandra_select_version() -> Result<(), Box> { - let image = Cassandra::default().with_startup_timeout(Duration::from_secs(240)); + pretty_env_logger::init(); + let image = Cassandra::default().with_startup_timeout(Duration::from_secs(120)); let instance = image.start().await?; let host = instance.get_host().await?; let port = instance.get_host_port_ipv4(9042).await?; let hostname = format!("{host}:{port}"); - let session: Session = SessionBuilder::new().known_node(hostname).build().await?; - let prepared_statement = session - .prepare("SELECT release_version FROM system.local") + let cluster_config = NodeTcpConfigBuilder::new() + .with_contact_point(hostname.into()) + .build() .await?; - let rows = session - .execute_unpaged(&prepared_statement, &[]) - .await? - .into_rows_result()?; - let (version,) = rows.single_row::<(String,)>()?; - assert_eq!(version, "5.0.6"); + + let session = + TcpSessionBuilder::new(RoundRobinLoadBalancingStrategy::new(), cluster_config) + .build() + .await?; + + let result = session + .query("SELECT release_version FROM system.local") + .await?; + + let body = result.response_body()?; + let rows = body.into_rows().unwrap(); + let version = rows + .first() + .unwrap() + .by_name::("release_version")? + .unwrap(); + + assert_eq!(version, TAG); Ok(()) } }