/ 
EV-Nats-0.06
/ EV::Nats::KV

NAME

EV::Nats::KV - Key-Value store on top of NATS JetStream

SYNOPSIS

use EV;
use EV::Nats;
use EV::Nats::JetStream;
use EV::Nats::KV;

my $nats = EV::Nats->new(host => '127.0.0.1');
my $js   = EV::Nats::JetStream->new(nats => $nats);
my $kv   = EV::Nats::KV->new(js => $js, bucket => 'config');

$kv->create_bucket({}, sub {
    $kv->put('app.setting', 'on', sub {
        $kv->get('app.setting', sub {
            my ($val, $err) = @_;
            print "got: $val\n";
        });
    });
});

# Live updates
$kv->watch('app.>', sub {
    my ($key, $value, $op) = @_;
    print "$op $key = $value\n";
});

EV::run;

DESCRIPTION

A KV bucket is a JetStream stream named KV_<bucket> with subjects $KV.<bucket>.>, history of 1, rollup headers allowed, and deletes denied (purge tombstones are used instead). This module wraps that convention: put/get become single calls that hide the underlying js_publish + STREAM.MSG.GET dance.

Keys must match [-/_=.a-zA-Z0-9]+ without leading or trailing dots, and are validated up front.

METHODS

All callbacks fire on the EV loop, not synchronously.

new

new(js => $js, bucket => $name, [timeout => $ms])

The bucket need not exist yet; call "create_bucket" first to provision it. timeout defaults to the timeout of $js. The bucket name must match [a-zA-Z0-9_-]+.

get

get($key, $cb)

Fetch the latest value for $key. Callback: ($value, $err). $value is undef if the key does not exist, or has been deleted or purged (the tombstone is recognised and surfaces as a clean miss).

put

put($key, $value, $cb)

Set $key to $value. Callback: ($seq, $err), where $seq is the JetStream sequence number assigned by the server.

create

create($key, $value, [$cb])

Like put, but only succeeds if $key is missing. With a callback, deleted/purged keys can be recreated using the tombstone's revision; concurrent creators see a wrong-last-sequence error. Callback: ($seq, $err). Without $cb the headered publish is fire-and-forget.

update

update($key, $value, $rev, [$cb])

Like put, but only succeeds if $key's current revision is $rev (compare-and-swap). Callback: ($seq, $err).

delete

delete($key, [$cb])

Mark $key as deleted by publishing a KV-Operation: DEL tombstone, followed by a flush fence so a subsequent get won't race the publish. Callback: ($ok, $err); $err is set if the connection dropped before the PONG arrived, or if the client is not currently connected.

purge

purge($key, [$cb])

Like delete, but emits Nats-Rollup: sub too -- the prior history of $key is rolled up and replaced by the tombstone, freeing storage. Callback: ($ok, $err); $err is set if the client is not currently connected.

keys

keys($cb)

List all keys currently stored in the bucket. Callback: (\@keys, $err). Tombstoned keys are not filtered out -- check with get if needed.

watch

watch($pattern, $cb)

Subscribe to live changes. $pattern is a NATS subject suffix relative to the bucket (e.g. > for all keys, app.> for everything under app.); it must be non-empty and contain no whitespace. Callback receives ($key, $value, $op) where $op is PUT, DEL, or PURGE. Returns the underlying subscription id; pass to "unsubscribe" in EV::Nats to stop. Resubscribes on reconnect; updates published while disconnected are missed.

create_bucket

create_bucket(\%opts, $cb)

Provision the underlying stream. Recognised \%opts:

  • max_history - default 1; how many revisions to keep per key.

  • max_bytes - bucket-wide storage cap.

  • max_age - per-message TTL in nanoseconds.

  • replicas - cluster replication factor.

Callback: ($info, $err).

delete_bucket

delete_bucket($cb)

Tear down the underlying stream. Callback: ($info, $err).

status

status($cb)

Returns a snapshot hashref:

{ bucket => $name, values => $count, bytes => $n, history => $h }

Callback: (\%status, $err).

SEE ALSO

EV::Nats, EV::Nats::JetStream, EV::Nats::ObjectStore, NATS KV documentation.