Packages

@kama/redis 0.1.0

Redis and Valkey client in pure kama: RESP2 and RESP3, TLS, pipelining, pub/sub, cluster and sentinel

Install

kama pkg add kama.json @kama/redis --version ^0.1.0

or in kama.json: "@kama/redis": { "version": "^0.1.0" }

License
MIT OR Apache-2.0
Repository
github.com/cosmic-canopy/kama-redis
Keywords
redis valkey resp database cache
Dependencies
@kama/tls ^0.1.3

Versions

VersionPublishedRevisionIntegrity
0.1.0146a4d834167sha256-6c66eef98f45…

@kama/redis

The Redis and Valkey client for kama. It speaks RESP2 and RESP3 directly, in kama, over std::net, as redis-rs, go-redis and Lettuce do. There is no hiredis and nothing to install. TLS comes from @kama/tls.

Status: 0.1.0, a complete client for one server. A connection authenticates, picks its protocol and database, connects over TLS, and sends commands, pipelines, transactions and scripts with deadlines; replies read into kama types. A subscriber listens to channels, patterns and shard channels. A lost connection is made again by its next call, a subscriber's with every subscription, and nothing is sent twice. It is tested live on Redis 6.2–8.10 and Valkey 7.2–9.1. Cluster and Sentinel come next; see the roadmap. Needs kama ≥ 0.9.667.

Connecting

import { core::println, redis::Config, redis::Connection, redis::Command, redis::RedisError, redis::cmd,
         redis::reply, redis::replyOpt };

fn int32 main() {
    Result<Config, RedisError> parsed = Config.parse(url: "redis://127.0.0.1:6379/0?client_name=example");
    Config config = match (give parsed) {
        case Ok(value: c): give c;
        case Err(error: e): { println(s: e.message()); return 1; }
    };
    Result<Connection, RedisError> opened = Connection.connect(config: config);
    Connection conn = match (give opened) {
        case Ok(value: c): give c;
        case Err(error: e): { println(s: e.message()); return 1; }
    };

    // Each hole is one whole argument, spaces and line breaks included: never text spliced into a word.
    string key = "greeting";
    string value = "hello, world";
    int32 ttl = 60;
    Command set = cmd"SET ${key} ${value} EX ${ttl}";
    Result<string, RedisError> stored = reply::<string>(result: conn.call(command: set));

    Command get = cmd"GET ${key}";
    Result<Optional<string>, RedisError> got = replyOpt::<string>(result: conn.call(command: get));
    match (give got) {
        case Ok(value: found): {
            match (give found) {
                case Some(value: text): { println(s: give text); }
                case None: { println(s: "(no such key)"); }
            };
        }
        case Err(error: e): { println(s: e.message()); return 1; }
    };
    conn.close();
    return 0;
}

TLS

rediss:// (and valkeys://) connect over TLS, from @kama/tls, which vendors Mbed TLS: still nothing to install.

import { core::println, redis::Config, redis::Connection, redis::RedisError, redis::cmd, redis::reply };

fn int32 main() {
    // The server's certificate must chain to this CA and name 127.0.0.1; a client certificate goes with it.
    string url = "rediss://127.0.0.1:6380?tls_ca=/etc/redis/ca.crt&tls_cert=/etc/redis/client.crt&tls_key=/etc/redis/client.key";
    Result<Config, RedisError> parsed = Config.parse(url: url);
    Config config = match (give parsed) {
        case Ok(value: c): give c;
        case Err(error: e): { println(s: e.message()); return 1; }
    };
    Result<Connection, RedisError> opened = Connection.connect(config: config);
    Connection conn = match (give opened) {
        case Ok(value: c): give c;
        case Err(error: e): { println(s: e.message()); return 1; }
    };
    Result<string, RedisError> pong = reply::<string>(result: conn.call(command: cmd"PING"));
    match (give pong) {
        case Ok(value: p): { println(s: give p); }
        case Err(error: e): { println(s: e.message()); return 1; }
    };
    conn.close();
    return 0;
}

