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
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
#![allow(non_camel_case_types)]
#![allow(dead_code)]
#![allow(missing_copy_implementations)]

use crate::cassandra::cluster::Cluster;
use crate::cassandra::error::*;
use crate::cassandra::future::CassFuture;
use crate::cassandra::metrics::SessionMetrics;
use crate::cassandra::prepared::PreparedStatement;
use crate::cassandra::result::CassResult;
use crate::cassandra::schema::schema_meta::SchemaMeta;
use crate::cassandra::statement::Statement;
use crate::cassandra::util::{Protected, ProtectedInner};
use crate::{cassandra::batch::Batch, BatchType};

use crate::cassandra_sys::cass_session_close;
use crate::cassandra_sys::cass_session_connect;
use crate::cassandra_sys::cass_session_connect_keyspace_n;
use crate::cassandra_sys::cass_session_execute;
use crate::cassandra_sys::cass_session_execute_batch;
use crate::cassandra_sys::cass_session_free;
use crate::cassandra_sys::cass_session_get_metrics;
use crate::cassandra_sys::cass_session_get_schema_meta;
use crate::cassandra_sys::cass_session_new;
use crate::cassandra_sys::cass_session_prepare_n;
use crate::cassandra_sys::CassSession as _Session;

use std::ffi::NulError;
use std::mem;
use std::os::raw::c_char;
use std::sync::Arc;

#[derive(Debug, Eq, PartialEq)]
struct SessionInner<T>(T, *mut _Session);

// The underlying C type has no thread-local state, and explicitly supports access
// from multiple threads: https://datastax.github.io/cpp-driver/topics/#thread-safety
unsafe impl<T: Send> Send for SessionInner<T> {}
unsafe impl<T: Sync> Sync for SessionInner<T> {}

impl<T> SessionInner<T> {
    fn new(bound: T, inner: *mut _Session) -> Arc<Self> {
        Arc::new(Self(bound, inner))
    }
}

/// The session scope defines if the session is bound [`session_scope::Bound`], meaning it is associated
/// with a keyspace.  Or unbound [`session_scope::Unbound`], meaning it is not associated with any keyspace.
///
/// Keyspace association is done by the `connect` and `connect_lazy` methods in [`Cluster`].
pub trait SessionScope: session_scope::sealed::Sealed + Clone + Send + Sync + 'static {}

/// See [`SessionScope`] for more information.
pub mod session_scope {
    pub use super::SessionScope;

    pub(super) mod sealed {
        pub trait Sealed {}
    }

    /// The session is bound to a given keyspace.
    #[derive(Debug, Clone)]
    pub struct Bound {
        pub(crate) keyspace: String,
    }

    impl Bound {
        pub(crate) fn new(keyspace: impl Into<String>) -> Self {
            Self {
                keyspace: keyspace.into(),
            }
        }
    }

    /// The session is not bound to any keyspace in particular.
    #[derive(Debug, Clone)]
    pub struct Unbound;

    impl sealed::Sealed for Bound {}
    impl sealed::Sealed for Unbound {}

    impl super::SessionScope for Bound {}
    impl super::SessionScope for Unbound {}
}
/// A session object is used to execute queries and maintains cluster state through
/// the control connection. The control connection is used to auto-discover nodes and
/// monitor cluster changes (topology and schema). Each session also maintains multiple
/// /pools of connections to cluster nodes which are used to query the cluster.
///
/// Instances of the session object are thread-safe to execute queries.
#[derive(Debug, Clone)]
pub struct Session<T = session_scope::Bound>(Arc<SessionInner<T>>);

impl<T> Eq for Session<T> {}

impl<T> PartialEq for Session<T> {
    fn eq(&self, other: &Self) -> bool {
        (self.0).1 == (other.0).1
    }
}

// The underlying C type has no thread-local state, and explicitly supports access
// from multiple threads: https://datastax.github.io/cpp-driver/topics/#thread-safety
unsafe impl<T: Send> Send for Session<T> {}
unsafe impl<T: Sync> Sync for Session<T> {}

impl<T> ProtectedInner<*mut _Session> for SessionInner<T> {
    fn inner(&self) -> *mut _Session {
        self.1
    }
}

impl<T> ProtectedInner<*mut _Session> for Session<T> {
    fn inner(&self) -> *mut _Session {
        self.0.inner()
    }
}

impl Protected<*mut _Session> for Session<session_scope::Unbound> {
    fn build(inner: *mut _Session) -> Self {
        if inner.is_null() {
            panic!("Unexpected null pointer")
        };
        Session(SessionInner::new(session_scope::Unbound, inner))
    }
}

impl<T> Drop for SessionInner<T> {
    /// Frees a session instance. If the session is still connected it will be synchronously
    /// closed before being deallocated.
    fn drop(&mut self) {
        unsafe { cass_session_free(self.1) }
    }
}

impl Default for Session<session_scope::Unbound> {
    fn default() -> Session<session_scope::Unbound> {
        Session::new(session_scope::Unbound)
    }
}

impl<S> Session<S>
where
    S: SessionScope,
{
    pub(crate) fn new(bound: S) -> Self {
        unsafe { Session(SessionInner::new(bound, cass_session_new())) }
    }

    /// Creates a statement with the given query.
    pub fn statement(&self, query: impl AsRef<str>) -> Statement<S> {
        let query = query.as_ref();
        let param_count = query.matches("?").count();
        Statement::new(self.clone(), query, param_count)
    }

    /// Executes a given query.
    pub async fn execute(&self, query: impl AsRef<str>) -> Result<CassResult> {
        let statement = self.statement(query);
        statement.execute().await
    }

    /// Creates a new batch that is bound to this session.
    pub fn batch(&self, batch_type: BatchType) -> Batch<S> {
        Batch::new(batch_type, self.clone())
    }

    /// Gets a snapshot of this session's schema metadata. The returned
    /// snapshot of the schema metadata is not updated. This function
    /// must be called again to retrieve any schema changes since the
    /// previous call.
    pub fn get_schema_meta(&self) -> SchemaMeta {
        unsafe { SchemaMeta::build(cass_session_get_schema_meta(self.inner())) }
    }

    /// Gets a copy of this session's performance/diagnostic metrics.
    pub fn get_metrics(&self) -> SessionMetrics {
        unsafe {
            let mut metrics = mem::zeroed();
            cass_session_get_metrics(self.inner(), &mut metrics);
            SessionMetrics::build(&metrics)
        }
    }

    /// Create a prepared statement with the given query.
    pub async fn prepare(&self, query: impl AsRef<str>) -> Result<PreparedStatement<S>> {
        let query = query.as_ref();
        let prepare_future = {
            let query_ptr = query.as_ptr() as *const c_char;
            CassFuture::build(self.clone(), unsafe {
                cass_session_prepare_n(self.inner(), query_ptr, query.len())
            })
        };
        prepare_future.await
    }

    //    pub fn get_schema(&self) -> Schema {
    //        unsafe { Schema(cass_session_get_schema(self.0)) }
    //    }
}

impl Session<session_scope::Bound> {
    /// Returns the keyspace that the session is bound to.
    pub fn keyspace(&self) -> &str {
        &(self.0).0.keyspace
    }
}