-module(glats). -compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch]). -export([publish/4, publish_message/2, request/5, respond/4, subscribe/4, unsubscribe/2, server_info/1, active_subscriptions/1, new_inbox/0, connect/3]). -export_type([message/0, publish_option/0, subscribe_option/0, server_info/0, connection_option/0, error/0, connection_message/0, subscription_message/0, state/0]). -type message() :: {message, binary(), gleam@dict:dict(binary(), binary()), gleam@option:option(binary()), binary()}. -type publish_option() :: {reply_to, binary()} | {headers, list({binary(), binary()})}. -type subscribe_option() :: {queue_group, binary()}. -type server_info() :: {server_info, binary(), binary(), binary(), binary(), binary(), integer(), boolean(), integer(), integer(), boolean(), gleam@option:option(boolean())}. -type connection_option() :: {user_pass, binary(), binary()} | {token, binary()} | {n_key_seed, binary()} | {jwt, binary()} | {ca_cert, binary()} | {client_cert, binary(), binary()} | {inbox_prefix, binary()} | {connection_timeout, integer()} | enable_no_responders. -type error() :: no_reply_topic | timeout | no_responders | unexpected. -opaque connection_message() :: {subscribe, gleam@erlang@process:subject({ok, integer()} | {error, error()}), gleam@erlang@process:pid_(), binary(), list(subscribe_option())} | {unsubscribe, gleam@erlang@process:subject({ok, nil} | {error, error()}), integer()} | {publish, gleam@erlang@process:subject({ok, nil} | {error, error()}), binary(), binary(), list(publish_option())} | {request, gleam@erlang@process:subject(fun(() -> {ok, message()} | {error, error()})), binary(), binary(), list(publish_option()), integer()} | {get_server_info, gleam@erlang@process:subject({ok, server_info()} | {error, error()})} | {get_active_subscriptions, gleam@erlang@process:subject({ok, integer()} | {error, error()})} | {exited, gleam@erlang@process:exit_message()}. -type subscription_message() :: {received_message, gleam@erlang@process:subject(connection_message()), integer(), gleam@option:option(integer()), message()}. -type state() :: {state, gleam@erlang@process:pid_(), gleam@dict:dict(gleam@erlang@process:pid_(), integer())}. -file("/home/arnar/Code/glats/src/glats.gleam", 276). -spec handle_server_info( gleam@erlang@process:subject({ok, server_info()} | {error, error()}), state() ) -> gleam@otp@actor:next(any(), state()). handle_server_info(From, State) -> _pipe = 'Elixir.Gnat':server_info(erlang:element(2, State)), _pipe@1 = 'Elixir.Glats':decode_server_info(_pipe), _pipe@2 = gleam@result:map_error(_pipe@1, fun(_) -> unexpected end), gleam@erlang@process:send(From, _pipe@2), {continue, State, none}. -file("/home/arnar/Code/glats/src/glats.gleam", 287). -spec handle_active_subscriptions( gleam@erlang@process:subject({ok, integer()} | {error, error()}), state() ) -> gleam@otp@actor:next(any(), state()). handle_active_subscriptions(From, State) -> _pipe = 'Elixir.Gnat':active_subscriptions(erlang:element(2, State)), _pipe@1 = gleam@result:map_error(_pipe, fun(_) -> unexpected end), gleam@erlang@process:send(From, _pipe@1), {continue, State, none}. -file("/home/arnar/Code/glats/src/glats.gleam", 297). -spec handle_publish( gleam@erlang@process:subject({ok, nil} | {error, error()}), binary(), binary(), list(publish_option()), state() ) -> gleam@otp@actor:next(any(), state()). handle_publish(From, Topic, Body, Opts, State) -> case begin _pipe = 'Elixir.Gnat':pub(erlang:element(2, State), Topic, Body, Opts), erlang:atom_to_binary(_pipe) end of <<"ok"/utf8>> -> gleam@erlang@process:send(From, {ok, nil}); _ -> gleam@erlang@process:send(From, {error, unexpected}) end, {continue, State, none}. -file("/home/arnar/Code/glats/src/glats.gleam", 351). -spec handle_unsubscribe( gleam@erlang@process:subject({ok, nil} | {error, error()}), integer(), state() ) -> gleam@otp@actor:next(any(), state()). handle_unsubscribe(From, Sid, State) -> case begin _pipe = 'Elixir.Gnat':unsub(erlang:element(2, State), Sid, []), erlang:atom_to_binary(_pipe) end of <<"ok"/utf8>> -> gleam@erlang@process:send(From, {ok, nil}); _ -> gleam@erlang@process:send(From, {error, unexpected}) end, {continue, State, none}. -file("/home/arnar/Code/glats/src/glats.gleam", 364). -spec handle_subscribe( gleam@erlang@process:subject({ok, integer()} | {error, error()}), gleam@erlang@process:pid_(), binary(), list(subscribe_option()), state() ) -> gleam@otp@actor:next(any(), state()). handle_subscribe(From, Subscriber, Topic, Opts, State) -> case 'Elixir.Gnat':sub(erlang:element(2, State), Subscriber, Topic, Opts) of {ok, Sid} -> case gleam_erlang_ffi:link(Subscriber) of true -> gleam@erlang@process:send(From, {ok, Sid}), {continue, erlang:setelement( 3, State, gleam@dict:insert( erlang:element(3, State), Subscriber, Sid ) ), none}; false -> gleam@erlang@process:send(From, {error, unexpected}), {continue, State, none} end; {error, _} -> gleam@erlang@process:send(From, {error, unexpected}), {continue, State, none} end. -file("/home/arnar/Code/glats/src/glats.gleam", 404). -spec take_reply_topic(list(publish_option()), message()) -> list(publish_option()). take_reply_topic(Prev, Message) -> case erlang:element(4, Message) of {some, Rt} -> gleam@list:prepend(Prev, {reply_to, Rt}); none -> Prev end. -file("/home/arnar/Code/glats/src/glats.gleam", 413). -spec publish( gleam@erlang@process:subject(connection_message()), binary(), binary(), list(publish_option()) ) -> {ok, nil} | {error, error()}. publish(Conn, Topic, Body, Opts) -> gleam@erlang@process:call( Conn, fun(_capture) -> {publish, _capture, Topic, Body, Opts} end, 5000 ). -file("/home/arnar/Code/glats/src/glats.gleam", 425). -spec publish_message( gleam@erlang@process:subject(connection_message()), message() ) -> {ok, nil} | {error, error()}. publish_message(Conn, Message) -> _pipe = [{headers, maps:to_list(erlang:element(3, Message))}], _pipe@1 = take_reply_topic(_pipe, Message), publish( Conn, erlang:element(2, Message), erlang:element(5, Message), _pipe@1 ). -file("/home/arnar/Code/glats/src/glats.gleam", 440). -spec request( gleam@erlang@process:subject(connection_message()), binary(), binary(), list(publish_option()), integer() ) -> {ok, message()} | {error, error()}. request(Conn, Topic, Body, Opts, Timeout) -> Make_request = gleam@erlang@process:call( Conn, fun(_capture) -> {request, _capture, Topic, Body, Opts, Timeout} end, 1000 ), Make_request(). -file("/home/arnar/Code/glats/src/glats.gleam", 460). -spec respond( gleam@erlang@process:subject(connection_message()), message(), binary(), list(publish_option()) ) -> {ok, nil} | {error, error()}. respond(Conn, Message, Body, Opts) -> case erlang:element(4, Message) of {some, Rt} -> publish(Conn, Rt, Body, Opts); none -> {error, no_reply_topic} end. -file("/home/arnar/Code/glats/src/glats.gleam", 476). -spec subscription_mapper( gleam@erlang@process:subject(connection_message()), glats@internal@subscription:raw_message() ) -> subscription_message(). subscription_mapper(Conn, Raw_msg) -> {received_message, Conn, erlang:element(2, Raw_msg), erlang:element(3, Raw_msg), {message, erlang:element(4, Raw_msg), erlang:element(5, Raw_msg), erlang:element(6, Raw_msg), erlang:element(7, Raw_msg)}}. -file("/home/arnar/Code/glats/src/glats.gleam", 506). -spec subscribe( gleam@erlang@process:subject(connection_message()), gleam@erlang@process:subject(subscription_message()), binary(), list(subscribe_option()) ) -> {ok, integer()} | {error, error()}. subscribe(Conn, Subscriber, Topic, Opts) -> case glats@internal@subscription:start_subscriber( Conn, Subscriber, fun subscription_mapper/2 ) of {ok, Sub} -> case gleam@erlang@process:call( Conn, fun(_capture) -> {subscribe, _capture, begin _pipe = Sub, gleam@erlang@process:subject_owner(_pipe) end, Topic, Opts} end, 5000 ) of {ok, Sid} -> gleam@erlang@process:unlink( begin _pipe@1 = Sub, gleam@erlang@process:subject_owner(_pipe@1) end ), {ok, Sid}; {error, Err} -> gleam@erlang@process:kill( begin _pipe@2 = Sub, gleam@erlang@process:subject_owner(_pipe@2) end ), {error, Err} end; {error, _} -> {error, unexpected} end. -file("/home/arnar/Code/glats/src/glats.gleam", 553). -spec unsubscribe(gleam@erlang@process:subject(connection_message()), integer()) -> {ok, nil} | {error, error()}. unsubscribe(Conn, Sid) -> gleam@erlang@process:call( Conn, fun(_capture) -> {unsubscribe, _capture, Sid} end, 5000 ). -file("/home/arnar/Code/glats/src/glats.gleam", 563). -spec server_info(gleam@erlang@process:subject(connection_message())) -> {ok, server_info()} | {error, error()}. server_info(Conn) -> gleam@erlang@process:call( Conn, fun(Field@0) -> {get_server_info, Field@0} end, 5000 ). -file("/home/arnar/Code/glats/src/glats.gleam", 573). -spec active_subscriptions(gleam@erlang@process:subject(connection_message())) -> {ok, integer()} | {error, error()}. active_subscriptions(Conn) -> gleam@erlang@process:call( Conn, fun(Field@0) -> {get_active_subscriptions, Field@0} end, 5000 ). -file("/home/arnar/Code/glats/src/glats.gleam", 583). -spec new_inbox() -> binary(). new_inbox() -> glats@internal@util:random_inbox(<<"_INBOX."/utf8>>). -file("/home/arnar/Code/glats/src/glats.gleam", 608). -spec apply_conn_option( gleam@dict:dict(binary(), gleam@dynamic:dynamic_()), connection_option() ) -> gleam@dict:dict(binary(), gleam@dynamic:dynamic_()). apply_conn_option(Prev, Opt) -> case Opt of {user_pass, User, Pass} -> _pipe = Prev, _pipe@1 = gleam@dict:insert( _pipe, <<"username"/utf8>>, gleam_stdlib:identity(User) ), gleam@dict:insert( _pipe@1, <<"password"/utf8>>, gleam_stdlib:identity(Pass) ); {token, Token} -> _pipe@2 = Prev, gleam@dict:insert( _pipe@2, <<"token"/utf8>>, gleam_stdlib:identity(Token) ); {n_key_seed, Seed} -> _pipe@3 = Prev, gleam@dict:insert( _pipe@3, <<"nkey_seed"/utf8>>, gleam_stdlib:identity(Seed) ); {jwt, Jwt} -> _pipe@4 = Prev, gleam@dict:insert( _pipe@4, <<"jwt"/utf8>>, gleam_stdlib:identity(Jwt) ); {ca_cert, Path} -> _pipe@5 = Prev, _pipe@6 = gleam@dict:insert( _pipe@5, <<"tls"/utf8>>, gleam_stdlib:identity(true) ), gleam@dict:insert( _pipe@6, <<"cacertfile"/utf8>>, gleam_stdlib:identity(Path) ); {client_cert, Cert, Key} -> _pipe@7 = Prev, _pipe@8 = gleam@dict:insert( _pipe@7, <<"tls"/utf8>>, gleam_stdlib:identity(true) ), _pipe@9 = gleam@dict:insert( _pipe@8, <<"certfile"/utf8>>, gleam_stdlib:identity(Cert) ), gleam@dict:insert( _pipe@9, <<"keyfile"/utf8>>, gleam_stdlib:identity(Key) ); {inbox_prefix, Prefix} -> gleam@dict:insert( Prev, <<"inbox_prefix"/utf8>>, gleam_stdlib:identity(Prefix) ); {connection_timeout, Timeout} -> gleam@dict:insert( Prev, <<"connection_timeout"/utf8>>, gleam_stdlib:identity(Timeout) ); enable_no_responders -> gleam@dict:insert( Prev, <<"no_responders"/utf8>>, gleam_stdlib:identity(true) ) end. -file("/home/arnar/Code/glats/src/glats.gleam", 643). -spec add_ssl_opts(gleam@dict:dict(binary(), gleam@dynamic:dynamic_())) -> gleam@dict:dict(binary(), gleam@dynamic:dynamic_()). add_ssl_opts(Prev) -> _pipe = gleam@dict:take( Prev, [<<"cacertfile"/utf8>>, <<"certfile"/utf8>>, <<"keyfile"/utf8>>] ), _pipe@1 = maps:to_list(_pipe), _pipe@2 = gleam@list:map( _pipe@1, fun(O) -> {erlang:binary_to_atom(erlang:element(1, O)), erlang:element(2, O)} end ), _pipe@3 = gleam_stdlib:identity(_pipe@2), gleam@dict:insert(Prev, <<"ssl_opts"/utf8>>, _pipe@3). -file("/home/arnar/Code/glats/src/glats.gleam", 590). -spec build_settings(binary(), integer(), list(connection_option())) -> gleam@dict:dict(gleam@erlang@atom:atom_(), gleam@dynamic:dynamic_()). build_settings(Host, Port, Opts) -> _pipe = [{<<"host"/utf8>>, gleam_stdlib:identity(Host)}, {<<"port"/utf8>>, gleam_stdlib:identity(Port)}], _pipe@1 = maps:from_list(_pipe), _pipe@2 = gleam@list:fold(Opts, _pipe@1, fun apply_conn_option/2), _pipe@3 = add_ssl_opts(_pipe@2), _pipe@4 = gleam@dict:take( _pipe@3, [<<"host"/utf8>>, <<"port"/utf8>>, <<"tls"/utf8>>, <<"ssl_opts"/utf8>>, <<"inbox_prefix"/utf8>>, <<"connection_timeout"/utf8>>, <<"no_responders"/utf8>>, <<"username"/utf8>>, <<"password"/utf8>>, <<"token"/utf8>>, <<"nkey_seed"/utf8>>, <<"jwt"/utf8>>] ), _pipe@5 = maps:to_list(_pipe@4), _pipe@6 = gleam@list:map( _pipe@5, fun(I) -> {erlang:binary_to_atom(erlang:element(1, I)), erlang:element(2, I)} end ), maps:from_list(_pipe@6). -file("/home/arnar/Code/glats/src/glats.gleam", 686). -spec atom_field( binary(), fun((gleam@dynamic:dynamic_()) -> {ok, IDY} | {error, list(gleam@dynamic:decode_error())}) ) -> fun((gleam@dynamic:dynamic_()) -> {ok, IDY} | {error, list(gleam@dynamic:decode_error())}). atom_field(Key, Value) -> gleam@dynamic:field(erlang:binary_to_atom(Key), Value). -file("/home/arnar/Code/glats/src/glats.gleam", 668). -spec headers(gleam@dynamic:dynamic_()) -> {ok, gleam@dict:dict(binary(), binary())} | {error, list(gleam@dynamic:decode_error())}. headers(Data) -> _pipe = Data, _pipe@1 = (atom_field( <<"headers"/utf8>>, gleam@dynamic:list( gleam@dynamic:tuple2( fun gleam@dynamic:string/1, fun gleam@dynamic:string/1 ) ) ))(_pipe), _pipe@2 = gleam@result:map(_pipe@1, fun maps:from_list/1), gleam@result:'or'(_pipe@2, {ok, maps:new()}). -file("/home/arnar/Code/glats/src/glats.gleam", 680). -spec reply_to(gleam@dynamic:dynamic_()) -> {ok, gleam@option:option(binary())} | {error, list(gleam@dynamic:decode_error())}. reply_to(Data) -> _pipe = Data, _pipe@1 = (gleam@dynamic:optional( atom_field(<<"reply_to"/utf8>>, fun gleam@dynamic:string/1) ))(_pipe), gleam@result:'or'(_pipe@1, {ok, none}). -file("/home/arnar/Code/glats/src/glats.gleam", 654). -spec decode_msg(gleam@dynamic:dynamic_()) -> {ok, message()} | {error, list(gleam@dynamic:decode_error())}. decode_msg(Data) -> _pipe = Data, (gleam@dynamic:decode4( fun(Field@0, Field@1, Field@2, Field@3) -> {message, Field@0, Field@1, Field@2, Field@3} end, atom_field(<<"topic"/utf8>>, fun gleam@dynamic:string/1), fun headers/1, fun reply_to/1, atom_field(<<"body"/utf8>>, fun gleam@dynamic:string/1) ))(_pipe). -file("/home/arnar/Code/glats/src/glats.gleam", 316). -spec handle_request( gleam@erlang@process:subject(fun(() -> {ok, message()} | {error, error()})), binary(), binary(), list(publish_option()), integer(), state() ) -> gleam@otp@actor:next(any(), state()). handle_request(From, Topic, Body, _, Timeout, State) -> Opts = [{erlang:binary_to_atom(<<"receive_timeout"/utf8>>), gleam_stdlib:identity(Timeout)}], Req_func = fun() -> case 'Elixir.Gnat':request(erlang:element(2, State), Topic, Body, Opts) of {ok, Msg} -> _pipe = decode_msg(Msg), gleam@result:map_error(_pipe, fun(_) -> unexpected end); {error, Err} -> case erlang:atom_to_binary(Err) of <<"timeout"/utf8>> -> {error, timeout}; <<"no_responders"/utf8>> -> {error, no_responders}; _ -> {error, unexpected} end end end, gleam@erlang@process:send(From, Req_func), {continue, State, none}. -file("/home/arnar/Code/glats/src/glats.gleam", 236). -spec handle_command(connection_message(), state()) -> gleam@otp@actor:next(any(), state()). handle_command(Message, State) -> case Message of {exited, Em} -> case erlang:element(2, Em) =:= erlang:element(2, State) of true -> {stop, erlang:element(3, Em)}; false -> case gleam_stdlib:map_get( erlang:element(3, State), erlang:element(2, Em) ) of {ok, Sid} -> 'Elixir.Gnat':unsub( erlang:element(2, State), Sid, [] ), {continue, erlang:setelement( 3, State, gleam@dict:delete( erlang:element(3, State), erlang:element(2, Em) ) ), none}; {error, nil} -> gleam_stdlib:println( <<"exited process not found in map of subscribers"/utf8>> ), {continue, State, none} end end; {publish, From, Topic, Body, Opts} -> handle_publish(From, Topic, Body, Opts, State); {request, From@1, Topic@1, Body@1, Opts@1, Timeout} -> handle_request(From@1, Topic@1, Body@1, Opts@1, Timeout, State); {subscribe, From@2, Subscriber, Topic@2, Opts@2} -> handle_subscribe(From@2, Subscriber, Topic@2, Opts@2, State); {unsubscribe, From@3, Sid@1} -> handle_unsubscribe(From@3, Sid@1, State); {get_server_info, From@4} -> handle_server_info(From@4, State); {get_active_subscriptions, From@5} -> handle_active_subscriptions(From@5, State) end. -file("/home/arnar/Code/glats/src/glats.gleam", 210). -spec connect(binary(), integer(), list(connection_option())) -> {ok, gleam@erlang@process:subject(connection_message())} | {error, gleam@otp@actor:start_error()}. connect(Host, Port, Opts) -> gleam@otp@actor:start_spec( {spec, fun() -> gleam_erlang_ffi:trap_exits(true), Selector = begin _pipe = gleam_erlang_ffi:new_selector(), gleam@erlang@process:selecting_trapped_exits( _pipe, fun(Field@0) -> {exited, Field@0} end ) end, case 'Elixir.Gnat':start_link(build_settings(Host, Port, Opts)) of {ok, Pid} -> {ready, {state, Pid, maps:new()}, Selector}; {error, _} -> {failed, <<"starting connection failed"/utf8>>} end end, 5000, fun handle_command/2} ).