02d8172
use std::time::Duration;
use redis::{Cmd, Pipeline, ToRedisArgs, Value, cmd};
use serde::Serialize;
use crate::error::{CacheError, CacheResult};
use crate::json;
use crate::kv::{Cache, millis};
pub struct CachePipeline {
cache: Cache,
pipeline: Pipeline,
}
impl CachePipeline {
pub(crate) fn new(cache: Cache) -> Self {
Self {
cache,
pipeline: Pipeline::new(),
}
}
pub fn pipeline(&self) -> &Pipeline {
&self.pipeline
}
pub fn len(&self) -> usize {
self.pipeline.len()
}
pub fn is_empty(&self) -> bool {
self.pipeline.is_empty()
}
pub fn clear(&mut self) {
self.pipeline.clear();
}
pub fn atomic(&mut self) -> CacheResult<&mut Self> {
if self.cache.is_cluster() {
return Err(CacheError::unsupported(
"MULTI/EXEC",
"transactions are unsupported by Redis Cluster",
));
}
self.pipeline.atomic();
Ok(self)
}
pub fn add(&mut self, command: &Cmd) -> &mut Self {
self.pipeline.add_command(command.clone());
self
}
pub fn command(&mut self, name: &str) -> Cmd {
cmd(name)
}
pub fn get(&mut self, key: &str) -> &mut Self {
self.add(cmd("GET").arg(key))
}
pub fn mget<K: ToRedisArgs + Send + Sync>(&mut self, keys: K) -> &mut Self {
self.add(cmd("MGET").arg(keys))
}
pub fn set(&mut self, key: &str, value: &[u8]) -> &mut Self {
self.add(cmd("SET").arg(key).arg(value))
}
pub fn set_ex(&mut self, key: &str, value: &[u8], ttl: Duration) -> &mut Self {
self.add(cmd("SET").arg(key).arg(value).arg("PX").arg(millis(ttl)))
}
pub fn set_nx_ex(&mut self, key: &str, value: &[u8], ttl: Duration) -> &mut Self {
self.add(
cmd("SET")
.arg(key)
.arg(value)
.arg("PX")
.arg(millis(ttl))
.arg("NX"),
)
}
pub fn del<K: ToRedisArgs + Send + Sync>(&mut self, keys: K) -> &mut Self {
self.add(cmd("DEL").arg(keys))
}
pub fn unlink<K: ToRedisArgs + Send + Sync>(&mut self, keys: K) -> &mut Self {
self.add(cmd("UNLINK").arg(keys))
}
pub fn exists<K: ToRedisArgs + Send + Sync>(&mut self, keys: K) -> &mut Self {
self.add(cmd("EXISTS").arg(keys))
}
pub fn expire(&mut self, key: &str, ttl: Duration) -> &mut Self {
self.add(cmd("PEXPIRE").arg(key).arg(millis(ttl)))
}
pub fn persist(&mut self, key: &str) -> &mut Self {
self.add(cmd("PERSIST").arg(key))
}
pub fn ttl(&mut self, key: &str) -> &mut Self {
self.add(cmd("PTTL").arg(key))
}
pub fn incr(&mut self, key: &str) -> &mut Self {
self.add(cmd("INCR").arg(key))
}
pub fn incr_by(&mut self, key: &str, delta: i64) -> &mut Self {
self.add(cmd("INCRBY").arg(key).arg(delta))
}
pub fn decr(&mut self, key: &str) -> &mut Self {
self.add(cmd("DECR").arg(key))
}
pub fn decr_by(&mut self, key: &str, delta: i64) -> &mut Self {
self.add(cmd("DECRBY").arg(key).arg(delta))
}
pub fn hset(&mut self, key: &str, field: &str, value: &[u8]) -> &mut Self {
self.add(cmd("HSET").arg(key).arg(field).arg(value))
}
pub fn hset_multiple<V: AsRef<[u8]>>(&mut self, key: &str, fields: &[(&str, V)]) -> &mut Self {
let mut command = cmd("HSET");
command.arg(key);
for (field, value) in fields {
command.arg(*field).arg(value.as_ref());
}
self.add(&command)
}
pub fn hget(&mut self, key: &str, field: &str) -> &mut Self {
self.add(cmd("HGET").arg(key).arg(field))
}
pub fn hgetall(&mut self, key: &str) -> &mut Self {
self.add(cmd("HGETALL").arg(key))
}
pub fn hdel<F: ToRedisArgs + Send + Sync>(&mut self, key: &str, fields: F) -> &mut Self {
self.add(cmd("HDEL").arg(key).arg(fields))
}
pub fn hexists(&mut self, key: &str, field: &str) -> &mut Self {
self.add(cmd("HEXISTS").arg(key).arg(field))
}
pub fn hlen(&mut self, key: &str) -> &mut Self {
self.add(cmd("HLEN").arg(key))
}
pub fn hkeys(&mut self, key: &str) -> &mut Self {
self.add(cmd("HKEYS").arg(key))
}
pub fn hvals(&mut self, key: &str) -> &mut Self {
self.add(cmd("HVALS").arg(key))
}
pub fn lpush<V: ToRedisArgs + Send + Sync>(&mut self, key: &str, values: V) -> &mut Self {
self.add(cmd("LPUSH").arg(key).arg(values))
}
pub fn rpush<V: ToRedisArgs + Send + Sync>(&mut self, key: &str, values: V) -> &mut Self {
self.add(cmd("RPUSH").arg(key).arg(values))
}
pub fn lpop(&mut self, key: &str, count: Option<usize>) -> &mut Self {
let mut command = cmd("LPOP");
command.arg(key);
if let Some(count) = count {
command.arg(count);
}
self.add(&command)
}
pub fn rpop(&mut self, key: &str, count: Option<usize>) -> &mut Self {
let mut command = cmd("RPOP");
command.arg(key);
if let Some(count) = count {
command.arg(count);
}
self.add(&command)
}
pub fn lrange(&mut self, key: &str, start: isize, stop: isize) -> &mut Self {
self.add(cmd("LRANGE").arg(key).arg(start).arg(stop))
}
pub fn ltrim(&mut self, key: &str, start: isize, stop: isize) -> &mut Self {
self.add(cmd("LTRIM").arg(key).arg(start).arg(stop))
}
pub fn llen(&mut self, key: &str) -> &mut Self {
self.add(cmd("LLEN").arg(key))
}
pub fn sadd<V: ToRedisArgs + Send + Sync>(&mut self, key: &str, members: V) -> &mut Self {
self.add(cmd("SADD").arg(key).arg(members))
}
pub fn srem<V: ToRedisArgs + Send + Sync>(&mut self, key: &str, members: V) -> &mut Self {
self.add(cmd("SREM").arg(key).arg(members))
}
pub fn scard(&mut self, key: &str) -> &mut Self {
self.add(cmd("SCARD").arg(key))
}
pub fn sismember(&mut self, key: &str, member: &str) -> &mut Self {
self.add(cmd("SISMEMBER").arg(key).arg(member))
}
pub fn smembers(&mut self, key: &str) -> &mut Self {
self.add(cmd("SMEMBERS").arg(key))
}
pub fn zadd(&mut self, key: &str, member: &str, score: f64) -> &mut Self {
self.add(cmd("ZADD").arg(key).arg(score).arg(member))
}
pub fn zrem<M: ToRedisArgs + Send + Sync>(&mut self, key: &str, members: M) -> &mut Self {
self.add(cmd("ZREM").arg(key).arg(members))
}
pub fn zcard(&mut self, key: &str) -> &mut Self {
self.add(cmd("ZCARD").arg(key))
}
pub fn zscore(&mut self, key: &str, member: &str) -> &mut Self {
self.add(cmd("ZSCORE").arg(key).arg(member))
}
pub fn zincr(&mut self, key: &str, member: &str, delta: f64) -> &mut Self {
self.add(cmd("ZINCRBY").arg(key).arg(delta).arg(member))
}
pub fn zrange(&mut self, key: &str, start: isize, stop: isize) -> &mut Self {
self.add(cmd("ZRANGE").arg(key).arg(start).arg(stop))
}
pub fn set_json_ex<T: Serialize>(
&mut self,
key: &str,
value: &T,
ttl: Duration,
) -> CacheResult<&mut Self> {
let encoded = json::encode(key, value)?;
Ok(self.set_ex(key, &encoded, ttl))
}
pub async fn execute(&self) -> CacheResult<Vec<Value>> {
if self.pipeline.is_empty() {
return Ok(Vec::new());
}
let mut conn = self.cache.conn().clone();
self.pipeline
.query_async(&mut conn)
.await
.map_err(CacheError::from)
}
}