diff --git a/META6.json b/META6.json index 21643bc..7902744 100644 --- a/META6.json +++ b/META6.json @@ -1,7 +1,7 @@ { "auth": "zef:FCO", "authors": [ - "Fernando Corrêa de Oliveira" + "Fernando Corr\u00eaa de Oliveira" ], "build-depends": [ "Test::Mock", @@ -11,7 +11,7 @@ "JSON::Fast", "URL" ], - "description": "NATS client for Raku with JetStream support", + "description": "NATS client for Raku with JetStream and KV support", "license": "Artistic-2.0", "name": "Nats", "perl": "6.d", @@ -27,17 +27,17 @@ "Nats::Message": "lib/Nats/Message.rakumod", "Nats::Replyable": "lib/Nats/Replyable.rakumod", "Nats::Subscription": "lib/Nats/Subscription.rakumod", - "Nats::Subscriptions": "lib/Nats/Subscriptions.rakumod" + "Nats::Subscriptions": "lib/Nats/Subscriptions.rakumod", + "Nats::KV": "lib/Nats/KV.rakumod" }, - "resources": [ - ], + "resources": [], "source-url": "https://github.com/FCO/nats.raku.git", "tags": [ "nats", "jetstream", - "messaging" - ], - "test-depends": [ + "messaging", + "kv" ], + "test-depends": [], "version": "0.1.0" } diff --git a/lib/Nats.rakumod b/lib/Nats.rakumod index bb2a7da..10d6f3c 100644 --- a/lib/Nats.rakumod +++ b/lib/Nats.rakumod @@ -8,6 +8,7 @@ use Nats::Data; use Nats::Message; use Nats::Subscription; use Nats::JetStream; +use Nats::KV; has $.socket-class = IO::Socket::Async; has %!subs; @@ -236,6 +237,10 @@ method stream($name, *@subjects, |c) { Nats::Stream.new: :nats(self), :$name, |(:@subjects if @subjects), |c } +method kv(Str $bucket, |c) { + Nats::KV.new: :nats(self), :$bucket, |c +} + method !in(|c) { self!debug(">>", |c) } @@ -383,6 +388,14 @@ my $stream = $nats.stream: 'mystream', :subjects['foo.>']; Creates a C object for JetStream operations. +=head2 kv + +=begin code :lang +my $kv = $nats.kv: 'mybucket'; +=end code + +Creates a C object for JetStream Key-Value operations. + =head1 JetStream See C for stream and consumer management. @@ -401,6 +414,30 @@ react whenever $consumer.msgs(:batch, :no-wait) { } =end code +=head1 Key-Value Store (JetStream KV) + +See C for key-value operations built on JetStream. + +=begin code :lang +use Nats::KV; + +my $kv = $nats.kv: 'mybucket'; +await $kv.create; + +$kv.put: 'foo', 'bar'; +say $kv.get: 'foo'; # bar + +$kv.delete: 'foo'; + +# List all keys +.say for $kv.keys; + +# Watch for changes +react whenever $kv.watch.supply -> $msg { + say \"Key changed: { $msg.subject }\"; +} +=end code + =head1 AUTHOR Fernando Corrêa de Oliveira diff --git a/lib/Nats/KV.rakumod b/lib/Nats/KV.rakumod new file mode 100644 index 0000000..438897f --- /dev/null +++ b/lib/Nats/KV.rakumod @@ -0,0 +1,115 @@ +use JSON::Fast; + +# ═════════════════════════════════════════════ +# NATS JetStream Key-Value Store +# ═════════════════════════════════════════════ +# +# Built on top of JetStream streams. Each KV bucket is a stream +# with max_msgs_per_subject=1 (last-write-wins) and discard=new. +# +# API subjects: $KV.. +# +# Usage: +# my $kv = $nats.kv('mybucket'); +# await $kv.create; +# $kv.put('foo', 'bar'); +# say $kv.get('foo'); # bar +# $kv.delete('foo'); + +class Nats::KV { + has $.nats is required; + has Str $.bucket is required; + has Str $.description; + has Int $.max-age = 0; # 0 = no TTL + has $.stream; + + # Full subject prefix for this bucket + method prefix { "\$KV.{ $!bucket }" } + + # Build a stream object configured as a KV bucket + method !build-stream { + Nats::Stream.new: + :$!nats, + :name("KV_{ $!bucket }"), + :subjects(["{ self.prefix }.>"]), + :retention, + :discard, + :max-msgs-per-subject(1), + :allow-direct, + :max-age($!max-age), + |($!description ?? :$!description !! Empty), + } + + # Create the KV bucket (idempotent — existing buckets are reused) + method create { + $!stream = self!build-stream; + $!stream.create; + } + + # Get stream info + method info { + $!stream.info; + } + + # Delete the entire bucket + method destroy { + $!stream.delete; + } + + # ── CRUD operations ── + + # Put a value for a key + method put(Str $key, Str() $value) { + $!nats.publish: "{ self.prefix }.{ $key }", $value; + } + + # Get the current value for a key (returns Str or Nil) + method get(Str $key --> Str) { + my $resp = $!stream.get-last-msg("{ self.prefix }.{ $key }"); + return Nil without $resp; + my $msg = await $resp.Promise; + return Nil unless $msg && $msg.payload && $msg.payload.chars > 0; + $msg.payload; + } + + # Delete a key (publishes a tombstone — empty payload) + method delete(Str $key) { + $!nats.publish: "{ self.prefix }.{ $key }", ""; + } + + # ── Bulk operations ── + + # List all keys in the bucket + method keys(--> Seq) { + my $resp = $!stream.info; + return ().Seq without $resp; + my $msg = await $resp.Promise; + return ().Seq unless $msg && $msg.payload; + my %info = try from-json($msg.payload); + return ().Seq if $!; + my $prefix = "{ self.prefix }."; + my $len = $prefix.chars; + gather { + for %info.List -> $subject { + next unless $subject.starts-with($prefix); + take $subject.substr($len); + } + } + } + + # ── Watcher ── + + # Subscribe to all changes in the bucket. + # Returns a Nats::Subscription. Use $sub.supply to react. + method watch { + $!nats.subscribe: "{ self.prefix }.>"; + } + + # ── History ── + + # Get the history of a key (requires the stream to have max_msgs_per_subject > 1). + # Not supported in this implementation (KV buckets use max_msgs_per_subject=1). + method history(Str $key) { + die "history() requires max-msgs-per-subject > 1. This KV bucket uses the default (1)."; + } +} diff --git a/t/kv.rakutest b/t/kv.rakutest new file mode 100644 index 0000000..16ac9a5 --- /dev/null +++ b/t/kv.rakutest @@ -0,0 +1,86 @@ +use Test; +use Nats; +use Nats::KV; + +plan 9; + +my $nats = Nats.new: :servers[%*ENV // 'nats://localhost:4222']; +await $nats.start; +$nats.connect; + +my $bucket = 'test-kv-' ~ (^10000).pick; + +subtest 'create bucket', { + my $kv = $nats.kv($bucket); + my $resp = $kv.create; + my $msg = await $resp.Promise; + ok $msg, 'bucket created'; + ok $msg.payload, 'response has payload'; +} + +subtest 'put and get', { + my $kv = $nats.kv($bucket); + $kv.put('hello', 'world'); + sleep 0.1; # give NATS time to process + my $val = $kv.get('hello'); + is $val, 'world', 'get returns put value'; +} + +subtest 'get missing key', { + my $kv = $nats.kv($bucket); + my $val = $kv.get('nonexistent'); + ok !$val.defined, 'get missing key returns Nil'; +} + +subtest 'overwrite value', { + my $kv = $nats.kv($bucket); + $kv.put('counter', '1'); + $kv.put('counter', '2'); + sleep 0.1; + is $kv.get('counter'), '2', 'last write wins'; +} + +subtest 'delete key', { + my $kv = $nats.kv($bucket); + $kv.put('temp', 'data'); + sleep 0.1; + $kv.delete('temp'); + sleep 0.1; + my $val = $kv.get('temp'); + ok !$val.defined || $val eq '', 'delete removes value'; +} + +subtest 'list keys', { + my $kv = $nats.kv($bucket); + $kv.put('alpha', '1'); + $kv.put('beta', '2'); + $kv.put('gamma', '3'); + sleep 0.2; + my @keys = $kv.keys; + ok @keys.elems >= 3, "has at least 3 keys (got {@keys.elems})"; + ok @keys.grep('alpha'), 'contains alpha'; + ok @keys.grep('beta'), 'contains beta'; +} + +subtest 'watch changes', { + my $kv = $nats.kv($bucket); + my $sub = $kv.watch; + my $p = $sub.supply.head.Promise; + $kv.put('watched', 'changed'); + await Promise.anyof: $p, Promise.in(2); + ok $p.so, 'watch received change'; + $nats.unsubscribe: $sub.sid; +} + +subtest 'kv via Nats.kv method', { + my $kv = $nats.kv($bucket); + ok $kv ~~ Nats::KV, 'Nats.kv returns Nats::KV'; +} + +subtest 'destroy bucket', { + my $kv = $nats.kv($bucket); + my $resp = $kv.destroy; + ok $resp, 'bucket destroyed'; +} + +done-testing;