voltdb-client-rust 0.1.11

A socket client library for Voltdb server
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
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

use std::{
    fmt,
    sync::{
        Arc,
        atomic::{AtomicUsize, Ordering},
    },
};
use std::time::SystemTime;

use crate::{block_for_result, Node, node, NodeOpt, Opts, Value, VoltError, VoltTable};

#[derive(Debug)]
struct InnerPool {
    opts: Opts,
    pool: Vec<Node>,
}


impl InnerPool {
    pub fn node_sizes(&self) -> usize {
        return self.opts.0.ip_ports.len();
    }

    fn to_node_opt(&self, i: usize) -> NodeOpt {
        return NodeOpt {
            ip_port: self.opts.0.ip_ports.get(i).cloned().unwrap(),
            pass: self.opts.0.pass.clone(),
            user: self.opts.0.user.clone(),
        };
    }
    fn get_node(&mut self, idx: usize) -> &mut Node {
        return self.pool.get_mut(idx).unwrap();
    }
    fn new(size: usize, opts: Opts) -> Result<InnerPool, VoltError> {
        let mut pool = InnerPool {
            opts,
            pool: Vec::with_capacity(size),
        };
        let total = pool.node_sizes();
        for i in 0..size {
            let z = i % total;
            pool.new_conn(z)?;
        }
        Ok(pool)
    }
    fn new_conn(&mut self, idx: usize) -> Result<(), VoltError> {
        match node::Node::new(self.to_node_opt(idx)) {
            Ok(conn) => {
                self.pool.push(conn);
                Ok(())
            }
            Err(err) => Err(err),
        }
    }
}


pub struct Pool {
    size: usize,
    total: Arc<AtomicUsize>,
    inner_pool: InnerPool,
}

impl fmt::Debug for Pool {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(
            f,
            "Pool total: {}, size: {}",
            self.total.load(Ordering::Relaxed),
            self.size
        )
    }
}

impl Pool {
    fn _get_conn(&mut self) -> Result<PooledConn, VoltError> {
        let total = self.total.fetch_add(1, Ordering::Relaxed);
        let idx = total % self.size;
        Ok(PooledConn {
            created: SystemTime::now(),
            conn: self.inner_pool.get_node(idx),
        })
    }

    pub fn new<T: Into<Opts>>(opts: T) -> Result<Pool, VoltError> {
        Pool::new_manual(10, opts)
    }

    pub fn new_manual<T: Into<Opts>>(size: usize, opts: T) -> Result<Pool, VoltError> {
        let pool = InnerPool::new(size, opts.into())?;
        Ok(Pool {
            inner_pool: pool,
            size,
            total: Arc::new(AtomicUsize::from(0 as usize)),

        })
    }

    pub fn get_conn(&mut self) -> Result<PooledConn, VoltError> {
        self._get_conn()
    }
}

#[derive(Debug)]
pub struct PooledConn<'a> {
    created: SystemTime,
    conn: &'a mut Node,
}

impl<'a> Drop for PooledConn<'a> {
    fn drop(&mut self) {
//        let since = SystemTime::now().duration_since(self.created);
        // TODO record error ,
        //   println!("used {:?} ", since)
    }
}

impl<'a> PooledConn<'a> {
    pub fn query(&mut self, sql: &str) -> Result<VoltTable, VoltError> {
        return block_for_result(&self.conn.query(sql)?);
    }
    pub fn list_procedures(&mut self) -> Result<VoltTable, VoltError> {
        return block_for_result(&self.conn.list_procedures()?);
    }
    pub fn call_sp(&mut self, query: &str, param: Vec<&dyn Value>) -> Result<VoltTable, VoltError> {
        return block_for_result(&self.conn.call_sp(query, param)?);
    }
    pub fn upload_jar(&mut self, bs: Vec<u8>) -> Result<VoltTable, VoltError> {
        return block_for_result(&self.conn.upload_jar(bs)?);
    }
}