Java client

Overview

ruvio-client is the Java client for Ruvio. One socket is shared by all threads: commands are pipelined and replies are matched in order. It has a typed method for every command in the compatibility matrix, a blocking API and a CompletableFuture API, and helpers for locks, rate limits, cache-aside and idempotency.

Install

The client is not published to Maven Central yet, and no artifact exists in any public repository. That is pending. Today you build it from the repository into your local Maven repository:

git clone https://github.com/salihcantekin/ruvio
cd ruvio/clients/java
mvn install          # JDK 17+ and Maven; add -DskipTests to skip the tests

Then use it from your own project:

<dependency>
  <groupId>io.github.salihcantekin</groupId>
  <artifactId>ruvio-client</artifactId>
  <version>VERSION</version>   <!-- the version in clients/java/pom.xml, or the release you want -->
</dependency>

The tests that talk to a server are skipped unless RUVIO_TEST_ADDR is set, so mvn install works without a server.

Quick start

The client owns the socket and the two helper threads. Close it when you are done.

import io.github.salihcantekin.ruvio.RuvioClient;
import io.github.salihcantekin.ruvio.args.SetOptions;

try (RuvioClient<String, String> c = RuvioClient.connect("127.0.0.1", 6379)) {
    c.set("greeting", "hello");
    String v    = c.get("greeting");      // "hello"
    String none = c.get("nope");          // null (nil reply)
    long visits = c.incr("hits");         // 1

    // SET k v EX 60 NX: false when the key already existed
    boolean first = c.set("k", "v", SetOptions.ex(60).ifAbsent());   // true
    boolean again = c.set("k", "v", SetOptions.ex(60).ifAbsent());   // false

    // the same connection as a future
    CompletableFuture<Long> n = c.async().incr("hits");               // 2
}

Options go through RuvioOptions:

RuvioOptions options = RuvioOptions.builder()
    .host("127.0.0.1").port(6379)
    .credentials("app", "secret")     // or .password("secret"); AUTH on connect (not run here)
    .database(4)                      // standalone only; cluster has database 0 only
    .clientName("orders")
    .connectTimeoutMillis(2000)
    .commandTimeoutMillis(5000)       // how long a blocking call waits; 0 (default) waits forever
    .maxInFlight(256)                 // replies outstanding per connection (default 16384)
    .build();
RuvioClient<String, String> c = RuvioClient.connect(options);

RuvioOptions.of("host:port") is a shortcut. A command timeout stops your wait, not the command: the server may still have run it. TLS is not supported.

Values are String (UTF-8) by default. For bytes use c.binaryValues(), which is a view on the same connection, or RuvioClient.connect(options, RuvioCodec.BYTES, RuvioCodec.BYTES):

RuvioClient<String, byte[]> b = c.binaryValues();
b.set("bin", new byte[] {0, 1, (byte) 255});
int len = b.get("bin").length;        // 3

Performance

How the client stays light, from the README and the code:

• One socket is shared by all threads. A caller encodes its frame straight into the shared output buffer under a short lock and adds its pending future to a FIFO ring in the same step. Do not open a connection per request.
• Write coalescing: when nothing is in flight the caller writes the buffer itself, so a serial call is sent at once with no thread hand-off. Otherwise a writer thread sends everything queued in one write. There is no timer and no wait to fill a batch.
• Command names are static final byte[] fragments such as $3\r\nGET\r\n (class Cmd, generated by tools/generate.py). Integers are written as ASCII digits straight into the buffer. Fixed-shape hot commands (GET, SET, INCR, HGET and similar) encode with no intermediate objects; option-heavy commands go through a small private Args buffer (one extra copy).
• A reader thread parses each reply in place from one reusable buffer and completes futures in FIFO order. Scalar replies (integer, OK, nil, bulk) allocate only the returned value. An incomplete reply rewinds and waits for more bytes.
• The per-call allocation of a typed command is the CompletableFuture plus the decoded value. The blocking methods are generated from the async ones and wait on that future.
• Futures complete on the reader thread, so keep thenApply callbacks short.

