redis#

A Redis client close to ioredis: strings with expiry, counters, hashes, lists, sets, sorted sets, pipelines, MULTI/EXEC transactions and pub/sub, over redis:// or rediss:// (TLS). The client is the package's native library: RESP2 in Rust, on Velt's own sockets and rustls.

import { connect, subscribe, psubscribe, RedisClient, RedisPipeline, RedisReply, RedisSubscriber, RedisMessage, RedisError, ZMember, SetOptions, RedisConnectOptions } from "redis";

Adding it to a program#

There is no public package registry yet. Depend on the package in the Velt repository, from git or from a local checkout, in package.vlt:

import type { Package } from "velt/package";

export const pkg: Package = {
  name: "app",
  version: "0.1.0",
  dependencies: {
    redis: { git: "https://github.com/velt-lang/velt", branch: "main", path: "packages/redis" },
    // Or a local checkout of the repository:
    // redis: { path: "../velt/packages/redis" },
  },
};

The client is a Rust crate (the package's native/ directory), so both forms build it with cargo the first time and need a Rust toolchain. A path dependency builds it directly; a git dependency builds it only when you allow that with VELT_NATIVE_FROM_SOURCE=1, since the build runs the package's build scripts. A package published to a registry carries prebuilt libraries, and then needs no Rust (packages with native code).

A first program#

import { connect, subscribe } from "redis";

const redis = await connect("redis://127.0.0.1:6379/0");
await redis.set("greeting", "hello", { ex: 60 });
console.log(await redis.get("greeting")); // hello
console.log(await redis.incr("visits")); // 1, 2, …

const p = redis.pipeline();
p.hset("user:1", "name", "Ada");
p.hget("user:1", "name");
const replies = await p.exec();
console.log(replies[1].asString()); // Ada

const sub = await subscribe(redis, ["chat"]);
await redis.publish("chat", "hi");
for await (const m of sub) {
  console.log(`${m.channel}: ${m.message}`); // chat: hi
  sub.close(); // the loop ends: the next `next()` returns null
}
redis.close();

Connecting#

connect(url, opts: RedisConnectOptions = {}): Promise<RedisClient>. The URL is redis://[[user]:password@]host[:port][/db]; AUTH and SELECT are sent for you. rediss:// uses TLS, trusting the built-in roots and the PEM certificates in opts.ca.

One RedisClient is one multiplexed connection: commands from any number of tasks are pipelined on it in order, so it already works as a connection pool (there is no createPool). It is a handle: pass it around and capture it in handlers and tasks. duplicate(): Promise<RedisClient> opens a second connection to the same server and database, for blocking commands (BLPOP, …) that must not hold up the others. close() releases it; closing again does nothing.

A lost connection is reopened automatically. Commands already sent on it throw RedisError ECONNRESET (they may have run, so they are not resent); later commands wait while the client reconnects with exponential backoff (50 ms doubling to 2 s, for up to 10 s; then they throw the connect error and the next command tries again). Subscribers do not reconnect.

Commands#

Every command is an async method. Commands that take several keys or values take a string[]: del(["a", "b"]). Values are UTF-8 strings (invalid bytes read back as U+FFFD).

Keys and strings#

get(key): string | null, set(key, value, opts: SetOptions = {}): bool with SetOptions { ex?, px?, nx?, xx? } (false when nx/xx prevented it), del(keys), exists(keys), expire(key, seconds): bool, pexpire(key, ms): bool, ttl(key), pttl(key) (-1 no expiry, -2 missing), persist(key): bool, incr, incrBy(key, n), decr, decrBy (all i64), mget(keys): (string | null)[], mset(entries: Map<string, string>), keys(pattern): string[], ping(): string.

• Hashes: hget(key, field): string | null, hset(key, field, value): i64, hmset(key, fields: Map<string, string>): i64, hgetall(key): Map<string, string>, hdel(key, fields): i64, hincrBy(key, field, n): i64, hlen(key): i64.
• Lists: lpush(key, values), rpush(key, values) (the new length), lpop(key), rpop(key) (string | null), lrange(key, start, stop): string[] (negative indexes count from the end), llen(key).
• Sets: sadd(key, members), srem(key, members) (i64), smembers(key): string[], sismember(key, member): bool, scard(key).

Sorted sets#

zadd(key, members: ZMember[]): i64 with ZMember { member, score: f64 }, zrange(key, start, stop): string[], zrangeWithScores(key, start, stop): ZMember[], zscore(key, member): f64 | null, zrem(key, members): i64, zincrby(key, increment, member): f64, zcard(key). Infinite scores map to ±inf.

• Anything else: call(args: string[]): RedisReply runs any command (args[0] is its name).

RedisReply { kind; text; int; items }, where kind is "nil", "status", "error", "int", "string" or "array", has isNull(), isError(), asString(): string | null, asInt(): i64 | null and asStrings(): string[].

const reply = await redis.call(["OBJECT", "ENCODING", "visits"]);
console.log(reply.asString()); // int

Pipelines and transactions#

pipeline() and multi() return a RedisPipeline. Queue commands with get, set, del, incr, incrBy, expire, hget, hset, lpush, rpush, sadd, zadd(key, score, member), publish or call(args) (they return nothing), then exec(): Promise<RedisReply[]> sends them and returns one reply per command. length counts the queued commands.

In a pipeline a rejected command is an "error" reply, not a throw. multi() wraps the commands in MULTI/EXEC, so they run as one transaction; one the server aborts throws RedisError EXECABORT.

const tx = redis.multi();
tx.incr("orders");
tx.rpush("queue", ["order:42"]);
const [count, length] = await tx.exec();
console.log(count.asInt(), length.asInt());

Pub/sub#

subscribe(target: string | RedisClient, channels: string[], opts = {}): Promise<RedisSubscriber> and psubscribe(target, patterns, opts = {}) (glob patterns) open a dedicated connection (to the URL, or to the server and database a client uses) and resolve once the server confirmed. publish(channel, message): i64 on a client returns the number of receivers.

A RedisSubscriber is a handle with next(): Promise<RedisMessage | null> (null after close()), subscribe(channels), unsubscribe(channels), psubscribe(patterns), punsubscribe(patterns) and close() (from any task; closing again does nothing). It is an AsyncIterable<RedisMessage, RedisError>: for await (const m of sub) receives messages until close(); leaving the loop early keeps the subscription. RedisMessage { channel; message; pattern: string | null }.

Messages are pulled with next() or for await; the client stores no callbacks.

Errors#

RedisError { code, message }. code is the server's error code for error replies ("WRONGTYPE", "ERR", "NOAUTH", "WRONGPASS", "EXECABORT", …) or an I/O code ("ECONNREFUSED", "ECONNRESET", "ETIMEDOUT", "EINVAL" for a bad URL or argument).

Testing it#

Most of the package's tests (velt test in packages/redis) need a server: set VELT_TEST_REDIS_URL (redis://host:6379), and for TLS VELT_TEST_REDISS_URL and VELT_TEST_REDISS_CA (a PEM file). Without them those tests do nothing. Tests of your own program that use Redis need a server the same way; pass its URL in an environment variable.