diff options
Diffstat (limited to 'src/database')
| -rw-r--r-- | src/database/redis/mod.rs | 44 |
1 files changed, 27 insertions, 17 deletions
diff --git a/src/database/redis/mod.rs b/src/database/redis/mod.rs index b709125..eded583 100644 --- a/src/database/redis/mod.rs +++ b/src/database/redis/mod.rs @@ -4,8 +4,11 @@ use futures_util::StreamExt; use futures::stream; use lazy_static::lazy_static; use log::error; -use redis::AsyncCommands; +use mobc_redis::redis; +use mobc_redis::redis::AsyncCommands; use serde_json::json; +use mobc::Pool; +use mobc_redis::RedisConnectionManager; use crate::database::{Storage, Result, StorageError, ErrorKind, MicropubChannel}; use crate::indieauth::User; @@ -14,8 +17,17 @@ struct RedisScripts { edit_post: redis::Script } -impl From<redis::RedisError> for StorageError { - fn from(err: redis::RedisError) -> Self { +impl From<mobc_redis::redis::RedisError> for StorageError { + fn from(err: mobc_redis::redis::RedisError) -> Self { + Self { + msg: format!("{}", err), + source: Some(Box::new(err)), + kind: ErrorKind::Backend + } + } +} +impl From<mobc::Error<mobc_redis::redis::RedisError>> for StorageError { + fn from(err: mobc::Error<mobc_redis::redis::RedisError>) -> Self { Self { msg: format!("{}", err), source: Some(Box::new(err)), @@ -32,8 +44,7 @@ lazy_static! { #[derive(Clone)] pub struct RedisStorage { - // TODO: use mobc crate to create a connection pool and reuse connections for efficiency - redis: redis::Client, + redis: mobc::Pool<RedisConnectionManager>, } fn filter_post(mut post: serde_json::Value, user: &'_ Option<String>) -> Option<serde_json::Value> { @@ -65,17 +76,17 @@ fn filter_post(mut post: serde_json::Value, user: &'_ Option<String>) -> Option< #[async_trait] impl Storage for RedisStorage { async fn delete_post<'a>(&self, url: &'a str) -> Result<()> { - let mut conn = self.redis.get_async_std_connection().await?; + let mut conn = self.redis.get().await?; Ok(conn.hdel::<&str, &str, ()>("posts", url).await?) } async fn post_exists(&self, url: &str) -> Result<bool> { - let mut conn = self.redis.get_async_std_connection().await?; + let mut conn = self.redis.get().await?; Ok(conn.hexists::<&str, &str, bool>(&"posts", url).await?) } async fn get_post(&self, url: &str) -> Result<Option<serde_json::Value>> { - let mut conn = self.redis.get_async_std_connection().await?; + let mut conn = self.redis.get().await?; match conn.hget::<&str, &str, Option<String>>(&"posts", url).await? { Some(val) => { let parsed = serde_json::from_str::<serde_json::Value>(&val)?; @@ -93,7 +104,7 @@ impl Storage for RedisStorage { } async fn get_channels(&self, user: &User) -> Result<Vec<MicropubChannel>> { - let mut conn = self.redis.get_async_std_connection().await?; + let mut conn = self.redis.get().await?; let channels = conn.smembers::<String, Vec<String>>("channels_".to_string() + user.me.as_str()).await?; // TODO: use streams here instead of this weird thing... how did I even write this?! Ok(futures_util::future::join_all(channels.iter() @@ -111,7 +122,7 @@ impl Storage for RedisStorage { } async fn put_post<'a>(&self, post: &'a serde_json::Value) -> Result<()> { - let mut conn = self.redis.get_async_std_connection().await?; + let mut conn = self.redis.get().await?; let key: &str; match post["properties"]["uid"][0].as_str() { Some(uid) => key = uid, @@ -133,7 +144,7 @@ impl Storage for RedisStorage { } async fn read_feed_with_limit<'a>(&self, url: &'a str, after: &'a Option<String>, limit: usize, user: &'a Option<String>) -> Result<Option<serde_json::Value>> { - let mut conn = self.redis.get_async_std_connection().await?; + let mut conn = self.redis.get().await?; let mut feed; match conn.hget::<&str, &str, Option<String>>(&"posts", url).await? { Some(post) => feed = serde_json::from_str::<serde_json::Value>(&post)?, @@ -161,9 +172,7 @@ impl Storage for RedisStorage { } let posts = stream::iter(posts_iter) .map(|url| async move { - // Is it rational to use a new connection for every post fetched? - // TODO: Use a connection pool here - match self.redis.get_async_std_connection().await { + match self.redis.get().await { Ok(mut conn) => match conn.hget::<&str, &str, Option<String>>("posts", &url).await { Ok(post) => match post { Some(post) => match serde_json::from_str::<serde_json::Value>(&post) { @@ -196,6 +205,7 @@ impl Storage for RedisStorage { // It will probably depend on how often can you encounter a private post on the page // It shouldn't be too large, or we'll start fetching too many posts from the database // It MUST NOT be larger than the typical page size + // It MUST NOT be a significant amount of the connection pool size .buffered(std::cmp::min(3, limit)) // Hack to unwrap the Option and sieve out broken links // Broken links return None, and Stream::filter_map skips all Nones. @@ -214,7 +224,7 @@ impl Storage for RedisStorage { } async fn update_post<'a>(&self, mut url: &'a str, update: serde_json::Value) -> Result<()> { - let mut conn = self.redis.get_async_std_connection().await?; + let mut conn = self.redis.get().await?; if !conn.hexists::<&str, &str, bool>("posts", url).await.unwrap() { return Err(StorageError::new(ErrorKind::NotFound, "can't edit a non-existent post")) } @@ -222,7 +232,7 @@ impl Storage for RedisStorage { if let Some(new_url) = post["see_other"].as_str() { url = new_url } - Ok(SCRIPTS.edit_post.key("posts").arg(url).arg(update.to_string()).invoke_async::<_, ()>(&mut conn).await?) + Ok(SCRIPTS.edit_post.key("posts").arg(url).arg(update.to_string()).invoke_async::<_, ()>(&mut conn as &mut redis::aio::Connection).await?) } } @@ -231,7 +241,7 @@ impl RedisStorage { /// Create a new RedisDatabase that will connect to Redis at `redis_uri` to store data. pub async fn new(redis_uri: String) -> Result<Self> { match redis::Client::open(redis_uri) { - Ok(client) => Ok(Self { redis: client }), + Ok(client) => Ok(Self { redis: Pool::builder().max_open(20).build(RedisConnectionManager::new(client)) }), Err(e) => Err(e.into()) } } |