Measured in the README (4-vCPU Linux container, JDK 21, loopback, one shared connection, 256-byte values, single run each; shared noisy VM, orders of magnitude, not a comparison, not release evidence):

For best speed: share one client between threads, use pipeline() or the async API when you have many independent commands, keep values small, and use binaryValues() for non-text data. Allocation counts for parsing and encoding were reasoned from the code, not measured with an allocation profiler. I did not re-run the benchmark for this page.

Commands

Method names follow the Redis command in lower case: get, hset, zadd. Subcommands are camel-cased: xgroupCreate, clientList, aclSetuser, scriptLoad. A nil reply is null. These calls were run on a standalone server.

c.hset("user:1", Map.of("name", "Ada", "age", "36"));
String name = c.hget("user:1", "name");                          // "Ada"

c.rpush("queue", "a", "b", "c");
List<String> items = c.lrange("queue", 0, -1);                    // [a, b, c]

c.sadd("tags", "x", "y");
boolean has = c.sismember("tags", "x");                          // true

Map<String, Double> board = new LinkedHashMap<>();
board.put("ada", 100d); board.put("bob", 50d);
c.zadd("board", board);
List<ScoredValue<String>> top = c.zrevrangeWithScores("board", 0, -1);   // ada=100.0, bob=50.0

String id = c.xadd("events", Map.of("type", "signup"));
List<StreamEntry<String, String>> entries = c.xrange("events", "-", "+");   // 1 entry

c.jsonSet("doc", "$", "{\"city\":\"Izmir\"}");
String city = c.jsonGet("doc", "$.city");                        // "Izmir" (JSON text, quotes included)

c.geoadd("places", GeoValue.of(13.361389, 38.115556, "Palermo"),
                   GeoValue.of(15.087269, 37.502669, "Catania"));
Double km = c.geodist("places", "Palermo", "Catania", GeoUnit.KM);   // 166.27...
List<GeoResult<String>> near =
    c.geosearch("places", GeoSearchArgs.fromLonLat(15, 37).byRadius(200, GeoUnit.KM).asc().withDist());   // Catania, Palermo

c.setbit("flags", 7, true);
long bits = c.bitcount("flags");                                 // 1
c.pfadd("uniq", "a", "b", "a");
long unique = c.pfcount("uniq");                                 // 2

Anything without a typed method goes through execute with the same words as redis-cli. Integers come back as Long, bulk strings as byte[], arrays as List; executeText decodes bulk strings to String. On the async API it returns a future.

Object len = c.execute("STRLEN", "greeting");     // 5L

Ruvio commands

These replace Lua scripts or multi-step Redis patterns with one atomic command.

c.set("cfg", "v1");
c.setIfEq("cfg", "v2", "v1");                   // true
c.setIfEq("cfg", "v3", "v1");                   // false, the value is now v2
c.delex("cfg", "v2");                           // true

Long a = c.incrbyMax("seats", 3, 4);            // 3
Long b = c.incrbyMax("seats", 2, 4);            // null: would pass the limit

Long left1 = c.limit("login", 2, 60_000);       // 1 (remaining)
Long left2 = c.limit("login", 2, 60_000);       // 0
Long left3 = c.limit("login", 2, 60_000);       // null: refused

FillResult<String> f = c.fillget("page", 10_000);       // state FILL: you build it
if (f.state() == FillResult.State.FILL)
    c.fillset("page", "<html/>", 60_000, f.token());   // true
// the next fillget returns state HIT with the value

Long lease = c.lease("job", 30_000);            // fencing token
Long other = c.lease("job", 30_000);            // null while held
c.release("job", lease);                        // true

c.once("pay", 60_000);                          // true
c.once("pay", 60_000);                          // false

c.set("tickets", "5");
Long t1 = c.take("tickets", 2);                 // 3 left
Long t2 = c.take("tickets", 9);                 // null: not enough

