actix_cloud/memorydb/
redis.rs

1use std::time::Duration;
2
3use async_trait::async_trait;
4use redis::{aio::ConnectionManager, AsyncCommands, Expiry};
5
6use super::interface::MemoryDB;
7use crate::Result;
8
9/// Redis memory database backend.
10///
11/// Uses a [`ConnectionManager`](redis::aio::ConnectionManager), which reconnects to
12/// Redis automatically, so the backend can be cloned and shared freely.
13#[derive(Clone)]
14pub struct RedisBackend {
15    client: ConnectionManager,
16}
17
18impl RedisBackend {
19    /// Connect to Redis with a DSN like `redis://user:pass@127.0.0.1:6379/0`.
20    pub async fn new(dsn: &str) -> Result<Self> {
21        let client = ConnectionManager::new(redis::Client::open(dsn)?).await?;
22        Ok(Self { client })
23    }
24}
25
26#[async_trait]
27impl MemoryDB for RedisBackend {
28    async fn set(&self, key: &str, value: &str) -> Result<()> {
29        self.client
30            .clone()
31            .set(key, value)
32            .await
33            .map_err(Into::into)
34    }
35
36    async fn get(&self, key: &str) -> Result<Option<String>> {
37        self.client.clone().get(key).await.map_err(Into::into)
38    }
39
40    async fn get_del(&self, key: &str) -> Result<Option<String>> {
41        self.client.clone().get_del(key).await.map_err(Into::into)
42    }
43
44    async fn get_ex(&self, key: &str, ttl: &Duration) -> Result<Option<String>> {
45        self.client
46            .clone()
47            .get_ex(key, Expiry::PX(ttl.as_millis().try_into()?))
48            .await
49            .map_err(Into::into)
50    }
51
52    async fn set_ex(&self, key: &str, value: &str, ttl: &Duration) -> Result<()> {
53        self.client
54            .clone()
55            .pset_ex(key, value, ttl.as_millis().try_into()?)
56            .await
57            .map_err(Into::into)
58    }
59
60    async fn del(&self, key: &str) -> Result<bool> {
61        self.client.clone().del(key).await.map_err(Into::into)
62    }
63
64    async fn expire(&self, key: &str, ttl: &Duration) -> Result<bool> {
65        self.client
66            .clone()
67            .pexpire(key, ttl.as_millis().try_into()?)
68            .await
69            .map_err(Into::into)
70    }
71
72    async fn flush(&self) -> Result<()> {
73        redis::cmd("FLUSHDB")
74            .query_async(&mut self.client.clone())
75            .await
76            .map_err(Into::into)
77    }
78
79    async fn keys(&self, key: &str) -> Result<Vec<String>> {
80        self.client.clone().keys(key).await.map_err(Into::into)
81    }
82
83    async fn dels(&self, keys: &[String]) -> Result<u64> {
84        let mut p = redis::pipe();
85        let mut p = p.atomic();
86        for i in keys {
87            p = p.del(i);
88        }
89        let res: Vec<u64> = p.query_async(&mut self.client.clone()).await?;
90        Ok(res.into_iter().sum())
91    }
92
93    async fn ttl(&self, key: &str) -> Result<Option<Duration>> {
94        let ret: i64 = self.client.clone().pttl(key).await?;
95        if ret < 0 {
96            Ok(None)
97        } else {
98            Ok(Some(Duration::from_millis(ret.try_into()?)))
99        }
100    }
101}
actix_cloud/memorydb/redis.rs

actix_cloud/memorydb/
redis.rs
