### Rust: Basic Redis Connection and String Operations

Source: https://context7.com/dahomey-technologies/rustis/llms.txt

Demonstrates connecting to a Redis server and performing fundamental string operations like SET, GET, and MSET/MGET. It includes automatic type conversion for retrieved values and uses the `rustis` library with `tokio` for asynchronous execution. Ensure `rustis` and `tokio` are added as dependencies.

```rust
use rustis::
    client::Client,
    commands::{FlushingMode, ServerCommands, StringCommands},
    Result,
};

#[tokio::main]
async fn main() -> Result<()> {
    // Connect to Redis using IP:port
    let client = Client::connect("127.0.0.1:6379").await?;

    // Flush all existing data
    client.flushdb(FlushingMode::Sync).await?;

    // Set a key-value pair
    client.set("key", "value").await?;

    // Get a value with automatic type conversion
    let value: String = client.get("key").await?;
    println!("value: {}", value);

    // SET and GET with different data types
    client.set("number", 42).await?;
    let number: i32 = client.get("number").await?;
    assert_eq!(42, number);

    // Multiple SET operations
    client.mset([("key1", "value1"), ("key2", "value2")]).await?;

    // Multiple GET operations
    let values: Vec<String> = client.mget(["key1", "key2"]).await?;
    assert_eq!(vec!["value1", "value2"], values);

    Ok(())
}
```

--------------------------------

### Redis Sorted Set Operations with Score Queries using Rustis

Source: https://context7.com/dahomey-technologies/rustis/llms.txt

Illustrates how to manage Redis sorted sets with the Rustis client, suitable for leaderboards or priority queues. This includes adding members with scores, optionally only adding new members or updating scores if greater. It covers retrieving ranges of members by rank and score, both in ascending and descending order, and fetching members along with their scores. Additionally, it shows how to find a member's rank and score, increment scores, remove members by key or rank range, and get the total count of members.

```rust
use rustis::{
    client::Client,
    commands::{SortedSetCommands, ZAddOptions, ZRangeOptions},
    Result,
};

#[tokio::main]
async fn main() -> Result<()> {
    let client = Client::connect("127.0.0.1:6379").await?;

    // Add members with scores to sorted set
    client.zadd(
        "leaderboard",
        [(100.0, "player1"), (250.0, "player2"), (180.0, "player3")],
        ZAddOptions::default(),
    ).await?;

    // Add with options (NX = only add new, GT = only update if new score is greater)
    client.zadd(
        "leaderboard",
        [(300.0, "player4")],
        ZAddOptions::default().nx(),
    ).await?;

    // Get range by rank (0-based index)
    let top_players: Vec<String> = client.zrange("leaderboard", 0, 2, ZRangeOptions::default()).await?;
    println!("Top 3 players: {:?}", top_players);

    // Get range with scores (highest to lowest)
    let top_with_scores: Vec<(String, f64)> = client
        .zrange_with_scores("leaderboard", 0, 2, ZRangeOptions::default().rev())
        .await?;
    for (player, score) in top_with_scores {
        println!("{}: {}", player, score);
    }

    // Get range by score
    let mid_range: Vec<String> = client.zrangebyscore("leaderboard", 100.0, 200.0).await?;

    // Get rank of member (0-based, lowest score = rank 0)
    let rank: Option<usize> = client.zrank("leaderboard", "player2").await?;

    // Get score of member
    let score: Option<f64> = client.zscore("leaderboard", "player2").await?;

    // Increment score
    let new_score: f64 = client.zincrby("leaderboard", 50.0, "player1").await?;

    // Remove members
    client.zrem("leaderboard", "player3").await?;

    // Remove by rank range
    client.zremrangebyrank("leaderboard", 0, 0).await?;  // Remove lowest ranked

    // Get cardinality (number of members)
    let count: usize = client.zcard("leaderboard").await?;

    Ok(())
}

```

--------------------------------

### Redis Hash Operations with Field Expiration using Rustis

Source: https://context7.com/dahomey-technologies/rustis/llms.txt

Demonstrates how to perform various operations on Redis hashes using the Rustis client. This includes setting and getting single or multiple fields, retrieving all fields, and crucially, setting field-level expiration (supported in Redis 7.4+) and checking the Time To Live (TTL) of a field. It also covers incrementing field values, deleting fields, and checking for field existence.

