Crate sabi_redis

Expand description

§sabi-redis

This crate provides sabi data sources for Redis. It supports standalone, Sentinel, and Cluster configurations, in both synchronous and asynchronous (Tokio) versions.

§Usage

§Standalone Configuration

To use this crate, add it to your Cargo.toml with the desired features:

[dependencies]
sabi-redis = { version = "0.1", features = ["standalone", "standalone-async"] }

§Synchronous

#[cfg(feature = "standalone")]
mod standalone {
    use override_macro::{overridable, override_with};
    use sabi::{setup, uses, DataHub, DataAcc};
    use sabi_redis::{RedisDataSrc, RedisDataConn};
    use redis::TypedCommands;

    uses!("redis", RedisDataSrc::new("redis://127.0.0.1:6379/0"));

    #[overridable]
    pub trait MyData {
        fn set_name(&mut self, name: &str) -> errs::Result<()>;
    }

    fn my_logic(data: &mut impl MyData) -> errs::Result<()> {
        data.set_name("Tom")
    }

    #[overridable]
    pub trait RedisDataAcc: DataAcc {
        fn set_name(&mut self, name: &str) -> errs::Result<()> {
            let data_conn = self.get_data_conn::<RedisDataConn>("redis")?;
            let redis_conn = data_conn.get_connection();
            redis_conn.set("name", name)
                .map_err(|e| errs::Err::with_source("Redis SET command failed", e))?;
            data_conn.add_force_back(|redis_conn| {
                redis_conn.del("name")
                    .map_err(|e| errs::Err::with_source("Redis DEL command failed", e))?;
                Ok(())
            });
            Ok(())
        }
    }

    impl RedisDataAcc for DataHub {}

    #[override_with(RedisDataAcc)]
    impl MyData for DataHub {}

    fn my_app() -> errs::Result<()> {
        let _auto_shutdown = setup()?;

        let mut hub = DataHub::new();
        hub.txn(my_logic)
    }
}

§Asynchronous (Tokio)

#[cfg(feature = "standalone-async")]
mod standalone_async {
    use override_macro::{overridable, override_with};
    use sabi::tokio::{setup_async, uses, logic, DataHub, DataAcc};
    use sabi_redis::{RedisDataSrcAsync, RedisDataConnAsync};
    use redis::AsyncTypedCommands;

    uses!("redis", RedisDataSrcAsync::new("redis://127.0.0.1:6379/0"));

    #[overridable]
    pub trait MyDataAsync {
        async fn set_name_async(&mut self, name: &str) -> errs::Result<()>;
    }

    async fn my_logic_async(data: &mut impl MyDataAsync) -> errs::Result<()> {
        data.set_name_async("Tom").await
    }

    #[overridable]
    pub trait RedisDataAccAsync: DataAcc {
        async fn set_name_async(&mut self, name: &str) -> errs::Result<()> {
            let data_conn = self.get_data_conn_async::<RedisDataConnAsync>("redis").await?;
            let redis_conn = data_conn.get_connection();
            redis_conn.set("name", name).await
                .map_err(|e| errs::Err::with_source("Redis SET command failed", e))?;
            data_conn.add_force_back_async(async |mut redis_conn| {
                redis_conn.del("name").await
                    .map_err(|e| errs::Err::with_source("Redis DEL command failed", e))?;
                Ok(())
            }).await;
            Ok(())
        }
    }

    impl RedisDataAccAsync for DataHub {}

    #[override_with(RedisDataAccAsync)]
    impl MyDataAsync for DataHub {}

    async fn my_app() -> errs::Result<()> {
        let _auto_shutdown = setup_async().await?;

        let mut hub = DataHub::new();
        hub.txn_async(logic!(my_logic_async)).await
    }
}

§Pub/Sub

This crate also provides a way to receive Redis Pub/Sub messages and process them within a sabi transaction.

§Synchronous

#[cfg(feature = "standalone")]
mod standalone {
    use override_macro::{overridable, override_with};
    use sabi::{DataHub, DataAcc};
    use sabi_redis::{RedisPubSubSubscriber, RedisPubSubMsgDataSrc, RedisPubSubMsgDataConn};
    use redis::ControlFlow;

    #[overridable]
    pub trait MyData {
        fn receive_msg(&mut self) -> errs::Result<String>;
    }

    fn my_logic(data: &mut impl MyData) -> errs::Result<()> {
        let s = data.receive_msg()?;
        // ...process a message string
        Ok(())
    }

    #[overridable]
    pub trait RedisPubSubDataAcc: DataAcc {
        fn receive_msg(&mut self) -> errs::Result<String> {
            let data_conn = self.get_data_conn::<RedisPubSubMsgDataConn>("redis:pubsub")?;
            let msg = data_conn.get_message();
            msg.get_payload::<String>()
                .map_err(|e| errs::Err::with_source("Fail to get a Redis PubSub message", e))
        }
    }

