b6f7f73
ZhenYi
refactor: reorganize configuration files and update module paths for improved structure
b6f7f73
use std::time::Duration;
use config::redis::RedisConfig;
use redis::{AsyncCommands, Cmd, FromRedisValue, ToRedisArgs, ToSingleRedisArg};
use crate::conn::{CacheConn, CacheConnection};
use crate::error::{CacheError, CacheResult};
#[derive(Clone)]
pub struct Cache {
conn: CacheConn,
prefix: String,
default_ttl: Option<Duration>,
}
impl Cache {
pub async fn connect(cfg: &RedisConfig) -> CacheResult<Self> {
let conn = CacheConn::init(cfg).await?;
Ok(Self {
conn,
prefix: cfg.key_prefix().to_string(),
default_ttl: cfg.default_ttl(),
})
}
pub fn from_conn(conn: CacheConn) -> Self {
Self {
conn,
prefix: String::new(),
default_ttl: None,
}
}
pub fn with_prefix(mut self, prefix: impl Into<String>) -> Self {
self.prefix = prefix.into();
self
}
pub fn with_default_ttl(mut self, ttl: Duration) -> Self {
self.default_ttl = Some(ttl);
self
}
pub fn conn(&self) -> &CacheConn {
&self.conn
}
pub fn key_prefix(&self) -> &str {
&self.prefix
}
pub fn default_ttl(&self) -> Option<Duration> {
self.default_ttl
}
pub fn is_cluster(&self) -> bool {
self.conn.is_cluster()
}
pub fn database(&self) -> i64 {
self.conn.database()
}
pub async fn acquire(&self) -> CacheResult<CacheConnection> {
self.conn.acquire().await
}
pub async fn ping(&self) -> CacheResult<()> {
let mut cmd = Cmd::new();
cmd.arg("PING");
self.exec_unit(&cmd).await
}
pub fn pipeline(&self) -> crate::pipeline::CachePipeline {
crate::pipeline::CachePipeline::new(self.clone())
}
pub fn scripts(&self) -> crate::scripts::LuaScripts {
crate::scripts::LuaScripts::new(self.clone())
}
pub(crate) async fn exec<T: FromRedisValue + Send>(&self, cmd: &Cmd) -> CacheResult<T> {
let mut conn = self.conn.clone();
cmd.query_async::<T>(&mut conn)
.await
.map_err(CacheError::from)
}
pub(crate) async fn exec_unit(&self, cmd: &Cmd) -> CacheResult<()> {
self.exec::<()>(cmd).await
}
pub async fn get<T: FromRedisValue + Send>(&self, key: &str) -> CacheResult<Option<T>> {
let mut conn = self.conn.clone();
conn.get(key).await.map_err(CacheError::from)
}
pub async fn get_bytes(&self, key: &str) -> CacheResult<Option<Vec<u8>>> {
self.get::<Vec<u8>>(key).await
}
pub async fn get_del<T: FromRedisValue + Send>(&self, key: &str) -> CacheResult<Option<T>> {
let mut cmd = Cmd::new();
cmd.arg("GETDEL").arg(key);
self.exec(&cmd).await
}
pub async fn set<V: ToRedisArgs + Send + Sync>(&self, key: &str, value: V) -> CacheResult<()> {
self.set_with(key, value, None, false, false).await
}
pub async fn set_ex<V: ToRedisArgs + Send + Sync>(
&self,
key: &str,
value: V,
ttl: Duration,
) -> CacheResult<()> {
self.set_with(key, value, Some(ttl), false, false).await
}
pub async fn set_default<V: ToRedisArgs + Send + Sync>(
&self,
key: &str,
value: V,
) -> CacheResult<()> {
self.set_with(key, value, self.default_ttl, false, false)
.await
}
pub async fn set_nx<V: ToRedisArgs + Send + Sync>(
&self,
key: &str,
value: V,
) -> CacheResult<bool> {
self.set_flag(key, value, None, true, false).await
}
pub async fn set_nx_ex<V: ToRedisArgs + Send + Sync>(
&self,
key: &str,
value: V,
ttl: Duration,
) -> CacheResult<bool> {
self.set_flag(key, value, Some(ttl), true, false).await
}
pub async fn set_xx<V: ToRedisArgs + Send + Sync>(
&self,
key: &str,
value: V,
) -> CacheResult<bool> {
self.set_flag(key, value, None, false, true).await
}
pub async fn set_xx_ex<V: ToRedisArgs + Send + Sync>(
&self,
key: &str,
value: V,
ttl: Duration,
) -> CacheResult<bool> {
self.set_flag(key, value, Some(ttl), false, true).await
}
async fn set_with<V: ToRedisArgs + Send + Sync>(
&self,
key: &str,
value: V,
ttl: Option<Duration>,
nx: bool,
xx: bool,
) -> CacheResult<()> {
if nx || xx {
self.set_flag(key, value, ttl, nx, xx).await?;
return Ok(());
}
let cmd = set_cmd(key, value, ttl, false, false);
self.exec_unit(&cmd).await
}
async fn set_flag<V: ToRedisArgs + Send + Sync>(
&self,
key: &str,
value: V,
ttl: Option<Duration>,
nx: bool,
xx: bool,
) -> CacheResult<bool> {
let cmd = set_cmd(key, value, ttl, nx, xx);
self.exec(&cmd).await
}
pub async fn get_or_set<V, F, Fut>(
&self,
key: &str,
ttl: Option<Duration>,
fill: F,
) -> CacheResult<Option<V>>
where
V: ToRedisArgs + FromRedisValue + Send + Sync,
F: FnOnce() -> Fut,
Fut: std::future::Future<Output = CacheResult<V>>,
{
if let Some(value) = self.get::<V>(key).await? {
return Ok(Some(value));
}
let value = fill().await?;
match ttl {
Some(ttl) => self.set_ex(key, &value, ttl).await?,
None => self.set(key, &value).await?,
}
Ok(Some(value))
}
pub async fn append<V: ToSingleRedisArg + Send + Sync>(
&self,
key: &str,
value: V,
) -> CacheResult<usize> {
let mut conn = self.conn.clone();
conn.append(key, value).await.map_err(CacheError::from)
}
pub async fn strlen(&self, key: &str) -> CacheResult<usize> {
let mut conn = self.conn.clone();
conn.strlen(key).await.map_err(CacheError::from)
}
pub async fn get_range(&self, key: &str, start: isize, stop: isize) -> CacheResult<Vec<u8>> {
let mut conn = self.conn.clone();
conn.getrange(key, start, stop)
.await
.map_err(CacheError::from)
}
pub async fn incr(&self, key: &str) -> CacheResult<i64> {
self.incr_by(key, 1).await
}
pub async fn incr_by(&self, key: &str, delta: i64) -> CacheResult<i64> {
let mut conn = self.conn.clone();
conn.incr(key, delta).await.map_err(CacheError::from)
}
pub async fn decr(&self, key: &str) -> CacheResult<i64> {
self.decr_by(key, 1).await
}
pub async fn decr_by(&self, key: &str, delta: i64) -> CacheResult<i64> {
let mut conn = self.conn.clone();
conn.decr(key, delta).await.map_err(CacheError::from)
}
pub async fn exists(&self, key: &str) -> CacheResult<bool> {
let mut conn = self.conn.clone();
conn.exists(key).await.map_err(CacheError::from)
}
pub async fn exists_many<K: ToRedisArgs + Send + Sync>(&self, keys: K) -> CacheResult<i64> {
let mut conn = self.conn.clone();
conn.exists(keys).await.map_err(CacheError::from)
}
pub async fn expire(&self, key: &str, ttl: Duration) -> CacheResult<bool> {
let mut conn = self.conn.clone();
conn.pexpire(key, millis(ttl))
.await
.map_err(CacheError::from)
}
pub async fn expire_at(&self, key: &str, unix_millis: i64) -> CacheResult<bool> {
let mut cmd = Cmd::new();
cmd.arg("PEXPIREAT").arg(key).arg(unix_millis);
self.exec(&cmd).await
}
pub async fn persist(&self, key: &str) -> CacheResult<bool> {
let mut conn = self.conn.clone();
conn.persist(key).await.map_err(CacheError::from)
}
pub async fn ttl(&self, key: &str) -> CacheResult<Option<Duration>> {
let mut conn = self.conn.clone();
let seconds: i64 = conn.ttl(key).await.map_err(CacheError::from)?;
Ok(duration_from_seconds(seconds))
}
pub async fn p_ttl(&self, key: &str) -> CacheResult<Option<Duration>> {
let mut conn = self.conn.clone();
let millis: i64 = conn.pttl(key).await.map_err(CacheError::from)?;
Ok(duration_from_millis(millis))
}
pub async fn delete(&self, key: &str) -> CacheResult<bool> {
let mut conn = self.conn.clone();
let removed: usize = conn.del(key).await.map_err(CacheError::from)?;
Ok(removed > 0)
}
pub async fn delete_many<K: ToRedisArgs + Send + Sync>(&self, keys: K) -> CacheResult<usize> {
let mut conn = self.conn.clone();
conn.del(keys).await.map_err(CacheError::from)
}
pub async fn unlink<K: ToRedisArgs + Send + Sync>(&self, keys: K) -> CacheResult<usize> {
let mut conn = self.conn.clone();
conn.unlink(keys).await.map_err(CacheError::from)
}
pub async fn mget<T: FromRedisValue + Send, K: ToRedisArgs + Send + Sync>(
&self,
keys: K,
) -> CacheResult<Vec<Option<T>>> {
let mut conn = self.conn.clone();
conn.mget(keys).await.map_err(CacheError::from)
}
pub async fn mset<K: ToRedisArgs + Send + Sync, V: ToRedisArgs + Send + Sync>(
&self,
items: &[(K, V)],
) -> CacheResult<()> {
let mut conn = self.conn.clone();
conn.mset(items).await.map_err(CacheError::from)
}
pub async fn rename(&self, key: &str, new_key: &str) -> CacheResult<()> {
let mut conn = self.conn.clone();
conn.rename(key, new_key).await.map_err(CacheError::from)
}
}
fn set_cmd<V: ToRedisArgs + Send + Sync>(
key: &str,
value: V,
ttl: Option<Duration>,
nx: bool,
xx: bool,
) -> Cmd {
let mut cmd = Cmd::new();
cmd.arg("SET").arg(key).arg(value);
if let Some(ttl) = ttl {
cmd.arg("PX").arg(millis(ttl));
}
if nx {
cmd.arg("NX");
}
if xx {
cmd.arg("XX");
}
cmd
}
pub(crate) fn millis(ttl: Duration) -> i64 {
let millis = ttl.as_millis();
if millis == 0 {
1
} else if millis > i64::MAX as u128 {
i64::MAX
} else {
millis as i64
}
}
fn duration_from_seconds(seconds: i64) -> Option<Duration> {
if seconds < 0 {
None
} else {
Some(Duration::from_secs(seconds as u64))
}
}
fn duration_from_millis(millis: i64) -> Option<Duration> {
if millis < 0 {
None
} else {
Some(Duration::from_millis(millis as u64))
}
}