tinyredis 1.0.0

A Redis-compatible server written in Rust. Uses RESP2, persists writes to an append-only file, and accepts connections from any standard Redis client.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129

use std::sync::Arc;
use std::sync::atomic::Ordering;

use crate::commands::utils::{aof_log, to_str};
use crate::error::{Error, Result};
use crate::parser::{Command, Frame};
use crate::persistence::AofSender;
use crate::stats::SharedStats;
use crate::store::Store;
use bytes::Bytes;
use tokio::sync::Mutex;

type SharedStore = Arc<Mutex<Store>>;

pub async fn keys(cmd: &Command, store: &SharedStore) -> Result<Frame> {
    let pattern = to_str(cmd.arg(0, "KEYS")?, "KEYS")?;
    let mut keys = store.lock().await.keys(&pattern);
    keys.sort(); // deterministic order
    Ok(Frame::Array(
        keys.into_iter()
            .map(|k| Frame::Bulk(Bytes::from(k)))
            .collect(),
    ))
}

/// SCAN cursor [MATCH pattern] [COUNT count]
/// Always completes in one call (returns cursor "0").
pub async fn scan(cmd: &Command, store: &SharedStore) -> Result<Frame> {
    // cursor argument is accepted but ignored — we always do a full scan
    let _ = cmd.arg(0, "SCAN")?;

    let mut pattern = "*".to_string();
    let mut i = 1;
    while i < cmd.args.len() {
        match cmd.args[i].to_ascii_uppercase().as_slice() {
            b"MATCH" => {
                i += 1;
                pattern = to_str(cmd.arg(i, "SCAN")?, "SCAN")?;
            }
            b"COUNT" => {
                i += 1; // accept and ignore
            }
            _ => return Err(Error::Syntax),
        }
        i += 1;
    }

    let mut keys = store.lock().await.keys(&pattern);
    keys.sort();
    let elements: Vec<Frame> = keys
        .into_iter()
        .map(|k| Frame::Bulk(Bytes::from(k)))
        .collect();
    Ok(Frame::Array(vec![
        Frame::Bulk(Bytes::from_static(b"0")),
        Frame::Array(elements),
    ]))
}

pub async fn dbsize(store: &SharedStore) -> Result<Frame> {
    Ok(Frame::Integer(store.lock().await.dbsize() as i64))
}

pub async fn rename(cmd: &Command, store: &SharedStore, aof: &Option<AofSender>) -> Result<Frame> {
    let key = to_str(cmd.arg(0, "RENAME")?, "RENAME")?;
    let newkey = to_str(cmd.arg(1, "RENAME")?, "RENAME")?;
    store.lock().await.rename(&key, &newkey)?;
    aof_log(
        aof,
        Frame::Array(vec![
            Frame::Bulk(Bytes::from_static(b"RENAME")),
            Frame::Bulk(Bytes::from(key)),
            Frame::Bulk(Bytes::from(newkey)),
        ]),
    )
    .await;
    Ok(Frame::Simple("OK".into()))
}

pub async fn flushdb(store: &SharedStore, aof: &Option<AofSender>) -> Result<Frame> {
    store.lock().await.flushdb();
    aof_log(
        aof,
        Frame::Array(vec![Frame::Bulk(Bytes::from_static(b"FLUSHDB"))]),
    )
    .await;
    Ok(Frame::Simple("OK".into()))
}

pub async fn flushall(store: &SharedStore, aof: &Option<AofSender>) -> Result<Frame> {
    store.lock().await.flushdb();
    aof_log(
        aof,
        Frame::Array(vec![Frame::Bulk(Bytes::from_static(b"FLUSHALL"))]),
    )
    .await;
    Ok(Frame::Simple("OK".into()))
}

pub async fn bgrewriteaof(stats: &SharedStats) -> Result<Frame> {
    match &stats.rewrite_tx {
        None => Ok(Frame::Error(
            "ERR AOF is not enabled or rewrite channel unavailable".into(),
        )),
        Some(tx) => {
            if stats
                .aof_rewrite_in_progress
                .compare_exchange(false, true, Ordering::Relaxed, Ordering::Relaxed)
                .is_err()
            {
                return Ok(Frame::Error(
                    "ERR Background append only file rewriting already in progress".into(),
                ));
            }
            // Best-effort send; if the channel is full the rewrite is already queued.
            if tx.try_send(()).is_err() {
                stats
                    .aof_rewrite_in_progress
                    .store(false, Ordering::Relaxed);
                return Ok(Frame::Error(
                    "ERR Background rewrite already scheduled".into(),
                ));
            }
            Ok(Frame::Simple(
                "Background append only file rewriting started".into(),
            ))
        }
    }
}