armature_cache/
parallel.rs

1//! Parallel batch operations for cache stores.
2
3use crate::error::{CacheError, CacheResult};
4use crate::traits::CacheStore;
5use futures::future::{join_all, try_join_all};
6use serde::{Serialize, de::DeserializeOwned};
7use std::collections::HashMap;
8use std::time::Duration;
9
10/// Parallel batch operations for cache stores.
11///
12/// This module provides high-performance batch operations that execute
13/// multiple cache operations concurrently, significantly reducing total latency.
14///
15/// # Performance
16///
17/// - **get_many**: 10-100x faster than sequential gets (depending on network latency)
18/// - **set_many**: 10-100x faster than sequential sets
19/// - **delete_many**: Similar performance gains
20///
21/// # Examples
22///
23/// ```no_run
24/// use armature_cache::*;
25/// use armature_cache::parallel::*;
26///
27/// # async fn example() -> CacheResult<()> {
28/// let cache = RedisCache::new(CacheConfig::redis("redis://localhost:6379")?).await?;
29///
30/// // Get multiple keys in parallel
31/// let keys = vec!["user:1", "user:2", "user:3"];
32/// let values = get_many_json(&cache, &keys).await?;
33///
34/// // Set multiple keys in parallel
35/// let items = vec![
36///     ("key1", "value1".to_string()),
37///     ("key2", "value2".to_string()),
38/// ];
39/// set_many_json(&cache, &items, None).await?;
40/// # Ok(())
41/// # }
42/// ```
43pub struct ParallelCacheOps;
44
45impl ParallelCacheOps {
46    /// Get multiple JSON values in parallel.
47    ///
48    /// # Arguments
49    ///
50    /// * `store` - The cache store
51    /// * `keys` - Slice of keys to fetch
52    ///
53    /// # Returns
54    ///
55    /// A vector of optional values in the same order as keys.
56    ///
57    /// # Examples
58    ///
59    /// ```no_run
60    /// use armature_cache::*;
61    /// use armature_cache::parallel::ParallelCacheOps;
62    ///
63    /// # async fn example() -> CacheResult<()> {
64    /// let cache = RedisCache::new(CacheConfig::redis("redis://localhost:6379")?).await?;
65    ///
66    /// let keys = vec!["key1", "key2", "key3"];
67    /// let values = ParallelCacheOps::get_many_json(&cache, &keys).await?;
68    ///
69    /// for (key, value) in keys.iter().zip(values.iter()) {
70    ///     println!("{}: {:?}", key, value);
71    /// }
72    /// # Ok(())
73    /// # }
74    /// ```
75    pub async fn get_many_json<S: CacheStore>(
76        store: &S,
77        keys: &[&str],
78    ) -> CacheResult<Vec<Option<String>>> {
79        let futures = keys.iter().map(|key| store.get_json(key));
80        let results: Vec<CacheResult<Option<String>>> = join_all(futures).await;
81
82        results.into_iter().collect()
83    }
84
85    /// Get multiple typed values in parallel.
86    ///
87    /// # Type Parameters
88    ///
89    /// * `T` - The type to deserialize into
90    ///
91    /// # Examples
92    ///
93    /// ```no_run
94    /// use armature_cache::*;
95    /// use armature_cache::parallel::ParallelCacheOps;
96    /// use serde::{Deserialize, Serialize};
97    ///
98    /// #[derive(Serialize, Deserialize)]
99    /// struct User {
100    ///     id: u64,
101    ///     name: String,
102    /// }
103    ///
104    /// # async fn example() -> CacheResult<()> {
105    /// let cache = RedisCache::new(CacheConfig::redis("redis://localhost:6379")?).await?;
106    ///
107    /// let keys = vec!["user:1", "user:2", "user:3"];
108    /// let users: Vec<Option<User>> = ParallelCacheOps::get_many(&cache, &keys).await?;
109    /// # Ok(())
110    /// # }
111    /// ```
112    pub async fn get_many<S: CacheStore, T: DeserializeOwned>(
113        store: &S,
114        keys: &[&str],
115    ) -> CacheResult<Vec<Option<T>>> {
116        let json_values = Self::get_many_json(store, keys).await?;
117
118        json_values
119            .into_iter()
120            .map(|opt_json| {
121                opt_json
122                    .map(|json| {
123                        serde_json::from_str(&json)
124                            .map_err(|e| CacheError::Deserialization(e.to_string()))
125                    })
126                    .transpose()
127            })
128            .collect()
129    }
130
131    /// Set multiple JSON values in parallel.
132    ///
133    /// # Arguments
134    ///
135    /// * `store` - The cache store
136    /// * `items` - Slice of (key, value) tuples
137    /// * `ttl` - Optional time-to-live for all items
138    ///
139    /// # Examples
140    ///
141    /// ```no_run
142    /// use armature_cache::*;
143    /// use armature_cache::parallel::ParallelCacheOps;
144    /// use std::time::Duration;
145    ///
146    /// # async fn example() -> CacheResult<()> {
147    /// let cache = RedisCache::new(CacheConfig::redis("redis://localhost:6379")?).await?;
148    ///
149    /// let items = vec![
150    ///     ("key1", r#"{"value": 1}"#.to_string()),
151    ///     ("key2", r#"{"value": 2}"#.to_string()),
152    /// ];
153    ///
154    /// ParallelCacheOps::set_many_json(&cache, &items, Some(Duration::from_secs(3600))).await?;
155    /// # Ok(())
156    /// # }
157    /// ```
158    pub async fn set_many_json<S: CacheStore>(
159        store: &S,
160        items: &[(&str, String)],
161        ttl: Option<Duration>,
162    ) -> CacheResult<()> {
163        let futures = items
164            .iter()
165            .map(|(key, value)| store.set_json(key, value.clone(), ttl));
166
167        try_join_all(futures).await?;
168        Ok(())
169    }
170
171    /// Set multiple typed values in parallel.
172    ///
173    /// # Type Parameters
174    ///
175    /// * `T` - The type to serialize from
176    ///
177    /// # Examples
178    ///
179    /// ```no_run
180    /// use armature_cache::*;
181    /// use armature_cache::parallel::ParallelCacheOps;
182    /// use serde::{Deserialize, Serialize};
183    ///
184    /// #[derive(Serialize, Deserialize)]
185    /// struct Counter {
186    ///     count: u64,
187    /// }
188    ///
189    /// # async fn example() -> CacheResult<()> {
190    /// let cache = RedisCache::new(CacheConfig::redis("redis://localhost:6379")?).await?;
191    ///
192    /// let items = vec![
193    ///     ("counter:1", Counter { count: 10 }),
194    ///     ("counter:2", Counter { count: 20 }),
195    /// ];
196    ///
197    /// ParallelCacheOps::set_many(&cache, &items, None).await?;
198    /// # Ok(())
199    /// # }
200    /// ```
201    pub async fn set_many<S: CacheStore, T: Serialize>(
202        store: &S,
203        items: &[(&str, T)],
204        ttl: Option<Duration>,
205    ) -> CacheResult<()> {
206        let json_items: Result<Vec<_>, _> = items
207            .iter()
208            .map(|(key, value)| {
209                serde_json::to_string(value)
210                    .map(|json| (*key, json))
211                    .map_err(|e| CacheError::Serialization(e.to_string()))
212            })
213            .collect();
214
215        let json_items = json_items?;
216        let item_refs: Vec<_> = json_items.iter().map(|(k, v)| (*k, v.clone())).collect();
217
218        Self::set_many_json(store, &item_refs, ttl).await
219    }
220
221    /// Delete multiple keys in parallel.
222    ///
223    /// # Examples
224    ///
225    /// ```no_run
226    /// use armature_cache::*;
227    /// use armature_cache::parallel::ParallelCacheOps;
228    ///
229    /// # async fn example() -> CacheResult<()> {
230    /// let cache = RedisCache::new(CacheConfig::redis("redis://localhost:6379")?).await?;
231    ///
232    /// let keys = vec!["key1", "key2", "key3"];
233    /// ParallelCacheOps::delete_many(&cache, &keys).await?;
234    /// # Ok(())
235    /// # }
236    /// ```
237    pub async fn delete_many<S: CacheStore>(store: &S, keys: &[&str]) -> CacheResult<()> {
238        let futures = keys.iter().map(|key| store.delete(key));
239        try_join_all(futures).await?;
240        Ok(())
241    }
242
243    /// Check if multiple keys exist in parallel.
244    ///
245    /// # Returns
246    ///
247    /// A vector of booleans indicating existence, in the same order as keys.
248    ///
249    /// # Examples
250    ///
251    /// ```no_run
252    /// use armature_cache::*;
253    /// use armature_cache::parallel::ParallelCacheOps;
254    ///
255    /// # async fn example() -> CacheResult<()> {
256    /// let cache = RedisCache::new(CacheConfig::redis("redis://localhost:6379")?).await?;
257    ///
258    /// let keys = vec!["key1", "key2", "key3"];
259    /// let exists = ParallelCacheOps::exists_many(&cache, &keys).await?;
260    ///
261    /// for (key, exists) in keys.iter().zip(exists.iter()) {
262    ///     println!("{}: {}", key, exists);
263    /// }
264    /// # Ok(())
265    /// # }
266    /// ```
267    pub async fn exists_many<S: CacheStore>(store: &S, keys: &[&str]) -> CacheResult<Vec<bool>> {
268        let futures = keys.iter().map(|key| store.exists(key));
269        let results: Vec<CacheResult<bool>> = join_all(futures).await;
270
271        results.into_iter().collect()
272    }
273
274    /// Get TTL for multiple keys in parallel.
275    ///
276    /// # Returns
277    ///
278    /// A vector of optional durations, in the same order as keys.
279    ///
280    /// # Examples
281    ///
282    /// ```no_run
283    /// use armature_cache::*;
284    /// use armature_cache::parallel::ParallelCacheOps;
285    ///
286    /// # async fn example() -> CacheResult<()> {
287    /// let cache = RedisCache::new(CacheConfig::redis("redis://localhost:6379")?).await?;
288    ///
289    /// let keys = vec!["key1", "key2", "key3"];
290    /// let ttls = ParallelCacheOps::ttl_many(&cache, &keys).await?;
291    ///
292    /// for (key, ttl) in keys.iter().zip(ttls.iter()) {
293    ///     println!("{}: {:?}", key, ttl);
294    /// }
295    /// # Ok(())
296    /// # }
297    /// ```
298    pub async fn ttl_many<S: CacheStore>(
299        store: &S,
300        keys: &[&str],
301    ) -> CacheResult<Vec<Option<Duration>>> {
302        let futures = keys.iter().map(|key| store.ttl(key));
303        let results: Vec<CacheResult<Option<Duration>>> = join_all(futures).await;
304
305        results.into_iter().collect()
306    }
307
308    /// Cache warming: preload multiple keys into cache.
309    ///
310    /// # Type Parameters
311    ///
312    /// * `T` - The type to serialize
313    /// * `F` - Factory function that returns data for a given key
314    ///
315    /// # Examples
316    ///
317    /// ```ignore
318    /// use armature_cache::*;
319    /// use armature_cache::parallel::ParallelCacheOps;
320    /// use std::time::Duration;
321    ///
322    /// # async fn example() -> CacheResult<()> {
323    /// let cache = RedisCache::new(CacheConfig::redis("redis://localhost:6379")?).await?;
324    ///
325    /// let keys = vec!["user:1", "user:2", "user:3"];
326    ///
327    /// ParallelCacheOps::warm_cache(
328    ///     &cache,
329    ///     &keys,
330    ///     Some(Duration::from_secs(3600)),
331    ///     |key: &str| async move {
332    ///         // Fetch from database
333    ///         let data = format!("Data for {}", key);
334    ///         Ok::<String, CacheError>(data)
335    ///     },
336    /// ).await?;
337    /// # Ok(())
338    /// # }
339    /// ```
340    pub async fn warm_cache<S, T, F, Fut>(
341        store: &S,
342        keys: &[&str],
343        ttl: Option<Duration>,
344        factory: F,
345    ) -> CacheResult<()>
346    where
347        S: CacheStore,
348        T: Serialize,
349        F: Fn(&str) -> Fut,
350        Fut: std::future::Future<Output = CacheResult<T>>,
351    {
352        let mut futures = Vec::new();
353
354        for key in keys {
355            let fut = async {
356                let value = factory(key).await?;
357                let json = serde_json::to_string(&value)
358                    .map_err(|e| CacheError::Serialization(e.to_string()))?;
359                store.set_json(key, json, ttl).await?;
360                Ok::<(), CacheError>(())
361            };
362            futures.push(fut);
363        }
364
365        try_join_all(futures).await?;
366        Ok(())
367    }
368}
369
370/// Helper functions for parallel cache operations.
371///
372/// These functions provide a more convenient API than `ParallelCacheOps` methods.
373/// Get multiple JSON values in parallel.
374pub async fn get_many_json<S: CacheStore>(
375    store: &S,
376    keys: &[&str],
377) -> CacheResult<Vec<Option<String>>> {
378    ParallelCacheOps::get_many_json(store, keys).await
379}
380
381/// Get multiple typed values in parallel.
382pub async fn get_many<S: CacheStore, T: DeserializeOwned>(
383    store: &S,
384    keys: &[&str],
385) -> CacheResult<Vec<Option<T>>> {
386    ParallelCacheOps::get_many(store, keys).await
387}
388
389/// Set multiple JSON values in parallel.
390pub async fn set_many_json<S: CacheStore>(
391    store: &S,
392    items: &[(&str, String)],
393    ttl: Option<Duration>,
394) -> CacheResult<()> {
395    ParallelCacheOps::set_many_json(store, items, ttl).await
396}
397
398/// Set multiple typed values in parallel.
399pub async fn set_many<S: CacheStore, T: Serialize>(
400    store: &S,
401    items: &[(&str, T)],
402    ttl: Option<Duration>,
403) -> CacheResult<()> {
404    ParallelCacheOps::set_many(store, items, ttl).await
405}
406
407/// Delete multiple keys in parallel.
408pub async fn delete_many<S: CacheStore>(store: &S, keys: &[&str]) -> CacheResult<()> {
409    ParallelCacheOps::delete_many(store, keys).await
410}
411
412/// Build a HashMap from multiple keys fetched in parallel.
413pub async fn get_many_as_map<S: CacheStore, T: DeserializeOwned>(
414    store: &S,
415    keys: &[&str],
416) -> CacheResult<HashMap<String, T>> {
417    let values = get_many(store, keys).await?;
418
419    let map: HashMap<String, T> = keys
420        .iter()
421        .zip(values.into_iter())
422        .filter_map(|(key, opt_value)| opt_value.map(|value| (key.to_string(), value)))
423        .collect();
424
425    Ok(map)
426}
427
428#[cfg(test)]
429mod tests {
430    #[test]
431    fn test_parallel_ops_exist() {
432        // Ensure the module compiles - this test validates the module is correctly defined
433    }
434}
armature_cache/parallel.rs

armature_cache/
parallel.rs