long v1 = c.setv("d", "a");                     // 1
long v2 = c.setv("d", "b", v1);                 // 2 (a wrong version throws ERR version mismatch)

c.getex("greeting", 5_000);                     // value, and the TTL is now 5000 ms

c.lpushcap("recent", 3, "1", "2", "3", "4", "5");   // list keeps 3: [5, 4, 3]

c.bfReserve("bf", 0.01, 1000);
c.bfMadd("bf", "a", "b");
List<Boolean> seen = c.bfMexists("bf", "a", "z");   // [true, false]

// JSON.INDEX CREATE only runs in database 0
c.jsonIndexCreate("byCity", "user:", "$.city", "string");
c.jsonSet("user:1", "$", "{\"city\":\"Izmir\"}");
// the index builds in the background: jsonFind throws "ERR index is building" until it is ready
List<String> keys = c.jsonFind("byCity", "Izmir");   // [user:1]

long cursor = c.changesStart();
c.set("note", "x");
List<ChangeNote> notes = c.changes(cursor);      // one note: <cursor + 1>:SET note, e.g. 73101:SET note

semaphore(key, max, ttlMillis) and fillgetWait are typed too; the semaphore is exercised through the helper below. Do not call fillgetWait on the shared client: it parks the whole socket (use dedicated()).

Helpers

Methods on RuvioClient itself. They only send single-key commands, so they work unchanged in cluster mode (checked, see below).

// Lock: SET NX PX to take, SET IFEQ to renew, DELEX IFEQ to release.
try (RuvioLock<String> lock = c.tryLock("lock:invoice", 10_000)) {
    if (lock == null) return;                 // someone else has it
    boolean still = lock.renew(10_000);       // false when the lock was lost
    boolean mine  = lock.isHeld();
}                                             // close() releases only if the token still matches
RuvioLock<String> waited = c.lock("lock:invoice", 10_000, 2_000);   // wait up to 2 s, null after that

// Fenced lease and semaphore slot
try (RuvioToken<String> lease = c.tryLease("lease:job", 30_000)) {
    if (lease != null) { long fence = lease.token(); }
}
try (RuvioToken<String> slot = c.tryAcquirePermit("sem:export", 4, 30_000)) { }

// Rate limit: refused calls are not counted
RateLimitResult d = c.rateLimit("rl:ada", 2, 60_000);   // allowed, remaining 1 (then 0, then rejected)

// Cache-aside with stampede protection (FILLGET/FILLSET)
String page = c.getOrCreate("cache:home", 300_000, () -> "<home/>");   // factory ran once

// Idempotency: the action runs once; a repeat gets the stored result
String r = c.idempotent("idem:o:1", "fingerprint", 600_000, 2_000, () -> "done");   // "done"
String r2 = c.idempotent("idem:o:1", "fingerprint", 600_000, 2_000, () -> "again"); // "done"

// Lazy scans: scanIterable/scanStream, hscanIterable/hscanStream, sscan*, zscan*
for (String key : c.scanIterable(ScanArgs.match("scan:*").limit(10))) { }
long n = c.scanStream(ScanArgs.match("scan:*")).count();

// Typed JSON through a JsonCodec you plug in
c.setJsonCodec(myCodec);
c.jsonSetObject("u:1", "$", new Person("Ada", 36));
Person p = c.jsonGetObject("u:1", "$", Person.class);

// Recent items and "seen before"
c.pushRecent("recent:ada", 50, "item-41", "item-42");
List<String> newest = c.recent("recent:ada", 10);             // [item-42, item-41]
List<Boolean> seen  = c.seenBefore("seen:emails", "a@x", "b@x");   // [false, false], then records them
List<Boolean> maybe = c.mightContain("seen:emails", "c@x");      // read-only

The JDK has no JSON library and the client has no dependencies, so the typed JSON helpers need a two-method JsonCodec (toJson, fromJson). The default is a passthrough that takes JSON text and returns String. The run above used a small hand-written codec for a Person record. A Jackson codec (from the class Javadoc, not run here):