```rust
use rustis::{
    client::Client,
    commands::{ExpireOption, HashCommands},
    Result,
};
use std::collections::HashMap;

#[tokio::main]
async fn main() -> Result<()> {
    let client = Client::connect("127.0.0.1:6379").await?;

    // Set single field in hash
    client.hset("user:1000", ("name", "John Doe")).await?;

    // Set multiple fields at once
    client.hset("user:1000", [
        ("email", "john@example.com"),
        ("age", "30"),
        ("city", "San Francisco"),
    ]).await?;

    // Get single field
    let name: String = client.hget("user:1000", "name").await?;
    assert_eq!("John Doe", name);

    // Get multiple fields
    let values: Vec<String> = client.hmget("user:1000", ["name", "email"]).await?;

    // Get all fields and values
    let all_fields: HashMap<String, String> = client.hgetall("user:1000").await?;
    println!("User data: {:?}", all_fields);

    // Set field expiration (Redis 7.4+)
    client.hexpire("user:1000", 3600, ExpireOption::None, "email").await?;

    // Check field TTL
    let ttl: Vec<i64> = client.httl("user:1000", "email").await?;
    println!("Email field TTL: {:?}", ttl);

    // Increment field value
    let new_age: i64 = client.hincrby("user:1000", "age", 1).await?;

    // Delete specific fields
    client.hdel("user:1000", ["city", "age"]).await?;

    // Check if field exists
    let exists: bool = client.hexists("user:1000", "name").await?;

    Ok(())
}

```

--------------------------------

### Connect and Basic Operations with Tokio

Source: https://github.com/dahomey-technologies/rustis/blob/main/README.md

Demonstrates connecting to a Redis server using Tokio, flushing the database, setting a string key, and retrieving its value. It utilizes the `rustis` crate and requires the `tokio` runtime.

```rust
use rustis::
     client::Client, 
     commands::{FlushingMode, ServerCommands, StringCommands},
     Result,
};

#[tokio::main]
async fn main() -> Result<()> {
     // Connect the client to a Redis server from its IP and port
     let client = Client::connect("127.0.0.1:6379").await?;
 
     // Flush all existing data in Redis
     client.flushdb(FlushingMode::Sync).await?;

     // sends the command SET to Redis. This command is defined in the StringCommands trait
     client.set("key", "value").await?;
 
     // sends the command GET to Redis. This command is defined in the StringCommands trait
     let value: String = client.get("key").await?;
     println!("value: {value:?}");

     Ok(())
}
```

--------------------------------

### Rust: Execute Generic Redis Commands with Rustis

Source: https://context7.com/dahomey-technologies/rustis/llms.txt

Demonstrates how to send arbitrary Redis commands using the Rustis client. This includes building commands with the `cmd()` builder, setting keys with expiration, executing module-specific commands, sending commands without expecting a response, and processing batch requests. It also shows how to handle raw `Value` responses from Redis.

```rust
use rustis::{
    client::Client,
    resp::{cmd, Command, Value},
    Result,
};

#[tokio::main]
async fn main() -> Result<()> {
    let client = Client::connect("127.0.0.1:6379").await?;

    // Build custom command with cmd() builder
    let response: String = client.send(
        cmd("GET").arg("mykey"),
        None
    ).await?;

    // Custom command with multiple arguments
    let response: Value = client.send(
        cmd("SET")
            .arg("mykey")
            .arg("myvalue")
            .arg("EX")
            .arg(3600),
        None
    ).await?;

    // Module-specific or new Redis commands
    let response: Vec<String> = client.send(
        cmd("CUSTOMCOMMAND")
            .arg("param1")
            .arg("param2"),
        None
    ).await?;

    // Send and forget (no response needed)
    client.send_and_forget(cmd("SET").arg("key").arg("value")).await?;

    // Send batch of commands
    let commands = vec![
        cmd("SET").arg("key1").arg("value1"),
        cmd("SET").arg("key2").arg("value2"),
        cmd("SET").arg("key3").arg("value3"),
    ];

    let responses: Vec<Value> = client.send_batch(commands).await?;

    // Raw Value response handling
    let raw_response: Value = client.send(cmd("INFO").arg("server"), None).await?;
    match raw_response {
        Value::BulkString(data) => {
            let info = String::from_utf8_lossy(data.as_bytes());
            println!("Server info: {}", info);
        }
        Value::Array(items) => {
            println!("Got array with {} items", items.len());
        }
        _ => println!("Other response type"),
    }

    Ok(())
}

```

--------------------------------

### Rust: Configuring Redis Connections via URL

Source: https://context7.com/dahomey-technologies/rustis/llms.txt

Illustrates various ways to configure Redis client connections using URL strings with the `rustis` library. It covers basic connections, password authentication, username/password pairs, database selection, TLS, timeouts, command names, Redis Cluster, and Redis Sentinel configurations. Ensure the appropriate `rustis` features (e.g., `tokio-rustls` or `tokio-native-tls` for TLS) are enabled.

```rust
use rustis::{client::Client, Result};

#[tokio::main]
async fn main() -> Result<()> {
    // Basic connection
    let client = Client::connect("redis://127.0.0.1:6379").await?;

    // Connection with password authentication
    let client = Client::connect("redis://:password@127.0.0.1:6379").await?;

    // Connection with username and password
    let client = Client::connect("redis://username:password@127.0.0.1:6379").await?;

    // Select specific database (default is 0)
    let client = Client::connect("redis://127.0.0.1:6379/1").await?;

    // TLS connection (requires tokio-rustls or tokio-native-tls feature)
    let client = Client::connect("rediss://127.0.0.1:6379").await?;

    // Connection with timeouts and configuration
    let client = Client::connect(
        "redis://127.0.0.1:6379?connect_timeout=5000&command_timeout=10000&connection_name=myapp"
    ).await?;

    // Redis Cluster connection
    let client = Client::connect(
        "redis+cluster://node1:6379,node2:6379,node3:6379"
    ).await?;

    // Redis Sentinel connection
    let client = Client::connect(
        "redis+sentinel://sentinel1:26379/mymaster/0"
    ).await?;

    Ok(())
}
```

