1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
//! Shared helper for the "list all items referenced by a Redis set" pattern.
//!
//! Both `AuthExt::list_policies` (in [`crate::auth`]) and `SchemaRegistryExt::list_schemas`
//! (in [`crate::schema::registry_impl`]) connect to Redis, read the members of a set that
//! tracks known item names, then fetch each item individually — skipping any that are
//! missing. [`collect_from_redis_set`] factors that shape out so each call site only
//! supplies the set key and a per-item fetcher.
use crate::context::Context;
use crate::error::{Mecha10Error, Result};
use redis::AsyncCommands;
use std::future::Future;
/// Connects to Redis (via `Context::get_redis_url`) and returns a multiplexed async
/// connection.
///
/// Every Redis-backed call site in this crate (schema registry, auth policies, and this
/// module's own `collect_from_redis_set`/`get_optional_from_redis` helpers) needs exactly
/// this connect-then-multiplex step before doing any actual work; factoring it out here
/// collapses what was a ~13-line duplicated block at each call site into a single call.
pub(crate) async fn get_redis_connection() -> Result<redis::aio::MultiplexedConnection> {
let redis_url = Context::get_redis_url()?;
let client = redis::Client::open(redis_url.as_str()).map_err(|e| Mecha10Error::MessagingError {
message: format!("Failed to connect to Redis: {}", e),
suggestion: "Ensure Redis is running".to_string(),
})?;
client
.get_multiplexed_async_connection()
.await
.map_err(|e| Mecha10Error::MessagingError {
message: format!("Failed to get Redis connection: {}", e),
suggestion: "Ensure Redis is running".to_string(),
})
}
/// Connects to Redis, reads the members of `list_key`, and fetches each one via
/// `get_item`, collecting the ones that are found.
pub(crate) async fn collect_from_redis_set<T, F, Fut>(list_key: &str, get_item: F) -> Result<Vec<T>>
where
F: Fn(String) -> Fut,
Fut: Future<Output = Result<Option<T>>> + Send,
{
let mut conn = get_redis_connection().await?;
let keys: Vec<String> = conn
.smembers(list_key)
.await
.map_err(|e| Mecha10Error::MessagingError {
message: format!("Failed to get list: {}", e),
suggestion: "Check Redis connection".to_string(),
})?;
let mut items = Vec::new();
for key in keys {
if let Some(item) = get_item(key).await? {
items.push(item);
}
}
Ok(items)
}
/// Connects to Redis and reads a single optional value for `key`.
///
/// Shared by `SchemaRegistryExt::get_latest_schema` (in [`crate::schema::registry_impl`]) and
/// `AuthExt::get_role` (in [`crate::auth`]), which both connect to Redis and fetch one optional
/// value for a key, differing only in the key, the value type, and the error message used if
/// the `GET` itself fails.
pub(crate) async fn get_optional_from_redis<V>(key: &str, get_error: &str) -> Result<Option<V>>
where
V: redis::FromRedisValue,
{
let mut conn = get_redis_connection().await?;
conn.get(key).await.map_err(|e| Mecha10Error::MessagingError {
message: format!("{}: {}", get_error, e),
suggestion: "Check Redis connection".to_string(),
})
}