use deadpool::managed::{self, Manager, Metrics, RecycleError, RecycleResult};
use oracle_rs::{Config, Connection, Error};
use std::time::Duration;
pub struct OracleConnectionManager {
config: Config,
}
impl OracleConnectionManager {
pub fn new(config: Config) -> Self {
Self { config }
}
}
impl Manager for OracleConnectionManager {
type Type = Connection;
type Error = Error;
async fn create(&self) -> Result<Connection, Error> {
Connection::connect_with_config(self.config.clone()).await
}
async fn recycle(
&self,
conn: &mut Connection,
_metrics: &Metrics,
) -> RecycleResult<Error> {
if conn.is_closed() {
return Err(RecycleError::message("connection closed"));
}
conn.rollback().await.ok();
conn.ping().await.map_err(RecycleError::Backend)?;
Ok(())
}
}
pub type Pool = managed::Pool<OracleConnectionManager>;
pub type Object = managed::Object<OracleConnectionManager>;
pub struct PoolBuilder {
config: Config,
max_size: usize,
wait_timeout: Option<Duration>,
create_timeout: Option<Duration>,
recycle_timeout: Option<Duration>,
}
impl PoolBuilder {
pub fn new(config: Config) -> Self {
Self {
config,
max_size: num_cpus() * 4,
wait_timeout: Some(Duration::from_secs(30)),
create_timeout: Some(Duration::from_secs(30)),
recycle_timeout: Some(Duration::from_secs(5)),
}
}
pub fn max_size(mut self, size: usize) -> Self {
self.max_size = size;
self
}
pub fn wait_timeout(mut self, timeout: Option<Duration>) -> Self {
self.wait_timeout = timeout;
self
}
pub fn create_timeout(mut self, timeout: Option<Duration>) -> Self {
self.create_timeout = timeout;
self
}
pub fn recycle_timeout(mut self, timeout: Option<Duration>) -> Self {
self.recycle_timeout = timeout;
self
}
pub fn build(self) -> Result<Pool, BuildError> {
let manager = OracleConnectionManager::new(self.config);
let builder = managed::Pool::builder(manager)
.max_size(self.max_size)
.runtime(deadpool::Runtime::Tokio1)
.timeouts(managed::Timeouts {
wait: self.wait_timeout,
create: self.create_timeout,
recycle: self.recycle_timeout,
});
builder.build().map_err(BuildError)
}
}
#[derive(Debug)]
pub struct BuildError(managed::BuildError);
impl std::fmt::Display for BuildError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "failed to build connection pool: {}", self.0)
}
}
impl std::error::Error for BuildError {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
Some(&self.0)
}
}
pub type PoolError = managed::PoolError<Error>;
fn num_cpus() -> usize {
std::thread::available_parallelism()
.map(|p| p.get())
.unwrap_or(4)
}
pub trait ConfigExt {
fn into_pool(self) -> Result<Pool, BuildError>;
fn into_pool_with_size(self, max_size: usize) -> Result<Pool, BuildError>;
}
impl ConfigExt for Config {
fn into_pool(self) -> Result<Pool, BuildError> {
PoolBuilder::new(self).build()
}
fn into_pool_with_size(self, max_size: usize) -> Result<Pool, BuildError> {
PoolBuilder::new(self).max_size(max_size).build()
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_pool_builder_defaults() {
let config = Config::new("localhost", 1521, "FREEPDB1", "test", "test");
let builder = PoolBuilder::new(config);
assert!(builder.max_size > 0);
assert!(builder.wait_timeout.is_some());
assert!(builder.create_timeout.is_some());
assert!(builder.recycle_timeout.is_some());
}
#[test]
fn test_pool_builder_configuration() {
let config = Config::new("localhost", 1521, "FREEPDB1", "test", "test");
let builder = PoolBuilder::new(config)
.max_size(5)
.wait_timeout(Some(Duration::from_secs(10)))
.create_timeout(None)
.recycle_timeout(Some(Duration::from_secs(2)));
assert_eq!(builder.max_size, 5);
assert_eq!(builder.wait_timeout, Some(Duration::from_secs(10)));
assert_eq!(builder.create_timeout, None);
assert_eq!(builder.recycle_timeout, Some(Duration::from_secs(2)));
}
#[test]
fn test_pool_build_lazy() {
let config = Config::new("localhost", 1521, "FREEPDB1", "test", "test");
let pool = PoolBuilder::new(config).max_size(10).build();
assert!(pool.is_ok());
let pool = pool.unwrap();
let status = pool.status();
assert_eq!(status.size, 0);
assert_eq!(status.available, 0);
}
}