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
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577

use cassandra::future::CassFuture;
use cassandra::policy::retry::RetryPolicy;
use cassandra::session::Session;
use cassandra::ssl::Ssl;
use cassandra::time::TimestampGen;
use cassandra::util::Protected;
use cassandra::error::*;

use cassandra_sys::CassCluster as _Cluster;
use cassandra_sys::cass_bool_t;
use cassandra_sys::cass_cluster_free;
use cassandra_sys::cass_cluster_new;
use cassandra_sys::cass_cluster_set_connect_timeout;
use cassandra_sys::cass_cluster_set_connection_heartbeat_interval;
use cassandra_sys::cass_cluster_set_connection_idle_timeout;
use cassandra_sys::cass_cluster_set_contact_points;
use cassandra_sys::cass_cluster_set_core_connections_per_host;
use cassandra_sys::cass_cluster_set_credentials;
use cassandra_sys::cass_cluster_set_latency_aware_routing;
use cassandra_sys::cass_cluster_set_latency_aware_routing_settings;
use cassandra_sys::cass_cluster_set_load_balance_dc_aware;
use cassandra_sys::cass_cluster_set_load_balance_round_robin;
use cassandra_sys::cass_cluster_set_max_concurrent_creation;
use cassandra_sys::cass_cluster_set_max_concurrent_requests_threshold;
use cassandra_sys::cass_cluster_set_max_connections_per_host;
use cassandra_sys::cass_cluster_set_max_requests_per_flush;
use cassandra_sys::cass_cluster_set_num_threads_io;
use cassandra_sys::cass_cluster_set_pending_requests_high_water_mark;
use cassandra_sys::cass_cluster_set_pending_requests_low_water_mark;
use cassandra_sys::cass_cluster_set_port;
use cassandra_sys::cass_cluster_set_protocol_version;
use cassandra_sys::cass_cluster_set_queue_size_event;
use cassandra_sys::cass_cluster_set_queue_size_io;
use cassandra_sys::cass_cluster_set_reconnect_wait_time;
use cassandra_sys::cass_cluster_set_request_timeout;
use cassandra_sys::cass_cluster_set_retry_policy;
use cassandra_sys::cass_cluster_set_ssl;
use cassandra_sys::cass_cluster_set_tcp_keepalive;
use cassandra_sys::cass_cluster_set_tcp_nodelay;
use cassandra_sys::cass_cluster_set_timestamp_gen;
use cassandra_sys::cass_cluster_set_token_aware_routing;
use cassandra_sys::cass_cluster_set_use_schema;
use cassandra_sys::cass_cluster_set_whitelist_filtering;
use cassandra_sys::cass_cluster_set_write_bytes_high_water_mark;
use cassandra_sys::cass_cluster_set_write_bytes_low_water_mark;
use cassandra_sys::cass_false;
use cassandra_sys::cass_future_error_code;
use cassandra_sys::cass_session_connect;
use cassandra_sys::cass_session_new;
use cassandra_sys::cass_true;

use std::ffi::{CString, CStr};
use std::ffi::NulError;
use std::fmt;
use std::iter::Map;
use std::net::AddrParseError;
use std::net::Ipv4Addr;
use std::result;
use std::fmt::Display;
use std::str::FromStr;
use time::Duration;
use futures::Future;

/// A CQL protocol version is just an integer.
pub type CqlProtocol = i32;

///
/// The main class to use when interacting with a Cassandra cluster.
/// Typically, one instance of this class will be created for each separate
/// Cassandra cluster that your application interacts with.
///
/// # Examples
/// ```
/// use cassandra_cpp::Cluster;
/// let mut cluster = Cluster::default();
/// cluster.set_contact_points("127.0.0.1").unwrap();
/// let _session = cluster.connect().unwrap();
/// ```
#[derive(Debug)]
pub struct Cluster(pub *mut _Cluster);

// The underlying C type has no thread-local state, but does not support access
// from multiple threads: https://datastax.github.io/cpp-driver/topics/#thread-safety
unsafe impl Send for Cluster {}

impl Drop for Cluster {
    /// Frees a cluster instance.
    fn drop(&mut self) { unsafe { cass_cluster_free(self.0) } }
}

