flippico_cache/
lib.rs

1use crate::types::provider::{Provider, ProviderEither};
2
3pub mod core;
4pub mod providers;
5pub mod types;
6
7pub struct Cache {
8    provider: ProviderEither,
9}
10
11impl Cache {
12    pub fn new() -> Self {
13        Self {
14            provider: Provider::Redis.create_provider(),
15        }
16    }
17
18    pub fn pub_sub(
19        self,
20    ) -> Result<
21        Box<
22            dyn types::provider::ConnectablePubSubProvider<
23                    Channels = types::channels::SubscriptionChannel,
24                > + Send
25                + Sync,
26        >,
27        &'static str,
28    > {
29        self.provider.pub_sub()
30    }
31}
32
33#[cfg(test)]
34mod tests {
35    use crate::types::channels::{ChannelMessage, SubscriptionChannel};
36    use std::collections::HashMap;
37
38    use super::*;
39
40    fn get_test_msg() -> ChannelMessage {
41        ChannelMessage {
42            channel: SubscriptionChannel::BajkomatApi,
43            body: HashMap::new(),
44        }
45    }
46
47    #[test]
48    fn publish_test() {
49        let mut cache = Cache::new().pub_sub().unwrap();
50        cache.publish(SubscriptionChannel::BajkomatApi, get_test_msg());
51    }
52
53    #[tokio::test]
54    async fn subscribe_test() {
55        let subscriber_cache = Cache::new().pub_sub().unwrap();
56        let mut publisher_cache = Cache::new().pub_sub().unwrap();
57
58        // Use a channel to signal when the callback is executed
59        let (tx, rx) = tokio::sync::oneshot::channel();
60        let tx = std::sync::Arc::new(std::sync::Mutex::new(Some(tx)));
61
62        let sub_clbk = move |msg: ChannelMessage| {
63            let msg2 = get_test_msg();
64            assert_eq!(msg, msg2);
65            if let Ok(mut tx_guard) = tx.lock() {
66                if let Some(sender) = tx_guard.take() {
67                    let _ = sender.send(());
68                }
69            }
70        };
71
72        // Start the subscription task by moving the cache into the spawned task
73        let _handle = tokio::spawn(async move {
74            let subscription_task =
75                subscriber_cache.subscribe(SubscriptionChannel::BajkomatApi, Box::new(sub_clbk));
76            let _ = subscription_task.await;
77        });
78
79        // Wait a moment for subscription to be established
80        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
81
82        // Publish the message
83        publisher_cache.publish(SubscriptionChannel::BajkomatApi, get_test_msg());
84
85        // Wait for the callback to be executed with timeout
86        let timeout_result = tokio::time::timeout(tokio::time::Duration::from_secs(10), rx).await;
87        match timeout_result {
88            Ok(Ok(())) => {
89                // Test passed
90            }
91            Ok(Err(e)) => {
92                panic!("Callback channel error: {:?}", e);
93            }
94            Err(error) => {
95                // Check if Redis is available - if not, skip the test
96                println!("Test timed out - Redis may not be available, skipping test");
97                return;
98            }
99        }
100    }
101}
flippico_cache/lib.rs

flippico_cache/
lib.rs