--------------------------------

### Redis Connection Pool with BB8 (Rust)

Source: https://context7.com/dahomey-technologies/rustis/llms.txt

Illustrates setting up and using a Redis connection pool with BB8. This is an alternative to multiplexing, managing multiple connections efficiently. It requires `rustis` and `tokio`. The input is Redis connection details and pool configuration, and the output is successfully executed Redis commands.

```rust
use rustis::{
    bb8::Pool,
    client::PooledClientManager,
    commands::StringCommands,
    Result,
};

#[tokio::main]
async fn main() -> Result<()> {
    // Create a connection pool manager
    let manager = PooledClientManager::new("redis://127.0.0.1:6379")?;

    // Build the pool with custom configuration
    let pool = Pool::builder()
        .max_size(15)                    // Maximum pool size
        .min_idle(Some(5))               // Minimum idle connections
        .build(manager)
        .await?;

    // Get a client from the pool
    let client = pool.get().await.unwrap();

    // Use the client (automatically returned to pool when dropped)
    client.set("key", "value").await?;
    let value: String = client.get("key").await?;
    assert_eq!("value", value);

    // Pool can be shared across threads
    let pool_clone = pool.clone();
    tokio::spawn(async move {
        let client = pool_clone.get().await.unwrap();
        client.set("another_key", "another_value").await
    })
    .await??;

    Ok(())
}

```

--------------------------------

### Connect and Operate on Redis Cluster with Rustis

Source: https://context7.com/dahomey-technologies/rustis/llms.txt

This snippet demonstrates how to connect to a Redis Cluster using the `rustis` library and perform various operations. It highlights automatic slot routing for normal commands, using hash tags to ensure keys land on the same shard, and transparently handling multi-key operations across shards. It also shows how to retrieve cluster-specific information like node details and slot mappings, and perform cluster-wide administrative tasks such as flushing all databases.

```rust
use rustis::{
    client::Client,
    commands::{ClusterCommands, ServerCommands, StringCommands},
    Result,
};

#[tokio::main]
async fn main() -> Result<()> {
    // Connect to Redis Cluster (automatically discovers topology)
    let client = Client::connect(
        "redis+cluster://node1:7000,node2:7001,node3:7002"
    ).await?;

    // Normal operations work transparently (automatic slot routing)
    client.set("user:1000", "John").await?;
    client.set("user:2000", "Jane").await?;

    let value: String = client.get("user:1000").await?;

    // Hash tags to force keys to same slot: {hashtag}
    client.set("user:{group1}:1000", "data1").await?;
    client.set("user:{group1}:2000", "data2").await?;
    // These keys will be on the same shard

    // Multi-key operations across shards (automatically split and merged)
    client.mset([
        ("key1", "value1"),
        ("key2", "value2"),
        ("key3", "value3"),
    ]).await?;

    let values: Vec<String> = client.mget(["key1", "key2", "key3"]).await?;

    // Cluster-specific commands
    let cluster_info = client.cluster_info().await?;
    println!("Cluster state: {}", cluster_info.cluster_state);

    // Get cluster nodes information
    let nodes = client.cluster_nodes().await?;
    for node in nodes {
        println!("Node {} - {} - Slots: {:?}",
            node.node_id, node.ip_port, node.slots);
    }

    // Get cluster slots mapping
    let slots = client.cluster_slots().await?;
    for slot_range in slots {
        println!("Slots {:?} -> Master: {:?}",
            slot_range.start_slot, slot_range.master);
    }

    // Flush all databases across all cluster nodes
    use rustis::commands::FlushingMode;
    client.flushall(FlushingMode::Sync).await?;

    // Get total database size across cluster
    let total_keys: usize = client.dbsize().await?;

    Ok(())
}

```

--------------------------------

### Rust Redis List Operations for Queues and Stacks

Source: https://context7.com/dahomey-technologies/rustis/llms.txt

Demonstrates using Rustis to manage Redis lists as queues (FIFO) and stacks (LIFO). It covers push, pop, range retrieval, length checking, insertion, element setting, blocking pops, trimming, and moving elements between lists. Dependencies include 'rustis' and 'tokio'.

