diff --git a/lib/Nats/JetStream.rakumod b/lib/Nats/JetStream.rakumod index 40681ba..4f748d1 100644 --- a/lib/Nats/JetStream.rakumod +++ b/lib/Nats/JetStream.rakumod @@ -17,7 +17,7 @@ constant STREAM-PURGE = JS-API ~ '.STREAM.PURGE.%s'; # Direct Message Subjects constant DIRECT-GET = JS-API ~ '.DIRECT.GET.%s'; -constant DIRECT-GET-LAST = JS-API ~ '.DIRECT.GET.%s.%s'; +constant DIRECT-GET-LAST = JS-API ~ '.DIRECT.GET.%s'; # Consumer Subjects constant CONSUMER-CREATE = JS-API ~ '.CONSUMER.CREATE.%s.%s'; @@ -33,6 +33,7 @@ sub to-map($obj, *%pars --> Map()) { next if %pars{$name}:e && !%pars{$name}; my $val = $attr.get_value: $obj; next unless $val.defined && $val ~~ Str | Int | Positional | Associative; + next if $val ~~ Bool && !$val; # skip False booleans (defaults) next if $val ~~ Associative && $val.elems == 0; next if $val ~~ Positional && $val.elems == 0; $name => $val @@ -75,6 +76,7 @@ class Nats::Stream { has Int() $.num-replicas = 1; has Int() $.duplicate-window; has Bool() $.no-ack = False; + has Bool() $.allow-direct = False; has Str() $.template-owner; has Str() $.compression = 'none'; has UInt() $.first-seq; @@ -94,17 +96,15 @@ class Nats::Stream { method purge { $!nats.request: $.subject(STREAM-PURGE, $!name) } # Direct message get by sequence number - method get-msg(UInt $seq, Str :$subject) { - my $api-subject = $subject - ?? sprintf(DIRECT-GET-LAST, $!name, $subject) - !! sprintf(DIRECT-GET, $!name); - my %payload = :last_by_subj($seq); + method get-msg(UInt $seq) { + my $api-subject = sprintf(DIRECT-GET, $!name); + my %payload = :last_by_seq($seq); $!nats.request: $api-subject, to-json %payload } # Direct get last message for a subject method get-last-msg(Str $subject) { - $!nats.request: sprintf(DIRECT-GET-LAST, $!name, $subject), to-json { :last_by_subj($subject) } + $!nats.request: sprintf(DIRECT-GET-LAST, $!name), to-json { :last_by_subj($subject) } } method consumer(Str $name, |c) { diff --git a/t/direct-get-last.rakutest b/t/direct-get-last.rakutest new file mode 100644 index 0000000..8ca143e --- /dev/null +++ b/t/direct-get-last.rakutest @@ -0,0 +1,55 @@ +#!/usr/bin/env raku +use Nats::JetStream; +use Nats; +use Test::Mock; +use JSON::Fast; + +my $pass = 0; +my $fail = 0; + +# TEST 1: get-last-msg body-only (no subject in path) +{ + my $nats = mocked Nats, overriding => { + request => -> $subject, $payload { + if $subject ne q⌁$JS.API.DIRECT.GET.MY⌁ { + $fail++; note "❌ get-last-msg subject: got {$subject}"; + } else { + $pass++; note "✅ get-last-msg body-only format"; + } + } + }; + Nats::Stream.new(:nats($nats), name => "MY").get-last-msg("foo"); +} + +# TEST 2: allow-direct present when True +{ + my $nats = mocked Nats, overriding => { + request => -> $subject, $payload { + my %cfg = from-json($payload); + if %cfg:!exists || %cfg !== True { + $fail++; note "❌ allow-direct: exists={%cfg:exists}, val={%cfg}"; + } else { + $pass++; note "✅ allow-direct present and True"; + } + } + }; + Nats::Stream.new(:nats($nats), name => "MY", :allow-direct, subjects => ["foo"]).create; +} + +# TEST 3: allow-direct omitted when False (default) +{ + my $nats = mocked Nats, overriding => { + request => -> $subject, $payload { + my %cfg = from-json($payload); + if %cfg:exists { + $fail++; note "❌ allow-direct should be omitted when False"; + } else { + $pass++; note "✅ allow-direct omitted (default False)"; + } + } + }; + Nats::Stream.new(:nats($nats), name => "MY", subjects => ["foo"]).create; +} + +note "\n{$pass} passed, {$fail} failed"; +exit $fail ?? 1 !! 0; diff --git a/t/jetstream.rakutest b/t/jetstream.rakutest index c161d94..42d243d 100644 --- a/t/jetstream.rakutest +++ b/t/jetstream.rakutest @@ -364,7 +364,7 @@ use-ok 'Nats::JetStream'; request => -> $subject, $payload { is $subject, '$JS.API.DIRECT.GET.MY', 'direct get subject'; my %p = from-json($payload); - is %p, 5, 'sequence number in payload'; + is %p, 5, 'sequence number via last_by_seq'; } }; Nats::Stream.new(:nats($nats), name => 'MY').get-msg(5); @@ -372,18 +372,46 @@ use-ok 'Nats::JetStream'; } # Stream direct get last per subject +# Uses body-only format: $JS.API.DIRECT.GET. with last_by_subj in payload +# (subject in path + body = 408 Bad Request in NATS v2.14+) { my $nats = mocked Nats, overriding => { request => -> $subject, $payload { - is $subject, '$JS.API.DIRECT.GET.MY.foo', 'direct get last subject'; + is $subject, '$JS.API.DIRECT.GET.MY', 'direct get last — body-only, no subject in path'; my %p = from-json($payload); - is %p, 'foo', 'subject in payload'; + is %p, 'foo', 'subject in payload via last_by_subj'; } }; Nats::Stream.new(:nats($nats), name => 'MY').get-last-msg('foo'); check-mock $nats, *.called('request', :once); } +# Stream with allow-direct enabled +{ + my $nats = mocked Nats, overriding => { + request => -> $subject, $payload { + is $subject, '$JS.API.STREAM.CREATE.MY', 'stream create with allow-direct'; + my %cfg = from-json($payload); + ok %cfg:exists, 'allow_direct present in config'; + is %cfg, True, 'allow_direct is True'; + } + }; + Nats::Stream.new(:nats($nats), name => 'MY', :allow-direct, subjects => ['foo']).create; + check-mock $nats, *.called('request', :once); +} + +# Stream without allow-direct (default false, attribute omitted from config) +{ + my $nats = mocked Nats, overriding => { + request => -> $subject, $payload { + my %cfg = from-json($payload); + nok %cfg:exists, 'allow_direct omitted when False (default)'; + } + }; + Nats::Stream.new(:nats($nats), name => 'MY', subjects => ['foo']).create; + check-mock $nats, *.called('request', :once); +} + # Consumer update { my $nats = mocked Nats, overriding => {