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    pub fn fifo(
33        self,
34    ) -> Result<Box<dyn types::provider::ConnectableFifoProvider + Send + Sync>, &'static str> {
35        self.provider.fifo()
36    }
37
38    pub fn cache(
39        self,
40    ) -> Result<Box<dyn types::provider::ConnectableCacheProvider + Send + Sync>, &'static str>
41    {
42        self.provider.cache()
43    }
44}
45
46#[cfg(test)]
47mod tests {
48    use crate::types::channels::{
49        CacheSpace, ChannelMessage, ListChannel, ListMessage, SubscriptionChannel,
50    };
51    use std::sync::Arc;
52
53    use super::*;
54
55    fn get_test_msg() -> ChannelMessage {
56        ChannelMessage {
57            channel: SubscriptionChannel::BajkomatApi,
58            meta: None,
59            body: None,
60        }
61    }
62
63    fn get_test_list_msg() -> ListMessage {
64        ListMessage {
65            meta: None,
66            body: serde_json::json!({"body": "test"}).into(),
67        }
68    }
69
70    #[tokio::test]
71    async fn fifo_test() {
72        let cache = Arc::new(Cache::new().fifo().unwrap());
73        let cache_clone = cache.clone();
74        cache.push(
75            ListChannel::BajkomatApi,
76            "job_key",
77            serde_json::json!({"body": "test"}),
78        );
79        let fifo_value = cache_clone.pop(ListChannel::BajkomatApi, "job_key");
80        let msg2 = get_test_list_msg();
81        assert_eq!(fifo_value.unwrap(), msg2);
82    }
83
84    #[test]
85    fn publish_test() {
86        let mut cache = Cache::new().pub_sub().unwrap();
87        cache.publish(SubscriptionChannel::BajkomatApi, get_test_msg());
88    }
89
90    #[test]
91    fn cache_test_set() {
92        let cache = Cache::new().cache().unwrap();
93        let test_data = serde_json::json!({"data": "test"});
94        cache.set(CacheSpace::Shopify, "SomeKey", test_data, None);
95        let value = cache.get(CacheSpace::Shopify, "SomeKey", None);
96        assert_eq!("{\"meta\":null,\"value\":{\"data\":\"test\"}}", value);
97    }
98
99    #[test]
100    fn cache_test_del() {
101        let cache = Cache::new().cache().unwrap();
102        let test_data = serde_json::json!({"data": "test"});
103        cache.set(CacheSpace::Shopify, "SomeKey", test_data, None);
104        let value = cache.get(CacheSpace::Shopify, "SomeKey", None);
105        assert_eq!("{\"meta\":null,\"value\":{\"data\":\"test\"}}", value);
106        cache.delete(CacheSpace::Shopify, "SomeKey", None);
107        let value = cache.get(CacheSpace::Shopify, "SomeKey", None);
108        assert_eq!("", value);
109    }
110
111    // #[tokio::test]
112    // async fn subscribe_test() {
113    //     let subscriber_cache = Cache::new().pub_sub().unwrap();
114    //     let mut publisher_cache = Cache::new().pub_sub().unwrap();
115
116    //     // Use a channel to signal when the callback is executed
117    //     let (tx, rx) = tokio::sync::oneshot::channel();
118    //     let tx = std::sync::Arc::new(std::sync::Mutex::new(Some(tx)));
119
120    //     let sub_clbk = move |msg: ChannelMessage| {
121    //         let tx_clone = tx.clone();
122    //         Box::pin(async move {
123    //             let msg2 = get_test_msg();
124    //             assert_eq!(msg, msg2);
125    //             if let Ok(mut tx_guard) = tx_clone.lock() {
126    //                 if let Some(sender) = tx_guard.take() {
127    //                     let _ = sender.send(());
128    //                 }
129    //             }
130    //         }) as Pin<Box<dyn std::future::Future<Output = ()> + Send + Sync>>
131    //     };
132
133    //     // Start the subscription task by moving the cache into the spawned task
134    //     let _handle = tokio::spawn(async move {
135    //         let subscription_task =
136    //             subscriber_cache.subscribe(Box::new(sub_clbk), SubscriptionChannel::BajkomatApi);
137    //         let _ = subscription_task.await;
138    //     });
139
140    //     // Wait a moment for subscription to be established
141    //     tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
142
143    //     // Publish the message
144    //     publisher_cache.publish(SubscriptionChannel::BajkomatApi, get_test_msg());
145
146    //     // Wait for the callback to be executed with timeout
147    //     let timeout_result = tokio::time::timeout(tokio::time::Duration::from_secs(10), rx).await;
148    //     match timeout_result {
149    //         Ok(Ok(())) => {
150    //             // Test passed
151    //         }
152    //         Ok(Err(e)) => {
153    //             panic!("Callback channel error: {:?}", e);
154    //         }
155    //         Err(_error) => {
156    //             // Check if Redis is available - if not, skip the test
157    //             println!("Test timed out - Redis may not be available, skipping test");
158    //             return;
159    //         }
160    //     }
161    // }
162}
flippico_cache/lib.rs

flippico_cache/
lib.rs