```rust
use rustis::{
    client::Client,
    commands::{ListCommands, BlockingCommands},
    Result,
};

#[tokio::main]
async fn main() -> Result<()> {
    let client = Client::connect("127.0.0.1:6379").await?;

    // Push to end of list (right side) - Queue pattern
    client.rpush("tasks", ["task1", "task2", "task3"]).await?;

    // Push to beginning of list (left side) - Stack pattern
    client.lpush("notifications", "urgent_alert").await?;

    // Pop from left (FIFO queue with RPUSH/LPOP)
    let task: Option<String> = client.lpop("tasks", 1).await?;
    println!("Processing task: {:?}", task);

    // Pop from right (LIFO stack with RPUSH/RPOP)
    let item: Option<String> = client.rpop("tasks", 1).await?;

    // Get range of elements (0 = first, -1 = last)
    let all_tasks: Vec<String> = client.lrange("tasks", 0, -1).await?;

    // Get list length
    let len: usize = client.llen("tasks").await?;

    // Insert before or after a pivot element
    use rustis::commands::LInsertWhere;
    client.linsert("tasks", LInsertWhere::Before, "task2", "task1.5").await?;

    // Set element at index
    client.lset("tasks", 0, "updated_task").await?;

    // Blocking pop (waits up to timeout for element)
    let result: Option<(String, String)> = client.blpop("tasks", 5.0).await?;
    match result {
        Some((key, value)) => println!("Popped {} from {}", value, key),
        None => println!("Timeout, no elements available"),
    }

    // Trim list to specific range
    client.ltrim("tasks", 0, 99).await?;  // Keep only first 100 elements

    // Move element from one list to another
    client.lmove("source", "destination",
        rustis::commands::LMoveWhere::Left,
        rustis::commands::LMoveWhere::Right).await?;

    Ok(())
}

```

--------------------------------

### Handle Binary Data and Custom Serialization with Rustis

Source: https://context7.com/dahomey-technologies/rustis/llms.txt

Demonstrates how to store and retrieve raw binary data using `BulkString` in Rustis, preserving data integrity for formats like CBOR. It also shows how to serialize and deserialize custom Rust structures to JSON using the `serde_json` feature.

```rust
use rustis::{
    client::Client,
    commands::StringCommands,
    resp::BulkString,
    Result,
};

#[tokio::main]
async fn main() -> Result<()> {
    let client = Client::connect("127.0.0.1:6379").await?;

    // Store binary data (e.g., CBOR, MessagePack, Protobuf)
    let cbor_data = b"\xa1\x63\x66\x6f\x6f\x63\x62\x61\x72";  // {"foo": "bar"} in CBOR
    client.set("binary_key", cbor_data).await?;

    // Retrieve as BulkString to preserve binary data
    let value: BulkString = client.get("binary_key").await?;
    assert_eq!(cbor_data, value.as_bytes());

    // Store byte slices
    let image_data: Vec<u8> = vec![0xFF, 0xD8, 0xFF, 0xE0];  // JPEG header
    client.set("image:123", &image_data).await?;

    // Retrieve binary data
    let retrieved: BulkString = client.get("image:123").await?;
    let bytes: &[u8] = retrieved.as_bytes();

    // Store serialized structures with serde_json (requires "json" feature)
    #[derive(serde::Serialize, serde::Deserialize)]
    struct User {
        id: u64,
        name: String,
        email: String,
    }

    let user = User {
        id: 1000,
        name: "Alice".to_string(),
        email: "alice@example.com".to_string(),
    };

    let json_str = serde_json::to_string(&user)?;
    client.set("user:1000:json", json_str).await?;

    let retrieved: String = client.get("user:1000:json").await?;
    let user: User = serde_json::from_str(&retrieved)?;

    println!("User: {} - {}", user.name, user.email);

    Ok(())
}

```

--------------------------------

### Multiplexed Redis Client for Axum Web Apps (Rust)

Source: https://context7.com/dahomey-technologies/rustis/llms.txt

Demonstrates how to use a multiplexed Redis client with the Axum web framework. This approach allows a single Redis connection to be shared across multiple concurrent requests, improving efficiency. It requires `axum`, `rustis`, and `tokio`. The input is Redis keys and values, and the output is fetched values or status confirmations.

