Rust client

Overview

ruvio-client is a blocking client for Ruvio. It uses only the Rust standard library. Each method is named after the command it sends, and execute sends anything else.

Install

The crate is published on crates.io. The repository can contain features that are not in the latest release yet; its README lists them under "Unreleased".

# the latest published release
cargo add ruvio-client

# everything on this page: use the repository checkout (this is what was tested)
git clone https://github.com/salihcantekin/ruvio
# then in your Cargo.toml:
ruvio-client = { path = "../ruvio/clients/ruvio-client", features = ["serde"] }

A git = dependency on the repository was not tried. Registry release of the new version is pending.

Quick start

use std::time::Duration;
use ruvio_client::{Client, ClientOptions};

let mut db = Client::connect("127.0.0.1", 17350)?;
db.set("session:ada", "hello")?;
println!("{:?}", db.get_string("session:ada")?); // => Some("hello")
println!("{:?}", db.get_string("nope")?);        // => None
println!("{}", db.incr("visits")?);              // => 1
db.expire("session:ada", Duration::from_secs(60))?;
println!("{}", db.ttl("session:ada")?);          // => 60

// Options: database, timeouts, credentials.
let mut db = Client::connect_with(ClientOptions {
    host: "127.0.0.1".into(),
    port: 17350,
    database: 2,                                  // SELECT 2 before connect returns
    connect_timeout: Duration::from_secs(2),
    ..ClientOptions::default()                    // username/password: None
})?;
db.set_read_timeout(Some(Duration::from_secs(3)))?;
// The connection closes when `db` is dropped. There is no close() method.

ClientOptions also has username and password (sent as AUTH only when set), discover_cluster, and the reconnect settings (reconnect, max_reconnect_attempts, reconnect_base_delay, reconnect_max_delay).

Shortcuts: connect_database(host, port, db), connect_ip, connect_addr. TLS is not supported. The AUTH and TLS paths were not run on this page.

Performance

