Redis RS Async Connection Pool

use redis::{aio::MultiplexedConnection, Client, RedisResult, cmd};
use std::time::Duration;

async fn warm_client(url: &str) -> RedisResult<MultiplexedConnection> {
    let client = Client::open(url)?;

    let connection = tokio::time::timeout(
        Duration::from_secs(2),
        client.get_multiplexed_connection(),
    )
    .await
    .map_err(|_| redis::RedisError::from((redis::ErrorKind::IoError, "connect timeout")))??;

    Ok(connection)
}

async fn cache_aside(
    con: &mut MultiplexedConnection,
    key: &str,
    loader: impl FnOnce() -> String,
) -> RedisResult<String> {
    // pipeline 一并发出 GET 与 TTL，单次往返
    let (value, ttl): (Option<String>, i64) = redis::pipe()
        .cmd("GET")
        .arg(key)
        .ignore()
        .cmd("TTL")
        .arg(key)
        .query_async(con)
        .await?;

    if let Some(hit) = value {
        println!("{key} cached, ttl {ttl}s");
        return Ok(hit);
    }

    let fresh = loader();
    let _: () = cmd("SET")
        .arg(key)
        .arg(&fresh)
        .arg("EX")
        .arg(300)
        .query_async(con)
        .await?;
    println!("{key} loaded from source, cached 300s");
    Ok(fresh)
}

#[tokio::main]
async fn main() -> RedisResult<()> {
    let mut con = warm_client("redis://127.0.0.1:6379/0").await?;

    let value = cache_aside(&mut con, "rates:usd", || "1.0".to_string()).await?;
    println!("rate: {value}");

    let pattern = "rates:*";
    let keys: Vec<String> = cmd("KEYS").arg(pattern).query_async(&mut con).await?;
    println!("cached keys: {keys:?}");

    let _: () = cmd("DEL").arg("rates:usd").query_async(&mut con).await?;
    Ok(())
}