```rust
use axum::{
    Router,
    extract::{Path, State},
    http::StatusCode,
    response::{IntoResponse, Response},
    routing::get,
};
use rustis::{
    client::Client,
    commands::{GenericCommands, StringCommands},
};
use std::net::SocketAddr;
use tokio::net::TcpListener;
use std::sync::Arc;

#[tokio::main]
async fn main() {
    // Create a single multiplexed client (cloneable, shares one connection)
    let redis = Arc::new(Client::connect("redis://127.0.0.1:6379").await.unwrap());

    // Build Axum application
    let app = Router::new()
        .route("/:key", get(read).post(update).delete(del))
        .with_state(redis);

    // Run server
    let addr = SocketAddr::from(([127, 0, 0, 1], 3000));
    println!("listening on {}", addr);
    let listener = TcpListener::bind(&addr).await.unwrap();
    axum::serve(listener, app).await.unwrap();
}

async fn read(
    State(redis): State<Arc<Client>>,
    Path(key): Path<String>,
) -> Result<String, ServiceError> {
    let value: Option<String> = redis.get(&key).await?;
    value.ok_or_else(|| ServiceError::new(
        StatusCode::NOT_FOUND,
        format!("Key `{}` does not exist.", key),
    ))
}

async fn update(
    State(redis): State<Arc<Client>>,
    Path(key): Path<String>,
    value: String,
) -> Result<(), ServiceError> {
    redis.set(key, value).await?;
    Ok(())
}

async fn del(
    State(redis): State<Arc<Client>>,
    Path(key): Path<String>,
) -> Result<(), ServiceError> {
    let deleted = redis.del(&key).await?;
    if deleted > 0 {
        Ok(())
    } else {
        Err(ServiceError::new(StatusCode::NOT_FOUND, "Key not found"))
    }
}

struct ServiceError(StatusCode, String);
impl ServiceError {
    fn new(status_code: StatusCode, description: impl ToString) -> Self {
        Self(status_code, description.to_string())
    }
}
impl IntoResponse for ServiceError {
    fn into_response(self) -> Response {
        (self.0, self.1).into_response()
    }
}
impl From<rustis::Error> for ServiceError {
    fn from(e: rustis::Error) -> Self {
        eprintln!("rustis error: {}", e);
        ServiceError::new(StatusCode::INTERNAL_SERVER_ERROR, "Internal Server Error")
    }
}

```

--------------------------------

### Rust Redis Stream Operations for Event Logging

Source: https://context7.com/dahomey-technologies/rustis/llms.txt

Illustrates using Rustis to manage Redis streams for append-only logs and distributed message processing. Covers adding entries (with auto/specific IDs, maxlen), reading entries, creating consumer groups, reading as a consumer, acknowledging messages, checking pending messages, and retrieving stream information. Dependencies include 'rustis' and 'tokio'.

```rust
use rustis::{
    client::Client,
    commands::{StreamCommands, XAddOptions, XReadOptions, XReadGroupOptions, XGroupCreateOptions},
    Result,
};

#[tokio::main]
async fn main() -> Result<()> {
    let client = Client::connect("127.0.0.1:6379").await?;

    // Add entries to stream (* = auto-generate ID)
    let id1: String = client.xadd(
        "events",
        "*",
        [("user", "john"), ("action", "login"), ("timestamp", "1234567890")],
        XAddOptions::default(),
    ).await?;
    println!("Added event with ID: {}", id1);

    // Add with specific ID
    let id2: String = client.xadd(
        "events",
        "1234567890000-0",
        [("user", "jane"), ("action", "logout")],
        XAddOptions::default(),
    ).await?;

    // Add with maxlen to limit stream size
    client.xadd(
        "events",
        "*",
        [("user", "bob"), ("action", "purchase")],
        XAddOptions::default().maxlen(1000),  // Keep last 1000 entries
    ).await?;

    // Read entries from stream
    let entries: Vec<(String, Vec<(String, Vec<(String, String)>)>)> = 
        client.xread(["events"], ["0"], XReadOptions::default().count(10)).await?;

    for (stream_key, stream_entries) in entries {
        for (entry_id, fields) in stream_entries {
            println!("Stream: {}, ID: {}, Fields: {:?}", stream_key, entry_id, fields);
        }
    }

    // Create consumer group for distributed processing
    client.xgroup_create("events", "processors", "0", XGroupCreateOptions::default()).await?;

    // Read as consumer in group
    let group_entries: Vec<(String, Vec<(String, Vec<(String, String)>)>)> = 
        client.xreadgroup(
            "processors",      // group name
            "worker1",         // consumer name
            ["events"],        // streams
            [">"],             // read only new messages
            XReadGroupOptions::default().count(5).block(1000),  // block for 1 second
        ).await?;

    // Process entries and acknowledge
    for (stream_key, stream_entries) in group_entries {
        for (entry_id, fields) in stream_entries {
            // Process the entry...
            println!("Processing: {:?}", fields);

            // Acknowledge successful processing
            client.xack(&stream_key, "processors", &entry_id).await?;
        }
    }

    // Check pending messages in group
    let pending = client.xpending("events", "processors").await?;

    // Get stream info
    use rustis::commands::XInfoStreamOptions;
    let info = client.xinfo_stream("events", XInfoStreamOptions::default()).await?;
    println!("Stream length: {}, First ID: {}, Last ID: {}",
        info.length, info.first_entry.stream_id, info.last_entry.stream_id);

    Ok(())
}

```

--------------------------------

### Execute Lua Scripts and Redis Functions with Rustis

Source: https://context7.com/dahomey-technologies/rustis/llms.txt

This snippet demonstrates executing inline Lua scripts using `EVAL`, loading and executing scripts using `EVALSHA`, and interacting with Redis Functions (introduced in Redis 7.0) for atomic server-side operations. It covers loading libraries, calling functions, and managing function lifecycles. Dependencies include the `rustis` crate and `tokio` for async operations. It handles various return types from scripts and functions.

