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

#![allow(non_camel_case_types)]
#![allow(dead_code)]
#![allow(missing_copy_implementations)]

use crate::cassandra::batch::Batch;
use crate::cassandra::cluster::Cluster;
use crate::cassandra::custom_payload::CustomPayloadResponse;
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;
use crate::CustomPayload;

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;

/// 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)]
pub struct Session(pub *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 Send for Session {}
unsafe impl Sync for Session {}

impl Protected<*mut _Session> for Session {
    fn inner(&self) -> *mut _Session {
        self.0
    }
    fn build(inner: *mut _Session) -> Self {
        if inner.is_null() {
            panic!("Unexpected null pointer")
        };
        Session(inner)
    }
}

impl Drop for Session {
    /// 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.0) }
    }
}

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

impl Session {
    /// Create a new Cassanda session.
    /// It's recommended to use Cluster.connect() instead
    pub fn new() -> Session {
        unsafe { Session(cass_session_new()) }
    }

    //    pub fn new2() -> *mut _Session {
    //        unsafe { cass_session_new() }
    //    }

    /// Connects a session.
    pub fn connect(self, cluster: &Cluster) -> CassFuture<()> {
        unsafe { <CassFuture<()>>::build(cass_session_connect(self.0, cluster.inner())) }
    }

    /// Connects a session and sets the keyspace.
    pub fn connect_keyspace(&self, cluster: &Cluster, keyspace: &str) -> Result<CassFuture<()>> {
        unsafe {
            let keyspace_ptr = keyspace.as_ptr() as *const c_char;
            Ok(<CassFuture<()>>::build(cass_session_connect_keyspace_n(
                self.0,
                cluster.inner(),
                keyspace_ptr,
                keyspace.len(),
            )))
        }
    }

    /// Closes the session instance, outputs a close future which can
    /// be used to determine when the session has been terminated. This allows
    /// in-flight requests to finish.
    pub fn close(self) -> CassFuture<()> {
        unsafe { <CassFuture<()>>::build(cass_session_close(self.0)) }
    }

    /// Create a prepared statement.
    pub fn prepare(&self, query: &str) -> Result<CassFuture<PreparedStatement>> {
        unsafe {
            let query_ptr = query.as_ptr() as *const c_char;
            Ok(<CassFuture<PreparedStatement>>::build(
                cass_session_prepare_n(self.0, query_ptr, query.len()),
            ))
        }
    }

    //    ///Execute a query or bound statement.
    //    pub fn execute(&self, statement: &str, parameter_count: u64) -> CassFuture {
    //        unsafe {
    //            CassFuture::build(cass_session_execute(self.0,
    //            Statement::new(statement, parameter_count).inner()))
    //        }
    //    }

    /// Execute a batch statement.
    pub fn execute_batch(&self, batch: &Batch) -> CassFuture<CassResult> {
        <CassFuture<CassResult>>::build(unsafe {
            cass_session_execute_batch(self.0, batch.inner())
        })
    }

    /// Execute a batch statement and get any custom payloads from the response.
    pub fn execute_batch_with_payloads(
        &self,
        batch: &Batch,
    ) -> CassFuture<(CassResult, CustomPayloadResponse)> {
        <CassFuture<(CassResult, CustomPayloadResponse)>>::build(unsafe {
            cass_session_execute_batch(self.0, batch.inner())
        })
    }

    /// Execute a statement.
    pub fn execute(&self, statement: &Statement) -> CassFuture<CassResult> {
        unsafe { <CassFuture<CassResult>>::build(cass_session_execute(self.0, statement.inner())) }
    }

    /// Execute a statement and get any custom payloads from the response.
    pub fn execute_with_payloads(
        &self,
        statement: &Statement,
    ) -> CassFuture<(CassResult, CustomPayloadResponse)> {
        unsafe {
            <CassFuture<(CassResult, CustomPayloadResponse)>>::build(cass_session_execute(
                self.0,
                statement.inner(),
            ))
        }
    }

    /// 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.0)) }
    }

    /// 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.0, &mut metrics);
            SessionMetrics::build(&metrics)
        }
    }

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