Standard library · Databases

db/oxiresp

Imported as import "db/oxiresp" as oxiresp, its names are then oxiresp.…. Every signature below is the one the checker infers.

A client for OxiDB's in-memory key-value engine, written in Rill.

This engine is not on the wire the other two clients here use. It listens on a port of its own — the server is started with OXIDB_OXIMEM_PORT — and it speaks RESP, the protocol Redis speaks, so what is written here is what any Redis client writes and the engine cannot tell them apart.

RESP2, and the whole of it:

+OKa simple string -ERR no such keyan error :42an integer $3a bulk string: a length, then exactly that many bytes $-1no value 2 an array -1no array — what EXEC answers when a transaction broke

A command is an array of bulk strings and nothing else, which is why every argument below is a Str: numbers go out as their digits and come back as whatever the command decided to answer with.

Unlike a length-prefixed wire, RESP gives no frame size up front — a reader has to parse to find out whether it has the whole reply. So decoding answers RMore for a reply that has not all arrived, and the socket is read again. One command, one reply, except while subscribed — see the bottom of this file, where the reply to a command and a message nobody asked for arrive through the same socket and have to be told apart.

# not run
import "db/oxiresp"
fd = kv_connect("127.0.0.1", 6379)?              # the in-memory engine, or any Redis
kv_set(fd, "greeting", "hello")
kv_text(kv_get(fd, "greeting")?)                # "hello"
kv_int(kv_incr(fd, "visits")?)                  # 1
kv_texts(kv_keys(fd, "*")?)                     # every key
kv_hang_up(fd)

Types

Resp

• RSimple(Str)
• RError(Str)
• RInt(Int)
• RBulk(Str)
• RNil
• RNilArray
• RArray(List(Resp))

RespRead

• RGot(value: Resp, pos: Int)
• RMore
• RBad(why: Str, pos: Int)

Sub

A subscription is a socket and what has been read off it but not yet used. Every command until now has been one write and one reply, so whatever came after the reply could be thrown away; a subscriber cannot do that, because one read may carry two messages and the second is not going to be sent again. So the leftover is carried, and kv_next answers a new subscription along with the message — the bytes have to live somewhere, and a binding does not change.

• Sub(fd: Int, buf: Str)

Delivery

• Message(sub: Sub, channel: Str, payload: Str)
• PatternMessage(sub: Sub, pattern: Str, channel: Str, payload: Str)
• Note(sub: Sub, kind: Str, name: Str, count: Int)
• Ended(why: Str)

Functions

fn resp_command(args: List(Str)) -> Str

import "db/oxiresp"
resp_command(Cons("GET", Cons("k", Nil)))   # => *2\r\n$3\r\nGET\r\n$1\r\nk\r\n

fn resp_decode(s: Str, i: Int) -> RespRead

RMore and RBad are different answers and the difference is the whole reason this can be driven off a socket: a reply that has not finished arriving is not a reply that is wrong.

import "db/oxiresp"
resp_decode("+OK\r\n", 0)                       # => RGot(RSimple(OK), 5)
resp_decode(":42\r\n", 0)                       # => RGot(RInt(42), 5)
resp_decode("$3\r\nada\r\n", 0)                 # => RGot(RBulk(ada), 9)
resp_decode("$-1\r\n", 0)                       # => RGot(RNil, 5)
resp_decode("*2\r\n:1\r\n:2\r\n", 0)             # => RGot(RArray(Cons(RInt(1), Cons(RInt(2), Nil))), 12)
resp_decode("-ERR no such key\r\n", 0)          # => RGot(RError(ERR no such key), 18)
resp_decode("$3\r\nad", 0)                      # => RMore
resp_decode("?x\r\n", 0)                          # => RBad(no RESP value begins with `?`, 0)

fn kv_connect(host: Str, port: Int) -> Result(Int, Str)

# not run
import "db/oxiresp"
fd = kv_connect("127.0.0.1", 6379)?

fn kv_hang_up(fd: Int) -> Unit

# not run
import "db/oxiresp"
kv_hang_up(fd)