ObjectMapper mapper = new ObjectMapper();
c.setJsonCodec(new JsonCodec() {
    public String toJson(Object value) {
        try { return mapper.writeValueAsString(value); } catch (Exception e) { throw new RuntimeException(e); }
    }
    public <T> T fromJson(String json, Class<T> type) {
        try { return mapper.readValue(json, type); } catch (Exception e) { throw new RuntimeException(e); }
    }
});

Failure behavior:

• Lock. tryLock returns null when the lock is held; lock(key, ttl, wait) retries and returns null after wait. There is no automatic renewal: call renew yourself (false means you no longer own it). Renew and release compare the token, so they never touch another owner's lock. A release that fails is not retried; the key expires by its TTL.
• getOrCreate. Losers poll plain FILLGET with a backoff, never FILLGET ... WAIT, because a server-side wait parks the whole shared socket. If your factory throws, the exception reaches you and the fill marker lapses. There is also a stale-while-revalidate form with staleMillis and fillMillis. In the run, 10 threads asked for the same key and the factory ran once.
• Rate limit and idempotency. Server errors and socket failures throw. Nothing is retried for you, and an unknown outcome after a timeout stays unknown.

Pipelines, transactions, pub/sub

Concurrent calls on one client are already pipelined, and so are futures from async(). pipeline() buffers commands locally and sends them in one write on flush() (or sync(), which also waits). A pipeline is not thread safe and can be reused after a flush. Do not wait on a pipeline future before flushing it.

Pipeline<String, String> pl = c.pipeline();
CompletableFuture<Void>   s = pl.set("p1", "1");
CompletableFuture<Long>   i = pl.incr("p1");
CompletableFuture<String> g = pl.get("p1");
pl.flush();                              // one write
long n = i.join();                       // 2
String v = g.join();                     // "2"

// MULTI/EXEC on a pooled dedicated connection
try (Transaction<String, String> tx = c.multi()) {
    CompletableFuture<Long>   n2 = tx.incr("{t}:a");
    CompletableFuture<String> g2 = tx.get("{t}:a");
    boolean ok = tx.exec();              // true; false when a WATCHed key changed
}

// WATCH: a change from another connection makes exec() return false
try (Transaction<String, String> tx = c.multi()) {
    tx.watch("{t}:w");
    c.set("{t}:w", "changed");
    tx.set("{t}:w", "mine");
    boolean ok = tx.exec();              // false
}

// Pub/Sub uses its own connection and resubscribes after a drop
BlockingQueue<PubSubMessage> got = new LinkedBlockingQueue<>();
try (RuvioPubSub sub = c.pubSub(got::add)) {
    sub.subscribe("news");               // also psubscribe, ssubscribe
    c.publish("news", "hello");
    PubSubMessage m = got.poll(2, TimeUnit.SECONDS);   // channel "news", payloadString() "hello"
}

// Blocking commands go on a second client, so the shared socket is not held
try (RuvioClient<String, String> d = c.dedicated()) {
    KeyValue<String, String> item = d.blpop(0.1, "nolist");   // null on timeout
}

RESP3 and client-side caching are not supported by the typed API. RuvioOptions.pushListener exists for push frames, but I did not run it. Commands refused while queued in MULTI fail on their own future; the rest still run (the README says so; on the prebuilt server I used, one test of this fails, see Testing).

Cluster

Give the client one seed address and turn on cluster(true). On connect it sends CLUSTER SLOTS, builds a 16384-entry slot table (CRC16, {hashtags} honoured) and opens one lazy multiplexed connection per node. A standalone server answers with an error, and the client then behaves exactly like the standalone client, including database(n). In cluster mode only database 0 exists and the option is ignored.

