deadpool_redis/cluster/
mod.rs

1//! This module extends the library to support Redis Cluster.
2mod config;
3
4use std::{
5    ops::{Deref, DerefMut},
6    sync::atomic::{AtomicUsize, Ordering},
7};
8
9use deadpool::managed;
10use redis::{
11    IntoConnectionInfo, RedisError, RedisResult, aio::ConnectionLike,
12    cluster_read_routing::RandomReplicaStrategy,
13};
14
15use redis;
16pub use redis::cluster::{ClusterClient, ClusterClientBuilder};
17pub use redis::cluster_async::ClusterConnection;
18
19pub use self::config::{Config, ConfigError};
20
21pub use deadpool::managed::reexports::*;
22deadpool::managed_reexports!(
23    "redis_cluster",
24    Manager,
25    Connection,
26    RedisError,
27    ConfigError
28);
29
30type RecycleResult = managed::RecycleResult<RedisError>;
31
32/// Wrapper around [`redis::cluster_async::ClusterConnection`].
33///
34/// This structure implements [`redis::aio::ConnectionLike`] and can therefore
35/// be used just like a regular [`redis::cluster_async::ClusterConnection`].
36#[allow(missing_debug_implementations)] // `redis::cluster_async::ClusterConnection: !Debug`
37pub struct Connection {
38    conn: Object,
39}
40
41impl Connection {
42    /// Takes this [`Connection`] from its [`Pool`] permanently.
43    ///
44    /// This reduces the size of the [`Pool`].
45    #[must_use]
46    pub fn take(this: Self) -> ClusterConnection {
47        Object::take(this.conn)
48    }
49
50    /// Returns the unique [`ObjectId`] of this [`Connection`].
51    pub fn id(this: &Self) -> ObjectId {
52        Object::id(&this.conn)
53    }
54
55    /// Returns the [`Metrics`] of this [`Connection`].
56    pub fn metrics(this: &Self) -> &Metrics {
57        Object::metrics(&this.conn)
58    }
59}
60
61impl From<Object> for Connection {
62    fn from(conn: Object) -> Self {
63        Self { conn }
64    }
65}
66
67impl Deref for Connection {
68    type Target = ClusterConnection;
69
70    fn deref(&self) -> &ClusterConnection {
71        &self.conn
72    }
73}
74
75impl DerefMut for Connection {
76    fn deref_mut(&mut self) -> &mut ClusterConnection {
77        &mut self.conn
78    }
79}
80
81impl AsRef<ClusterConnection> for Connection {
82    fn as_ref(&self) -> &ClusterConnection {
83        &self.conn
84    }
85}
86
87impl AsMut<ClusterConnection> for Connection {
88    fn as_mut(&mut self) -> &mut ClusterConnection {
89        &mut self.conn
90    }
91}
92
93impl ConnectionLike for Connection {
94    fn req_packed_command<'a>(
95        &'a mut self,
96        cmd: &'a redis::Cmd,
97    ) -> redis::RedisFuture<'a, redis::Value> {
98        self.conn.req_packed_command(cmd)
99    }
100
101    fn req_packed_commands<'a>(
102        &'a mut self,
103        cmd: &'a redis::Pipeline,
104        offset: usize,
105        count: usize,
106    ) -> redis::RedisFuture<'a, Vec<redis::Value>> {
107        self.conn.req_packed_commands(cmd, offset, count)
108    }
109
110    fn get_db(&self) -> i64 {
111        self.conn.get_db()
112    }
113}
114
115/// [`Manager`] for creating and recycling [`redis::cluster_async`] connections.
116///
117/// [`Manager`]: managed::Manager
118pub struct Manager {
119    client: ClusterClient,
120    ping_number: AtomicUsize,
121}
122
123// `redis::cluster_async::ClusterClient: !Debug`
124impl std::fmt::Debug for Manager {
125    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
126        f.debug_struct("Manager")
127            .field("client", &format!("{:p}", &self.client))
128            .field("ping_number", &self.ping_number)
129            .finish()
130    }
131}
132
133impl Manager {
134    /// Creates a new [`Manager`] from the given `params`.
135    ///
136    /// # Errors
137    ///
138    /// If establishing a new [`ClusterClientBuilder`] fails.
139    pub fn new<T: IntoConnectionInfo>(
140        params: Vec<T>,
141        read_from_replicas: bool,
142    ) -> RedisResult<Self> {
143        let mut client = ClusterClientBuilder::new(params);
144        if read_from_replicas {
145            client = client.read_routing_strategy(RandomReplicaStrategy);
146        }
147        Ok(Self {
148            client: client.build()?,
149            ping_number: AtomicUsize::new(0),
150        })
151    }
152}
153
154impl managed::Manager for Manager {
155    type Type = ClusterConnection;
156    type Error = RedisError;
157
158    async fn create(&self) -> Result<ClusterConnection, RedisError> {
159        let conn = self.client.get_async_connection().await?;
160        Ok(conn)
161    }
162
163    async fn recycle(&self, conn: &mut ClusterConnection, _: &Metrics) -> RecycleResult {
164        let ping_number = self.ping_number.fetch_add(1, Ordering::Relaxed).to_string();
165        let n = redis::cmd("PING")
166            .arg(&ping_number)
167            .query_async::<String>(conn)
168            .await?;
169        if n == ping_number {
170            Ok(())
171        } else {
172            Err(managed::RecycleError::message("Invalid PING response"))
173        }
174    }
175}
deadpool_redis/cluster/mod.rs

deadpool_redis/cluster/
mod.rs