impl Protected<*mut _Cluster> for Cluster {
    fn inner(&self) -> *mut _Cluster { self.0 }
    fn build(inner: *mut _Cluster) -> Self { Cluster(inner) }
}

impl Default for Cluster {
    /// Creates a new cluster
    fn default() -> Cluster { unsafe { Cluster(cass_cluster_new()) } }
}

impl Cluster {
    /// Sets/Appends contact points. This *MUST* be set. The first call sets
    /// the contact points and any subsequent calls appends additional contact
    /// points. Passing an empty string will clear the contact points. White space
    /// is stripped from the contact points.
    ///
    ///
    /// {contact points: "127.0.0.1" "127.0.0.1,127.0.0.2", "server1.domain.com"}
    ///
    ///
    pub fn set_contact_points(&mut self, contact_points: &str) -> Result<&mut Self> {
        unsafe {
            let s = CString::new(contact_points.clone())?;
            let err = cass_cluster_set_contact_points(self.0, s.as_ptr());
            err.to_result(self)
        }
    }


    /// Sets the port
    ///
    ///
    /// Default: 9042
    ///
    pub fn set_port(&mut self, port: u16) -> Result<&mut Self> {
        unsafe { cass_cluster_set_port(self.0, port as i32).to_result(self) }
    }


    /// Sets the SSL context and enables SSL
    pub fn set_ssl(&mut self, ssl: &mut Ssl) -> &Self {
        unsafe {
            cass_cluster_set_ssl(self.0, ssl.inner());
            self
        }
    }

    /// Performs a blocking call to connect to Cassandra cluster
    pub fn connect(&mut self) -> Result<Session> {
        unsafe {
            let session = Session(cass_session_new());
            let connect_future = <CassFuture<()>>::build(cass_session_connect(session.0, self.0));
            connect_future.wait().map(|_| session)
        }
    }

    /// Sets the protocol version. This will automatically downgrade to the lowest
    /// supported protocol version.
    ///
    ///
    /// Default: version 4
    ///
    pub fn set_protocol_version(&mut self, protocol_version: CqlProtocol) -> Result<&mut Self> {
        unsafe {
            cass_cluster_set_protocol_version(self.0, protocol_version as i32)
                .to_result(self)
        }
    }

    /// Sets the number of IO threads. This is the number of threads
    /// that will handle query requests.
    ///
    ///
    /// Default: 1
    ///
    pub fn set_num_threads_io(&mut self, num_threads: u32) -> Result<&mut Self> {
        unsafe {
            cass_cluster_set_num_threads_io(self.0, num_threads)
                .to_result(self)
        }
    }

    /// Sets the size of the fixed size queue that stores pending requests.
    ///
    ///
    /// Default: 8192
    ///
    pub fn set_queue_size_io(&mut self, queue_size: u32) -> Result<&mut Self> {
        unsafe {
            cass_cluster_set_queue_size_io(self.0, queue_size)
                .to_result(self)
        }
    }

    /// Sets the size of the fixed size queue that stores events.
    ///
    ///
    /// Default: 8192
    ///
    pub fn set_queue_size_event(&mut self, queue_size: u32) -> Result<&mut Self> {
        unsafe {
            cass_cluster_set_queue_size_event(self.0, queue_size)
                .to_result(self)
        }
    }

    /// Sets the number of connections made to each server in each
    /// IO thread.
    ///
    ///
    /// Default: 1
    ///
    pub fn set_core_connections_per_host(&mut self, num_connections: u32) -> Result<&mut Self> {
        unsafe {
            cass_cluster_set_core_connections_per_host(self.0, num_connections)
                .to_result(self)
        }
    }

    /// Sets the maximum number of connections made to each server in each
    /// IO thread.
    ///
    ///
    /// Default: 2
    ///
    pub fn set_max_connections_per_host(&mut self, num_connections: u32) -> Result<&mut Self> {
        unsafe {
            cass_cluster_set_max_connections_per_host(self.0, num_connections)
                .to_result(self)
        }
    }

    /// Sets the amount of time to wait before attempting to reconnect.
    ///
    ///
    /// Default: 1000ms
    ///
    pub fn set_reconnect_wait_time(&mut self, wait_time: u32) -> &Self {
        unsafe {
            cass_cluster_set_reconnect_wait_time(self.0, wait_time);
        }
        self
    }

