cdrs 2.2.0

Cassandra DB driver written in Rust
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79

use openssl::ssl::SslConnector;
use r2d2::{Builder, ManageConnection, Pool};
use std::cell::RefCell;
use std::error::Error;
use std::io::Write;

use crate::authenticators::Authenticator;
use crate::cluster::{startup, NodeSslConfig};
use crate::compression::Compression;
use crate::error;
use crate::frame::parser::parse_frame;
use crate::frame::{Frame, IntoBytes};
use crate::transport::CDRSTransport;
use crate::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, A: Authenticator + Send + Sync + 'static>(
  node_config: NodeSslConfig<'a, 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: String,
  ssl_connector: SslConnector,
  auth: A,
}

impl<A> SslConnectionsManager<A> {
  pub fn new<S: ToString>(addr: S, auth: A, ssl_connector: SslConnector) -> Self {
    SslConnectionsManager {
      addr: addr.to_string(),
      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();
    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()
  }
}