Current section
Files
Jump to
Current section
Files
src/glats@jetstream@stream.erl
-module(glats@jetstream@stream).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch]).
-export([info/2, create/4, update/3, delete/2, purge/2, find_stream_name_by_subject/2, get_message/3]).
-export_type([retention_policy/0, discard_policy/0, stream_option/0, stream_info/0, stream_config/0, stream_state/0, stream_message/0, access_method/0, raw_stream_message/0]).
-if(?OTP_RELEASE >= 27).
-define(MODULEDOC(Str), -moduledoc(Str)).
-define(DOC(Str), -doc(Str)).
-else.
-define(MODULEDOC(Str), -compile([])).
-define(DOC(Str), -compile([])).
-endif.
-type retention_policy() :: limits_policy | interest_policy | work_queue_policy.
-type discard_policy() :: discard_old | discard_new.
-type stream_option() :: {description, binary()} |
{retention, retention_policy()} |
{max_consumers, integer()} |
{max_messages, integer()} |
{max_bytes, integer()} |
{max_age, integer()} |
{max_messages_per_subject, integer()} |
{max_message_size, integer()} |
{discard, discard_policy()} |
{storage, glats@jetstream:storage_type()} |
{num_replicas, integer()} |
{duplicate_window, integer()} |
{allow_direct, boolean()} |
{mirror_direct, boolean()} |
{deny_delete, boolean()} |
{deny_purge, boolean()} |
{allow_rollup, boolean()} |
{discard_new_per_subject, boolean()}.
-type stream_info() :: {stream_info, binary(), stream_config(), stream_state()}.
-type stream_config() :: {stream_config,
binary(),
list(binary()),
gleam@option:option(binary()),
gleam@option:option(integer()),
gleam@option:option(integer()),
gleam@option:option(integer()),
gleam@option:option(integer()),
gleam@option:option(integer()),
gleam@option:option(integer()),
gleam@option:option(binary()),
gleam@option:option(binary()),
gleam@option:option(integer()),
gleam@option:option(integer()),
gleam@option:option(boolean()),
gleam@option:option(boolean()),
gleam@option:option(boolean()),
gleam@option:option(boolean()),
gleam@option:option(boolean()),
gleam@option:option(boolean())}.
-type stream_state() :: {stream_state,
integer(),
integer(),
integer(),
binary(),
integer(),
binary(),
integer()}.
-type stream_message() :: {stream_message, integer(), binary(), glats:message()}.
-type access_method() :: {sequence_id, integer()} | {last_by_subject, binary()}.
-type raw_stream_message() :: {raw_stream_message,
binary(),
integer(),
gleam@option:option(binary()),
binary(),
binary()}.
-file("src/glats/jetstream/stream.gleam", 249).
-spec stream_state_decoder() -> gleam@dynamic@decode:decoder(stream_state()).
stream_state_decoder() ->
gleam@dynamic@decode:field(
<<"messages"/utf8>>,
{decoder, fun gleam@dynamic@decode:decode_int/1},
fun(Messages) ->
gleam@dynamic@decode:field(
<<"bytes"/utf8>>,
{decoder, fun gleam@dynamic@decode:decode_int/1},
fun(Bytes) ->
gleam@dynamic@decode:field(
<<"first_seq"/utf8>>,
{decoder, fun gleam@dynamic@decode:decode_int/1},
fun(First_seq) ->
gleam@dynamic@decode:field(
<<"first_ts"/utf8>>,
{decoder,
fun gleam@dynamic@decode:decode_string/1},
fun(First_ts) ->
gleam@dynamic@decode:field(
<<"last_seq"/utf8>>,
{decoder,
fun gleam@dynamic@decode:decode_int/1},
fun(Last_seq) ->
gleam@dynamic@decode:field(
<<"last_ts"/utf8>>,
{decoder,
fun gleam@dynamic@decode:decode_string/1},
fun(Last_ts) ->
gleam@dynamic@decode:field(
<<"consumer_count"/utf8>>,
{decoder,
fun gleam@dynamic@decode:decode_int/1},
fun(Consumer_count) ->
gleam@dynamic@decode:success(
{stream_state,
Messages,
Bytes,
First_seq,
First_ts,
Last_seq,
Last_ts,
Consumer_count}
)
end
)
end
)
end
)
end
)
end
)
end
)
end
).
-file("src/glats/jetstream/stream.gleam", 269).
-spec decode_optional_field(
binary(),
gleam@dynamic@decode:decoder(HCQ),
fun((gleam@option:option(HCQ)) -> gleam@dynamic@decode:decoder(HGA))
) -> gleam@dynamic@decode:decoder(HGA).
decode_optional_field(Key, Field_decoder, Next) ->
gleam@dynamic@decode:optional_field(
Key,
none,
gleam@dynamic@decode:optional(Field_decoder),
Next
).
-file("src/glats/jetstream/stream.gleam", 199).
-spec stream_config_decoder() -> gleam@dynamic@decode:decoder(stream_config()).
stream_config_decoder() ->
gleam@dynamic@decode:field(
<<"name"/utf8>>,
{decoder, fun gleam@dynamic@decode:decode_string/1},
fun(Name) ->
gleam@dynamic@decode:field(
<<"subjects"/utf8>>,
gleam@dynamic@decode:list(
{decoder, fun gleam@dynamic@decode:decode_string/1}
),
fun(Subjects) ->
decode_optional_field(
<<"retention"/utf8>>,
{decoder, fun gleam@dynamic@decode:decode_string/1},
fun(Retention) ->
decode_optional_field(
<<"max_consumers"/utf8>>,
{decoder, fun gleam@dynamic@decode:decode_int/1},
fun(Max_consumers) ->
decode_optional_field(
<<"max_msgs"/utf8>>,
{decoder,
fun gleam@dynamic@decode:decode_int/1},
fun(Max_msgs) ->
decode_optional_field(
<<"max_bytes"/utf8>>,
{decoder,
fun gleam@dynamic@decode:decode_int/1},
fun(Max_bytes) ->
decode_optional_field(
<<"max_age"/utf8>>,
{decoder,
fun gleam@dynamic@decode:decode_int/1},
fun(Max_age) ->
decode_optional_field(
<<"max_msgs_per_subject"/utf8>>,
{decoder,
fun gleam@dynamic@decode:decode_int/1},
fun(
Max_msgs_per_subject
) ->
decode_optional_field(
<<"max_msg_size"/utf8>>,
{decoder,
fun gleam@dynamic@decode:decode_int/1},
fun(
Max_msg_size
) ->
decode_optional_field(
<<"discard"/utf8>>,
{decoder,
fun gleam@dynamic@decode:decode_string/1},
fun(
Discard
) ->
decode_optional_field(
<<"storage"/utf8>>,
{decoder,
fun gleam@dynamic@decode:decode_string/1},
fun(
Storage
) ->
decode_optional_field(
<<"num_replicas"/utf8>>,
{decoder,
fun gleam@dynamic@decode:decode_int/1},
fun(
Num_replicas
) ->
decode_optional_field(
<<"duplicate_window"/utf8>>,
{decoder,
fun gleam@dynamic@decode:decode_int/1},
fun(
Duplicate_window
) ->
decode_optional_field(
<<"allow_direct"/utf8>>,
{decoder,
fun gleam@dynamic@decode:decode_bool/1},
fun(
Allow_direct
) ->
decode_optional_field(
<<"mirror_direct"/utf8>>,
{decoder,
fun gleam@dynamic@decode:decode_bool/1},
fun(
Mirror_direct
) ->
decode_optional_field(
<<"sealed"/utf8>>,
{decoder,
fun gleam@dynamic@decode:decode_bool/1},
fun(
Sealed
) ->
decode_optional_field(
<<"deny_delete"/utf8>>,
{decoder,
fun gleam@dynamic@decode:decode_bool/1},
fun(
Deny_delete
) ->
decode_optional_field(
<<"deny_purge"/utf8>>,
{decoder,
fun gleam@dynamic@decode:decode_bool/1},
fun(
Deny_purge
) ->
decode_optional_field(
<<"allow_rollup_hdrs"/utf8>>,
{decoder,
fun gleam@dynamic@decode:decode_bool/1},
fun(
Allow_rollup_hdrs
) ->
gleam@dynamic@decode:success(
{stream_config,
Name,
Subjects,
Retention,
Max_consumers,
Max_msgs,
Max_bytes,
Max_age,
Max_msgs_per_subject,
Max_msg_size,
Discard,
Storage,
Num_replicas,
Duplicate_window,
Allow_direct,
Mirror_direct,
Sealed,
Deny_delete,
Deny_purge,
Allow_rollup_hdrs}
)
end
)
end
)
end
)
end
)
end
)
end
)
end
)
end
)
end
)
end
)
end
)
end
)
end
)
end
)
end
)
end
)
end
)
end
)
end
).
-file("src/glats/jetstream/stream.gleam", 185).
-spec decode_info(binary()) -> {ok, stream_info()} |
{error, glats@jetstream:jetstream_error()}.
decode_info(Body) ->
Decoder = begin
gleam@dynamic@decode:field(
<<"created"/utf8>>,
{decoder, fun gleam@dynamic@decode:decode_string/1},
fun(Created) ->
gleam@dynamic@decode:field(
<<"config"/utf8>>,
stream_config_decoder(),
fun(Config) ->
gleam@dynamic@decode:field(
<<"state"/utf8>>,
stream_state_decoder(),
fun(State) ->
gleam@dynamic@decode:success(
{stream_info, Created, Config, State}
)
end
)
end
)
end
)
end,
_pipe = gleam@json:parse(Body, Decoder),
_pipe@1 = gleam@result:map_error(
_pipe,
fun(_) -> {-1, <<"decode error"/utf8>>} end
),
gleam@result:map_error(_pipe@1, fun glats@internal@js:map_code_to_error/1).
-file("src/glats/jetstream/stream.gleam", 490).
-spec encode_get_message_body(access_method()) -> binary().
encode_get_message_body(Method) ->
_pipe = case Method of
{sequence_id, Seq} ->
[{<<"seq"/utf8>>, gleam@json:int(Seq)}];
{last_by_subject, Subj} ->
[{<<"last_by_subj"/utf8>>, gleam@json:string(Subj)}]
end,
_pipe@1 = gleam@json:object(_pipe),
gleam@json:to_string(_pipe@1).
-file("src/glats/jetstream/stream.gleam", 524).
-spec raw_to_stream_message(raw_stream_message()) -> {ok, stream_message()} |
{error, nil}.
raw_to_stream_message(Msg) ->
gleam@result:then(
begin
_pipe = gleam@bit_array:base64_decode(erlang:element(5, Msg)),
gleam@result:map(_pipe, fun gleam@bit_array:to_string/1)
end,
fun(Body) ->
Hdrs = case erlang:element(4, Msg) of
{some, Data} ->
_pipe@1 = glats@internal@js:decode_b64_headers(Data),
gleam@result:unwrap(_pipe@1, maps:new());
none ->
maps:new()
end,
{ok,
{stream_message,
erlang:element(3, Msg),
erlang:element(6, Msg),
{message,
erlang:element(2, Msg),
Hdrs,
none,
begin
_pipe@2 = Body,
gleam@result:unwrap(_pipe@2, <<""/utf8>>)
end}}}
end
).
-file("src/glats/jetstream/stream.gleam", 553).
-spec json_parse_jetstream_errorable(
binary(),
gleam@dynamic@decode:decoder(HDN)
) -> {ok, HDN} | {error, glats@jetstream:jetstream_error()}.
json_parse_jetstream_errorable(Body, Decoder) ->
_pipe@1 = gleam@json:parse(
Body,
gleam@dynamic@decode:one_of(
begin
_pipe = Decoder,
gleam@dynamic@decode:map(
_pipe,
fun(Field@0) -> {ok, Field@0} end
)
end,
[begin
gleam@dynamic@decode:subfield(
[<<"error"/utf8>>, <<"err_code"/utf8>>],
{decoder, fun gleam@dynamic@decode:decode_int/1},
fun(Err_code) ->
gleam@dynamic@decode:subfield(
[<<"error"/utf8>>, <<"description"/utf8>>],
{decoder,
fun gleam@dynamic@decode:decode_string/1},
fun(Description) ->
gleam@dynamic@decode:success(
{error, {Err_code, Description}}
)
end
)
end
)
end]
)
),
_pipe@2 = gleam@result:map_error(
_pipe@1,
fun(_) -> {-1, <<"decode error"/utf8>>} end
),
_pipe@3 = gleam@result:flatten(_pipe@2),
gleam@result:map_error(_pipe@3, fun glats@internal@js:map_code_to_error/1).
-file("src/glats/jetstream/stream.gleam", 351).
-spec decode_delete(binary()) -> {ok, nil} |
{error, glats@jetstream:jetstream_error()}.
decode_delete(Body) ->
Decoder = begin
gleam@dynamic@decode:field(
<<"success"/utf8>>,
{decoder, fun gleam@dynamic@decode:decode_bool/1},
fun(_) -> gleam@dynamic@decode:success(nil) end
)
end,
_pipe = Body,
json_parse_jetstream_errorable(_pipe, Decoder).
-file("src/glats/jetstream/stream.gleam", 378).
-spec decode_purge(binary()) -> {ok, integer()} |
{error, glats@jetstream:jetstream_error()}.
decode_purge(Body) ->
_pipe = Body,
json_parse_jetstream_errorable(
_pipe,
begin
gleam@dynamic@decode:field(
<<"success"/utf8>>,
{decoder, fun gleam@dynamic@decode:decode_bool/1},
fun(_) ->
gleam@dynamic@decode:field(
<<"purged"/utf8>>,
{decoder, fun gleam@dynamic@decode:decode_int/1},
fun(Purged) -> gleam@dynamic@decode:success(Purged) end
)
end
)
end
).
-file("src/glats/jetstream/stream.gleam", 421).
-spec decode_names(binary()) -> {ok, list(binary())} |
{error, glats@jetstream:jetstream_error()}.
decode_names(Body) ->
_pipe = Body,
json_parse_jetstream_errorable(
_pipe,
begin
gleam@dynamic@decode:field(
<<"streams"/utf8>>,
gleam@dynamic@decode:optional(
gleam@dynamic@decode:list(
{decoder, fun gleam@dynamic@decode:decode_string/1}
)
),
fun(Streams) ->
gleam@dynamic@decode:success(
begin
_pipe@1 = Streams,
gleam@option:unwrap(_pipe@1, [])
end
)
end
)
end
).
-file("src/glats/jetstream/stream.gleam", 501).
-spec decode_raw_message(binary()) -> {ok, raw_stream_message()} |
{error, glats@jetstream:jetstream_error()}.
decode_raw_message(Body) ->
_pipe = Body,
json_parse_jetstream_errorable(
_pipe,
begin
gleam@dynamic@decode:field(
<<"message"/utf8>>,
begin
gleam@dynamic@decode:field(
<<"subject"/utf8>>,
{decoder, fun gleam@dynamic@decode:decode_string/1},
fun(Subject) ->
gleam@dynamic@decode:field(
<<"seq"/utf8>>,
{decoder, fun gleam@dynamic@decode:decode_int/1},
fun(Seq) ->
gleam@dynamic@decode:field(
<<"data"/utf8>>,
{decoder,
fun gleam@dynamic@decode:decode_string/1},
fun(Data) ->
gleam@dynamic@decode:field(
<<"time"/utf8>>,
{decoder,
fun gleam@dynamic@decode:decode_string/1},
fun(Time) ->
gleam@dynamic@decode:optional_field(
<<"hdrs"/utf8>>,
none,
gleam@dynamic@decode:optional(
{decoder,
fun gleam@dynamic@decode:decode_string/1}
),
fun(Hdrs) ->
gleam@dynamic@decode:success(
{raw_stream_message,
Subject,
Seq,
Hdrs,
Data,
Time}
)
end
)
end
)
end
)
end
)
end
)
end,
fun(Message) -> gleam@dynamic@decode:success(Message) end
)
end
).
-file("src/glats/jetstream/stream.gleam", 634).
-spec ret_pol_to_string(retention_policy()) -> binary().
ret_pol_to_string(Pol) ->
case Pol of
limits_policy ->
<<"limits"/utf8>>;
interest_policy ->
<<"interest"/utf8>>;
work_queue_policy ->
<<"workqueue"/utf8>>
end.
-file("src/glats/jetstream/stream.gleam", 642).
-spec dis_pol_to_string(discard_policy()) -> binary().
dis_pol_to_string(Pol) ->
case Pol of
discard_old ->
<<"old"/utf8>>;
discard_new ->
<<"new"/utf8>>
end.
-file("src/glats/jetstream/stream.gleam", 649).
-spec storage_to_string(glats@jetstream:storage_type()) -> binary().
storage_to_string(Storage) ->
case Storage of
file_storage ->
<<"file"/utf8>>;
memory_storage ->
<<"memory"/utf8>>
end.
-file("src/glats/jetstream/stream.gleam", 597).
-spec apply_stream_option(list({binary(), gleam@json:json()}), stream_option()) -> list({binary(),
gleam@json:json()}).
apply_stream_option(Prev, Opt) ->
Pair = case Opt of
{description, Desc} ->
{<<"description"/utf8>>, gleam@json:string(Desc)};
{retention, Pol} ->
{<<"retention"/utf8>>,
begin
_pipe = ret_pol_to_string(Pol),
gleam@json:string(_pipe)
end};
{discard, Pol@1} ->
{<<"discard"/utf8>>,
begin
_pipe@1 = dis_pol_to_string(Pol@1),
gleam@json:string(_pipe@1)
end};
{storage, Storage} ->
{<<"storage"/utf8>>,
begin
_pipe@2 = storage_to_string(Storage),
gleam@json:string(_pipe@2)
end};
{max_consumers, Num} ->
{<<"max_consumers"/utf8>>, gleam@json:int(Num)};
{max_messages, Num@1} ->
{<<"max_msgs"/utf8>>, gleam@json:int(Num@1)};
{max_bytes, Num@2} ->
{<<"max_bytes"/utf8>>, gleam@json:int(Num@2)};
{max_age, Num@3} ->
{<<"max_age"/utf8>>, gleam@json:int(Num@3)};
{max_messages_per_subject, Num@4} ->
{<<"max_msgs_per_subject"/utf8>>, gleam@json:int(Num@4)};
{max_message_size, Num@5} ->
{<<"max_msg_size"/utf8>>, gleam@json:int(Num@5)};
{num_replicas, Num@6} ->
{<<"num_replicas"/utf8>>, gleam@json:int(Num@6)};
{duplicate_window, Num@7} ->
{<<"duplicate_window"/utf8>>, gleam@json:int(Num@7)};
{allow_direct, Val} ->
{<<"allow_direct"/utf8>>, gleam@json:bool(Val)};
{mirror_direct, Val@1} ->
{<<"mirror_direct"/utf8>>, gleam@json:bool(Val@1)};
{deny_delete, Val@2} ->
{<<"deny_delete"/utf8>>, gleam@json:bool(Val@2)};
{deny_purge, Val@3} ->
{<<"deny_purge"/utf8>>, gleam@json:bool(Val@3)};
{allow_rollup, Val@4} ->
{<<"allow_rollup_hdrs"/utf8>>, gleam@json:bool(Val@4)};
{discard_new_per_subject, Val@5} ->
{<<"discard_new_per_subject"/utf8>>, gleam@json:bool(Val@5)}
end,
gleam@list:prepend(Prev, Pair).
-file("src/glats/jetstream/stream.gleam", 590).
-spec apply_stream_options(
list({binary(), gleam@json:json()}),
list(stream_option())
) -> list({binary(), gleam@json:json()}).
apply_stream_options(Prev, Opts) ->
gleam@list:fold(Opts, Prev, fun apply_stream_option/2).
-file("src/glats/jetstream/stream.gleam", 580).
-spec stream_options_to_json(
list({binary(), gleam@json:json()}),
list(stream_option())
) -> binary().
stream_options_to_json(Prev, Opts) ->
_pipe = Prev,
_pipe@1 = apply_stream_options(_pipe, Opts),
_pipe@2 = gleam@json:object(_pipe@1),
gleam@json:to_string(_pipe@2).
-file("src/glats/jetstream/stream.gleam", 173).
?DOC(" Get info about a stream by name.\n").
-spec info(gleam@erlang@process:subject(glats:connection_message()), binary()) -> {ok,
stream_info()} |
{error, glats@jetstream:jetstream_error()}.
info(Conn, Name) ->
Topic = <<<<"$JS.API.STREAM"/utf8, ".INFO."/utf8>>/binary, Name/binary>>,
case glats:request(Conn, Topic, <<""/utf8>>, [], 1000) of
{ok, Msg} ->
decode_info(erlang:element(5, Msg));
{error, _} ->
{error, {stream_not_found, <<""/utf8>>}}
end.
-file("src/glats/jetstream/stream.gleam", 295).
?DOC(
" Creates a new stream.\n"
"\n"
" Calling this when a stream by the same already exists will run successfully\n"
" when no mutable field in the config is different.\n"
"\n"
" ## Example\n"
"\n"
" ```gleam\n"
" create(\n"
" conn,\n"
" \"samplestream\",\n"
" [\"sample.subject.>\"],\n"
" [\n"
" Storage(MemoryStorage),\n"
" Retention(WorkQueuePolicy),\n"
" ],\n"
" )\n"
" ```\n"
).
-spec create(
gleam@erlang@process:subject(glats:connection_message()),
binary(),
list(binary()),
list(stream_option())
) -> {ok, stream_info()} | {error, glats@jetstream:jetstream_error()}.
create(Conn, Name, Subjects, Opts) ->
Topic = <<<<"$JS.API.STREAM"/utf8, ".CREATE."/utf8>>/binary, Name/binary>>,
Body = begin
_pipe = [{<<"name"/utf8>>, gleam@json:string(Name)},
{<<"subjects"/utf8>>,
gleam@json:array(Subjects, fun gleam@json:string/1)}],
stream_options_to_json(_pipe, Opts)
end,
case glats:request(Conn, Topic, Body, [], 1000) of
{ok, Msg} ->
decode_info(erlang:element(5, Msg));
{error, _} ->
{error, {stream_not_found, <<""/utf8>>}}
end.
-file("src/glats/jetstream/stream.gleam", 322).
?DOC(" Updates the config of a stream.\n").
-spec update(
gleam@erlang@process:subject(glats:connection_message()),
binary(),
list(stream_option())
) -> {ok, stream_info()} | {error, glats@jetstream:jetstream_error()}.
update(Conn, Name, Opts) ->
Topic = <<<<"$JS.API.STREAM"/utf8, ".UPDATE."/utf8>>/binary, Name/binary>>,
Body = begin
_pipe = [{<<"name"/utf8>>, gleam@json:string(Name)}],
stream_options_to_json(_pipe, Opts)
end,
case glats:request(Conn, Topic, Body, [], 1000) of
{ok, Msg} ->
decode_info(erlang:element(5, Msg));
{error, _} ->
{error, {stream_not_found, <<""/utf8>>}}
end.
-file("src/glats/jetstream/stream.gleam", 341).
?DOC(" Deletes a stream.\n").
-spec delete(gleam@erlang@process:subject(glats:connection_message()), binary()) -> {ok,
nil} |
{error, glats@jetstream:jetstream_error()}.
delete(Conn, Name) ->
Topic = <<<<"$JS.API.STREAM"/utf8, ".DELETE."/utf8>>/binary, Name/binary>>,
case glats:request(Conn, Topic, <<""/utf8>>, [], 1000) of
{ok, Msg} ->
decode_delete(erlang:element(5, Msg));
{error, _} ->
{error, {stream_not_found, <<""/utf8>>}}
end.
-file("src/glats/jetstream/stream.gleam", 368).
?DOC(" Purges all of the data in a Stream, leaves the Stream.\n").
-spec purge(gleam@erlang@process:subject(glats:connection_message()), binary()) -> {ok,
integer()} |
{error, glats@jetstream:jetstream_error()}.
purge(Conn, Name) ->
Topic = <<<<"$JS.API.STREAM"/utf8, ".PURGE."/utf8>>/binary, Name/binary>>,
case glats:request(Conn, Topic, <<""/utf8>>, [], 1000) of
{ok, Msg} ->
decode_purge(erlang:element(5, Msg));
{error, _} ->
{error, {stream_not_found, <<""/utf8>>}}
end.
-file("src/glats/jetstream/stream.gleam", 394).
?DOC(" Tries to find a stream name by subject.\n").
-spec find_stream_name_by_subject(
gleam@erlang@process:subject(glats:connection_message()),
binary()
) -> {ok, binary()} | {error, glats@jetstream:jetstream_error()}.
find_stream_name_by_subject(Conn, Subject) ->
Topic = <<"$JS.API.STREAM"/utf8, ".NAMES"/utf8>>,
Body = begin
_pipe = [{<<"subject"/utf8>>, gleam@json:string(Subject)}],
_pipe@1 = gleam@json:object(_pipe),
gleam@json:to_string(_pipe@1)
end,
case glats:request(Conn, Topic, Body, [], 1000) of
{ok, Msg} ->
case decode_names(erlang:element(5, Msg)) of
{ok, Names} ->
_pipe@2 = gleam@list:first(Names),
gleam@result:map_error(
_pipe@2,
fun(_) ->
{stream_not_found, <<"stream not found"/utf8>>}
end
);
{error, Err} ->
{error, Err}
end;
{error, Err@1} ->
_pipe@3 = glats@internal@js:map_glats_error_to_jetstream(Err@1),
{error, _pipe@3}
end.
-file("src/glats/jetstream/stream.gleam", 472).
?DOC(
" Directly fetches a message from a stream either by sequence ID\n"
" (by passing `SequenceID(Int)`) or by subject (by passing\n"
" `LastBySubject(String)`).\n"
"\n"
" A subject can have wildcards (e.g. `orders.*.item.>`), please\n"
" refer to [Subject-Based messaging docs](https://docs.nats.io/nats-concepts/subjects)\n"
" for more info.\n"
"\n"
" Keep in mind that `allow_direct` has to be enabled in stream config\n"
" for this to work.\n"
).
-spec get_message(
gleam@erlang@process:subject(glats:connection_message()),
binary(),
access_method()
) -> {ok, stream_message()} | {error, glats@jetstream:jetstream_error()}.
get_message(Conn, Stream, Method) ->
Topic = <<<<"$JS.API.STREAM"/utf8, ".MSG.GET."/utf8>>/binary,
Stream/binary>>,
Body = encode_get_message_body(Method),
case glats:request(Conn, Topic, Body, [], 1000) of
{ok, Msg} ->
_pipe = decode_raw_message(erlang:element(5, Msg)),
gleam@result:then(
_pipe,
fun(D) -> _pipe@1 = raw_to_stream_message(D),
gleam@result:map_error(
_pipe@1,
fun(_) -> {decode_error, erlang:element(5, Msg)} end
) end
);
{error, _} ->
{error, {unknown, -1, <<"unknown error"/utf8>>}}
end.