    /// Sets the maximum number of connections that will be created concurrently.
    /// Connections are created when the current connections are unable to keep up with
    /// request throughput.
    ///
    ///
    /// Default: 1
    pub fn set_max_concurrent_creation(&mut self, num_connections: u32) -> Result<&mut Self> {
        unsafe {
            cass_cluster_set_max_concurrent_creation(self.0, num_connections)
                .to_result(self)
        }
    }

    /// Sets the threshold for the maximum number of concurrent requests in-flight
    /// on a connection before creating a new connection. The number of new connections
    /// created will not exceed max_connections_per_host.
    ///
    ///
    /// Default: 100
    pub fn set_max_concurrent_requests_threshold(&mut self, num_requests: u32) -> Result<&mut Self> {
        unsafe {
            cass_cluster_set_max_concurrent_requests_threshold(self.0, num_requests)
                .to_result(self)
        }
    }

    /// Sets the maximum number of requests processed by an IO worker
    /// per flush.
    ///
    ///
    /// Default: 128
    pub fn set_max_requests_per_flush(&mut self, num_requests: u32) -> Result<&mut Self> {
        unsafe {
            cass_cluster_set_max_requests_per_flush(self.0, num_requests)
                .to_result(self)
        }
    }

    /// Sets the high water mark for the number of bytes outstanding
    /// on a connection. Disables writes to a connection if the number
    /// of bytes queued exceed this value.
    ///
    ///
    /// Default: 64KB
    pub fn set_write_bytes_high_water_mark(&mut self, num_bytes: u32) -> Result<&mut Self> {
        unsafe {
            cass_cluster_set_write_bytes_high_water_mark(self.0, num_bytes)
                .to_result(self)
        }
    }

    /// Sets the low water mark for the number of bytes outstanding
    /// on a connection. Disables writes to a connection if the number
    /// of bytes queued fall below this value.
    ///
    ///
    /// Default: 32KB
    pub fn set_write_bytes_low_water_mark(&mut self, num_bytes: u32) -> Result<&mut Self> {
        unsafe {
            cass_cluster_set_write_bytes_low_water_mark(self.0, num_bytes)
                .to_result(self)
        }
    }

    /// Sets the high water mark for the number of requests queued waiting
    /// for a connection in a connection pool. Disables writes to a
    /// host on an IO worker if the number of requests queued exceed this
    /// value.
    ///
    ///
    /// Default: 256
    pub fn set_pending_requests_high_water_mark(&mut self, num_requests: u32) -> Result<&mut Self> {
        unsafe {
            cass_cluster_set_pending_requests_high_water_mark(self.0, num_requests)
                .to_result(self)
        }
    }

    /// Sets the low water mark for the number of requests queued waiting
    /// for a connection in a connection pool. After exceeding high water mark
    /// requests, writes to a host will only resume once the number of requests
    /// fall below this value.
    ///
    ///
    /// Default: 128
    pub fn set_pending_requests_low_water_mark(&mut self, num_requests: u32) -> Result<&mut Self> {
        unsafe {
            cass_cluster_set_pending_requests_low_water_mark(self.0, num_requests)
                .to_result(self)
        }
    }

    /// Sets the timeout for connecting to a node.
    ///
    ///
    /// Default: 5000ms
    #[allow(cast_possible_truncation,cast_sign_loss)]
    pub fn set_connect_timeout(&mut self, timeout: Duration) -> &Self {
        unsafe {
            cass_cluster_set_connect_timeout(self.0, timeout.num_milliseconds() as u32);
        }
        self
    }

    /// Sets the timeout for waiting for a response from a node.
    ///
    ///
    /// Default: 12000ms
    #[allow(cast_possible_truncation,cast_sign_loss)]
    pub fn set_request_timeout(&mut self, timeout: Duration) -> &Self {
        unsafe {
            cass_cluster_set_request_timeout(self.0, timeout.num_milliseconds() as u32);
        }
        self
    }