fn kv_command(fd: Int, args: List(Str)) -> Result(Resp, Str)

Every call answers Ok(value) or Err(message). An error the server sent and a connection that broke are both failures, and both belong here.

# not run
import "db/oxiresp"
kv_command(fd, Cons("CONFIG", Cons("GET", Cons("maxmemory", Nil))))   # any command, as its words

fn kv_args1(a: 'a) -> List('a)

import "db/oxiresp"
kv_args1("PING")   # => Cons(PING, Nil)

fn kv_args2(a: 'a, b: 'a) -> List('a)

import "db/oxiresp"
kv_args2("GET", "k")   # => Cons(GET, Cons(k, Nil))

fn kv_args3(a: 'a, b: 'a, c: 'a) -> List('a)

import "db/oxiresp"
kv_args3("SET", "k", "v")   # => Cons(SET, Cons(k, Cons(v, Nil)))

fn kv_args4(a: 'a, b: 'a, c: 'a, d: 'a) -> List('a)

import "db/oxiresp"
kv_args4("HSET", "h", "f", "v")   # => Cons(HSET, Cons(h, Cons(f, Cons(v, Nil))))

fn kv_ping(fd: Int) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_ping(fd)   # Ok(RSimple("PONG"))

fn kv_echo(fd: Int, text: Str) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_echo(fd, "hi")   # Ok(RBulk("hi"))

fn kv_set(fd: Int, key: Str, value: Str) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_set(fd, "k", "v")   # Ok(RSimple("OK"))

fn kv_setex(fd: Int, key: Str, seconds: Int, value: Str) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_setex(fd, "session", 3600, "token")   # the value, gone in an hour

fn kv_get(fd: Int, key: Str) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_text(kv_get(fd, "k")?)   # "v", or "" when there is no such key: kv_missing tells the two apart

fn kv_getset(fd: Int, key: Str, value: Str) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_getset(fd, "k", "new")   # the old value

fn kv_append(fd: Int, key: Str, value: Str) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_int(kv_append(fd, "k", "more")?)   # the new length

fn kv_strlen(fd: Int, key: Str) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_int(kv_strlen(fd, "k")?)   # how long the value is

fn kv_incr(fd: Int, key: Str) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_int(kv_incr(fd, "visits")?)   # the count after

fn kv_incrby(fd: Int, key: Str, by: Int) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_int(kv_incrby(fd, "visits", 10)?)   #

fn kv_mset(fd: Int, pairs: List(Str)) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_mset(fd, Cons("a", Cons("1", Cons("b", Cons("2", Nil)))))   # key, value, key, value

fn kv_mget(fd: Int, keys: List(Str)) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_texts(kv_mget(fd, Cons("a", Cons("b", Nil)))?)   # Cons("1", Cons("2", Nil))

fn kv_del(fd: Int, key: Str) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_int(kv_del(fd, "k")?)   # 1 if it was there

fn kv_exists(fd: Int, key: Str) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_int(kv_exists(fd, "k")?)   # 1 or 0

fn kv_type(fd: Int, key: Str) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_text(kv_type(fd, "k")?)   # "string", "list", "hash", "set", "zset" or "none"

fn kv_rename(fd: Int, key: Str, to: Str) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_rename(fd, "old", "new")   #

fn kv_expire(fd: Int, key: Str, seconds: Int) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_expire(fd, "session", 60)   # a minute more

fn kv_ttl(fd: Int, key: Str) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_int(kv_ttl(fd, "session")?)   # seconds left; -1 with no expiry, -2 with no key

fn kv_persist(fd: Int, key: Str) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_persist(fd, "session")   # the expiry taken off

fn kv_keys(fd: Int, pattern: Str) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_texts(kv_keys(fd, "user:*")?)   # every matching key, all at once — kv_scan_all for a big store

fn kv_dbsize(fd: Int) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_int(kv_dbsize(fd)?)   # how many keys

fn kv_scan(fd: Int, cursor: Str, pattern: Str, count: Int) -> Result(Resp, Str)

One page of the keyspace: the cursor to carry on from, and the keys on it.

KEYS builds the whole answer in one go and holds the engine while it does, which is fine for a handful of keys and wrong for a million. SCAN is the same question asked a page at a time, and the cursor is opaque — it is not a count and not an index, and the only thing to do with it is hand it back.

A walk sees every key that was there for the whole of it. One that arrives during the walk may or may not turn up, and one that goes may still be listed; that is what a scan costs, and it is the trade every keyspace this size makes.

# not run
import "db/oxiresp"
page = kv_scan(fd, "0", "user:*", 100)?   # a cursor and a page of keys: kv_page_cursor, kv_page_keys

fn kv_scan_args(cursor: Str, pattern: Str, count: Int) -> List(Str)

import "db/oxiresp"
kv_scan_args("0", "user:*", 100)   # => Cons(SCAN, Cons(0, Cons(MATCH, Cons(user:*, Cons(COUNT, Cons(100, Nil))))))

fn kv_page_cursor(v: Resp) -> Str

A page is a two-element array: the cursor, then the keys.

import "db/oxiresp"
kv_page_cursor(RArray(Cons(RBulk("17"), Cons(RArray(Nil), Nil))))   # => 17
kv_page_cursor(RNil)                                               # => 0

fn kv_page_keys(v: Resp) -> List(Str)

import "db/oxiresp"
kv_page_keys(RArray(Cons(RBulk("0"), Cons(RArray(Cons(RBulk("a"), Cons(RBulk("b"), Nil))), Nil))))   # => Cons(a, Cons(b, Nil))

fn kv_scan_all(fd: Int, pattern: Str) -> List(Str)

Every key that matches, by walking the cursor to its end.

The guard is not decoration: a cursor that never reaches "0" — an engine with a bug in it, or a page that answers with the cursor it was given — is an endless loop inside a caller that only asked for a list of keys.

# not run
import "db/oxiresp"
kv_scan_all(fd, "user:*")   # every matching key, page by page

fn kv_flushdb(fd: Int) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_flushdb(fd)   # everything gone

fn kv_select(fd: Int, n: Int) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_select(fd, 1)   # another database of the same server

fn kv_hset(fd: Int, key: Str, field_name: Str, value: Str) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_hset(fd, "user:1", "name", "ada")   #

fn kv_hget(fd: Int, key: Str, field_name: Str) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_text(kv_hget(fd, "user:1", "name")?)   # "ada"

fn kv_hdel(fd: Int, key: Str, field_name: Str) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_hdel(fd, "user:1", "name")   #

fn kv_hgetall(fd: Int, key: Str) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_texts(kv_hgetall(fd, "user:1")?)   # field, value, field, value

fn kv_hkeys(fd: Int, key: Str) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_texts(kv_hkeys(fd, "user:1")?)   #

fn kv_hvals(fd: Int, key: Str) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_texts(kv_hvals(fd, "user:1")?)   #

fn kv_hlen(fd: Int, key: Str) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_int(kv_hlen(fd, "user:1")?)   #

fn kv_lpush(fd: Int, key: Str, value: Str) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_lpush(fd, "queue", "job")   # onto the front

fn kv_rpush(fd: Int, key: Str, value: Str) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_rpush(fd, "queue", "job")   # onto the back

fn kv_lpop(fd: Int, key: Str) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_text(kv_lpop(fd, "queue")?)   # the front one, or "" — kv_missing says which

fn kv_rpop(fd: Int, key: Str) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_text(kv_rpop(fd, "queue")?)   #

fn kv_llen(fd: Int, key: Str) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_int(kv_llen(fd, "queue")?)   #

fn kv_lrange(fd: Int, key: Str, start: Int, stop: Int) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_texts(kv_lrange(fd, "queue", 0, -1)?)   # all of it

fn kv_sadd(fd: Int, key: Str, member: Str) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_sadd(fd, "tags", "rill")   #

fn kv_srem(fd: Int, key: Str, member: Str) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_srem(fd, "tags", "rill")   #

fn kv_sismember(fd: Int, key: Str, member: Str) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_int(kv_sismember(fd, "tags", "rill")?)   # 1 or 0

fn kv_smembers(fd: Int, key: Str) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_texts(kv_smembers(fd, "tags")?)   #

fn kv_scard(fd: Int, key: Str) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_int(kv_scard(fd, "tags")?)   #

fn kv_zadd(fd: Int, key: Str, score: Float, member: Str) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_zadd(fd, "board", 1500.0, "ada")   # a member with a score

fn kv_zscore(fd: Int, key: Str, member: Str) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_float(kv_zscore(fd, "board", "ada")?)   # 1500

fn kv_zrem(fd: Int, key: Str, member: Str) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_zrem(fd, "board", "ada")   #

fn kv_zcard(fd: Int, key: Str) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_int(kv_zcard(fd, "board")?)   #

fn kv_zrange(fd: Int, key: Str, start: Int, stop: Int) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_texts(kv_zrange(fd, "board", 0, 9)?)   # the ten lowest

fn kv_multi(fd: Int) -> Result(Resp, Str)

MULTI queues what follows and EXEC answers an array of every queued reply. A WATCHed key that somebody else changed makes EXEC answer the null array, which is kv_broke_off below rather than an error.

# not run
import "db/oxiresp"
kv_multi(fd)
kv_incr(fd, "a")
kv_incr(fd, "b")
kv_list(kv_exec(fd)?)   # both answers, once, together

fn kv_exec(fd: Int) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_exec(fd)   # the queued answers, or RNilArray when a watched key changed

fn kv_discard(fd: Int) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_discard(fd)   # the queue dropped

fn kv_watch(fd: Int, key: Str) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_watch(fd, "balance")   # the transaction breaks off if it changes first

fn kv_unwatch(fd: Int) -> Result(Resp, Str)

# not run
import "db/oxiresp"
kv_unwatch(fd)   #

fn kv_text(v: Resp) -> Str

A simple string and a bulk string are different things on the wire and the same thing to a caller who wanted text.

import "db/oxiresp"
kv_text(RBulk("ada"))     # => ada
kv_text(RSimple("OK"))    # => OK
kv_text(RInt(3))          # => 3
kv_text(RNil)             # =>

fn kv_int(v: Resp) -> Int

import "db/oxiresp"
kv_int(RInt(42))       # => 42
kv_int(RBulk("42"))    # => 42

fn kv_float(v: Resp) -> Float

import "db/oxiresp"
kv_float(RBulk("1500.5"))   # => 1500.5

fn kv_list(v: Resp) -> List(Resp)

import "db/oxiresp"
kv_list(RArray(Cons(RInt(1), Nil)))   # => Cons(RInt(1), Nil)
kv_list(RNil)                        # => Nil

fn kv_texts(v: Resp) -> List(Str)

import "db/oxiresp"
kv_texts(RArray(Cons(RBulk("a"), Cons(RInt(2), Nil))))   # => Cons(a, Cons(2, Nil))

fn kv_failed(v: Resp) -> Bool

A refusal that arrived inside a reply.

A command that fails answers Err — resp_answer lifts the engine's -ERR out of the value and into the Result, so the ordinary caller never sees one. EXEC is where they survive: it answers one array holding the answer to every queued command, and any of those may have failed on its own while the transaction as a whole did not. Walking that array is the only place this is the question, and there it is the only way to ask it.

kv_text of one is the engine's own words: WRONGTYPE Operation against a key holding the wrong kind of value.

import "db/oxiresp"
kv_failed(RError("ERR wrong type"))   # => true
kv_failed(RSimple("OK"))             # => false

fn kv_or(r: Result('a, 'b), fallback: 'a) -> 'a

The value, for a caller that has already decided a broken connection is not worth branching on — a counter it can live without, a line on a status page.

import "db/oxiresp"
kv_or(Ok(RInt(1)), RNil)                 # => RInt(1)
kv_or(Err("connection lost"), RNil)      # => RNil

fn kv_why(r: Result('a, Str)) -> Str

Why it failed, or "" if it did not. The engine's message, or this file's when the socket was the thing that broke.

import "db/oxiresp"
kv_why(Err("connection lost"))   # => connection lost
kv_why(Ok(RNil))                 # =>

fn kv_missing(v: Resp) -> Bool

A key that is not there. Distinct from the empty string, which is a value.

import "db/oxiresp"
kv_missing(RNil)         # => true
kv_missing(RBulk(""))    # => false

fn kv_broke_off(v: Resp) -> Bool

What EXEC answers when a WATCHed key moved under the transaction.

import "db/oxiresp"
kv_broke_off(RNilArray)   # => true

fn kv_publish(fd: Int, channel: Str, payload: Str) -> Result(Resp, Str)

PUBLISH answers how many subscribers were handed the message, and it is an ordinary command on an ordinary connection.

SUBSCRIBE is not. The server hands that connection to a loop of its own and never gives it back, so a program that both publishes and subscribes needs two — which is why kv_publish takes a socket and everything below takes a Sub.

# not run
import "db/oxiresp"
kv_int(kv_publish(fd, "news", "hello")?)   # how many subscribers got it

fn kv_subscribe(fd: Int, channels: List(Str)) -> Result(Sub, Str)

# not run
import "db/oxiresp"
sub = kv_subscribe(fd, Cons("news", Nil))?
match kv_next_message(sub)
  Message(sub2, channel, payload) -> payload
  Ended(why) -> why
  _ -> ""

fn kv_psubscribe(fd: Int, patterns: List(Str)) -> Result(Sub, Str)

Patterns are globs: news.* takes news.home and news.sport, and what arrives says which pattern caught it as well as which channel it came from.

# not run
import "db/oxiresp"
sub = kv_psubscribe(fd, Cons("news.*", Nil))?

fn kv_subscribe_more(sub: Sub, channels: List(Str)) -> Result(Sub, Str)

More channels on a subscription that already has some. Do this before anything is published to it: the notes are read back here, and a message that arrives in the middle of them is one this cannot put down again — it says so rather than dropping it.

# not run
import "db/oxiresp"
sub2 = kv_subscribe_more(sub, Cons("alerts", Nil))?

fn kv_psubscribe_more(sub: Sub, patterns: List(Str)) -> Result(Sub, Str)

# not run
import "db/oxiresp"
sub2 = kv_psubscribe_more(sub, Cons("alerts.*", Nil))?

fn kv_unsubscribe(sub: Sub, channels: List(Str)) -> Result(Sub, Str)

# not run
import "db/oxiresp"
sub2 = kv_unsubscribe(sub, Cons("news", Nil))?

fn kv_punsubscribe(sub: Sub, patterns: List(Str)) -> Result(Sub, Str)

# not run
import "db/oxiresp"
sub2 = kv_punsubscribe(sub, Cons("news.*", Nil))?

fn kv_sub_hang_up(sub: Sub) -> Unit

# not run
import "db/oxiresp"
kv_sub_hang_up(sub)

fn kv_next(sub: Sub) -> Delivery

Waits for whatever comes next, which may be a message or the server's word about a subscription. A strand blocked here parks, so a subscriber costs nothing between messages.

# not run
import "db/oxiresp"
match kv_next(sub)
  Message(s, channel, payload) -> payload
  Note(s, kind, name, count) -> kind + " " + name    # "subscribe news", and the like
  Ended(why) -> why
  _ -> ""

fn kv_next_message(sub: Sub) -> Delivery

The same as kv_next, with the server's paperwork passed over — for a caller that subscribed once and only wants what was published.

# not run
import "db/oxiresp"
match kv_next_message(sub)                              # notes are skipped
  Message(s, channel, payload) -> payload
  Ended(why) -> why
  _ -> ""

fn kv_shown(v: Resp) -> Str

For printing a reply whatever shape it came in.

import "db/oxiresp"
kv_shown(RBulk("ada"))                          # => "ada"
kv_shown(RArray(Cons(RInt(1), Cons(RNil, Nil))))   # => [1, nil]
kv_shown(RError("ERR x"))                       # => error: ERR x