• One connection, one caller. There is no multiplexing. To use several cores, give each thread its own Client.
• No allocation for fixed-arity commands. GET, INCR, EXPIRE, DELEX, ZCOUNT and similar borrow their arguments into a stack array and format numbers into a stack buffer. Variadic commands allocate one Vec<&str>, never owned copies.
• One encode buffer and one buffered reader per connection. Replies are parsed in place. A bulk string moves into the returned String or Vec<u8> without a second copy.
• Pipelines. execute_many writes a batch in one write and reads all replies. This is the main way to get more out of one connection.
• Binary API. get returns bytes; set_bytes and execute_binary take raw bytes.
• Not done: pre-encoded command fragments and write coalescing across callers (the C# client has these; this client has one caller per connection, so there is nothing to coalesce).

Measured with examples/alloc_bench.rs (release build, one-shard server on loopback, 4-vCPU shared VM). The numbers are noisy and are not release evidence. Allocation counts are exact (counting allocator); times vary from run to run, so compare them only inside one run. I did not re-run the benchmark for this page; these are the figures in the client README.

For best speed: reuse the client, batch with execute_many, use a Ruvio command that replaces a round trip (for example set_if_eq instead of GET then SET), and keep one connection per thread.

Commands

Every command in the compatibility matrix has a typed method. This table lists the families; the full method list is in the client README and on docs.rs. Command syntax is on Commands.

Typed examples, run against a standalone server:

use ruvio_client::{Client, SetMode, SortedSetAddOptions, StreamReadOptions};

let mut db = Client::connect_database("127.0.0.1", 17350, 3)?;

// Strings and expiry
db.set_ex("token:1", "abc", 30)?;
db.set_with("token:1", "zzz", None, SetMode::IfNotExists)?; // => false (key exists)
db.ttl("token:1")?;                                           // => 30
db.mset(&[("a", "1"), ("b", "2")])?;
db.mget(&["a", "b", "c"])?;                                   // => [Some("1"), Some("2"), None]
db.get("a")?;                                                 // => Some([49])   (bytes)

// Hashes
db.hset("user:1", &[("name", "Ada"), ("plan", "pro")])?;
db.hget("user:1", "name")?;                                   // => Some("Ada")
db.hgetall("user:1")?;                                        // => HashMap name=Ada, plan=pro

// Lists
db.rpush("jobs", &["a", "b", "c"])?;
db.lrange("jobs", 0, -1)?;                                    // => ["a", "b", "c"]
db.lpop("jobs")?;                                             // => Some("a")

// Sets
db.sadd("tags", &["x", "y"])?;
db.smembers("tags")?;                                         // => ["x", "y"] (order not fixed)

// Sorted sets: ZADD ... GT CH
db.zadd("leaderboard", &[("player1", 100.0), ("player2", 80.0)])?;
let gt = SortedSetAddOptions { greater_than: true, changed: true, ..Default::default() };
db.zadd_with("leaderboard", gt, &[("player2", 90.0)])?;        // => 1
for e in db.zrevrange_with_scores("leaderboard", 0, 9)? {
    println!("{} {}", e.member, e.score);                     // => player1 100, player2 90
}
db.zscore("leaderboard", "player1")?;                         // => Some(100.0)

// Streams with a consumer group
let id = db.xadd("events", "*", &[("kind", "signup"), ("user", "ada")])?;
db.xgroup_create("events", "workers", "0", false)?;
let opts = StreamReadOptions { count: Some(10), ..Default::default() };
for stream in db.xreadgroup("workers", "w1", &opts, &[("events", ">")])? {
    for entry in &stream.entries {
        // entry.fields == [("kind", "signup"), ("user", "ada")]
        db.xack("events", "workers", &[entry.id.as_str()])?; // => 1
    }
}

// JSON (values are JSON text)
db.json_set("user:ada", "$", r#"{"name":"Ada","city":"Berlin","tags":["a"]}"#)?;
db.json_get("user:ada", "$.name")?;                           // => Some("\"Ada\"")
db.json_arrappend("user:ada", "$.tags", "\"b\"")?;            // => Some(2)

// Geo
db.geo_add("cities", &[(13.361389, 38.115556, "Palermo"), (15.087269, 37.502669, "Catania")])?;
db.geo_dist("cities", "Palermo", "Catania", Some("km"))?;     // => Some(166.27...)
db.geo_search("cities", 15.0, 37.0, 200.0, "km")?;            // => ["Catania", "Palermo"]

// Bitmaps and HyperLogLog
db.setbit("online", 7, true)?;
db.bitcount("online")?;                                       // => 1
db.pfadd("visitors", &["u1", "u2", "u3"])?;
db.pfcount(&["visitors"])?;                                   // => 3

// Anything else: send the arguments as redis-cli would
db.execute(&["OBJECT", "ENCODING", "a"])?;                    // => Bulk("int")

execute returns a RespValue (Simple, Error, Integer, Bulk, Array, Null) with helpers such as as_integer, into_string, into_bytes. GEORADIUS ... STORE, XADD ... MINID and XAUTOCLAIM ... COUNT/JUSTID have no typed option; use execute for those.

Ruvio commands

These are the commands Redis does not have, or has with different semantics. Replies below are from a live run.

use std::time::Duration;
use ruvio_client::{Client, JsonIndexType};

let mut db = Client::connect_database("127.0.0.1", 17350, 4)?;

// SET key value IFEQ expected PX ttl  (compare-and-set; replaces WATCH/MULTI or a Lua script)
db.set("cfg", "v1")?;
db.set_if_eq("cfg", "v2", "v1", Some(Duration::from_secs(60)))?; // => true
db.set_if_eq("cfg", "v3", "v1", None)?;                          // => false (value is now v2)

// DELEX key IFEQ value  (delete only if equal; the safe lock release, no Lua)
db.delex("cfg", "nope")?;                                        // => false
db.delex("cfg", "v2")?;                                          // => true

// INCRBY key delta MAX limit  (bounded counter; replaces a Lua script)
db.incr_by_max("stock", 1, 2)?;                                  // => Some(1), then Some(2)
db.incr_by_max("stock", 1, 2)?;
db.incr_by_max("stock", 1, 2)?;                                  // => None (would pass 2)

// LIMIT key max window_ms  (fixed-window rate limit; Some = calls left, None = refused)
db.limit("api:ada", 2, 60_000)?;                                 // => Some(1), Some(0), None

// FILLGET / FILLSET  (cache fill with stampede protection)
db.fill_get("page", Duration::from_secs(5))?;                    // => Fill("1")  (you fill it; "1" is the token)
db.fill_get("page", Duration::from_secs(5))?;                    // => Busy       (someone else is filling)
// db.fill_set("page", "<html>", Duration::from_secs(60), "1", None)?   -- stores it
// db.fill_get_wait(...) sends FILLGET ... WAIT and parks the connection; see Helpers.

// LEASE / SEMAPHORE / RELEASE  (fenced lease, counting semaphore)
db.lease("leader", 10_000)?;                                     // => Some(1)  (fence token)
db.lease("leader", 10_000)?;                                     // => None
db.semaphore("dl", 1, 10_000)?;                                  // => Some(token)
db.semaphore("dl", 1, 10_000)?;                                  // => None (full)
// db.release(key, token)? gives the slot or lease back.

// ONCE key ttl_ms  (1 on first call, 0 afterwards)
db.once("job:1", 60_000)?;                                       // => 1, then 0

// TAKE key count  (take from a numeric stock; None when not enough)
db.set("tickets", "5")?;
db.take("tickets", 2)?;                                          // => Some(3)  (left)
db.take("tickets", 9)?;                                          // => None

// SETV key value [version]  (versioned write)
db.setv("doc", "one", None)?;                                    // => 1
db.setv("doc", "two", Some(1))?;                                 // => 2

// GETEX key ttl_ms  (read and set a millisecond TTL)
db.getex("g", 5000)?;                                            // => Some("hello"), pttl is now 5000

// LPUSHCAP key cap v...  (push and trim in one command)
db.lpushcap("recent", 3, &["1", "2", "3", "4", "5"])?;           // => 3
db.lrange("recent", 0, -1)?;                                     // => ["5", "4", "3"]

// Bloom filter
db.bf_reserve("seen", 0.01, 1000)?;
db.bf_madd("seen", &["a", "b"])?;                                // => [true, true]
db.bf_mexists("seen", &["a", "z"])?;                             // => [true, false]

// JSON.INDEX / JSON.FIND (database 0 only; see the note below)
db.json_index_create("by_city", "user:", "$.city", Some(JsonIndexType::String))?;
db.json_find("by_city", "\"Berlin\"", Some(100), None)?;         // => ["user:1"]

// CHANGES
let cursor = db.changes_start()?;
db.set("watched", "1")?;
db.changes(cursor as u64)?;                                      // => [(1, "SET watched")]

Two things I hit while running this. JSON.INDEX CREATE only works in database 0 (ERR JSON.INDEX CREATE must run in database 0 elsewhere). And JSON.FIND right after the create can return ERR index is building; retry after a few milliseconds (I looped on Error::Server containing "building" with a 50 ms sleep).

The extra methods set_if_eq_keep_ttl, rpushcap, json_index_drop, json_index_list, fill_get_wait and idempotency_begin/idempotency_complete exist; only json_index_list was run here.

Helpers

Thin layers over the native commands. All of them work unchanged on a cluster client because they only send single-key commands (lock and rate limit were run on the cluster, see Cluster).

use std::time::Duration;
use ruvio_client::{Client, Error, IdemOutcome};

let mut db = Client::connect_database("127.0.0.1", 17350, 5)?;

// Lock: SET key token NX PX, renew with SET IFEQ, release with DELEX IFEQ on drop.
if let Some(mut lock) = db.try_lock("lock:report", Duration::from_secs(30))? {
    lock.set("report:status", "running")?;       // the guard derefs to the client
    lock.renew(Duration::from_secs(30))?;        // => true
}                                                // key is deleted here; GET lock:report => None
db.with_lock("lock:job", Duration::from_secs(30), |db| db.incr("jobs"))?; // => Some(1)
// lock() polls until a deadline: Ok(None) if it could not get the lock in time.
db.lock("lock:busy", Duration::from_secs(5), Duration::from_millis(300))?; // => None after ~300 ms

Failure behavior. try_lock and with_lock return None when the lock is taken, they do not wait. lock returns None when the wait ends. If the lock expired and someone else holds it, renew returns false and the release on drop does nothing: DELEX IFEQ only deletes your own token, so it can never delete another holder's lock (checked: a lock held by someone else was still there after a failed lock()).

The guard cannot tell you mid-work that the lock expired; check renew if the work is long. Errors in Drop are ignored.

// Fenced lease and semaphore permit (RELEASE on drop)
if let Some(lease) = db.lease_guard("leader", Duration::from_secs(10))? {
    let fence = lease.fence();                   // => 1; pass it to the protected resource
}
let permit = db.semaphore_permit("downloads", 1, Duration::from_secs(60))?; // Some(permit)

// Rate limit: LIMIT -> RateLimit { allowed, remaining }
for _ in 0..3 {
    let v = db.rate_limit("login:ada", 2, Duration::from_secs(60))?;
    // => allowed=true remaining=1, allowed=true remaining=0, allowed=false remaining=0
}

// Cache-aside with stampede protection (FILLGET / FILLSET)
let page = db.get_or_create("page:home", Duration::from_secs(60), || {
    Ok::<_, Error>("<html>home</html>".to_owned())
})?;                                              // second call: factory is not run

lease_guard and semaphore_permit return None when the lease is held or the semaphore is full. get_or_create polls plain FILLGET with a 2 to 50 ms backoff when another caller is filling, because FILLGET ... WAIT would park the whole connection.

It gives up with an Io error of kind TimedOut after CacheOptions::max_wait. If the factory fails, nothing is stored and the fill marker lapses after fill_timeout. get_or_create_with takes CacheOptions (ttl, fill timeout, stale-while-revalidate window, max wait). The busy and timeout paths were not run here.

// Idempotency (IDEM BEGIN / COMPLETE)
for owner in ["worker-1", "worker-2"] {
    match db.idempotent("pay:42", "fp-1", owner, Duration::from_secs(3600),
                        || Ok::<_, Error>(b"charged".to_vec()))? {
        IdemOutcome::Executed(r) => { /* worker-1: ran the work */ }
        IdemOutcome::Replayed(r) => { /* worker-2: got b"charged" back, work not run */ }
        IdemOutcome::Pending => { /* another owner is running it */ }
        IdemOutcome::Conflict => { /* same key, different fingerprint */ }
    }
}
// Same key with fingerprint "other-fp" => Conflict

If the work closure fails, the reservation stays pending until the TTL passes. This is not exactly-once for external side effects. The Pending path was not run here.

// Lazy scan iterators: one page in memory at a time
for key in db.scan_iter(Some("session:*"), Some(2)) {
    let key = key?;                              // 5 keys seen; each item is a Result
}
for pair in db.hscan_iter("h", None, None) { let (field, value) = pair?; }

// Capped recent list (LPUSHCAP) and Bloom "seen before" (BF.MADD)
for i in 0..5 { db.push_recent("recent:ada", 3, &format!("item:{i}"))?; }
db.recent("recent:ada", 10)?;                    // => ["item:4", "item:3", "item:2"]
db.seen_before("emails:seen", &["a@x", "b@x"])?; // => [false, false]
db.seen_before("emails:seen", &["a@x", "c@x"])?; // => [true, false]  (Bloom: true means "probably")

Typed JSON needs the serde feature. The value is serialized with serde_json, so a plain string is sent with quotes.

#[derive(serde::Serialize, serde::Deserialize, Debug)]
struct User { name: String, city: String }

db.json_set_value("user:ada", "$", &User { name: "Ada".into(), city: "Izmir".into() })?;
let u: Option<User> = db.json_get_value("user:ada", "$")?; // => Some(User { name: "Ada", city: "Izmir" })
db.json_set_value("user:ada", "$.city", "Berlin")?;
db.json_get("user:ada", "$.city")?;                        // => Some("\"Berlin\"")

Pipelines, transactions, Pub/Sub, blocking

// Pipeline: one write, replies in order
let replies = db.execute_many(&[&["SET", "a", "1"], &["INCR", "a"], &["GET", "a"]])?;
// => [Simple("OK"), Integer(2), Bulk("2")]

// Transaction: inside MULTI the server answers QUEUED, so queue with execute()
db.set("balance", "100")?;
db.watch(&["balance"])?;
db.multi()?;
db.execute(&["INCRBY", "balance", "10"])?;
db.execute(&["GET", "balance"])?;
db.exec()?;                                       // => Some([Integer(110), Bulk("110")])
// If another client changed `balance` after WATCH, exec() returns None (checked).

// Pub/Sub: subscribe opens a second socket; PUBLISH and GET stay on the command socket
let t = std::thread::spawn(|| {
    let mut sub = Client::connect("127.0.0.1", 17350).unwrap();
    sub.subscribe(&["news"]).unwrap();
    sub.set_read_timeout(Some(Duration::from_secs(3))).unwrap();
    let m = sub.next_message().unwrap();          // kind=message channel=news payload=hello
});
db.publish("news", "hello")?;                     // => 1 receiver
t.join().unwrap();

// Blocking pop: waits on this connection
db.blpop(&["queue"], Duration::from_secs(2))?;    // => Some(("queue", "job-1")) after another client pushed
db.blpop_key("empty", Duration::from_millis(200))?; // => None

A blocking command occupies the connection until it returns, because the client has one socket per node. listen(callback) loops on next_message until an error. RESP3 and client-side caching (HELLO 3, tracking) are not supported by this client: it is RESP2 only, hello and client_tracking exist as plain commands.

Cluster

Pass one seed address. The client sends CLUSTER SLOTS to it. A standalone server answers ERR cluster mode is disabled and the client stays on one connection. On a sharded server it keeps a slot table (CRC16 of the key or of the {hash tag}, modulo 16384) and one socket per shard, opened as needed. Set ClientOptions::discover_cluster = false to stay on the seed socket.

• Single key: sent to the owner shard. Keyless commands (PING, INFO, ...) go to the seed. PUBLISH goes to one node.
• Split and gathered: MGET, MSET, DEL, EXISTS, TOUCH, UNLINK run per shard and the replies come back in your order. RENAME across slots is done client-side (read, write, delete). These are not atomic.
• Every shard: DBSIZE, FLUSHDB, FLUSHALL. SCAN walks the shards one after another with a cursor that encodes the shard, so scan_iter sees all keys.
• Other multi-key commands (SUNION, SINTER, ...) need one slot, otherwise the client returns CROSSSLOT without sending.
• Pipelines: grouped per shard, one write each, replies returned in your order.
• Transactions: WATCH/MULTI pin to the first key's shard. A key from another shard fails with the server's MOVED.
• MOVED: refreshes the table and retries, at most 3 redirects. A command that was in flight when a socket dropped is not replayed.
• Only database 0 exists in cluster mode.

Run against three shards (ports 17360 to 17362, seed 17360):

use ruvio_client::{Client, hash_slot};

let mut db = Client::connect("127.0.0.1", 17360)?;
db.is_cluster();                      // => true
db.shard_count();                     // => 3
hash_slot("somekey");                 // => 11058   (same as db.cluster_keyslot("somekey"))
hash_slot("{user1000}.following");    // => 3443    (same slot as "{user1000}.profile")

for k in ["alpha", "bravo", "charlie", "delta", "echo", "foxtrot"] { db.set(k, k)?; }
// slots 865, 8623, 1769, 9053, 14438, 3172: all three shards hold keys
db.mget(&["foxtrot", "alpha", "missing", "delta", "bravo"])?;
// => [Some("foxtrot"), Some("alpha"), None, Some("delta"), Some("bravo")]   (caller's order)

db.mset(&[("m1", "1"), ("m2", "2"), ("m3", "3"), ("m4", "4")])?;
db.del(&["m1", "m2", "m3", "m4", "nokey"])?;                  // => 4
db.exists(&["alpha", "bravo", "nokey"])?;                     // => 2

db.execute_many(&[&["SET", "p1", "1"], &["SET", "p2", "2"], &["GET", "p1"], &["GET", "p2"], &["INCR", "p3"]])?;
// => [OK, OK, Bulk("1"), Bulk("2"), Integer(1)]

db.sunion(&["s1", "s2"]);                                     // => Err("CROSSSLOT Keys in request do not hash to the same slot")
db.watch(&["{t}a"])?; db.multi()?;
db.execute(&["SET", "{t}a", "1"])?; db.execute(&["SET", "{t}b", "2"])?;
db.exec()?;                                                   // => Some([OK, OK])  (same hash tag, one slot)
db.scan_iter(None, None).count();                             // all 13 keys, dbsize() == 13

db.try_lock("lock:x", Duration::from_secs(5))?;               // works on the cluster client
db.rate_limit("rl:ada", 2, Duration::from_secs(60))?;         // allowed=true remaining=1

With discover_cluster off, get_string("somekey") on the seed returned Err("MOVED 11058 127.0.0.1:17362"), as expected. Inside a transaction, a second key on another shard (SET bravo after SET alpha) also came back as MOVED 8623 127.0.0.1:17361, so the "single slot" rule is enforced by the server's reply, not by a client check before sending.

Live test result. The repository's tests/cluster_live.rs was run against a real three-shard server (RUVIO_CLUSTER_ADDR pointing at the first shard) and both tests pass. The two tests share the one database a cluster has, so they now take turns. Not covered by that test file: forced MOVED recovery with a stale table, shard loss, and the keyless fan-out commands.

Errors

Every method returns Result<_, ruvio_client::Error> with three variants: Io (socket failed or timed out), Protocol (bytes were not valid RESP2, or the reply had an unexpected shape), and Server (a -ERR ... reply with its message). A server error leaves the connection usable.

db.set("str", "x")?;
match db.lpush("str", &["v"]) {
    Err(Error::Server(m)) => println!("{m}"),   // => WRONGTYPE Operation against a key holding the wrong kind of value
    _ => {}
}
db.ping()?;                                    // => "PONG", still usable
db.get("missing")?;                            // => None   (nil is Option::None, not an error)
Client::connect("127.0.0.1", 17399);           // => Err(Io(ConnectionRefused))

Nil maps to None (get, hget, lpop, blpop, zscore, ...). LIMIT, TAKE, LEASE and SEMAPHORE use None for "refused".

Reconnect. If a socket was dropped, the next command opens it again (up to max_reconnect_attempts, with reconnect_base_delay growing to reconnect_max_delay). The command that noticed the drop returns an error and is not replayed. on_connection_lost and on_connection_restored take a callback. Subscriptions are sent again on the new Pub/Sub socket. The reconnect path itself was not run on this page.

Timeouts. set_read_timeout(Some(d)) makes a slow read return Error::Io with kind WouldBlock (seen on Linux). There is no cancellation: calls are blocking, you stop one by timing it out.

After a read timeout the connection is opened again. A timed-out read on a command reply leaves that reply on its way, so the client marks the socket broken and the next call reconnects (when reconnect is on, the default). A late reply can never answer a later command. This was found while writing this page: before the fix the replies were shifted by one, and a live test now guards it. Only subscriber reads (read_message) may time out and carry on.

Not done or not verified

• Release. Features under "Unreleased" in the crate README are not on crates.io until the next release.
• No multiplexing. One client is one caller. No async API. No pre-encoded command fragments and no write coalescing.
• No TLS, no RESP3. AUTH is implemented but was not run here.
• No dedicated typed option for GEORADIUS ... STORE, XADD ... MINID, XAUTOCLAIM ... COUNT/JUSTID. Use execute.
• Cluster: the routing of newer commands (BITOP, SINTERCARD, JSON.MGET, XGROUP/XINFO, keyless RUVIO/JSON.INDEX/JSON.FIND) is unit-tested but was not run on three shards. MOVED recovery with a stale table, shard loss and reconnect of one shard were not run live. Cross-slot MSET, DEL and RENAME are not atomic.
• Benchmark numbers on this page are from the README, not re-run.
• Not run on this page: FILLGET busy/timeout path in get_or_create, Pending in idempotent, AUTH, reconnect, Pub/Sub on the cluster.

Testing

cd clients/ruvio-client
cargo test                                          # unit tests + coverage test, no server needed
RUVIO_TEST_ADDR=127.0.0.1:17350 cargo test          # also the live suites
RUVIO_CLUSTER_ADDR=127.0.0.1:17360 cargo test --test cluster_live -- --test-threads=1

Counts from my run (with a live standalone server, debug build): 27 unit tests, 1 coverage test, 12 tests in tests/live.rs (databases 0 to 7), 9 in tests/live_ext.rs (databases 0 and 8 to 15), 2 in tests/cluster_live.rs, and 5 doc tests.

All passed; the same counts passed with --features serde. The live suites flush those databases, so point them at a throwaway server. Without the environment variables the live tests return early and report as passed. tests/coverage.rs reads docs/compatibility-matrix.json and fails if a command has no typed method.

Start a server for the tests as on Start. For three shards, set shards = 3 and port_mode = "per-shard" in the config; shard i listens on base port + i.

See also: Clients overview, C#, Go, Java, Python, Node.js, Examples, Commands.