try (RuvioClient<String, String> c = RuvioClient.connect(
        RuvioOptions.builder().host("127.0.0.1").port(17450).cluster(true).build())) {   // any shard
    c.isCluster();         // true
    c.nodeAddresses();     // [127.0.0.1:17450, 127.0.0.1:17451, 127.0.0.1:17452]

    for (int i = 0; i < 30; i++) c.set("k" + i, "v" + i);       // spread over the 3 shards
    c.onNode("127.0.0.1:17451").dbsize();                       // per-node commands: 8, 8 and 14 keys here

    c.mget("k3", "k1", "missing", "k2");     // [v3, v1, null, v2]: split per slot, caller's order kept
    c.exists("a", "b", "c", "zz");           // counts summed across parts
    RuvioClient.hashSlot("somekey");         // 11058, the same as c.clusterKeyslot("somekey")

    try { c.sunion("a1", "b2"); }            // keys in different slots
    catch (RuvioServerException e) { e.code(); }   // "CROSSSLOT", nothing was sent

    try (RuvioLock<String> l = c.tryLock("job:42", 10_000)) { }   // helpers work unchanged
}

The split is per slot, not per node, because the server (like Redis) refuses a multi-key command whose keys are in different slots even when one shard owns them all. A large split MGET is therefore many small pipelined commands; use {hashtag} keys to keep related keys in one slot.