    impl RedisPubSubDataAcc for DataHub {}

    #[override_with(RedisPubSubDataAcc)]
    impl MyData for DataHub {}

    fn subscribe() -> errs::Result<()> {
        let mut subscriber = RedisPubSubSubscriber::new("redis://127.0.0.1:6379/0");
        subscriber.subscribe("my-channel");
        subscriber.receive(|msg| {
            let mut hub = DataHub::new();
            hub.uses("redis:pubsub", RedisPubSubMsgDataSrc::new(msg));
            hub.txn(my_logic).unwrap();
            ControlFlow::Continue
        })
    }
}

§Asynchronous (Tokio)

#[cfg(feature = "standalone-async")]
mod standalone_async {
    use override_macro::{overridable, override_with};
    use sabi::tokio::{logic, DataHub, DataAcc};
    use sabi_redis::{
        RedisPubSubSubscriberAsync, RedisPubSubMsgDataSrcAsync, RedisPubSubMsgDataConnAsync,
    };
    use redis::ControlFlow;

    #[overridable]
    pub trait MyDataAsync {
        async fn receive_msg_async(&mut self) -> errs::Result<String>;
    }

    async fn my_logic_async(data: &mut impl MyDataAsync) -> errs::Result<()> {
        let s = data.receive_msg_async().await?;
        // ...process a message string
        Ok(())
    }

    #[overridable]
    pub trait RedisPubSubDataAccAsync: DataAcc {
        async fn receive_msg_async(&mut self) -> errs::Result<String> {
            let data_conn = self.
                get_data_conn_async::<RedisPubSubMsgDataConnAsync>("redis:pubsub").await?;
            let msg = data_conn.get_message();
            msg.get_payload::<String>()
                .map_err(|e| errs::Err::with_source("Fail to get a Redis PubSub message", e))
        }
    }

    impl RedisPubSubDataAccAsync for DataHub {}

    #[override_with(RedisPubSubDataAccAsync)]
    impl MyDataAsync for DataHub {}

    async fn subscribe_async() -> errs::Result<()> {
        let mut subscriber = RedisPubSubSubscriberAsync::new("redis://127.0.0.1:6379/0");
        subscriber.subscribe("my-channel");
        subscriber.receive_async(async |msg| {
            let mut hub = DataHub::new();
            hub.uses("redis:pubsub", RedisPubSubMsgDataSrcAsync::new(msg));
            hub.txn_async(logic!(my_logic_async)).await.unwrap();
            ControlFlow::Continue
        }).await
    }
}

Modules§

clustercluster or cluster-async: A module for Redis Cluster.
sentinelsentinel or sentinel-async: A module for Redis Sentinel.

Structs§

RedisDataConnstandalone: A struct that holds a pooled Redis connection and transaction callbacks for standalone configuration.
RedisDataConnAsyncstandalone-async: A struct that holds a Redis connection and asynchronous transaction callbacks for standalone configuration.
RedisDataSrcstandalone: A struct that manages a Redis connection pool for standalone configuration.
RedisDataSrcAsyncstandalone-async: A struct that manages an asynchronous Redis connection pool for standalone configuration.
RedisPubSubMsgDataConnstandalone or sentinel or cluster: A struct that holds a Redis Pub/Sub message as a data connection.
RedisPubSubMsgDataConnAsyncstandalone-async or sentinel-async or cluster-async: A struct that holds a Redis Pub/Sub message as an asynchronous data connection.
RedisPubSubMsgDataSrcstandalone or sentinel or cluster: A struct that holds a Redis Pub/Sub message as a data source.
RedisPubSubMsgDataSrcAsyncstandalone-async or sentinel-async or cluster-async: A struct that holds a Redis Pub/Sub message as an asynchronous data source.
RedisPubSubSubscriberstandalone: A struct for subscribing to Redis channels and receiving messages for standalone configuration.
RedisPubSubSubscriberAsyncstandalone-async: A struct for subscribing to Redis channels and receiving messages asynchronously for standalone configuration.

Enums§

RedisErrorstandalone: Errors related to Redis data source and connection for standalone configuration.
RedisErrorAsyncstandalone-async: Errors related to Redis data source and connection for asynchronous standalone configuration.
RedisPubSubSubscriberErrorstandalone: Errors related to Redis Pub/Sub subscriber for standalone configuration.
RedisPubSubSubscriberErrorAsyncstandalone-async: Errors related to asynchronous Redis Pub/Sub subscriber for standalone configuration.

Crate sabi_redis

Crate sabi_redis Copy item path

§sabi-redis

§Usage

§Standalone Configuration

§Synchronous

§Asynchronous (Tokio)

§Pub/Sub

§Synchronous

§Asynchronous (Tokio)

Modules§

Structs§

Enums§

Crate sabi_redis