    /// Sets credentials for plain text authentication.
    pub fn set_credentials(&mut self, username: &str, password: &str) -> Result<&mut Self> {
        unsafe {
            cass_cluster_set_credentials(self.0,
                                         CString::new(username)?.as_ptr(),
                                         CString::new(password)?.as_ptr());
        }
        Ok(self)
    }

    /// Configures the cluster to use round-robin load balancing.
    ///
    /// The driver discovers all nodes in a cluster and cycles through
    /// them per request. All are considered 'local'.
    pub fn set_load_balance_round_robin(&mut self) -> &Self {
        unsafe {
            cass_cluster_set_load_balance_round_robin(self.0);
            self

        }
    }

    /// Configures the cluster to use DC-aware load balancing.
    /// For each query, all live nodes in a primary 'local' DC are tried first,
    /// followed by any node from other DCs.
    ///
    /// <b>Note:</b> This is the default, and does not need to be called unless
    /// switching an existing from another policy or changing settings.
    /// Without further configuration, a default local_dc is chosen from the
    /// first connected contact point, and no remote hosts are considered in
    /// query plans. If relying on this mechanism, be sure to use only contact
    /// points from the local DC.
    pub fn set_load_balance_dc_aware<S>(&mut self, local_dc: &str, used_hosts_per_remote_dc: u32,
        allow_remote_dcs_for_local_cl: cass_bool_t)
                                        -> Result<&mut Self> {
        unsafe {
            {
                    let local_dc = CString::new(local_dc)?;
                    cass_cluster_set_load_balance_dc_aware(self.0,
                                                           local_dc.as_ptr(),
                                                           used_hosts_per_remote_dc,
                                                           allow_remote_dcs_for_local_cl)
                }
                .to_result(self)
        }
    }

    /// Configures the cluster to use token-aware request routing or not.
    ///
    /// <b>Important:</b> Token-aware routing depends on keyspace information.
    /// For this reason enabling token-aware routing will also enable the usage
    /// of schema metadata.
    ///
    ///
    /// Default: true (enabled).
    ///
    ///
    /// This routing policy composes the base routing policy, routing
    /// requests first to replicas on nodes considered 'local' by
    /// the base load balancing policy.
    pub fn set_token_aware_routing(&mut self, enabled: bool) -> &Self {
        unsafe {
            cass_cluster_set_token_aware_routing(self.0, if enabled { cass_true } else { cass_false });
        }
        self
    }

    /// Configures the cluster to use latency-aware request routing or not.
    ///
    ///
    /// Default: false (disabled).
    ///
    ///
    /// This routing policy is a top-level routing policy. It uses the
    /// base routing policy to determine locality (dc-aware) and/or
    /// placement (token-aware) before considering the latency.
    pub fn set_latency_aware_routing(&mut self, enabled: bool) -> &Self {
        unsafe {
            cass_cluster_set_latency_aware_routing(self.0, if enabled { cass_true } else { cass_false });
        }
        self
    }


    /// Configures the settings for latency-aware request routing.
    ///
    ///
    /// Defaults:
    ///
    /// <ul>
    ///   <li>exclusion_threshold: 2.0</li>
    ///   <li>scale_ms: 100 milliseconds</li>
    ///  <li>retry_period_ms: 10,000 milliseconds (10 seconds)</li>
    ///  <li>update_rate_ms: 100 milliseconds</li>
    ///  <li>min_measured: 50</li>
    /// </ul>
    #[allow(cast_sign_loss)]
    pub fn set_latency_aware_routing_settings(&mut self, exclusion_threshold: f64, scale: Duration,
        retry_period: Duration, update_rate: Duration, min_measured: u64)
                                              -> &Self {
        unsafe {
            cass_cluster_set_latency_aware_routing_settings(self.0,
                                                            exclusion_threshold,
                                                            scale.num_milliseconds() as u64,
                                                            retry_period.num_milliseconds() as u64,
                                                            update_rate.num_milliseconds() as u64,
                                                            min_measured);
        }
        self
    }

