Java client

Package: io.github.salihcantekin:ruvio-client on Maven Central · Source on GitHub

Install

Maven coordinates:

<dependency>
  <groupId>io.github.salihcantekin</groupId>
  <artifactId>ruvio-client</artifactId>
  <version>0.1.0</version>
</dependency>

Published on Maven Central. To use the repository version instead, run mvn install in this directory.

Packages

Package	Contents
io.github.salihcantekin.ruvio	RuvioClient, RuvioAsync, Pipeline, Transaction, RuvioPubSub, RuvioOptions, codecs, exceptions
...ruvio.args	Option builders (SetOptions, XAddArgs, ZRangeArgs, …) and small enums
...ruvio.model	Result types (ScoredValue, StreamEntry, ScanResult, …)
...ruvio.commands	The typed command families the front-ends inherit
...ruvio.internal	Protocol, connection, multiplexing, cluster routing. Not supported API.

Quick start

import io.github.salihcantekin.ruvio.RuvioClient;
import io.github.salihcantekin.ruvio.RuvioLock;
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");                       // null when missing

    CompletableFuture<Long> n = c.async().incr("hits");  // same connection, async API
    c.set("k", "v", SetOptions.ex(60).ifAbsent());       // SET k v EX 60 NX -> false if it existed

    try (RuvioLock<String> lock = c.tryLock("job:42", 10_000)) {   // null if held elsewhere
        if (lock != null) { /* critical section */ }
    }                                                    // DELEX job:42 IFEQ token
}

Binary-safe: c.binaryValues() / c.withCodecs(RuvioCodec.BYTES, RuvioCodec.BYTES). Server errors are RuvioServerException (code() = WRONGTYPE, NOSCRIPT, …), socket failures RuvioConnectionException (the next call reconnects; nothing is replayed automatically). Anything without a typed method: c.async().execute("COMMAND", arg, ...).

Design

• One socket shared by all threads. A caller encodes its frame straight into the shared output buffer under a short lock and appends its pending future to a FIFO ring in the same critical section. When nothing is in flight the caller writes the buffer itself (serial calls are sent at once, no thread hand-off); otherwise the writer thread drains everything queued in a single write (no timers, no batch-fill wait).
• A reader thread reads into one reusable buffer, decodes each reply in place with the request’s decoder and completes futures in FIFO order. Scalars (integer, OK, nil, bulk) allocate only the returned value. Incomplete replies rewind and wait for more bytes.
• Command names are static final byte[] complete bulk fragments ($3\r\nGET\r\n, class Cmd, generated by tools/generate.py). Integers are written as ASCII digits directly into the buffer. Fixed-shape hot commands (GET, SET, INCR, HGET, … ) encode with no intermediate objects; option-heavy commands go through a small private Args buffer (one extra copy).
• The per-call allocation of a typed command is the CompletableFuture itself plus the decoded value.
• RuvioAsync (CompletableFuture) is the base API; the blocking facade RuvioSyncCommands (generated from it) calls the async method and waits. Futures complete on the reader thread, so keep thenApply callbacks short.
• Virtual threads: not used. Plain platform reader/writer threads are the right fit for one socket; blocking calls from virtual threads work as they park on the future.
• pipeline() buffers locally and sends one write on flush()/sync(). multi() runs MULTI/EXEC on a pooled dedicated connection. pubSub(listener) is a separate connection that resubscribes after a drop. dedicated() gives a second client for blocking commands (BLPOP, BLMOVE, XREAD BLOCK, fillgetWait), because they hold the shared socket.
• Two codec parameters: RuvioClient<K, V> (default String/String, UTF-8; byte[] available). Hash fields and stream field names use the key codec; members and field values the value codec.

Coverage

All 287 commands of docs/compatibility-matrix.json have a typed method (CoverageTest maps every matrix name to a method and fails otherwise), including the Ruvio options: SET ... IFEQ, INCRBY ... MAX, FILLGET ... WAIT, FILLSET ... SWR, GETEX, DELEX key IFEQ v, LPUSHCAP/RPUSHCAP, BF.*, JSON.* incl. JSON.INDEX/FIND, IDEM BEGIN/COMPLETE, LEASE/RELEASE, SEMAPHORE, ONCE, TAKE, SETV, CHANGES, LIMIT. Not typed (use execute): FLUSHDB ASYNC, HELLO ... SETNAME.

Helpers

