Release 0.5.2

Author: importcjj

.gitignore

@@ -1,4 +1,4 @@
-/target
+target/
 **/*.rs.bk
 Cargo.lock
 .cargo

CHANGELOG.md

@@ -1,10 +1,12 @@
 
 # CHANGELOG
 
-## v0.5.2 2020-01-15
+## v0.5.2 2020-01-17
 * Do health check for the connection before return it.
 * Add configure item `health_check`.
-* Impl `Debug` for State.
+* Impl Debug for State.
+* Add method is_brand_new to Connection, which returns true if the connection is newly established.
+* Skip health check of those new connections.
 
 ## v0.5.1 2020-01-07
 * Switch to `async-trait`

mobc-redis/Cargo.toml

@@ -8,9 +8,10 @@ edition = "2018"
 
 [dependencies]
 mobc = { version = "0.5", path = ".." }
-redis = { version = "0.13.1-alpha.0", git = "https://github.com/mitsuhiko/redis-rs" }
+redis = "0.14"
 
 [dev-dependencies]
 actix-web = "2.0.0"
 actix-rt = "1.0"
-actix-http = "1.0"
+actix-http = "1.0"
+tokio = { version = "0.2", features = ["full"]}

mobc-redis/examples/redis.rs

@@ -7,12 +7,12 @@ use std::time::Instant;
 async fn main() {
     let client = redis::Client::open("redis://127.0.0.1/").unwrap();
     let manager = RedisConnectionManager::new(client);
-    let pool = Pool::builder().max_open(100).build(manager);
+    let pool = Pool::builder().max_open(20).build(manager);
 
     const MAX: usize = 5000;
 
     let now = Instant::now();
-    let (tx, mut rx) = tokio::sync::mpsc::channel::<usize>(16);
+    let (tx, mut rx) = tokio::sync::mpsc::channel::<usize>(5000);
     for i in 0..MAX {
         let pool = pool.clone();
         let mut tx_c = tx.clone();

src/lib.rs

@@ -193,6 +193,7 @@ struct Conn<C, E> {
     #[allow(dead_code)]
     last_err: Mutex<Option<E>>,
     created_at: Instant,
+    brand_new: bool,
 }
 
 impl<C, E> Conn<C, E> {
@@ -436,35 +437,14 @@ impl<M: Manager> Pool<M> {
         loop {
             let timeout = delay_until(deadline);
             try_times += 1;
-            match self
-                .inner_get_ctx(GetStrategy::CachedOrNewConn, timeout)
-                .await
-            {
-                Ok(mut conn) => {
-                    if !self.0.config.health_check {
-                        return Ok(conn);
-                    }
-                    return {
-                        let raw_conn = conn.conn.as_mut().unwrap().raw.take().unwrap();
-                        match self.0.manager.check(raw_conn).await {
-                            Ok(raw) => conn.conn.as_mut().unwrap().raw = Some(raw),
-                            Err(e) => {
-                                if try_times == config.max_bad_conn_retries {
-                                    return Err(Error::Inner(e));
-                                }
-                                continue;
-                            }
-                        }
-                        Ok(conn)
-                    };
-                }
+            match self.get_ctx(GetStrategy::CachedOrNewConn, timeout).await {
+                Ok(conn) => return Ok(conn),
                 Err(Error::BadConn) => {
                     if try_times == config.max_bad_conn_retries {
                         let timeout = delay_until(deadline);
-                        return self
-                            .inner_get_ctx(GetStrategy::AlwaysNewConn, timeout)
-                            .await;
+                        return self.get_ctx(GetStrategy::AlwaysNewConn, timeout).await;
                     }
+                    continue;
                 }
                 Err(err) => return Err(err),
             }
@@ -477,46 +457,52 @@ impl<M: Manager> Pool<M> {
         loop {
             let never_ot = futures::future::pending();
             try_times += 1;
-            match self
-                .inner_get_ctx(GetStrategy::CachedOrNewConn, never_ot)
-                .await
-            {
-                Ok(mut conn) => {
-                    if !self.0.config.health_check {
-                        return Ok(conn);
-                    }
-                    return {
-                        let raw_conn = conn.conn.as_mut().unwrap().raw.take().unwrap();
-                        match self.0.manager.check(raw_conn).await {
-                            Ok(raw) => conn.conn.as_mut().unwrap().raw = Some(raw),
-                            Err(e) => {
-                                if try_times == config.max_bad_conn_retries {
-                                    return Err(Error::Inner(e));
-                                }
-                                continue;
-                            }
-                        }
-                        Ok(conn)
-                    };
-                }
+            match self.get_ctx(GetStrategy::CachedOrNewConn, never_ot).await {
+                Ok(conn) => return Ok(conn),
                 Err(Error::BadConn) => {
                     if try_times == config.max_bad_conn_retries {
                         let never_ot = futures::future::pending();
-                        return self
-                            .inner_get_ctx(GetStrategy::AlwaysNewConn, never_ot)
-                            .await;
+                        return self.get_ctx(GetStrategy::AlwaysNewConn, never_ot).await;
                     }
+                    continue;
                 }
                 Err(err) => return Err(err),
             }
         }
     }
 
-    async fn inner_get_ctx(
+    async fn get_ctx(
         &self,
         strategy: GetStrategy,
         ctx: impl Future<Output = ()> + Unpin,
     ) -> Result<Connection<M>, Error<M::Error>> {
+        let mut c = self.inner_get_ctx(strategy, ctx).await?;
+        let mut internals = self.0.internals.lock().await;
+        if !c.brand_new && c.expired(internals.config.max_lifetime) {
+            c.close(&mut internals);
+            return Err(Error::BadConn);
+        }
+
+        if !c.brand_new {
+            let raw = c.raw.take().unwrap();
+            match self.0.manager.check(raw).await {
+                Ok(raw) => c.raw = Some(raw),
+                Err(e) => return Err(Error::Inner(e)),
+            }
+        }
+
+        let conn = Connection {
+            pool: Some(self.clone()),
+            conn: Some(c),
+        };
+        Ok(conn)
+    }
+
+    async fn inner_get_ctx(
+        &self,
+        strategy: GetStrategy,
+        ctx: impl Future<Output = ()> + Unpin,
+    ) -> Result<Conn<M::Connection, M::Error>, Error<M::Error>> {
         let mut ctx = ctx.fuse();
 
         let mut internals = self.0.internals.lock().await;
@@ -529,18 +515,7 @@ impl<M: Manager> Pool<M> {
         let num_free = internals.free_conns.len();
         if strategy == GetStrategy::CachedOrNewConn && num_free > 0 {
             let c = internals.free_conns.swap_remove(0);
-
-            if c.expired(internals.config.max_lifetime) {
-                c.close(&mut internals);
-                return Err(Error::BadConn);
-            }
-
-            drop(internals);
-            let pooled = Connection {
-                pool: Some(self.clone()),
-                conn: Some(c),
-            };
-            return Ok(pooled);
+            return Ok(c);
         }
 
         if internals.config.max_open > 0 {
@@ -564,19 +539,7 @@ impl<M: Manager> Pool<M> {
                     }
                     conn = req_recv.fuse() => {
                         let c = conn.unwrap();
-
-                        let mut internals = self.0.internals.lock().await;
-                        if c.expired(internals.config.max_lifetime) {
-                            c.close(&mut internals);
-                            return Err(Error::BadConn);
-                        }
-                        let pooled = Connection {
-                            pool: Some(self.clone()),
-                            conn: Some(c),
-                        };
-                        return Ok(pooled)
-
-
+                        return Ok(c)
                     }
                 }
             }
@@ -592,13 +555,10 @@ impl<M: Manager> Pool<M> {
                     raw: Some(c),
                     last_err: Mutex::new(None),
                     created_at: Instant::now(),
-                };
-                let pooled = Connection {
-                    pool: Some(self.clone()),
-                    conn: Some(conn),
+                    brand_new: true,
                 };
 
-                return Ok(pooled);
+                return Ok(conn);
             }
             Err(e) => {
                 let internals = self.0.internals.lock().await;
@@ -649,10 +609,11 @@ async fn open_new_connection<M: Manager>(shared: &Weak<SharedPool<M>>) {
                 raw: Some(c),
                 last_err: Mutex::new(None),
                 created_at: Instant::now(),
+                brand_new: true,
             };
             return put_conn(&shared, internals, conn).await;
         }
-        Err(e) => {
+        Err(_) => {
             internals.num_open -= 1;
             return maybe_open_new_connection(&shared, internals).await;
         }
@@ -662,16 +623,20 @@ async fn open_new_connection<M: Manager>(shared: &Weak<SharedPool<M>>) {
 async fn put_conn<M: Manager>(
     shared: &Arc<SharedPool<M>>,
     mut internals: MutexGuard<'_, PoolInternals<M::Connection, M::Error>>,
-    conn: Conn<M::Connection, M::Error>,
+    mut conn: Conn<M::Connection, M::Error>,
 ) {
     if conn.raw.is_none() {
+        conn.close(&mut internals);
         return maybe_open_new_connection(shared, internals).await;
     }
 
     if internals.config.max_open > 0 && internals.num_open > internals.config.max_open {
+        conn.close(&mut internals);
         return;
     }
 
+    conn.brand_new = false;
+
     if internals.conn_requests.len() > 0 {
         let key = internals.conn_requests.keys().next().unwrap().clone();
         let req = internals.conn_requests.remove(&key).unwrap();
@@ -794,6 +759,13 @@ pub struct Connection<M: Manager> {
     conn: Option<Conn<M::Connection, M::Error>>,
 }
 
+impl<M: Manager> Connection<M> {
+    /// Returns true is the connection is newly established.
+    pub fn is_brand_new(&self) -> bool {
+        self.conn.as_ref().unwrap().brand_new
+    }
+}
+
 impl<M: Manager> Drop for Connection<M> {
     fn drop(&mut self) {
         let pool = self.pool.take().unwrap();

src/runtime.rs

@@ -74,7 +74,6 @@ mod runtime {
 mod runtime {
     use super::*;
     use async_std::task;
-    use async_std::task::JoinHandle;
     pub struct Runtime(TaskExecutor);
 
     impl Runtime {

src/time.rs

@@ -20,17 +20,20 @@ mod time {
         }
     }
 
+    /// Creates new Interval that yields with interval of duration.
     pub fn interval(duration: Duration) -> Interval {
         Interval {
             timer: Some(Delay::new(Duration::from_secs(0))),
             interval: duration,
         }
     }
 
+    /// Wait until duration has elapsed.
     pub fn delay_for(duration: Duration) -> Delay {
         Delay::new(duration)
     }
 
+    /// Wait until deadline is reached.
     pub fn delay_until(deadline: Instant) -> Delay {
         let mut delay = Delay::new(Duration::from_secs(1));
         delay.reset(deadline);

tests/mobc.rs

@@ -187,8 +187,9 @@ fn test_drop_on_broken() {
     rt.block_on(async {
         let pool = Pool::new(handler);
 
-        assert!(pool.get().await.is_err());
+        assert!(pool.get().await.is_ok());
         delay_for(Duration::from_secs(1)).await;
+        assert!(pool.get().await.is_err());
         assert!(DROPPED.load(Ordering::SeqCst));
         Ok::<(), Error<TestError>>(())
     })
@@ -228,8 +229,9 @@ fn test_invalid_conn() {
     rt.block_on(async {
         let pool = Pool::builder().max_open(1).build(Handler);
 
-        assert!(pool.get().await.is_err());
+        assert!(pool.get().await.is_ok());
         delay_for(Duration::from_secs(1)).await;
+        assert!(pool.get().await.is_err());
         assert!(DROPPED.load(Ordering::SeqCst));
         assert_eq!(1_u64, pool.state().await.connections);
         Ok::<(), Error<TestError>>(())
@@ -700,3 +702,20 @@ fn test_conns_drop_on_pool_drop() {
     })
     .unwrap();
 }
+
+#[test]
+fn test_is_brand_new() {
+    let mut rt = Runtime::new().unwrap();
+    let handler = OkManager;
+    rt.block_on(async {
+        let pool = Pool::builder().max_open(1).build(handler);
+
+        let conn = pool.get().await.ok().unwrap();
+        assert!(conn.is_brand_new());
+        drop(conn);
+        let conn = pool.get().await.ok().unwrap();
+        assert!(!conn.is_brand_new());
+        Ok::<(), Error<TestError>>(())
+    })
+    .unwrap();
+}