Commands

Replies, pipelines, transactions, scripts and SCAN

import { core::println, std::collections::DynamicArray, redis::Config, redis::Connection, redis::Command,
         redis::Pipeline, redis::Replies, redis::RedisError, redis::TxOutcome, redis::Script, redis::ScriptCall,
         redis::Scan, redis::cmd, redis::reply, redis::replyRecordOpt };

// A struct reads by its fields' names: HGETALL's map under RESP3, and its flat list under RESP2.
@generate(Deserializable)
type resource Account {
    @field string owner;
    @field int64 balance;
    public const fn string describe() { int64 b = this.balance; return copy this.owner.concat(other: ": ${b}"); }
}

// BLPOP's [key, value] reads by position, as a record.
@generate(Deserializable)
type resource Popped {
    @field string list;
    @field string item;
    public const fn string describe() { return copy this.list.concat(other: " gave ").concat(other: this.item); }
}

fn int32 main() {
    Result<Config, RedisError> parsed = Config.parse(url: "redis://127.0.0.1:6379");
    Config config = match (give parsed) {
        case Ok(value: c): give c;
        case Err(error: e): { println(s: e.message()); return 1; }
    };
    Result<Connection, RedisError> opened = Connection.connect(config: config);
    Connection conn = match (give opened) {
        case Ok(value: c): give c;
        case Err(error: e): { println(s: e.message()); return 1; }
    };
    string account = "account:42";
    string owner = "ada";
    int64 balance = 100;

    // A pipeline: every command in one write, every reply in order.
    Pipeline setup = Pipeline.make();
    Command clear = cmd"DEL ${account} jobs";
    Command fill = cmd"HSET ${account} owner ${owner} balance ${balance}";
    Command queue = cmd"RPUSH jobs first";
    setup.add(command: clear);
    setup.add(command: fill);
    setup.add(command: queue);
    Result<Replies, RedisError> ran = conn.run(pipeline: setup);
    match (ran) { case Ok(value: r): { } case Err(error: e): { println(s: e.message()); return 1; } };

    // Typed replies: a struct from a hash.
    Command all = cmd"HGETALL ${account}";
    Result<Account, RedisError> read = reply::<Account>(result: conn.call(command: all));
    match (read) {
        case Ok(value: a): { println(s: a.describe()); }
        case Err(error: e): { println(s: e.message()); return 1; }
    };

    // A transaction: MULTI, the commands and EXEC in one write. A key changed under WATCH is WatchConflict.
    int64 debit = -30;
    Pipeline tx = Pipeline.make();
    Command take = cmd"HINCRBY ${account} balance ${debit}";
    Command log = cmd"RPUSH jobs debited";
    tx.add(command: take);
    tx.add(command: log);
    Result<TxOutcome, RedisError> outcome = conn.transact(pipeline: tx);
    match (give outcome) {
        case Ok(value: o): {
            match (give o) {
                case Committed(replies: r): {
                    Replies replies = give r;
                    Result<int64, RedisError> left = reply::<int64>(result: replies.take(index: 0));
                    match (left) { case Ok(value: n): { println(s: "balance ${n}"); } case Err(error: e): { println(s: e.message()); } };
                }
                case WatchConflict: { println(s: "a watched key changed"); }
            };
        }
        case Err(error: e): { println(s: e.message()); return 1; }
    };

    // A script: by its SHA-1, and by its source the first time a server lacks it.
    Script bump = Script.of(source: "return redis.call('HINCRBY', KEYS[1], 'balance', ARGV[1])");
    ScriptCall call = ScriptCall.of(script: bump);
    int64 tip = 5;
    call.addKey(value: account);
    call.addArg(value: tip);
    Result<int64, RedisError> bumped = reply::<int64>(result: conn.eval(call: call));
    match (bumped) { case Ok(value: n): { println(s: "balance ${n}"); } case Err(error: e): { println(s: e.message()); return 1; } };

    // A blocking pop: a record, or None when nothing came within its timeout.
    Command pop = cmd"BLPOP jobs 1";
    Result<Optional<Popped>, RedisError> popped = replyRecordOpt::<Popped>(result: conn.call(command: pop));
    match (popped) {
        case Ok(value: o): { match (o) { case Some(value: p): { println(s: p.describe()); } case None: { println(s: "nothing"); } }; }
        case Err(error: e): { println(s: e.message()); return 1; }
    };

    // SCAN, a page at a time: each `next` is one round trip.
    Scan scan = Scan.keys();
    string pattern = "account:*";
    scan.matching(pattern: pattern);
    isize found = 0;
    while (!scan.isDone()) {
        Result<DynamicArray<string>, RedisError> page = reply::<DynamicArray<string>>(result: scan.next(conn: conn));
        match (page) { case Ok(value: keys): { found = found + keys.length(); } case Err(error: e): { println(s: e.message()); return 1; } };
    }
    println(s: "${found} accounts");
    conn.close();
    return 0;
}