Helper	Built on
tryLock / lock(key, ttl, wait) -> RuvioLock (renew, isHeld, close)	SET NX PX, SET IFEQ PX, DELEX IFEQ (never deletes another owner’s lock)
tryLease -> RuvioToken (fencing token()), tryAcquirePermit	LEASE/SEMAPHORE + RELEASE
rateLimit(key, max, window) -> RateLimitResult(allowed, remaining)	LIMIT
getOrCreate(key, ttl, factory) (+ stale-while-revalidate form)	FILLGET/FILLSET; losers poll plain FILLGET with backoff (WAIT would park the shared socket)
idempotent(key, fingerprint, ttl, wait, action)	IDEM BEGIN/COMPLETE
scanIterable/scanStream, hscan*, sscan*, zscan*	lazy cursor-driven pages
jsonSetObject/jsonGetObject/jsonMgetObjects + setJsonCodec(JsonCodec)	JSON.*; the JDK has no JSON, so plug in Jackson/Gson/… through the 2-method JsonCodec (default: String passthrough)
pushRecent / recent	LPUSHCAP
seenBefore / mightContain	BF.MADD / BF.MEXISTS

Cluster

RuvioClient<String, String> c = RuvioClient.connect(
        RuvioOptions.builder().host("127.0.0.1").port(7000).cluster(true).build());   // seed = any node
c.isCluster();       // false when the seed answered "cluster mode is disabled": plain standalone client
c.nodeAddresses();   // ["127.0.0.1:7000", ...] nodes that own slots
c.onNode("127.0.0.1:7001").dbsize();   // per-node commands (DBSIZE, KEYS, FLUSHDB, INFO, SCAN ...)

cluster(true) asks the seed for CLUSTER SLOTS on connect and builds a 16384-entry slot table. If the server is not in cluster mode (it answers with an error) nothing changes: the client is the standalone client, including database(n). In cluster mode the database option is ignored (the server only has database 0). Every node gets its own lazily opened multiplexed connection with the same fast path as the standalone one; the standalone path itself only pays one router == null check per command.

How it routes (the key table covers all 287 matrix commands, CoverageTest fails when one has no rule):

Command shape	Routing
single key (GET, SET, HSET, LEASE, LIMIT, FILLGET, IDEM, BF.*, JSON.*, XADD, SPUBLISH by channel, …)	node owning slot(key); {hashtag} aware, hashed in place from the encoded frame
keyless (PING, INFO, PUBLISH, CLUSTER, JSON.INDEX, JSON.FIND, CHANGES, SCRIPT, …)	any node (the seed while its connection is up)
several keys, same slot (RENAME, SUNION, LMOVE, BLPOP, BITOP, SINTERCARD, ZUNIONSTORE, EVAL/FCALL with numkeys, XREAD ... STREAMS, GEOSEARCHSTORE, GEORADIUS ... STORE, JSON.MGET, PFMERGE, …)	the shared slot’s node, using the real key lists
MGET, MSET, DEL, UNLINK, EXISTS, TOUCH over several slots	split per slot, each part sent to its owner, all concurrently, replies re-assembled in the caller’s order (null for a missing MGET key, DEL/EXISTS/… counts summed)
any other cross-slot command	fails at once with RuvioServerException (code CROSSSLOT); nothing is sent

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

• MOVED: that one slot is pointed at the node named in the reply and the command is sent there; a command follows at most 3 MOVED redirects, and the 4th MOVED is returned to the caller (no redirect loops). The whole table is re-read with CLUSTER SLOTS only when the named node is unknown, or from the second MOVED for the same command on; the retry still goes to the node the MOVED named. The retry runs on a small background thread, never on a connection reader thread. The router keeps a copy of each command’s frame for this (about 100 bytes allocated per routed command: the request object and the copy).
• Connection loss: only commands in flight on that node fail (RuvioConnectionException); nothing is replayed, the other nodes keep working, and the node’s connection reopens with its next command.
• Pipelines (pipeline()): commands are routed when queued and grouped per node; flush() sends one write per node and every future still completes with its own reply, so results come back in the caller’s order. A node that cannot be reached fails only its group.
• Transactions (multi()): bound to the node and slot of the first key (or WATCH); a command or WATCH with a key in another slot throws CROSSSLOT and is not queued.
• Blocking commands (BLPOP, BLMOVE, XREAD BLOCK, FILLGET WAIT): use dedicated(); in cluster mode it is a cluster client of its own, so the blocking call holds a connection to the owning node, not the shared one.
• Scan family: scanIterable/scanStream walk the nodes one after the other (each node has its own cursor) and see every key. A single scan(cursor) call only covers the node it lands on (use onNode(...) per node for manual paging). hscan/sscan/zscan route by key. KEYS, DBSIZE, FLUSHDB are per node.
• Pub/Sub: SUBSCRIBE/PSUBSCRIBE run on the seed connection and PUBLISH goes to the seed node too, which is why a publish from the same client reaches its subscribers; this server does not broadcast classic publishes between nodes. SSUBSCRIBE/SPUBLISH are routed by channel slot: the subscriber opens one extra connection per node and subscribes one slot per command.
• Helpers (locks, leases, semaphores, rate limit, getOrCreate, idempotency, capped list, Bloom, typed JSON) only issue single-key commands, so they work unchanged; ClusterLiveTest runs them on the cluster.
• Replicas, ASK redirects, READONLY reads and CLUSTERDOWN handling: not implemented.

