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(%opts)

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

    $nats->publish($msg->{reply}, '+WPI');   # work-in-progress --extend ack_wait

=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($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)

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