rbdc_oracle/connection/
mod.rs

1use futures_core::future::BoxFuture;
2use futures_util::future;
3use rbdc::Error;
4use rbdc::StatementCache;
5use std::fmt::{self, Debug, Formatter};
6use std::sync::atomic::Ordering;
7
8pub(crate) use handle::ConnectionHandle;
9
10use crate::OracleConnectOptions;
11use crate::connection::establish::EstablishParams;
12use crate::connection::worker::ConnectionWorker;
13use crate::statement::VirtualStatement;
14
15mod establish;
16mod execute;
17mod executor;
18mod handle;
19mod worker;
20
21pub use worker::Command;
22
23pub struct OracleConnection {
24    pub(crate) worker: ConnectionWorker,
25    pub(crate) row_channel_size: usize,
26}
27
28unsafe impl Sync for OracleConnection {}
29
30pub struct ConnectionState {
31    pub(crate) handle: ConnectionHandle,
32    pub(crate) transaction_active: bool,
33    pub(crate) statements: Statements,
34}
35
36pub(crate) struct Statements {
37    cached: StatementCache<VirtualStatement>,
38    temp: Option<VirtualStatement>,
39}
40
41impl OracleConnection {
42    pub(crate) async fn establish(options: &OracleConnectOptions) -> Result<Self, Error> {
43        let params = EstablishParams::from_options(options)?;
44        let worker = ConnectionWorker::establish(params).await?;
45        Ok(Self {
46            worker,
47            row_channel_size: options.row_channel_size,
48        })
49    }
50}
51
52impl Debug for OracleConnection {
53    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
54        f.debug_struct("OracleConnection")
55            .field("row_channel_size", &self.row_channel_size)
56            .field("cached_statements_size", &self.cached_statements_size())
57            .finish()
58    }
59}
60
61impl OracleConnection {
62    pub async fn do_close(&mut self) -> Result<(), Error> {
63        self.worker.shutdown().await
64    }
65
66    pub fn ping(&mut self) -> BoxFuture<'_, Result<(), Error>> {
67        Box::pin(self.worker.ping())
68    }
69
70    pub fn cached_statements_size(&self) -> usize {
71        self.worker
72            .shared
73            .cached_statements_size
74            .load(Ordering::Acquire)
75    }
76
77    pub fn clear_cached_statements(&mut self) -> BoxFuture<'_, Result<(), Error>> {
78        Box::pin(async move {
79            self.worker.clear_cache().await?;
80            Ok(())
81        })
82    }
83
84    #[doc(hidden)]
85    pub fn flush(&mut self) -> BoxFuture<'_, Result<(), Error>> {
86        Box::pin(future::ok(()))
87    }
88
89    #[doc(hidden)]
90    pub fn should_flush(&self) -> bool {
91        false
92    }
93}
94
95impl Drop for ConnectionState {
96    fn drop(&mut self) {
97        self.statements.clear();
98    }
99}
100
101impl Statements {
102    fn new(capacity: usize) -> Self {
103        Statements {
104            cached: StatementCache::new(capacity),
105            temp: None,
106        }
107    }
108
109    fn get(&mut self, query: &str, persistent: bool) -> Result<&mut VirtualStatement, Error> {
110        if !persistent || !self.cached.is_enabled() {
111            return Ok(self.temp.insert(VirtualStatement::new(query, false)?));
112        }
113
114        let exists = self.cached.contains_key(query);
115
116        if !exists {
117            let statement = VirtualStatement::new(query, true)?;
118            self.cached.insert(query, statement);
119        }
120
121        let statement = self.cached.get_mut(query).unwrap();
122        if exists {
123            statement.reset()?;
124        }
125
126        Ok(statement)
127    }
128
129    fn len(&self) -> usize {
130        self.cached.len()
131    }
132
133    fn clear(&mut self) {
134        self.cached.clear();
135        self.temp = None;
136    }
137}
rbdc_oracle/connection/mod.rs

rbdc_oracle/connection/
mod.rs