• MOVED. The slot is patched in the table and the command is retried once on the node in the reply. If the node is unknown, or a second MOVED arrives, the table is re-read with CLUSTER SLOTS. A second MOVED reaches the caller. The router keeps a copy of each command frame for this, about 100 bytes of allocation per routed command.
• Connection loss. Only commands in flight on that node fail (RuvioConnectionException). Nothing is replayed, other nodes keep working, and the node reconnects on its next command.
• Pipelines. Commands are routed when queued and grouped per node; flush() sends one write per node and each future gets its own reply, so results match the caller's order (9 INCRs over 9 keys returned 1 each). An unreachable node fails only its group.
• Transactions. Bound to the node and slot of the first key or WATCH. A command with a key in another slot throws CROSSSLOT and is not queued. Use a {hashtag} ({u}:a, {u}:b worked).
• Blocking commands. Use dedicated(); in cluster mode it is a cluster client of its own.
• Scan. scanIterable and scanStream walk the nodes one after the other and saw all 30 keys. A single scan(cursor) only covers the node it lands on. KEYS, DBSIZE and FLUSHDB go to one node: through the router dbsize() returned 8 (one node's count), not 30. Use onNode(...) per node.
• Pub/Sub. SUBSCRIBE, PSUBSCRIBE and PUBLISH use the seed node; this server does not broadcast classic publishes between nodes. SSUBSCRIBE and SPUBLISH are routed by channel slot. Shard pub/sub is covered by ClusterLiveTest; I did not run it by hand.

Fallback check: cluster(true) against the standalone server on port 17394 gave isCluster() false, an empty nodeAddresses(), and normal commands. Replicas, ASK, READONLY reads and CLUSTERDOWN are not implemented. See Not done.

README cluster numbers (same noisy shared 4-vCPU VM, the servers share the CPUs with the benchmark, 1.5 s per figure, not release evidence): routing one GET locally costs about 100 to 135 ns on top of about 30 ns of encoding. Standalone GET with 1/16/64 threads about 19k/91k/139k ops/s; cluster client on 3 shards about 17k/76k/114k.

On one node of the same server the routed client was within about 2 % of the unrouted view, which is inside the noise. The README itself says cluster-versus-standalone totals say little about scaling. I did not re-run these.

Errors, nil, reconnect, timeouts

• A server -ERR becomes RuvioServerException (unchecked) with the server text. code() is the first word: ERR, WRONGTYPE, NOSCRIPT, CROSSSLOT, MOVED. The connection stays usable. On the async API the exception is the cause of the failed future.
• Socket failures are RuvioConnectionException. The next call reconnects; nothing is replayed automatically, so a write is never applied twice.
• Nil maps to null (String, Long, Boolean). Commands that "refuse" (limit, take, lease, incrbyMax) return null; tryLock returns a null lock.
• Timeouts: commandTimeoutMillis bounds blocking calls; on the async API use CompletableFuture.orTimeout. Neither cancels the command on the server. maxInFlight bounds outstanding replies per connection; a caller blocks briefly when the limit is reached.
• Pub/Sub resubscribes after a drop (pubSubReconnects test).

try { c.incr("greeting"); }
catch (RuvioServerException e) { /* e.code() == "ERR", ERR value is not an integer or out of range */ }

try { c.execute("NOSUCH"); }
catch (RuvioServerException e) { /* ERR unknown command */ }

c.set("ro", "1");
try { c.lpush("ro", "x"); }
catch (RuvioServerException e) { /* e.code() == "WRONGTYPE" */ }

Not done / not verified

• Check Maven Central for the release status.
• No TLS, no RESP3 decoding in the typed API, no client-side caching. Replication, and the ACL file save/load success paths (no acl_file on the test server), were not verified.
• No automatic retry or replay after a socket drop.
• Cluster: replicas, READONLY reads, ASK, CLUSTERDOWN/TRYAGAIN retries, adding or removing nodes while running (the table refreshes on MOVED only, no periodic refresh), and a cluster-wide DBSIZE/KEYS/FLUSHALL are not implemented. Cluster mode was only tested against the 3-shard Ruvio server (one process, port_mode = "per-shard"), not against a Redis Cluster, and with no failover, resharding or node-loss run beyond the dropped-connection test.
• The cluster topology commands (clusterSlots and friends) succeed only on a server in per-shard cluster mode; against a standalone server the test asserts the typed "cluster mode is disabled" error, and the success path runs only with RUVIO_TEST_CLUSTER=1 (not verified by the author).
• Allocation counts were reasoned from the code, not profiled. Performance numbers are single noisy runs and were not compared with another client.
• Older server builds differed from the matrix text in places: SUNION was not sorted and a command refused while queued in MULTI did not abort EXEC. Both are fixed in the server now. EXPIRE key 0 deletes the key and returns 1, as Redis does (the matrix text was corrected).
• The Jackson snippet and the auth options were not run on this page.

Testing

mvn -q test                                              # unit tests; live tests are skipped
RUVIO_TEST_ADDR=127.0.0.1:6379 mvn -q test              # + live tests (59 per the README); logical DBs 9, 10, 11, 0 with FLUSHDB
RUVIO_CLUSTER_ADDR=127.0.0.1:17450 RUVIO_TEST_ADDR=127.0.0.1:17394 mvn -q test   # + cluster live tests (91 per the README)
RUVIO_TEST_ADDR=127.0.0.1:17394 mvn -q test -Dtest=BenchTest -DexcludedGroups=none   # few-second benchmark

My run on this page: mvn test without a server ran 41 tests with 17 skipped (the live ones). With RUVIO_TEST_ADDR=127.0.0.1:17394 and RUVIO_CLUSTER_ADDR=127.0.0.1:17450 (a 3-shard server with shards = 3 and port_mode = "per-shard") it ran 91 tests, all passed.

A command refused inside MULTI now makes EXEC answer EXECABORT and run nothing, as Redis does; the test for it was updated together with the server change.

What the suites cover: CodecTest (RESP encode and parse), CoverageTest (all 287 matrix names map to a method, and every command has a cluster key rule), live tests per family (strings, keys, hashes, lists, sets, sorted sets, geo, streams, JSON, Bloom, scripting, server), helpers and concurrency, pipelines, transactions, pub/sub with reconnect, ClusterUnitTest (slot table, hashtags, key extraction, split and reassembly, MOVED once, no replay after a drop; scripted fake servers, always runs), ClusterLiveTest (spread over all shards, CLUSTER KEYSLOT agreement, cross-slot MGET/MSET/DEL order, pipeline order, forced stale slot table, transactions, helpers, scan, shard pub/sub, dropped node connection) and ClusterFallbackLiveTest (cluster(true) against a plain server).

The live suites flush their databases, so point them at a disposable server. Start one as on Start; for three shards set shards = 3 and port_mode = "per-shard", and shard i listens on base port + i.

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