#![recursion_limit = "256"] use std::fmt::{Debug, Display}; use std::future::Future; use std::hash::Hash; use std::sync::Arc; use dashmap::DashMap; use eyre::{Result, WrapErr}; use prometheus::Registry; use redis::aio::ConnectionLike; use redis::{FromRedisValue, ToRedisArgs}; use serde::Serialize; use serde::de::DeserializeOwned; mod blocking; mod cache; mod commands; mod config; mod connection; mod key; mod metrics; mod pubsub; mod routing; mod util; use cache::{CacheManager, ConnectionProvider}; pub use cache::{ CacheSettings, Codec, EncodingFormat, InvalidCodec, InvalidEncodingFormat, RedisValue, }; pub use config::{ CacheLockingStrategy, InvalidCacheLockingStrategy, InvalidRedisConnectionType, InvalidRedisMode, InvalidRedisReadReplicaStrategy, ReadReplicaStrategy, RedisConfig, RedisConfigError, RedisConnectionType, RedisTopology, }; use connection::RedisBackend; pub use key::KeyBuilder; #[derive(Clone)] pub struct RedisPool { backend: RedisBackend, blocking: blocking::RedisBlockingPool, cache: CacheManager, config: RedisConfig, key_builder: KeyBuilder, } pub struct RedisConnection { inner: connection::RedisConnection, key_builder: KeyBuilder, settings: CacheSettings, } impl RedisPool { pub async fn new( meta_namespace: impl Into>, config: RedisConfig, cache_settings: CacheSettings, ) -> Result { tracing::info!( strategy = %config.cache_locking_strategy(), "configured Redis cache locking" ); let backend = RedisBackend::new(&config) .await .wrap_err("creating Redis command backend")?; let blocking = blocking::RedisBlockingPool::new(&config) .await .wrap_err("creating Redis blocking pool")?; let key_builder = KeyBuilder::new(meta_namespace, config.topology()); let cache = CacheManager::new(key_builder.clone(), cache_settings); Ok(Self { backend, blocking, cache, config, key_builder, }) } pub fn key(&self) -> &KeyBuilder { &self.key_builder } } impl RedisPool { pub async fn connect(&self) -> Result { Ok(RedisConnection { inner: self .backend .connect() .await .wrap_err("connecting to Redis")?, key_builder: self.key_builder.clone(), settings: self.cache.settings().clone(), }) } pub async fn register_and_set_metrics( &self, registry: &Registry, ) -> Result<()> { self.backend .register_metrics(registry) .wrap_err("registering Redis command pool metrics")?; self.blocking .register_metrics(registry) .wrap_err("registering Redis blocking pool metrics") } pub async fn get_cached_keys( &self, namespace: &str, keys: &[K], closure: F, ) -> Result> where F: FnOnce(Vec) -> Fut, Fut: Future, E>>, E: std::error::Error + Send + Sync + 'static, T: Serialize + DeserializeOwned, K: Display + Hash + Eq + PartialEq + Clone + DeserializeOwned + Serialize + Debug, { self.cache .get_cached_keys(self, namespace, keys, closure) .await } pub async fn get_cached_keys_raw( &self, namespace: &str, keys: &[K], closure: F, ) -> Result> where F: FnOnce(Vec) -> Fut, Fut: Future, E>>, E: std::error::Error + Send + Sync + 'static, T: Serialize + DeserializeOwned, K: Display + Hash + Eq + PartialEq + Clone + DeserializeOwned + Serialize + Debug, { self.cache .get_cached_keys_raw(self, namespace, keys, closure) .await } pub async fn get_cached_keys_with_slug( &self, namespace: &str, slug_namespace: &str, case_sensitive: bool, keys: &[I], closure: F, ) -> Result> where F: FnOnce(Vec) -> Fut, Fut: Future, T)>, E>>, E: std::error::Error + Send + Sync + 'static, T: Serialize + DeserializeOwned, I: Display + Hash + Eq + PartialEq + Clone + Debug, K: Display + Hash + Eq + PartialEq + Clone + DeserializeOwned + Serialize, S: Display + Clone + DeserializeOwned + Serialize + Debug, { self.cache .get_cached_keys_with_slug( self, namespace, slug_namespace, case_sensitive, keys, closure, ) .await } pub async fn get_cached_keys_raw_with_slug( &self, namespace: &str, slug_namespace: Option<&str>, case_sensitive: bool, keys: &[I], closure: F, ) -> Result> where F: FnOnce(Vec) -> Fut, Fut: Future, T)>, E>>, E: std::error::Error + Send + Sync + 'static, T: Serialize + DeserializeOwned, I: Display + Hash + Eq + PartialEq + Clone + Debug, K: Display + Hash + Eq + PartialEq + Clone + DeserializeOwned + Serialize, S: Display + Clone + DeserializeOwned + Serialize + Debug, { self.cache .get_cached_keys_raw_with_slug( self, namespace, slug_namespace, case_sensitive, keys, closure, ) .await } } impl ConnectionProvider for RedisPool { type Connection = RedisConnection; fn connect(&self) -> impl Future> + Send { RedisPool::connect(self) } } impl RedisConnection { pub fn key(&self) -> &KeyBuilder { &self.key_builder } pub async fn set( &mut self, key: &str, data: D, expiry: Option, ) -> Result<()> where D: ToRedisArgs + Send + Sync + Debug, { commands::set( &mut self.inner, key, data, expiry.unwrap_or(self.settings.default_expiry), ) .await } pub async fn set_serialized( &mut self, key: &str, data: D, expiry: Option, ) -> Result<()> where D: Serialize, { commands::set_serialized( &mut self.inner, key, data, expiry, &self.settings, ) .await } pub async fn get(&mut self, key: &str) -> Result> { commands::get(&mut self.inner, key).await } pub async fn get_many( &mut self, keys: &[String], ) -> Result>>> { commands::get_many(&mut self.inner, keys).await } pub async fn get_many_typed( &mut self, keys: &[String], ) -> Result>> where R: FromRedisValue, { commands::get_many_as(&mut self.inner, keys).await } pub async fn get_deserialized(&mut self, key: &str) -> Result> where R: for<'a> serde::Deserialize<'a>, { commands::get_deserialized(&mut self.inner, key, &self.settings).await } pub async fn get_many_deserialized( &mut self, keys: &[String], ) -> Result>> where R: for<'a> serde::Deserialize<'a>, { commands::get_many_deserialized(&mut self.inner, keys, &self.settings) .await } pub async fn delete(&mut self, key: &str) -> Result<()> { commands::delete(&mut self.inner, key).await } pub async fn delete_many(&mut self, keys: &[String]) -> Result<()> { commands::delete_many(&mut self.inner, keys).await } pub async fn lpush(&mut self, key: &str, value: D) -> Result<()> where D: ToRedisArgs + Send + Sync + Debug, { commands::lpush(&mut self.inner, key, value).await } pub async fn incr(&mut self, key: &str) -> Result> { commands::incr(&mut self.inner, key).await } } impl ConnectionLike for RedisConnection { fn req_packed_command<'a>( &'a mut self, command: &'a redis::Cmd, ) -> redis::RedisFuture<'a, redis::Value> { self.inner.req_packed_command(command) } fn req_packed_commands<'a>( &'a mut self, pipeline: &'a redis::Pipeline, offset: usize, count: usize, ) -> redis::RedisFuture<'a, Vec> { self.inner.req_packed_commands(pipeline, offset, count) } fn get_db(&self) -> i64 { self.inner.get_db() } } impl connection::RoutableConnection for RedisConnection { fn route_command<'a>( &'a mut self, command: redis::Cmd, routing: redis::cluster_routing::RoutingInfo, ) -> redis::RedisFuture<'a, redis::Value> { self.inner.route_command(command, routing) } }