```rust
use rustis::{
    client::Client,
    commands::{CallBuilder, ScriptingCommands, StringCommands},
    Result,
};

#[tokio::main]
async fn main() -> Result<()> {
    let client = Client::connect("127.0.0.1:6379").await?;

    // Execute inline Lua script with EVAL
    let result: String = client.eval(
        CallBuilder::script("return ARGV[1]")
            .args("hello")
    ).await?;
    assert_eq!("hello", result);

    // Script with keys and arguments
    client.set("mykey", "Hello").await?;
    let result: String = client.eval(
        CallBuilder::script("return redis.call('GET', KEYS[1])... \" "..ARGV[1]")
            .keys("mykey")
            .args("World!")
    ).await?;
    assert_eq!("Hello World!", result);

    // Complex script returning multiple values
    let lua_script = r#"
        redis.call("DEL", "numbers");
        redis.call("SADD", "numbers", 1, 2, 3, 4);
        local arr = redis.call("SMEMBERS", "numbers");
        redis.call("DEL", "numbers");
        return { ARGV[1], ARGV[2], 42, arr }
    "#;

    let result: (String, String, i32, Vec<i64>) = client.eval(
        CallBuilder::script(lua_script)
            .args("First")
            .args("Second")
    ).await?;
    assert_eq!(("First".to_string(), "Second".to_string(), 42, vec![1,2,3,4]), result);

    // Load script and execute with EVALSHA (more efficient)
    let sha1: String = client.script_load("return ARGV[1] * 2").await?;
    let doubled: i32 = client.evalsha(
        CallBuilder::sha1(&sha1).args(21)
    ).await?;
    assert_eq!(42, doubled);

    // Redis Functions (Redis 7.0+) - load a library
    let library_code = r#"#!lua name=mylib
        redis.register_function('double', function(keys, args)
            return args[1] * 2
        end)
    "#;

    client.function_load(library_code, false).await?;

    // Call the function
    let result: i32 = client.fcall("double", CallBuilder::new().args(21)).await?;
    assert_eq!(42, result);

    // Call function with read-only flag (can be routed to replicas)
    let result: i32 = client.fcall_ro("double", CallBuilder::new().args(10)).await?;

    // List loaded functions
    use rustis::commands::FunctionListOptions;
    let functions = client.function_list(FunctionListOptions::default()).await?;
    for lib in functions {
        println!("Library: {}, Functions: {:?}", lib.library_name, lib.functions);
    }

    // Delete function library
    client.function_delete("mylib").await?;

    Ok(())
}

```

--------------------------------

### Batch Commands with Rustis Pipelines

Source: https://context7.com/dahomey-technologies/rustis/llms.txt

Utilize Rustis pipelines to group multiple Redis commands into a single network round trip. This is beneficial for improving performance by reducing latency. Commands can be queued to either forget their responses or to retrieve them upon execution. The execute() method returns a tuple containing the results of the queued commands in order.

```rust
use rustis::{
    client::Client,
    commands::StringCommands,
    Result,
};

#[tokio::main]
async fn main() -> Result<()> {
    let client = Client::connect("127.0.0.1:6379").await?;

    // Create a pipeline
    let mut pipeline = client.create_pipeline();

    // Queue commands with .forget() if you don't need the response
    pipeline.set("key1", "value1").forget();
    pipeline.set("key2", "value2").forget();

    // Queue commands with .queue() to get responses
    pipeline.get::<()>( "key1").queue();
    pipeline.get::<()>( "key2").queue();
    pipeline.get::<()>( "key3").queue();

    // Execute all commands at once and get tuple of results
    let (value1, value2, value3): (String, String, Option<String>) = pipeline.execute().await?;

    assert_eq!("value1", value1);
    assert_eq!("value2", value2);
    assert_eq!(None, value3);

    // Complex pipeline with mixed commands
    let mut pipeline = client.create_pipeline();

    use rustis::commands::{HashCommands, ListCommands, GenericCommands};

    pipeline.set("counter", 0).forget();
    pipeline.incr("counter").queue();
    pipeline.hset("user:100", ("name", "Alice")).forget();
    pipeline.hget::<()>( "user:100", "name").queue();
    pipeline.rpush("logs", ["event1", "event2"]).queue();
    pipeline.llen("logs").queue();

    let (count, name, push_result, list_len): (i32, String, usize, usize) =
        pipeline.execute().await?;

    println!("Counter: {}, Name: {}, Pushed: {}, List length: {}",
        count, name, push_result, list_len);

    Ok(())
}

```

--------------------------------

### Atomic Transactions with Rustis MULTI/EXEC

Source: https://context7.com/dahomey-technologies/rustis/llms.txt

Implement atomic transactions using Rustis, which automatically wraps commands in MULTI/EXEC. This guarantees that all commands within a transaction are executed sequentially as a single isolated operation. The API for queuing commands in a transaction is identical to that of pipelines. Transactions can also incorporate WATCH for optimistic locking, ensuring execution only if watched keys remain unchanged.

