sabi_redis for Rust

The sabi data access library for Redis in Rust.

sabi_redis is a Rust crate that provides a streamlined way to access various Redis configurations within the sabi framework. It includes DataSrc and DataConn derived classes designed to make your development process more efficient

RedisDataSrc and RedisDataConn are designed for a standalone Redis server and provide synchronous connections for processing Redis commands. Additionally, RedisAsyncDataSrc and RedisAsyncDataConn are available for asynchronous command processing.

For Redis Sentinel configurations, RedisSentinelDataSrc and RedisSentinelDataConn provide synchronous connections, while RedisSentinelAsyncDataSrc and RedisSentinelAsyncDataConn provide asynchronous connections.

For Redis Cluster configurations, RedisClusterDataSrc and RedisClusterDataConn provide synchronous connections, while RedisClusterAsyncDataSrc and RedisClusterAsyncDataConn provide asynchronous connections.

For Redis Pub/Sub, RedisPubSub, RedisPubSubSentinel, and RedisPubSubCluster provide synchronous subscribers for standalone, Sentinel, and Cluster configurations, respectively. Similarly, RedisPubSubAsync, RedisPubSubSentinelAsync, and RedisPubSubClusterAsync provide asynchronous subscribers. These subscribers allow received messages to be processed as sabi data connections, facilitating consistent integration with your business logic.

Unlike relational databases, Redis does not support data rollbacks. This can lead to data inconsistency if a transaction involving both Redis and another database fails mid-process. To address this, sabi_redis offers three unique features to help developers manage Redis updates and revert changes when necessary: "force back", "pre-commit", and "post-commit".

Installation

In Cargo.toml, write this crate as a dependency:

[dependencies]
sabi_redis = "0.6.0" # `standalone-sync` feature is enabled by default.

# If you want to use the `standalone-async` feature:
# sabi_redis = { version = "0.6.0", default-features = false, features = ["standalone-async"] }

# If you want to use the `sentinel-sync` feature:
# sabi_redis = { version = "0.6.0", default-features = false, features = ["sentinel-sync"] }

# If you want to use the `sentinel-async` feature:
# sabi_redis = { version = "0.6.0", default-features = false, features = ["sentinel-async"] }

# If you want to use the `cluster-sync` feature:
# sabi_redis = { version = "0.6.0", default-features = false, features = ["cluster-sync"] }

# If you want to use the `cluster-async` feature:
# sabi_redis = { version = "0.6.0", default-features = false, features = ["cluster-async"] }

Usage

For Standalone Server And Synchronous Commands

The standalone-sync feature is required for this functionality, and it is enabled by default.

Here is an example of how to use RedisDataSrc and RedisDataConn to connect to Redis and execute a simple command.

use errs;
use override_macro::{overridable, override_with};
use redis::Commands;
use sabi::{uses, setup};
use sabi_redis::{RedisDataSrc, RedisDataConn};

fn main() -> Result<(), errs::Err> {
    // Register a `RedisDataSrc` instance to connect to a Redis server with the key "redis".
    uses("redis", RedisDataSrc::new("redis://127.0.0.1:6379/0"));

    // In this setup process, the registered `RedisDataSrc` instance connects to a Redis server.
    let _auto_shutdown = setup()?;

    my_app()
}

fn my_app() -> errs::Result<()> {
    let mut data = sabi::DataHub::new();
    data.txn(my_logic)
}

fn my_logic(data: &mut impl MyData) -> errs::Result<()> {
    let greeting = data.get_greeting()?;
    data.say_greeting(&greeting)
}

#[overridable]
trait MyData {
    fn get_greeting(&mut self) -> errs::Result<String>;
    fn say_greeting(&mut self, greeting: &str) -> errs::Result<()>;
}

#[overridable]
trait GettingDataAcc: sabi::DataAcc {
    fn get_greeting(&mut self) -> errs::Result<String> {
        Ok("Hello!".to_string())
    }
}

#[overridable]
trait RedisSayingDataAcc: sabi::DataAcc {
    fn say_greeting(&mut self, greeting: &str) -> errs::Result<()> {
        // Retrieve a `RedisDataConn` instance by the key "redis".
        let data_conn = self.get_data_conn::<RedisDataConn>("redis")?;

        // Get a Redis connection to execute Redis synchronous commands.
        let mut redis_conn = data_conn.get_connection()?;

        redis_conn.set("greeting", greeting)
            .map_err(|e| errs::Err::with_source("fail to set greeting", e))?;

        // Register a force back process to revert updates to Redis when an error occurs.
        data_conn.add_force_back(|redis_conn| {
            redis_conn.del("greeting")
                .map_err(|e| errs::Err::with_source("fail to force back", e))?;
            Ok(())
        });

        Ok(())
    }
}

impl GettingDataAcc for sabi::DataHub {}
impl RedisSayingDataAcc for sabi::DataHub {}

