EV-Nats

 view release on metacpan or  search on metacpan

lib/EV/Nats/JetStream.pm  view on Meta::CPAN

sub _api {
    my ($self, $subj, $payload, $cb) = @_;
    $payload //= '';
    $self->{nats}->request(
        "$self->{prefix}.$subj",
        $payload, $cb, $self->{timeout},
    );
}

## Public to sibling modules (KV, ObjectStore, ...) -- not part of the user
## API. Decode a JSON string; returns ($decoded, $error_or_undef) where
## $error already includes a "JSON decode error: ..." prefix when set.
## Gates on $@ so falsy-but-valid JSON (null/0/false/empty-string) doesn't
## get misreported as a decode failure.
sub decode_json_or_error {
    my ($json) = @_;
    local $@;
    my $r = eval { JSON::PP::decode_json($json) };
    return ($r, undef) unless $@;
    return (undef, "JSON decode error: $@");
}

## Public to sibling modules. True if a STREAM.MSG.GET response message
## carries a KV-Operation: DEL or PURGE tombstone header. $msg is the
## decoded {message} hash; its {hdrs} field is base64-encoded by the server.
sub msg_is_tombstone {
    my ($msg) = @_;
    return 0 unless $msg && $msg->{hdrs};
    require MIME::Base64;
    my $hdrs = MIME::Base64::decode_base64($msg->{hdrs});
    return $hdrs =~ /KV-Operation:\s*(?:DEL|PURGE)/i ? 1 : 0;
}

sub _json_api {
    my ($self, $subj, $data, $cb) = @_;
    my $payload = defined $data ? JSON::PP::encode_json($data) : '';
    $self->_api($subj, $payload, sub {
        my ($resp, $err) = @_;
        return $cb->(undef, $err) if $err;
        my ($decoded, $derr) = decode_json_or_error($resp);
        return $cb->(undef, $derr) if $derr;
        if ($decoded->{error}) {
            return $cb->(undef, "$decoded->{error}{description} (code $decoded->{error}{code})");
        }
        $cb->($decoded, undef);
    });
}

# Stream management

sub stream_create {
    my ($self, $config, $cb) = @_;
    my $name = $config->{name} || die "stream name required";
    $self->_json_api("STREAM.CREATE.$name", $config, $cb);
}

lib/EV/Nats/JetStream.pm  view on Meta::CPAN

            print "stored at seq=$ack->{seq}\n";
        });
    });

    EV::run;

=head1 DESCRIPTION

Thin async wrapper over the JetStream C<$JS.API.*> request/reply
endpoints. Each method is a single request whose callback is invoked
with the decoded JSON response (or an error string). The C<$nats>
connection passed to L</new> handles all the actual I/O.

L<EV::Nats::KV> and L<EV::Nats::ObjectStore> build on top of this
module -- see those for higher-level KV / blob APIs.

=head1 METHODS

All methods are async. Callbacks fire on the L<EV> loop.

=head2 new

lib/EV/Nats/JetStream.pm  view on Meta::CPAN

=head1 INTERNAL

These are exposed for sibling modules (L<EV::Nats::KV>,
L<EV::Nats::ObjectStore>) -- not part of the end-user API and subject
to change.

=head2 decode_json_or_error

    decode_json_or_error($json)

Decode C<$json>. Returns C<($decoded, $error_or_undef)>; the error
string already includes a C<"JSON decode error: "> prefix when set.
Gates on C<$@> so falsy-but-valid JSON (C<null>, C<0>, C<false>,
empty string) is reported as a clean decode rather than a failure.

=head2 msg_is_tombstone

    msg_is_tombstone($msg)

True if a C<STREAM.MSG.GET> response message carries a
C<KV-Operation: DEL> or C<KV-Operation: PURGE> header. C<$msg> is the
decoded C<message> hash from the response. Used by L<EV::Nats::KV>
and L<EV::Nats::ObjectStore> to surface deleted/purged entries as
clean misses rather than as malformed payloads.

=head1 SEE ALSO

L<EV::Nats>, L<EV::Nats::KV>, L<EV::Nats::ObjectStore>,
L<JetStream API reference|https://docs.nats.io/reference/reference-protocols/nats_api_reference>.

=cut

lib/EV/Nats/ObjectStore.pm  view on Meta::CPAN

}

# padded base64url of a byte/most-strings value, matching Go URLEncoding
sub _b64url {
    my $bytes = shift;
    utf8::encode($bytes) if utf8::is_utf8($bytes);   # names may be wide
    require MIME::Base64;
    (my $e = MIME::Base64::encode_base64($bytes, '')) =~ tr{+/}{-_};
    return $e;                                        # keeps '=' padding
}
# inverse: padded-or-unpadded base64url -> UTF-8-decoded string
sub _b64url_dec {
    my $s = shift;
    $s =~ tr{-_}{+/};
    require MIME::Base64;
    my $b = MIME::Base64::decode_base64($s);          # tolerant of padding
    utf8::decode($b);
    return $b;
}

# ADR-20 name encoding: padded base64url (nats.go). The 0.03-0.05 %XX

lib/EV/Nats/ObjectStore.pm  view on Meta::CPAN

object does not exist or was deleted (the deletion marker is
recognised). This is the recommended way to filter live objects out
of a L</list> result.

=head2 list

    list($cb)

List object names in the bucket. Callback: C<(\@names, $err)>.
Names written by 0.03-0.05 (legacy %XX encoding) and by 0.06+/nats.go
(base64url) are both decoded. Deleted entries still appear in the
listing -- filtering them would cost a per-name metadata round-trip,
so call C<info> to filter.

=head2 status

    status($cb)

Returns a snapshot hashref:

    { bucket => $name, bytes => $n, sealed => 0|1 }

t/18_object_kv.t  view on Meta::CPAN

    $os->put('report.txt', 'x', sub { });
    $js->drain;
    # plus a legacy-encoded name, as 0.05 would have written it
    my $legacy_meta = { name => 'a b.txt', bucket => 'B', nuid => 'N',
                        size => 0, chunks => 0, digest => '' };
    $js->{store}{'$O.B.M.a%20b.txt'} =
        { data => encode_base64(JSON::PP::encode_json($legacy_meta), ''), seq => ++$js->{seq} };
    push @{ $js->{order} }, '$O.B.M.a%20b.txt';
    my $names;
    $os->list(sub { $names = $_[0] });
    ok((grep { $_ eq 'report.txt' } @$names), 'base64url name decoded in list');
    ok((grep { $_ eq 'a b.txt' } @$names), 'legacy %XX name decoded in list');
};

subtest 'get() does not pin the connection' => sub {
    plan tests => 2;
    my $destroyed = 0;
    my $r;
    {
        my $js = MockJS->new;
        my $os = EV::Nats::ObjectStore->new(js => $js, bucket => 'B');
        # rides along inside the $self that get()'s closure captures