```rust
use rustis::{
    client::Client,
    commands::{StringCommands, TransactionCommands},
    Result,
};

#[tokio::main]
async fn main() -> Result<()> {
    let client = Client::connect("127.0.0.1:6379").await?;

    // Create a transaction
    let mut transaction = client.create_transaction();

    // Queue commands (same API as pipeline)
    transaction.set("account:1:balance", 1000).forget();
    transaction.set("account:2:balance", 500).forget();
    transaction.get::<()>( "account:1:balance").queue();
    transaction.get::<()>( "account:2:balance").queue();

    // Execute transaction atomically
    let (balance1, balance2): (i32, i32) = transaction.execute().await?;
    println!("Balances: {}, {}", balance1, balance2);

    // Transaction with WATCH for optimistic locking
    client.set("balance", 1000).await?;

    // Watch key for changes
    client.watch("balance").await?;

    // Read current value
    let mut balance: i32 = client.get("balance").await?;
    balance -= 100;  // Deduct amount

    // Create transaction to update
    let mut transaction = client.create_transaction();
    transaction.set("balance", balance).queue();

    // Execute transaction (will fail if "balance" was modified by another client)
    match transaction.execute::<()>().await {
        Ok(_) => println!("Transaction succeeded"),
        Err(e) => {
            println!("Transaction failed (key was modified): {}", e);
            client.unwatch().await?;
        }
    }

    // Transaction with conditional execution
    let mut transaction = client.create_transaction();

    use rustis::commands::ListCommands;
    transaction.lpop::<()>( "queue", 1).queue();
    transaction.rpush("processed", "item").queue();

    let result: Result<(Option<String>, usize)> = transaction.execute().await;
    match result {
        Ok((popped, _)) => println!("Processed: {:?}", popped),
        Err(e) => eprintln!("Transaction error: {}", e),
    }

    Ok(())
}

```

--------------------------------

### Real-time Pub/Sub Messaging with Rustis

Source: https://context7.com/dahomey-technologies/rustis/llms.txt

Subscribes to Redis channels and receives messages asynchronously. It supports single channel, multiple channels, pattern-based subscriptions, and shard channel subscriptions (for Redis 7.0+). Requires the `rustis` and `tokio` crates.

```rust
use rustis::
    client::Client,
    commands::{PubSubCommands, ListCommands},
;
use futures_util::StreamExt;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Create separate clients for pub/sub and regular operations
    let pub_sub_client = Client::connect("redis://127.0.0.1:6379").await?;
    let regular_client = Client::connect("redis://127.0.0.1:6379").await?;

    // Subscribe to a channel (returns async Stream)
    let mut sub_stream = pub_sub_client.subscribe("notifications").await?;

    // Publish messages from another client
    regular_client.publish("notifications", "Hello, World!").await?;

    // Receive messages from stream
    while let Some(message) = sub_stream.next().await {
        let msg = message?;
        let channel = String::from_utf8(msg.channel)?;
        let payload = String::from_utf8(msg.payload)?;
        println!("Received on {}: {}", channel, payload);

        // Break after first message for demo
        break;
    }

    // Close subscription
    sub_stream.close().await?;

    // Subscribe to multiple channels
    let mut multi_sub = pub_sub_client.subscribe(["channel1", "channel2", "channel3"]).await?;

    regular_client.publish("channel1", "message1").await?;
    regular_client.publish("channel2", "message2").await?;

    for _ in 0..2 {
        if let Some(Ok(msg)) = multi_sub.next().await {
            println!("Channel: {}, Payload: {}",
                String::from_utf8_lossy(&msg.channel),
                String::from_utf8_lossy(&msg.payload));
        }
    }

    // Pattern-based subscription
    let mut pattern_sub = pub_sub_client.psubscribe("user:*:events").await?;

    regular_client.publish("user:123:events", "login").await?;
    regular_client.publish("user:456:events", "logout").await?;

    if let Some(Ok(msg)) = pattern_sub.next().await {
        println!("Pattern: {}, Channel: {}, Payload: {}",
            String::from_utf8_lossy(&msg.pattern),
            String::from_utf8_lossy(&msg.channel),
            String::from_utf8_lossy(&msg.payload));
    }

    // Shard channel subscription (Redis 7.0+, for cluster)
    let mut shard_sub = pub_sub_client.ssubscribe("shard_channel").await?;
    regular_client.spublish("shard_channel", "shard_message").await?;

    Ok(())
}

```

--------------------------------

### Implement Reconnection and Error Handling with Rustis

Source: https://context7.com/dahomey-technologies/rustis/llms.txt

Details how to configure Rustis clients for robust Redis interactions, including automatic reconnection with exponential backoff and custom retry logic. It covers setting timeouts, handling various error types (Redis, Network), and subscribing to reconnection events.