    /// /Sets/Appends whitelist hosts. The first call sets the whitelist hosts and
    /// any subsequent calls appends additional hosts. Passing an empty string will
    /// clear and disable the whitelist. White space is striped from the hosts.
    ///
    /// This policy filters requests to all other policies, only allowing requests
    /// to the hosts contained in the whitelist. Any host not in the whitelist will
    /// be ignored and a connection will not be established. This policy is useful
    /// for ensuring that the driver will only connect to a predefined set of hosts.
    ///
    ///
    /// Examples: "127.0.0.1" "127.0.0.1,127.0.0.2", "server1.domain.com"
    pub fn set_whitelist_filtering(&mut self, hosts: Vec<String>) -> &Self {
        unsafe {
            cass_cluster_set_whitelist_filtering(self.0, hosts.join(",").as_ptr() as *const i8);
        }
        self
    }

    /// Enable/Disable Nagel's algorithm on connections.
    ///
    ///
    /// <b>Default:</b> true (disables Nagel's algorithm).
    pub fn set_tcp_nodelay(&mut self, enable: bool) -> &Self {
        unsafe {
            cass_cluster_set_tcp_nodelay(self.0, if enable { cass_true } else { cass_false });
        }
        self
    }

    /// Enable/Disable TCP keep-alive
    ///
    ///
    /// Default: false (disabled).
    #[allow(cast_possible_truncation,cast_sign_loss)]
    pub fn set_tcp_keepalive(&mut self, enable: bool, delay: Duration) -> &Self {
        unsafe {
            cass_cluster_set_tcp_keepalive(self.0,
                                           if enable { cass_true } else { cass_false },
                                           delay.num_seconds() as u32);
        }
        self
    }

    /// Sets the timestamp generator used to assign timestamps to all requests
    /// unless overridden by setting the timestamp on a statement or a batch.
    ///
    ///
    /// Default: server-side timestamp generator.
    pub fn set_timestamp_gen(&mut self, tsg: &TimestampGen) -> &mut Self {
        unsafe {
            cass_cluster_set_timestamp_gen(self.0, TimestampGen::inner(tsg));
            self
        }
    }

    /// Sets the amount of time between heartbeat messages and controls the amount
    /// of time the connection must be idle before sending heartbeat messages. This
    /// is useful for preventing intermediate network devices from dropping
    /// connections.
    ///
    ///
    /// Default: 30 seconds
    #[allow(cast_possible_truncation,cast_sign_loss)]
    pub fn set_connection_heartbeat_interval(&mut self, hearbeat: Duration) -> &mut Self {
        unsafe {
            cass_cluster_set_connection_heartbeat_interval(self.0, hearbeat.num_seconds() as u32);
            self
        }
    }

    /// Sets the amount of time a connection is allowed to be without a successful
    /// heartbeat response before being terminated and scheduled for reconnection.
    ///
    ///
    /// Default: 60 seconds
    #[allow(cast_possible_truncation,cast_sign_loss)]
    pub fn set_connection_idle_timeout(&mut self, timeout: Duration) -> &mut Self {
        unsafe {
            cass_cluster_set_connection_idle_timeout(self.0, timeout.num_seconds() as u32);
            self
        }
    }

    /// Sets the retry policy used for all requests unless overridden by setting
    /// a retry policy on a statement or a batch.
    ///
    ///
    /// Default: The same policy as would be created by the function:
    /// cass_retry_policy_default_new(). This policy will retry on a read timeout
    /// if there was enough replicas, but no data present, on a write timeout if a
    /// logged batch request failed to write the batch log, and on a unavailable
    /// error it retries using a new host. In all other cases the default policy
    /// will return an error.
    pub fn set_retry_policy(&mut self, retry_policy: RetryPolicy) -> &mut Self {
        unsafe {
            cass_cluster_set_retry_policy(self.0, retry_policy.inner());
            self
        }
    }

    /// Enable/Disable retrieving and updating schema metadata. If disabled
    /// this is allows the driver to skip over retrieving and updating schema
    /// metadata, but it also disables the usage of token-aware routing and
    /// cass_session_get_schema() will always return an empty object. This can be
    /// useful for reducing the startup overhead of short-lived sessions.
    ///
    ///
    /// Default: true (enabled).
    pub fn set_use_schema(&mut self, enabled: bool) -> &Self {
        unsafe {
            cass_cluster_set_use_schema(self.0, if enabled { cass_true } else { cass_false });
        }
        self
    }
}