deadpool_redis/
lib.rs

1#![doc = include_str!("../README.md")]
2#![cfg_attr(docsrs, feature(doc_cfg))]
3#![deny(
4    nonstandard_style,
5    rust_2018_idioms,
6    rustdoc::broken_intra_doc_links,
7    rustdoc::private_intra_doc_links
8)]
9#![forbid(non_ascii_idents, unsafe_code)]
10#![warn(
11    deprecated_in_future,
12    missing_copy_implementations,
13    missing_debug_implementations,
14    missing_docs,
15    unreachable_pub,
16    unused_import_braces,
17    unused_labels,
18    unused_lifetimes,
19    unused_qualifications,
20    unused_results
21)]
22#![allow(clippy::uninlined_format_args)]
23
24#[cfg(feature = "cluster")]
25pub mod cluster;
26mod config;
27
28#[cfg(feature = "sentinel")]
29pub mod sentinel;
30
31use std::{
32    ops::{Deref, DerefMut},
33    sync::atomic::{AtomicUsize, Ordering},
34};
35
36use deadpool::managed;
37use redis::{
38    AsyncConnectionConfig, Client, IntoConnectionInfo, RedisError, RedisResult,
39    aio::{ConnectionLike, MultiplexedConnection},
40};
41
42pub use redis;
43
44pub use self::config::{
45    Config, ConfigError, ConnectionAddr, ConnectionInfo, ProtocolVersion, RedisConnectionInfo,
46};
47
48pub use deadpool::managed::reexports::*;
49deadpool::managed_reexports!("redis", Manager, Connection, RedisError, ConfigError);
50
51/// Type alias for using [`deadpool::managed::RecycleResult`] with [`redis`].
52type RecycleResult = managed::RecycleResult<RedisError>;
53
54/// Wrapper around [`redis::aio::MultiplexedConnection`].
55///
56/// This structure implements [`redis::aio::ConnectionLike`] and can therefore
57/// be used just like a regular [`redis::aio::MultiplexedConnection`].
58#[allow(missing_debug_implementations)] // `redis::aio::MultiplexedConnection: !Debug`
59pub struct Connection {
60    conn: Object,
61}
62
63impl Connection {
64    /// Takes this [`Connection`] from its [`Pool`] permanently.
65    ///
66    /// This reduces the size of the [`Pool`].
67    #[must_use]
68    pub fn take(this: Self) -> MultiplexedConnection {
69        Object::take(this.conn)
70    }
71}
72
73impl From<Object> for Connection {
74    fn from(conn: Object) -> Self {
75        Self { conn }
76    }
77}
78
79impl Deref for Connection {
80    type Target = MultiplexedConnection;
81
82    fn deref(&self) -> &MultiplexedConnection {
83        &self.conn
84    }
85}
86
87impl DerefMut for Connection {
88    fn deref_mut(&mut self) -> &mut MultiplexedConnection {
89        &mut self.conn
90    }
91}
92
93impl AsRef<MultiplexedConnection> for Connection {
94    fn as_ref(&self) -> &MultiplexedConnection {
95        &self.conn
96    }
97}
98
99impl AsMut<MultiplexedConnection> for Connection {
100    fn as_mut(&mut self) -> &mut MultiplexedConnection {
101        &mut self.conn
102    }
103}
104
105impl ConnectionLike for Connection {
106    fn req_packed_command<'a>(
107        &'a mut self,
108        cmd: &'a redis::Cmd,
109    ) -> redis::RedisFuture<'a, redis::Value> {
110        self.conn.req_packed_command(cmd)
111    }
112
113    fn req_packed_commands<'a>(
114        &'a mut self,
115        cmd: &'a redis::Pipeline,
116        offset: usize,
117        count: usize,
118    ) -> redis::RedisFuture<'a, Vec<redis::Value>> {
119        self.conn.req_packed_commands(cmd, offset, count)
120    }
121
122    fn get_db(&self) -> i64 {
123        self.conn.get_db()
124    }
125}
126
127/// [`Manager`] for creating and recycling [`redis`] connections.
128///
129/// [`Manager`]: managed::Manager
130pub struct Manager {
131    client: Client,
132    connection_config: Option<AsyncConnectionConfig>,
133    ping_number: AtomicUsize,
134}
135
136impl std::fmt::Debug for Manager {
137    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
138        f.debug_struct("Manager")
139            .field("client", &self.client)
140            .field("ping_number", &self.ping_number)
141            .finish()
142    }
143}
144
145impl Manager {
146    /// Creates a new [`Manager`] from the given `params`.
147    ///
148    /// # Errors
149    ///
150    /// If establishing a new [`Client`] fails.
151    pub fn new<T: IntoConnectionInfo>(params: T) -> RedisResult<Self> {
152        Ok(Self {
153            client: Client::open(params)?,
154            connection_config: None,
155            ping_number: AtomicUsize::new(0),
156        })
157    }
158
159    /// Creates a new [`Manager`] from the given `params` and
160    /// [`AsyncConnectionConfig`].
161    ///
162    /// This allows configuring connection-level settings such as response timeouts and connection
163    /// timeouts.
164    ///
165    /// # Errors
166    ///
167    /// If establishing a new [`Client`] fails.
168    pub fn new_with_config<T: IntoConnectionInfo>(
169        params: T,
170        connection_config: AsyncConnectionConfig,
171    ) -> RedisResult<Self> {
172        Ok(Self {
173            client: Client::open(params)?,
174            connection_config: Some(connection_config),
175            ping_number: AtomicUsize::new(0),
176        })
177    }
178}
179
180impl managed::Manager for Manager {
181    type Type = MultiplexedConnection;
182    type Error = RedisError;
183
184    async fn create(&self) -> Result<MultiplexedConnection, RedisError> {
185        let conn = match &self.connection_config {
186            Some(config) => {
187                self.client
188                    .get_multiplexed_async_connection_with_config(config)
189                    .await?
190            }
191            None => self.client.get_multiplexed_async_connection().await?,
192        };
193
194        Ok(conn)
195    }
196
197    async fn recycle(&self, conn: &mut MultiplexedConnection, _: &Metrics) -> RecycleResult {
198        let ping_number = self.ping_number.fetch_add(1, Ordering::Relaxed).to_string();
199        // Using pipeline to avoid roundtrip for UNWATCH
200        let (n,) = redis::Pipeline::with_capacity(2)
201            .cmd("UNWATCH")
202            .ignore()
203            .cmd("PING")
204            .arg(&ping_number)
205            .query_async::<(String,)>(conn)
206            .await?;
207        if n == ping_number {
208            Ok(())
209        } else {
210            Err(managed::RecycleError::message("Invalid PING response"))
211        }
212    }
213}
deadpool_redis/lib.rs

deadpool_redis/
lib.rs