```rust
use rustis::{
    client::{Client, Config, ReconnectionConfig},
    commands::{ConnectionCommands, StringCommands},
    Result,
};
use std::time::Duration;

#[tokio::main]
async fn main() -> Result<()> {
    // Configure client with custom reconnection strategy
    let mut config = Config::from_url("redis://127.0.0.1:6379")?;

    // Set timeouts
    config.connect_timeout = Duration::from_secs(5);
    config.command_timeout = Duration::from_secs(10);

    // Configure exponential backoff reconnection
    config.reconnection = ReconnectionConfig::Exponential {
        initial_delay: Duration::from_millis(100),
        max_delay: Duration::from_secs(30),
        factor: 2.0,
        max_retries: Some(10),
    };

    // Enable automatic retry on network errors
    config.retry_on_error = true;

    let client = Client::connect(config).await?;

    // Subscribe to reconnection events
    let mut reconnect_rx = client.on_reconnect();

    tokio::spawn(async move {
        while reconnect_rx.recv().await.is_ok() {
            println!("Client reconnected to Redis!");
        }
    });

    // Normal operations with automatic retry
    match client.set("key", "value").await {
        Ok(_) => println!("Operation succeeded"),
        Err(e) => eprintln!("Operation failed: {}", e),
    }

    // Override retry behavior for specific command
    let result = client.get::<String>("key").retry_on_error(false).await;

    // Handle different error types
    use rustis::{Error, RedisErrorKind};
    match client.get::<String>("nonexistent").await {
        Ok(value) => println!("Found: {}", value),
        Err(Error::Redis(redis_err)) => {
            match redis_err.kind {
                RedisErrorKind::WrongType => eprintln!("Wrong data type"),
                RedisErrorKind::NoScript => eprintln!("Script not found"),
                _ => eprintln!("Redis error: {}", redis_err.description),
            }
        }
        Err(Error::Network(net_err)) => {
            eprintln!("Network error: {}", net_err);
            // Connection will auto-reconnect
        }
        Err(e) => eprintln!("Other error: {}", e),
    }

    // Gracefully close connection
    client.close().await?;

    Ok(())
}

```

--------------------------------

### Long Polling Pub/Sub for Web APIs with Rustis and Axum

Source: https://context7.com/dahomey-technologies/rustis/llms.txt

Implements a long-polling pattern for web APIs using Rustis and Axum. It leverages Redis lists for message queues and Redis pub/sub for notifications. The API allows clients to poll for messages on a channel and publish messages to a channel. Requires `rustis`, `axum`, and `tokio`.

```rust
use axum::
    Json,
    Router,
    extract::{Path, State},
    routing::get,
;
use futures_util::StreamExt;
use rustis::
    client::Client,
    commands::{ListCommands, PubSubCommands},
    Result,
;
use std::{net::SocketAddr, sync::Arc, time::Duration};
use tokio::net::TcpListener;

const POLL_TIMEOUT: Duration = Duration::from_secs(10);

pub struct RedisClients {
    pub regular: Client,  // For regular operations
    pub sub: Client,      // Dedicated for subscriptions
}

#[tokio::main]
async fn main() {
    let redis_uri = "redis://127.0.0.1:6379";
    let redis_clients = Arc::new(RedisClients {
        regular: Client::connect(redis_uri).await.unwrap(),
        sub: Client::connect(redis_uri).await.unwrap(),
    });

    let app = Router::new()
        .route("/:channel", get(poll_messages).post(publish))
        .with_state(redis_clients);

    let addr = SocketAddr::from(([127, 0, 0, 1], 3000));
    let listener = TcpListener::bind(&addr).await.unwrap();
    axum::serve(listener, app).await.unwrap();
}

async fn poll_messages(
    State(redis): State<Arc<RedisClients>>,
    Path(channel): Path<String>,
) -> Result<Json<Vec<String>>, rustis::Error> {
    // First check if messages are already in queue
    let messages: Vec<String> = redis.regular.lpop(&channel, i32::MAX as usize).await?;
    if !messages.is_empty() {
        return Ok(Json(messages));
    }

    // No messages, subscribe and wait
    let mut sub_stream = redis.sub.subscribe(&channel).await?;
    let msg = tokio::time::timeout(POLL_TIMEOUT, sub_stream.next()).await;

    match msg {
        Ok(Some(Ok(_))) => {
            // Notification received, fetch actual messages from queue
            let messages = redis.regular.lpop(&channel, i32::MAX as usize).await?;
            Ok(Json(messages))
        }
        _ => Ok(Json(Vec::new())),  // Timeout or error
    }
}

async fn publish(
    State(redis): State<Arc<RedisClients>>,
    Path(channel): Path<String>,
    message: String,
) -> Result<(), rustis::Error> {
    // Store actual message in list (queue)
    redis.regular.rpush(&channel, &message).await?;

    // Send notification via pub/sub
    redis.regular.publish(&channel, "new").await?;

    Ok(())
}

```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.