When the connection is lost

A connection is lost when the server restarts, kills it (CLIENT KILL, its idle timeout, a full output buffer), or the network fails. The call that meets the loss fails with what happened, and nothing is ever sent again on its own: a command that may have run is yours to retry or not. The next call makes the connection again, as the first was made, then sends.

Whether a failed call's command could have run:

error sent? what it means
Usage, Value, Config no refused before anything was written; the connection goes on
Reconnect no the connection was lost earlier and could not be made again; cause() says why, and the next call tries again
Closed no closed by close(), or lost with nothing to make it again with
Connect, Tls while connecting no no connection was made
Server yes the server answered with an error; LOADING, BUSY, TRYAGAIN and MASTERDOWN mean it did not run it
Io, Timeout, Protocol, Unexpected, Tls during a call maybe the connection was lost with the command sent or half answered: it may have run

A reliable queue shows one way to live with "maybe": BLMOVE into a processing list, so a job that may have been taken is found again.

Pub/sub

import { core::println, std::collections::DynamicArray, std::time::Duration, redis::Config, redis::Connection,
         redis::Subscriber, redis::Delivery, redis::Message, redis::Command, redis::RedisError, redis::cmd,
         redis::reply };

fn string describe(const ref Message m) {
    Result<string, RedisError> channel = m.channelText();
    Result<string, RedisError> payload = m.text();
    return match (channel) {
        case Ok(value: c): match (payload) { case Ok(value: p): c.concat(other: ": ").concat(other: p); case Err(error: e): e.message(); };
        case Err(error: e): e.message();
    };
}

