reool 0.30.0

An asynchrounous connection pool for Redis based on tokio and redis-rs
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

use std::time::Duration;

use futures::future::BoxFuture;
use tokio::time::sleep;

use crate::{CheckoutError, CheckoutErrorKind, Poolable};

use pool_internal::Managed;

mod checkout_constraint;
pub(crate) mod pool_internal;
mod pool_per_node;
mod single_pool;

pub(crate) use self::checkout_constraint::*;
pub(crate) use self::pool_per_node::PoolPerNode;
pub(crate) use self::single_pool::SinglePool;

/// Something that can checkout
pub(crate) trait CanCheckout<T: Poolable> {
    /// Directly do a checkout
    fn check_out<'a, M: Into<CheckoutConstraint> + Send + 'static>(
        &'a self,
        constraint: M,
    ) -> BoxFuture<'a, Result<Managed<T>, CheckoutError>>;
}

/// Retry the checkout if the checkout failed with a
/// `CheckoutLimitReached` as long as a retry is allowed
/// by the constraint
pub(crate) async fn check_out_maybe_retry_on_queue_limit_reached<P, T, M>(
    pool: &P,
    constraint: M,
    retry_enabled: bool,
) -> Result<Managed<T>, CheckoutError>
where
    P: CanCheckout<T> + Clone + Send + 'static,
    T: Poolable,
    M: Into<CheckoutConstraint> + Send + 'static,
{
    if !retry_enabled {
        return pool.check_out(constraint).await;
    }

    let constraint = constraint.into();

    match pool.check_out(constraint).await {
        Ok(conn) => Ok(conn),
        Err(err) => {
            if err.kind() != CheckoutErrorKind::CheckoutLimitReached {
                Err(err)
            } else {
                retry_on_queue_limit_reached(pool, constraint, err.kind()).await
            }
        }
    }
}

async fn retry_on_queue_limit_reached<P, T>(
    pool: &P,
    constraint: CheckoutConstraint,
    last_err: CheckoutErrorKind,
) -> Result<Managed<T>, CheckoutError>
where
    P: CanCheckout<T> + Send + 'static,
    T: Poolable,
{
    loop {
        if !constraint.can_wait_for_dispatch() {
            return Err(CheckoutError::from(last_err));
        }

        match pool.check_out(constraint).await {
            Ok(conn) => return Ok(conn),
            Err(err) => {
                if err.kind() != CheckoutErrorKind::CheckoutLimitReached {
                    return Err(err);
                }

                sleep(Duration::from_millis(1)).await;
            }
        }
    }
}