Spaces:
Running
Running
| //! This module provides the functionality to cache the aggregated results fetched and aggregated | |
| //! from the upstream search engines in a json format. | |
| use error_stack::Report; | |
| use mini_moka::sync::Cache as MokaCache; | |
| use std::time::Duration; | |
| use tokio::sync::Mutex; | |
| use crate::{config::parser::Config, models::aggregation_models::SearchResults}; | |
| use super::error::CacheError; | |
| use super::redis_cacher::RedisCache; | |
| /// Different implementations for caching, currently it is possible to cache in-memory or in Redis. | |
| pub enum Cache { | |
| /// Caching is disabled | |
| Disabled, | |
| /// Encapsulates the Redis based cache | |
| Redis(RedisCache), | |
| /// Contains the in-memory cache. | |
| InMemory(MokaCache<String, SearchResults>), | |
| /// Contains both the in-memory cache and Redis based cache | |
| Hybrid(RedisCache, MokaCache<String, SearchResults>), | |
| } | |
| impl Cache { | |
| /// A function that builds the cache from the given configuration. | |
| /// | |
| /// # Arguments | |
| /// | |
| /// * `config` - It takes the config struct as an argument. | |
| /// | |
| /// # Returns | |
| /// | |
| /// It returns a newly initialized variant based on the feature enabled by the user. | |
| pub async fn build(_config: &Config) -> Self { | |
| { | |
| log::info!("Using a hybrid cache"); | |
| Cache::new_hybrid( | |
| RedisCache::new(&_config.redis_url, 5) | |
| .await | |
| .expect("Redis cache configured"), | |
| ) | |
| } | |
| { | |
| log::info!("Listening redis server on {}", &_config.redis_url); | |
| Cache::new( | |
| RedisCache::new(&_config.redis_url, 5) | |
| .await | |
| .expect("Redis cache configured"), | |
| ) | |
| } | |
| { | |
| log::info!("Using an in-memory cache"); | |
| Cache::new_in_memory() | |
| } | |
| { | |
| log::info!("Caching is disabled"); | |
| Cache::Disabled | |
| } | |
| } | |
| /// A function that initializes a new connection pool struct. | |
| /// | |
| /// # Arguments | |
| /// | |
| /// * `redis_cache` - It takes the newly initialized connection pool struct as an argument. | |
| /// | |
| /// # Returns | |
| /// | |
| /// It returns a `Redis` variant with the newly initialized connection pool struct. | |
| pub fn new(redis_cache: RedisCache) -> Self { | |
| Cache::Redis(redis_cache) | |
| } | |
| /// A function that initializes the `in memory` cache which is used to cache the results in | |
| /// memory with the search engine thus improving performance by making retrieval and caching of | |
| /// results faster. | |
| /// | |
| /// # Returns | |
| /// | |
| /// It returns a `InMemory` variant with the newly initialized in memory cache type. | |
| pub fn new_in_memory() -> Self { | |
| let cache = MokaCache::builder() | |
| .max_capacity(1000) | |
| .time_to_live(Duration::from_secs(60)) | |
| .build(); | |
| Cache::InMemory(cache) | |
| } | |
| /// A function that initializes both in memory cache and redis client connection for being used | |
| /// for managing hybrid cache which increases resiliancy of the search engine by allowing the | |
| /// cache to switch to `in memory` caching if the `redis` cache server is temporarily | |
| /// unavailable. | |
| /// | |
| /// # Arguments | |
| /// | |
| /// * `redis_cache` - It takes `redis` client connection struct as an argument. | |
| /// | |
| /// # Returns | |
| /// | |
| /// It returns a tuple variant `Hybrid` storing both the in-memory cache type and the `redis` | |
| /// client connection struct. | |
| pub fn new_hybrid(redis_cache: RedisCache) -> Self { | |
| let cache = MokaCache::builder() | |
| .max_capacity(1000) | |
| .time_to_live(Duration::from_secs(60)) | |
| .build(); | |
| Cache::Hybrid(redis_cache, cache) | |
| } | |
| /// A function which fetches the cached json results as json string. | |
| /// | |
| /// # Arguments | |
| /// | |
| /// * `url` - It takes an url as a string. | |
| /// | |
| /// # Error | |
| /// | |
| /// Returns the `SearchResults` from the cache if the program executes normally otherwise | |
| /// returns a `CacheError` if the results cannot be retrieved from the cache. | |
| pub async fn cached_json(&mut self, _url: &str) -> Result<SearchResults, Report<CacheError>> { | |
| match self { | |
| Cache::Disabled => Err(Report::new(CacheError::MissingValue)), | |
| Cache::Redis(redis_cache) => { | |
| let json = redis_cache.cached_json(_url).await?; | |
| Ok(serde_json::from_str::<SearchResults>(&json) | |
| .map_err(|_| CacheError::SerializationError)?) | |
| } | |
| Cache::InMemory(in_memory) => match in_memory.get(&_url.to_string()) { | |
| Some(res) => Ok(res), | |
| None => Err(Report::new(CacheError::MissingValue)), | |
| }, | |
| Cache::Hybrid(redis_cache, in_memory) => match redis_cache.cached_json(_url).await { | |
| Ok(res) => Ok(serde_json::from_str::<SearchResults>(&res) | |
| .map_err(|_| CacheError::SerializationError)?), | |
| Err(_) => match in_memory.get(&_url.to_string()) { | |
| Some(res) => Ok(res), | |
| None => Err(Report::new(CacheError::MissingValue)), | |
| }, | |
| }, | |
| } | |
| } | |
| /// A function which caches the results by using the `url` as the key and | |
| /// `json results` as the value and stores it in the cache | |
| /// | |
| /// # Arguments | |
| /// | |
| /// * `json_results` - It takes the json results string as an argument. | |
| /// * `url` - It takes the url as a String. | |
| /// | |
| /// # Error | |
| /// | |
| /// Returns a unit type if the program caches the given search results without a failure | |
| /// otherwise it returns a `CacheError` if the search results cannot be cached due to a | |
| /// failure. | |
| pub async fn cache_results( | |
| &mut self, | |
| _search_results: &SearchResults, | |
| _url: &str, | |
| ) -> Result<(), Report<CacheError>> { | |
| match self { | |
| Cache::Disabled => Ok(()), | |
| Cache::Redis(redis_cache) => { | |
| let json = serde_json::to_string(_search_results) | |
| .map_err(|_| CacheError::SerializationError)?; | |
| redis_cache.cache_results(&json, _url).await | |
| } | |
| Cache::InMemory(cache) => { | |
| cache.insert(_url.to_string(), _search_results.clone()); | |
| Ok(()) | |
| } | |
| Cache::Hybrid(redis_cache, cache) => { | |
| let json = serde_json::to_string(_search_results) | |
| .map_err(|_| CacheError::SerializationError)?; | |
| match redis_cache.cache_results(&json, _url).await { | |
| Ok(_) => Ok(()), | |
| Err(_) => { | |
| cache.insert(_url.to_string(), _search_results.clone()); | |
| Ok(()) | |
| } | |
| } | |
| } | |
| } | |
| } | |
| } | |
| /// A structure to efficiently share the cache between threads - as it is protected by a Mutex. | |
| pub struct SharedCache { | |
| /// The internal cache protected from concurrent access by a mutex | |
| cache: Mutex<Cache>, | |
| } | |
| impl SharedCache { | |
| /// A function that creates a new `SharedCache` from a Cache implementation. | |
| /// | |
| /// # Arguments | |
| /// | |
| /// * `cache` - It takes the `Cache` enum variant as an argument with the prefered cache type. | |
| /// | |
| /// Returns a newly constructed `SharedCache` struct. | |
| pub fn new(cache: Cache) -> Self { | |
| Self { | |
| cache: Mutex::new(cache), | |
| } | |
| } | |
| /// A getter function which retrieves the cached SearchResulsts from the internal cache. | |
| /// | |
| /// # Arguments | |
| /// | |
| /// * `url` - It takes the search url as an argument which will be used as the key to fetch the | |
| /// cached results from the cache. | |
| /// | |
| /// # Error | |
| /// | |
| /// Returns a `SearchResults` struct containing the search results from the cache if nothing | |
| /// goes wrong otherwise returns a `CacheError`. | |
| pub async fn cached_json(&self, url: &str) -> Result<SearchResults, Report<CacheError>> { | |
| let mut mut_cache = self.cache.lock().await; | |
| mut_cache.cached_json(url).await | |
| } | |
| /// A setter function which caches the results by using the `url` as the key and | |
| /// `SearchResults` as the value. | |
| /// | |
| /// # Arguments | |
| /// | |
| /// * `search_results` - It takes the `SearchResults` as an argument which are results that | |
| /// needs to be cached. | |
| /// * `url` - It takes the search url as an argument which will be used as the key for storing | |
| /// results in the cache. | |
| /// | |
| /// # Error | |
| /// | |
| /// Returns an unit type if the results are cached succesfully otherwise returns a `CacheError` | |
| /// on a failure. | |
| pub async fn cache_results( | |
| &self, | |
| search_results: &SearchResults, | |
| url: &str, | |
| ) -> Result<(), Report<CacheError>> { | |
| let mut mut_cache = self.cache.lock().await; | |
| mut_cache.cache_results(search_results, url).await | |
| } | |
| } | |