sova-redis 0.1.4

Shared Redis/Valkey pool plugin for Sova
Documentation 
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
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108

use crate::error::RedisError;
use crate::fake::FakeRedis;
use crate::fake::FakeSubscriber;
use redis::aio::ConnectionManager;
use redis::Client;
use sova_core::Request;
use std::sync::{Arc, RwLock};

#[derive(Default)]
struct PoolInner {
    conn: Option<ConnectionManager>,
    url: Option<String>,
    fake: Option<Arc<FakeRedis>>,
}

/// Shared Redis handle filled during `on_startup` (mirrors [`sova_db::DbPool`]).
#[derive(Clone, Default)]
pub struct RedisPool {
    inner: Arc<RwLock<PoolInner>>,
}

impl RedisPool {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn is_fake(&self) -> bool {
        self.inner.read().unwrap().fake.is_some()
    }

    pub fn fake(&self) -> Option<Arc<FakeRedis>> {
        self.inner.read().unwrap().fake.clone()
    }

    pub fn set_fake(&self, fake: FakeRedis) {
        let mut g = self.inner.write().unwrap();
        g.fake = Some(Arc::new(fake));
        g.url = Some("fake://memory".into());
    }

    /// Remember `REDIS_URL` so Pub/Sub can open a dedicated connection.
    pub fn set_url(&self, url: impl Into<String>) {
        self.inner.write().unwrap().url = Some(url.into());
    }

    pub fn url(&self) -> Result<String, RedisError> {
        self.inner
            .read()
            .unwrap()
            .url
            .clone()
            .ok_or_else(|| RedisError::msg("redis url not set"))
    }

    pub fn set(&self, conn: ConnectionManager) {
        self.inner.write().unwrap().conn = Some(conn);
    }

    pub fn get(&self) -> Result<ConnectionManager, RedisError> {
        let g = self.inner.read().unwrap();
        if g.fake.is_some() {
            return Err(RedisError::msg("fake redis: use FakeRedis API"));
        }
        g.conn
            .clone()
            .ok_or_else(|| RedisError::msg("redis not connected"))
    }

    pub fn clear(&self) {
        let mut g = self.inner.write().unwrap();
        g.conn.take();
    }

    /// Connect and ping; used by the plugin and tests.
    pub async fn connect(url: &str) -> Result<ConnectionManager, RedisError> {
        let client = Client::open(url).map_err(RedisError::from)?;
        let mut conn = ConnectionManager::new(client)
            .await
            .map_err(RedisError::from)?;
        let _: String = redis::cmd("PING")
            .query_async(&mut conn)
            .await
            .map_err(RedisError::from)?;
        Ok(conn)
    }

    pub async fn subscribe_fake(
        &self,
        channel: impl AsRef<str>,
    ) -> Result<FakeSubscriber, RedisError> {
        let fake = self
            .fake()
            .ok_or_else(|| RedisError::msg("fake redis not configured"))?;
        Ok(fake.subscribe(channel.as_ref()))
    }
}

/// Convenient `req.redis()`.
pub trait RedisExt {
    fn redis(&self) -> &RedisPool;
}

impl RedisExt for Request {
    fn redis(&self) -> &RedisPool {
        self.get::<RedisPool>()
            .expect("Redis plugin is not installed (missing req.redis())")
    }
}