Tests and benchmark

mvn -q test                                              # unit tests; live tests are skipped
RUVIO_TEST_ADDR=127.0.0.1:17104 mvn -q test              # all; uses logical DB 9 (and 10, 11, 0 for a few) with FLUSHDB
RUVIO_TEST_ADDR=127.0.0.1:17104 mvn -q test -Dtest=BenchTest -DexcludedGroups=none   # few-second benchmark
RUVIO_CLUSTER_ADDR=127.0.0.1:17320 RUVIO_TEST_ADDR=127.0.0.1:17323 mvn -q test       # + cluster live tests
RUVIO_CLUSTER_ADDR=127.0.0.1:17320 RUVIO_TEST_ADDR=127.0.0.1:17323 mvn -q test -Dtest=ClusterBenchTest -DexcludedGroups=none

Cluster tests: ClusterUnitTest (slot table, hashtags, key extraction, split/reassembly, MOVED (up to 3, reloads), no replay after a drop, pipelines, transactions, standalone fallback; scripted fake servers, always runs), ClusterLiveTest (needs RUVIO_CLUSTER_ADDR = any shard of a 3-shard port_mode = "per-shard" server: 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, concurrency) and ClusterFallbackLiveTest (cluster(true) against a plain server, uses RUVIO_TEST_ADDR).

Notes: JSON.INDEX works on database 0 only, so that one test uses database 0 with unique names. Cluster topology commands (CLUSTER SLOTS …) need a server in per-shard cluster mode; against a standalone server the test asserts the typed “cluster mode is disabled” error and only runs the success path when RUVIO_TEST_CLUSTER=1 (not verified here).

Local numbers (4 vCPU Linux container, JDK 21, loopback, one shared connection, 256 B values, single run each, not release evidence): encode SET ~35-42 ns, GET ~21-29 ns, INCRBY ~62-68 ns (no allocation); parse bulk 256 B to String ~80-135 ns, integer ~16-19 ns, 10-element string array ~230-500 ns. Live blocking GET/SET: 1 thread ~17-19k ops/s, 16 threads ~110-118k, 64 threads ~185-210k; pipeline of 100 GET ~0.75-0.9M ops/s.

Cluster numbers (ClusterBenchTest, same machine, shared with other jobs at load ~6 on 4 vCPUs, so noisy; 1.5 s per figure, 256 B values, 4096 keys): routing one GET locally (parse + hash + table lookup + frame copy, no socket) ~100-135 ns on top of ~30 ns of encoding, ~103 bytes allocated per command. Standalone client on a 1-shard server: GET 1/16/64 threads ~19k/91k/139k ops/s, SET ~19k/86k/129k, pipeline x100 GET ~0.80M. Cluster client on the 3-shard server: GET ~17k/76k/114k, SET ~18k/79k/124k, pipeline x100 GET (grouped per node) ~0.79M. Routing overhead isolated on one node of the same 3-shard server with the same keys (16 threads): unrouted onNode view GET 108k / SET 102k vs routed cluster client GET 106k / SET 99k (about 2 %, inside the noise). The 3-shard server itself shares the 4 vCPUs with the benchmark, so cluster-vs-standalone totals say little about scaling; they are not release evidence.

Not done / not verified

• Cluster: replicas, READONLY replica reads, ASK, CLUSTERDOWN/TRYAGAIN retries, adding or removing nodes while running (the table is refreshed on MOVED only, there is no periodic refresh), and a cluster-wide DBSIZE/KEYS/FLUSHALL (use onNode) are not implemented. Cluster mode was only exercised against the 3-shard Ruvio server (one process, port_mode = "per-shard"), not against a Redis Cluster.
• No automatic retry or replay of commands after a socket drop; maxInFlight (default 16384) bounds outstanding replies per connection.
• Allocation counts for parsing/encoding were reasoned from the code, not measured with an allocation profiler.
• Older server builds differed from the matrix text in places (SUNION unsorted; a command refused while queued in MULTI did not abort EXEC). Both are fixed in the server. EXPIRE key 0 deletes the key, as Redis does.
• Replication, TLS, ACL-file save/load success paths (no acl_file configured on the test server) not verified.