rust_webx_core/cache/
cache_ext.rs

1//! Typed cache extensions — matches ASP.NET Core's `DistributedCacheExtensions`.
2//!
3//| ASP.NET Core                          | rust-webx                               |
4//|---------------------------------------|-----------------------------------------|
5//| `GetStringAsync(cache, key, token)`   | `get_string(&self, key)`                |
6//| `SetStringAsync(cache, key, val, opts)`| `set_string(&self, key, &val, opts)`    |
7
8use super::options::DistributedCacheEntryOptions;
9use super::trait_def::{CacheError, IDistributedCache, Result};
10use std::collections::HashMap;
11use std::future::Future;
12use std::sync::{Arc, OnceLock};
13use tokio::sync::Mutex;
14
15type KeyLockMap = Mutex<HashMap<String, Arc<Mutex<()>>>>;
16
17fn key_locks() -> &'static KeyLockMap {
18    static KEY_LOCKS: OnceLock<KeyLockMap> = OnceLock::new();
19    KEY_LOCKS.get_or_init(|| Mutex::new(HashMap::new()))
20}
21
22async fn with_key_lock<F, Fut, T>(key: &str, f: F) -> T
23where
24    F: FnOnce() -> Fut + Send,
25    Fut: Future<Output = T> + Send,
26    T: Send,
27{
28    let lock = {
29        let mut map = key_locks().lock().await;
30        map.entry(key.to_string())
31            .or_insert_with(|| Arc::new(Mutex::new(())))
32            .clone()
33    };
34    let _guard = lock.lock().await;
35    f().await
36}
37
38#[async_trait::async_trait]
39pub trait DistributedCacheExtensions: IDistributedCache {
40    async fn get_string<T: serde::de::DeserializeOwned + Send>(
41        &self,
42        key: &str,
43    ) -> Result<Option<T>>;
44    async fn set_string<T: serde::Serialize + Send + Sync>(
45        &self,
46        key: &str,
47        val: &T,
48        opts: &DistributedCacheEntryOptions,
49    ) -> Result<()>;
50    async fn get_or_create<T, F, Fut>(
51        &self,
52        key: &str,
53        factory: F,
54        opts: &DistributedCacheEntryOptions,
55    ) -> Result<T>
56    where
57        T: serde::Serialize + serde::de::DeserializeOwned + Send + Sync + Clone + 'static,
58        F: FnOnce() -> Fut + Send,
59        Fut: Future<Output = T> + Send;
60    async fn get_or_try_create<T, F, Fut, E>(
61        &self,
62        key: &str,
63        factory: F,
64        opts: &DistributedCacheEntryOptions,
65    ) -> Result<T>
66    where
67        T: serde::Serialize + serde::de::DeserializeOwned + Send + Sync + Clone + 'static,
68        F: FnOnce() -> Fut + Send,
69        Fut: Future<Output = std::result::Result<T, E>> + Send,
70        E: std::fmt::Display;
71}
72
73#[async_trait::async_trait]
74impl<T: IDistributedCache + ?Sized + Sync> DistributedCacheExtensions for T {
75    async fn get_string<U: serde::de::DeserializeOwned + Send>(
76        &self,
77        key: &str,
78    ) -> Result<Option<U>> {
79        let bytes = self.get(key).await?;
80        match bytes {
81            Some(data) => Ok(Some(
82                serde_json::from_slice(&data)
83                    .map_err(|e| CacheError::Serialization(e.to_string()))?,
84            )),
85            None => Ok(None),
86        }
87    }
88
89    async fn set_string<U: serde::Serialize + Send + Sync>(
90        &self,
91        key: &str,
92        val: &U,
93        opts: &DistributedCacheEntryOptions,
94    ) -> Result<()> {
95        let data = serde_json::to_vec(val).map_err(|e| CacheError::Serialization(e.to_string()))?;
96        if opts.size_limit > 0 && data.len() > opts.size_limit {
97            return Err(CacheError::Message(format!(
98                "value size {} exceeds limit {}",
99                data.len(),
100                opts.size_limit
101            )));
102        }
103        self.set(key, data, Some(opts)).await
104    }
105
106    async fn get_or_create<U, F, Fut>(
107        &self,
108        key: &str,
109        factory: F,
110        opts: &DistributedCacheEntryOptions,
111    ) -> Result<U>
112    where
113        U: serde::Serialize + serde::de::DeserializeOwned + Send + Sync + Clone + 'static,
114        F: FnOnce() -> Fut + Send,
115        Fut: Future<Output = U> + Send,
116    {
117        if let Some(val) = self.get_string::<U>(key).await? {
118            return Ok(val);
119        }
120        with_key_lock(key, || async {
121            if let Some(val) = self.get_string::<U>(key).await? {
122                return Ok(val);
123            }
124            let val = factory().await;
125            self.set_string(key, &val, opts).await?;
126            Ok(val)
127        })
128        .await
129    }
130
131    async fn get_or_try_create<U, F, Fut, E>(
132        &self,
133        key: &str,
134        factory: F,
135        opts: &DistributedCacheEntryOptions,
136    ) -> Result<U>
137    where
138        U: serde::Serialize + serde::de::DeserializeOwned + Send + Sync + Clone + 'static,
139        F: FnOnce() -> Fut + Send,
140        Fut: Future<Output = std::result::Result<U, E>> + Send,
141        E: std::fmt::Display,
142    {
143        if let Some(val) = self.get_string::<U>(key).await? {
144            return Ok(val);
145        }
146        with_key_lock(key, || async {
147            if let Some(val) = self.get_string::<U>(key).await? {
148                return Ok(val);
149            }
150            let val = factory()
151                .await
152                .map_err(|e| CacheError::Message(e.to_string()))?;
153            self.set_string(key, &val, opts).await?;
154            Ok(val)
155        })
156        .await
157    }
158}
rust_webx_core/cache/cache_ext.rs

rust_webx_core/cache/
cache_ext.rs