fn int32 main() {
    Result<Config, RedisError> parsed = Config.parse(url: "redis://127.0.0.1:6379");
    Config config = match (give parsed) {
        case Ok(value: c): give c;
        case Err(error: e): { println(s: e.message()); return 1; }
    };
    // A subscriber owns a connection of its own, under RESP3 and RESP2 alike.
    Result<Subscriber, RedisError> made = Subscriber.connect(config: config);
    Subscriber sub = match (give made) {
        case Ok(value: s): give s;
        case Err(error: e): { println(s: e.message()); return 1; }
    };
    // Each call returns once the server has confirmed every name.
    DynamicArray<string> channels = DynamicArray.empty();
    channels.add(item: "news");
    channels.add(item: "alerts");
    Result<Unit, RedisError> subscribed = sub.subscribe(names: channels);
    match (subscribed) { case Ok(value: u): { } case Err(error: e): { println(s: e.message()); return 1; } };
    Result<Unit, RedisError> patterned = sub.psubscribe(names: "log:*");
    match (patterned) { case Ok(value: u): { } case Err(error: e): { println(s: e.message()); return 1; } };

    // Publishing is an ordinary command, on any other connection. It answers how many subscribers got the message.
    Result<Connection, RedisError> opened = Connection.connect(config: config);
    Connection conn = match (give opened) {
        case Ok(value: c): give c;
        case Err(error: e): { println(s: e.message()); return 1; }
    };
    Command publish = cmd"PUBLISH news hello";
    Result<int64, RedisError> count = reply::<int64>(result: conn.call(command: publish));
    match (count) { case Ok(value: n): { println(s: "${n} subscriber got it"); } case Err(error: e): { println(s: e.message()); return 1; } };

    // Wait up to a second for the next message. None means the time passed; the subscriber goes on.
    Optional<Duration> second = Optional::Some(value: Duration.fromSecs(s: 1i64));
    Result<Optional<Delivery>, RedisError> next = sub.next(timeout: second);
    match (give next) {
        case Ok(value: o): {
            match (give o) {
                case Some(value: d): {
                    match (give d) {
                        case Message(message: m): { println(s: describe(m: m)); }
                        case Unsubscribed(kind: k, channel: c): { println(s: "the server ended a subscription"); }
                        case Reconnected(cause: e): { println(s: "reconnected, and messages may have been missed"); }
                    };
                }
                case None: { println(s: "nothing within a second"); }
            };
        }
        case Err(error: e): { println(s: e.message()); return 1; }
    };

    // Unsubscribe from everything and take the connection back, for commands.
    Result<Connection, RedisError> released = sub.release();
    Connection again = match (give released) {
        case Ok(value: c): give c;
        case Err(error: e): { println(s: e.message()); return 1; }
    };
    Command ping = cmd"PING";
    Result<string, RedisError> pong = reply::<string>(result: again.call(command: ping));
    match (give pong) { case Ok(value: p): { println(s: give p); } case Err(error: e): { println(s: e.message()); return 1; } };
    conn.close();
    again.close();
    return 0;
}

Underneath

Examples

Each is a program of its own under examples/, built by tools/test.sh and run against every server in the matrix by tools/check-examples.sh. Run one with REDIS_URL=redis://127.0.0.1:6379 kama run examples/quickstart/kama.json.

What comes next

On Redis 6.2 to 8.x and Valkey 7.2 to 9.x, tested on every one:

Tests

KAMA=/path/to/kama tools/test.sh                    # hermetic unit tests, debug and release
KAMA=/path/to/kama tools/test-integration.sh        # the live suite on every server, debug and release
tools/test-integration.sh --server valkey-9.1       # … on one
tools/redis.sh list                                 # the servers in the matrix
tools/redis.sh up --server valkey-9.1               # start one (podman or docker); its ports go to out/redistest-valkey-9.1.env
tools/redis.sh smoke --server valkey-9.1            # one check per authentication case and transport, with the server's own redis-cli
tools/redis.sh down --server valkey-9.1
tools/check-readme.sh --server redis-8.10           # build this README's programs and run them against the server
tools/check-examples.sh --server redis-8.10         # build examples/ and run them against the server
tools/mutate.sh tools/mutations/7a-s5.json          # each mutation of the source must make the case it targets fail

The unit tests need no server: a scripted server plays real servers' bytes to the connection, a byte at a time. Their vectors come from upstream by script: tools/gen-resp-vectors.sh, tools/gen-slot-vectors.sh, tools/gen-url-vectors.sh and tools/gen-server-vectors.sh. Each pins its sources by SHA256 and asks the servers and CLIs in throwaway containers.

Each test server is one container running two servers, configured from tests/integration/server/:

Its test certificates are generated into out/ and never committed.

License

@kama/redis is licensed under either of

at your option. It vendors nothing.

Contribution

Unless you explicitly state otherwise, any contribution intentionally submitted for inclusion in @kama/redis by you, as defined in the Apache-2.0 license, shall be dual licensed as above, without any additional terms or conditions.