#[override_with(GettingDataAcc, RedisSayingDataAcc)]
impl MyData for sabi::DataHub {}

For Standalone Server And Asynchronous Commands

The standalone-async feature is required for this functionality.

Here is an example of how to use RedisAsyncDataSrc and RedisAsyncDataConn to connect to Redis and execute a simple asynchronous command.

use errs;
use override_macro::{overridable, override_with};
use redis::AsyncCommands;
use sabi::{logic, uses, setup_async};
use sabi_redis::{RedisAsyncDataSrc, RedisAsyncDataConn};

#[tokio::main]
async fn main() -> Result<(), errs::Err> {
    // Register a `RedisAsyncDataSrc` instance to connect to a Redis server with the key "redis".
    uses("redis", RedisAsyncDataSrc::new("redis://127.0.0.1:6379/0"));

    // In this setup process, the registered `RedisAsyncDataSrc` instance connects to a Redis server.
    let _auto_shutdown = setup_async().await?;

    my_app_async().await
}

async fn my_app_async() -> errs::Result<()> {
    let mut data = sabi::tokio::DataHub::new();
    data.txn_async(logic!(my_logic_async)).await?;
    Ok(())
}

async fn my_logic_async(data: &mut impl MyData) -> errs::Result<()> {
    let greeting = data.get_greeting_async().await?;
    data.say_greeting_async(&greeting).await
}

#[overridable]
trait MyData {
    async fn get_greeting_async(&mut self) -> errs::Result<String>;
    async fn say_greeting_async(&mut self, greeting: &str) -> errs::Result<()>;
}

#[overridable]
trait GettingDataAcc: sabi::tokio::DataAcc {
    async fn get_greeting_async(&mut self) -> errs::Result<String> {
        Ok("Hello!".to_string())
    }
}

#[overridable]
trait RedisSayingDataAcc: sabi::tokio::DataAcc {
    async fn say_greeting_async(&mut self, greeting: &str) -> errs::Result<()> {
        // Retrieve a `RedisAsyncDataConn` instance by the key "redis".
        let data_conn = self.get_data_conn_async::<RedisAsyncDataConn>("redis").await?;

        // Get an asynchronous Redis connection to execute Redis commands.
        let mut redis_conn = data_conn.get_connection_async().await?;

        redis_conn.set("greeting", greeting)
            .await
            .map_err(|e| errs::Err::with_source("fail to set greeting", e))?;

        // Register an asynchronous force back process to revert updates to Redis when an error occurs.
        data_conn.add_force_back_async(|mut redis_conn| async move {
            redis_conn.del("greeting")
                .await
                .map_err(|e| errs::Err::with_source("fail to force back", e))?;
            Ok(())
        }).await;

        Ok(())
    }
}

impl GettingDataAcc for sabi::tokio::DataHub {}
impl RedisSayingDataAcc for sabi::tokio::DataHub {}

#[override_with(GettingDataAcc, RedisSayingDataAcc)]
impl MyData for sabi::tokio::DataHub {}

For Sentinel Configuration And Synchronous Commands

The sentinel-sync feature is required for this functionality.

use errs;
use sabi::{uses, setup};
use sabi_redis::sentinel::RedisSentinelDataSrc;

fn main() -> Result<(), errs::Err> {
    uses(
        "redis",
        RedisSentinelDataSrc::new(
            vec![
                "redis://127.0.0.1:26479",
                "redis://127.0.0.1:26480",
                "redis://127.0.0.1:26481",
            ],
            "mymaster",
        ),
    );

    let _auto_shutdown = setup()?;
    // ...
    Ok(())
}

For Sentinel Configuration And Asynchronous Commands

The sentinel-async feature is required for this functionality.

use errs;
use sabi::{uses, setup_async};
use sabi_redis::sentinel::RedisSentinelAsyncDataSrc;

#[tokio::main]
async fn main() -> Result<(), errs::Err> {
    uses(
        "redis",
        RedisSentinelAsyncDataSrc::new(
            vec![
                "redis://127.0.0.1:26479",
                "redis://127.0.0.1:26480",
                "redis://127.0.0.1:26481",
            ],
            "mymaster",
        ),
    );

    let _auto_shutdown = setup_async().await?;
    // ...
    Ok(())
}

For Cluster Configuration And Synchronous Commands

The cluster-sync feature is required for this functionality.

use errs;
use sabi::{uses, setup};
use sabi_redis::cluster::RedisClusterDataSrc;

fn main() -> Result<(), errs::Err> {
    uses(
        "redis",
        RedisClusterDataSrc::new(
            vec![
                "redis://127.0.0.1:7000",
                "redis://127.0.0.1:7001",
                "redis://127.0.0.1:7002",
            ],
        ),
    );

    let _auto_shutdown = setup()?;
    // ...
    Ok(())
}

