flippico_cache/

lib.rs

use crate::types::provider::{Provider, ProviderEither};

pub mod core;
pub mod providers;
pub mod types;

pub struct Cache {
    provider: ProviderEither,
}

impl Cache {
    pub fn new() -> Self {
        Self {
            provider: Provider::Redis.create_provider(),
        }
    }

    pub fn pub_sub(
        self,
    ) -> Result<
        Box<
            dyn types::provider::ConnectablePubSubProvider<
                    Channels = types::channels::SubscriptionChannel,
                > + Send
                + Sync,
        >,
        &'static str,
    > {
        self.provider.pub_sub()
    }

    pub fn fifo(
        self,
    ) -> Result<Box<dyn types::provider::ConnectableFifoProvider + Send + Sync>, &'static str> {
        self.provider.fifo()
    }

    pub fn cache(
        self,
    ) -> Result<Box<dyn types::provider::ConnectableCacheProvider + Send + Sync>, &'static str>
    {
        self.provider.cache()
    }

    pub fn sorted_set(
        self,
    ) -> Result<Box<dyn types::provider::ConnectableSortedSetProvider + Send + Sync>, &'static str>
    {
        self.provider.sorted_set()
    }

    /// Build the BullMQ capability.
    ///
    /// This builder is `async` while the other four are not. That is
    /// deliberate: BullMQ's job lifecycle cannot be driven from a blocking
    /// call. See the design spec, section 4.1.
    #[cfg(feature = "bullmq")]
    pub async fn bullmq(
        self,
    ) -> Result<Box<dyn types::bullmq::BullMqProvider + Send + Sync>, types::queues::BullMqError>
    {
        self.provider.bullmq().await
    }
}

#[cfg(test)]
mod tests {
    use crate::types::channels::{
        CacheSpace, ChannelMessage, ListChannel, ListMessage, SubscriptionChannel,
    };
    use std::sync::Arc;

    use super::*;

    fn get_test_msg() -> ChannelMessage {
        ChannelMessage {
            channel: SubscriptionChannel::BajkomatApi,
            meta: None,
            body: None,
        }
    }

    fn get_test_list_msg() -> ListMessage {
        ListMessage {
            meta: None,
            body: serde_json::json!({"body": "test"}).into(),
        }
    }

    #[tokio::test]
    async fn fifo_test() {
        let cache = Arc::new(Cache::new().fifo().unwrap());
        let cache_clone = cache.clone();
        cache.push(
            ListChannel::BajkomatApi,
            "job_key",
            serde_json::json!({"body": "test"}),
        );
        let fifo_value = cache_clone.pop(ListChannel::BajkomatApi, "job_key");
        let msg2 = get_test_list_msg();
        assert_eq!(fifo_value.unwrap(), msg2);
    }

    #[test]
    fn publish_test() {
        let mut cache = Cache::new().pub_sub().unwrap();
        cache.publish(SubscriptionChannel::BajkomatApi, get_test_msg());
    }

    #[test]
    fn cache_test_set() {
        let cache = Cache::new().cache().unwrap();
        let test_data = serde_json::json!({"data": "test"});
        cache.set(CacheSpace::Shopify, "SomeKey", test_data, None);
        let value = cache.get(CacheSpace::Shopify, "SomeKey", None);
        assert_eq!("{\"meta\":null,\"value\":{\"data\":\"test\"}}", value);
    }

    #[test]
    fn cache_test_del() {
        let cache = Cache::new().cache().unwrap();
        let test_data = serde_json::json!({"data": "test"});
        cache.set(CacheSpace::Shopify, "SomeKey", test_data, None);
        let value = cache.get(CacheSpace::Shopify, "SomeKey", None);
        assert_eq!("{\"meta\":null,\"value\":{\"data\":\"test\"}}", value);
        cache.delete(CacheSpace::Shopify, "SomeKey", None);
        let value = cache.get(CacheSpace::Shopify, "SomeKey", None);
        assert_eq!("", value);
    }

    // #[tokio::test]
    // async fn subscribe_test() {
    //     let subscriber_cache = Cache::new().pub_sub().unwrap();
    //     let mut publisher_cache = Cache::new().pub_sub().unwrap();

    //     // Use a channel to signal when the callback is executed
    //     let (tx, rx) = tokio::sync::oneshot::channel();
    //     let tx = std::sync::Arc::new(std::sync::Mutex::new(Some(tx)));

    //     let sub_clbk = move |msg: ChannelMessage| {
    //         let tx_clone = tx.clone();
    //         Box::pin(async move {
    //             let msg2 = get_test_msg();
    //             assert_eq!(msg, msg2);
    //             if let Ok(mut tx_guard) = tx_clone.lock() {
    //                 if let Some(sender) = tx_guard.take() {
    //                     let _ = sender.send(());
    //                 }
    //             }
    //         }) as Pin<Box<dyn std::future::Future<Output = ()> + Send + Sync>>
    //     };

    //     // Start the subscription task by moving the cache into the spawned task
    //     let _handle = tokio::spawn(async move {
    //         let subscription_task =
    //             subscriber_cache.subscribe(Box::new(sub_clbk), SubscriptionChannel::BajkomatApi);
    //         let _ = subscription_task.await;
    //     });

    //     // Wait a moment for subscription to be established
    //     tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;

    //     // Publish the message
    //     publisher_cache.publish(SubscriptionChannel::BajkomatApi, get_test_msg());

    //     // Wait for the callback to be executed with timeout
    //     let timeout_result = tokio::time::timeout(tokio::time::Duration::from_secs(10), rx).await;
    //     match timeout_result {
    //         Ok(Ok(())) => {
    //             // Test passed
    //         }
    //         Ok(Err(e)) => {
    //             panic!("Callback channel error: {:?}", e);
    //         }
    //         Err(_error) => {
    //             // Check if Redis is available - if not, skip the test
    //             println!("Test timed out - Redis may not be available, skipping test");
    //             return;
    //         }
    //     }
    // }
}