flippico_cache/providers/
redis.rs

1use futures_util::stream::StreamExt;
2use log::info;
3use redis::{Client, Commands, RedisResult};
4use serde_json;
5use std::future::Future;
6use std::pin::Pin;
7
8use crate::types::{
9    channels::{ChannelMessage, ListChannel, ListMessage, SubscriptionChannel},
10    provider::{
11        AsyncCallback, CacheProvider, Connectable, FifoProvider, ProviderTrait, PubSubProvider,
12        PubSubProviderError,
13    },
14};
15
16pub struct Redis {
17    pub url: String,
18    pub client: Option<Client>,
19}
20
21impl ProviderTrait for Redis {
22    fn new(url: String) -> Self {
23        Redis { url, client: None }
24    }
25}
26
27impl Connectable for Redis {
28    fn set_client(&mut self) {
29        let client = redis::Client::open(self.url.clone()).unwrap();
30        self.client = Some(client);
31    }
32
33    fn get_connection(&self) -> &Client {
34        self.client.as_ref().unwrap()
35    }
36
37    fn get_mut_connection(&mut self) -> &mut Client {
38        self.client.as_mut().unwrap()
39    }
40}
41
42impl PubSubProvider for Redis
43where
44    Redis: Connectable,
45{
46    type Channels = SubscriptionChannel;
47
48    fn subscribe(
49        &self,
50        callback: AsyncCallback<ChannelMessage>,
51        channel: Self::Channels,
52    ) -> Pin<Box<dyn Future<Output = Result<(), PubSubProviderError>> + Send + '_>> {
53        Box::pin(async move {
54            let client = self.get_connection();
55            let mut subscriber = client
56                .get_async_pubsub()
57                .await
58                .map_err(|_| "Failed to get async pubsub")?;
59            subscriber
60                .subscribe(&[channel.get_channel()])
61                .await
62                .map_err(|_| "Failed to subscribe to channel")?;
63            let mut stream = subscriber.into_on_message();
64            while let Some(msg) = stream.next().await {
65                let payload = msg.get_payload::<String>();
66                if let Ok(json_str) = payload {
67                    if let Ok(channel_msg) = serde_json::from_str::<ChannelMessage>(&json_str) {
68                        info!(
69                            "[{}] New message: {}",
70                            channel_msg.channel.get_channel(),
71                            &json_str
72                        );
73                        callback(channel_msg).await;
74                    }
75                }
76            }
77
78            Ok(())
79        })
80    }
81
82    fn publish(&mut self, channel: Self::Channels, message: ChannelMessage) {
83        let client = self.get_mut_connection();
84        if let Ok(json_str) = serde_json::to_string(&message) {
85            let _ = client.publish::<&str, String, String>(channel.get_channel(), json_str);
86        }
87    }
88}
89
90impl FifoProvider for Redis
91where
92    Redis: Connectable,
93{
94    fn pop(
95        &self,
96        channel_name: ListChannel,
97        job_key: &str,
98        callback: AsyncCallback<ListMessage>,
99    ) -> Pin<Box<dyn Future<Output = Result<(), PubSubProviderError>> + Send + '_>> {
100        let list_name = self.get_list_name(channel_name.get_channel(), job_key);
101
102        Box::pin(async move {
103            let client: &Client = self.get_connection();
104            let mut client = client.clone();
105            loop {
106                let result = client
107                    .brpop(&list_name, 0.0)
108                    .map(|mut items: Vec<String>| items.pop().unwrap_or_default());
109
110                match result {
111                    Ok(str) => {
112                        if let Ok(channel_msg) = serde_json::from_str::<ListMessage>(&str) {
113                            info!("Popped message: {}", &str);
114                            callback(channel_msg).await;
115                        }
116                    }
117                    Err(err) => {
118                        info!("Error popping from list {}: {}", list_name, err);
119                    }
120                }
121            }
122        })
123    }
124
125    fn push(&self, channel_name: ListChannel, job_key: &str, job_payload: serde_json::Value) {
126        let client: &Client = self.get_connection();
127        let mut client = client.clone();
128        if let Ok(json_str) = serde_json::to_string(&job_payload) {
129            let _: () = client
130                .rpush(
131                    self.get_list_name(channel_name.get_channel(), job_key),
132                    json_str,
133                )
134                .unwrap();
135        }
136    }
137}
138
139impl CacheProvider for Redis
140where
141    Redis: Connectable,
142{
143    fn get(
144        &self,
145        cache_space: crate::types::channels::CacheSpace,
146        key: &str,
147        app_name: Option<String>,
148    ) -> String {
149        let client: &Client = self.get_connection();
150        let mut client = client.clone();
151        let redis_key = self.get_key_name(cache_space, key, app_name);
152        let result: RedisResult<String> = client.get(&redis_key);
153        match result {
154            Ok(value) => value,
155            Err(_) => String::new(),
156        }
157    }
158
159    fn set(
160        &self,
161        cache_space: crate::types::channels::CacheSpace,
162        key: &str,
163        value: serde_json::Value,
164        app_name: Option<String>,
165    ) {
166        let client: &Client = self.get_connection();
167        let mut client = client.clone();
168        let redis_key = self.get_key_name(cache_space, key, app_name);
169        let cache_value = self.get_value(None, value);
170        if let Ok(json_str) = serde_json::to_string(&cache_value) {
171            let _: () = client.set(&redis_key, json_str).unwrap();
172        }
173    }
174
175    fn delete(
176        &self,
177        cache_space: crate::types::channels::CacheSpace,
178        key: &str,
179        app_name: Option<String>,
180    ) {
181        let client: &Client = self.get_connection();
182        let mut client = client.clone();
183        let redis_key = self.get_key_name(cache_space, key, app_name);
184        let _: () = client.del(&redis_key).unwrap();
185    }
186}
flippico_cache/providers/redis.rs

flippico_cache/providers/
redis.rs