For Cluster Configuration And Asynchronous Commands

The cluster-async feature is required for this functionality.

use errs;
use sabi::{uses, setup_async};
use sabi_redis::cluster::RedisClusterAsyncDataSrc;

#[tokio::main]
async fn main() -> Result<(), errs::Err> {
    uses(
        "redis",
        RedisClusterAsyncDataSrc::new(
            vec![
                "redis://127.0.0.1:7000",
                "redis://127.0.0.1:7001",
                "redis://127.0.0.1:7002",
            ],
        ),
    );

    let _auto_shutdown = setup_async().await?;
    // ...
    Ok(())
}

For Pub/Sub Subscribers And Synchronous Messages

One of standalone-sync, sentinel-sync, or cluster-sync features is required for this functionality.

Here is an example of how to use RedisPubSub to subscribe to a channel and process messages synchronously.

use redis::ControlFlow;
use sabi_redis::pubsub::{RedisPubSub, RedisPubSubMsgDataSrc, RedisPubSubMsgDataConn};

fn main() -> Result<(), errs::Err> {
    let mut pubsub = RedisPubSub::new("redis://127.0.0.1:6379/0");
    pubsub.subscribe("my-channel");

    pubsub.receive(|msg| {
        let mut data = sabi::DataHub::new();
        data.uses("redis/pubsub", RedisPubSubMsgDataSrc::new(msg));
        data.run(my_logic).unwrap();
        ControlFlow::Continue
    })
}

fn my_logic(data: &mut impl MyData) -> errs::Result<()> {
    let payload = data.get_message()?;
    println!("Received: {}", payload);
    Ok(())
}

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

#[overridable]
trait MyDataAcc: sabi::DataAcc {
    fn get_message(&mut self) -> errs::Result<String> {
        let data_conn = self.get_data_conn::<RedisPubSubMsgDataConn>("redis/pubsub")?;
        let msg = data_conn.get_message();
        let payload: String = msg.get_payload().unwrap();
        Ok(payload)
    }
}

impl MyDataAcc for sabi::DataHub {}

#[override_with(MyDataAcc)]
impl MyData for sabi::DataHub {}

For Pub/Sub Subscribers And Asynchronous Messages

One of standalone-async, sentinel-async, or cluster-async features is required for this functionality.

Here is an example of how to use RedisPubSubAsync to subscribe to a channel and process messages asynchronously.

use redis::ControlFlow;
use sabi::{logic, setup_async};
use sabi_redis::pubsub::{RedisPubSubAsync, RedisPubSubMsgAsyncDataSrc, RedisPubSubMsgAsyncDataConn};

#[tokio::main]
async fn main() -> Result<(), errs::Err> {
    let mut pubsub = RedisPubSubAsync::new("redis://127.0.0.1:6379/0");
    pubsub.subscribe("my-channel");

    pubsub.receive_async(async |msg| {
        let mut data = sabi::tokio::DataHub::new();
        data.uses("redis/pubsub", RedisPubSubMsgAsyncDataSrc::new(msg));
        data.run_async(logic!(my_logic_async)).await.unwrap();
        ControlFlow::Continue
    }).await
}

async fn my_logic_async(data: &mut impl MyData) -> errs::Result<()> {
    let message = data.get_message_async().await?;
    println!("Received: {}", message);
    Ok(())
}

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

#[overridable]
trait MyDataAcc: sabi::tokio::DataAcc {
    async fn get_message_async(&mut self) -> errs::Result<String> {
        let data_conn = self.get_data_conn_async::<RedisPubSubMsgAsyncDataConn>("redis/pubsub").await?;
        let msg = data_conn.get_message();
        let payload: String = msg.get_payload().unwrap();
        Ok(payload)
    }
}

impl MyDataAcc for sabi::tokio::DataHub {}

#[override_with(MyDataAcc)]
impl MyData for sabi::tokio::DataHub {}

Supported Rust versions

This crate supports Rust 1.87.0 or later.

% ./build.sh msrv
  [Meta]   cargo-msrv 0.18.4

Compatibility Check #1: Rust 1.75.0
  [FAIL]   Is incompatible

Compatibility Check #2: Rust 1.85.1
  [FAIL]   Is incompatible

Compatibility Check #3: Rust 1.90.0
  [OK]     Is compatible

Compatibility Check #4: Rust 1.87.0
  [OK]     Is compatible

Compatibility Check #5: Rust 1.86.0
  [FAIL]   Is incompatible

Result:
   Considered (min … max):   Rust 1.56.1 … Rust 1.94.1
   Search method:            bisect
   MSRV:                     1.87.0
   Target:                   x86_64-apple-darwin

License

Copyright (C) 2025-2026 Takayuki Sato

This program is free software under MIT License. See the file LICENSE in this distribution for more details.